340 lines
14 KiB
Markdown
340 lines
14 KiB
Markdown
|
# AGENT SPEC: OBF Live Switchover — MockOBProvider → HZOBProvider + step_live()
|
||
|
|
|
||
|
|
**Status**: Ready to implement
|
||
|
|
**Complexity**: Medium (~150 LOC across 2 files + tests)
|
||
|
|
**Blocking**: Live capital deployment (paper trading acceptable with Mock)
|
||
|
|
**Created**: 2026-03-26
|
||
|
|
|
||
|
|
---
|
||
|
|
|
||
|
|
## 1. Background & Current State
|
||
|
|
|
||
|
|
The OBF subsystem has **all infrastructure in place** but is wired with synthetic data:
|
||
|
|
|
||
|
|
| Component | Status |
|
||
|
|
|---|---|
|
||
|
|
| `obf_prefect_flow.py` | ✅ Running — pushes live L2 snapshots to `DOLPHIN_FEATURES["asset_{ASSET}_ob"]` at ~100ms |
|
||
|
|
| `HZOBProvider` (`hz_ob_provider.py`) | ✅ Exists — reads the correct HZ map and key format |
|
||
|
|
| `OBFeatureEngine` (`ob_features.py`) | ⚠️ Preload-only — no live streaming path |
|
||
|
|
| `nautilus_event_trader.py` | ❌ Wired to `MockOBProvider` with static biases |
|
||
|
|
|
||
|
|
**Root cause the switch is blocked**: `OBFeatureEngine.preload_date()` is the only ingestion path. It calls `provider.get_all_timestamps(asset)` to enumerate all snapshots upfront. `HZOBProvider.get_all_timestamps()` correctly returns `[]` (real-time has no history) — so `preload_date()` with `HZOBProvider` builds empty caches, and all downstream `get_placement/get_signal/get_market` calls return `None`.
|
||
|
|
|
||
|
|
---
|
||
|
|
|
||
|
|
## 2. HZ Payload Format (verified from `obf_prefect_flow.py`)
|
||
|
|
|
||
|
|
Key: `asset_{SYMBOL}_ob` in map `DOLPHIN_FEATURES`
|
||
|
|
|
||
|
|
```json
|
||
|
|
{
|
||
|
|
"timestamp": "2026-03-26T12:34:56.789000+00:00",
|
||
|
|
"bid_notional": [1234567.0, 987654.0, 876543.0, 765432.0, 654321.0],
|
||
|
|
"ask_notional": [1234567.0, 987654.0, 876543.0, 765432.0, 654321.0],
|
||
|
|
"bid_depth": [0.123, 0.456, 0.789, 1.012, 1.234],
|
||
|
|
"ask_depth": [0.123, 0.456, 0.789, 1.012, 1.234],
|
||
|
|
"_pushed_at": "2026-03-26T12:34:56.901000+00:00",
|
||
|
|
"_push_seq": 1711453296901
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
`HZOBProvider.get_snapshot()` already parses this and normalizes `timestamp` to a Unix float (ISO→float fix applied 2026-03-26).
|
||
|
|
|
||
|
|
---
|
||
|
|
|
||
|
|
## 3. What Needs to Be Built
|
||
|
|
|
||
|
|
### 3.1 Add `step_live()` to `OBFeatureEngine` (`ob_features.py`)
|
||
|
|
|
||
|
|
This is the **core change**. Add a new public method that:
|
||
|
|
1. Fetches fresh snapshots for all assets from the provider
|
||
|
|
2. Runs the same feature computation pipeline as `preload_date()`'s inner loop
|
||
|
|
3. Stores results in new live caches keyed by `bar_idx` (integer)
|
||
|
|
4. Updates `_median_depth_ref` incrementally via EMA
|
||
|
|
|
||
|
|
**Method signature**:
|
||
|
|
```python
|
||
|
|
def step_live(self, assets: List[str], bar_idx: int) -> None:
|
||
|
|
"""Fetch live snapshots and compute OBF features for the current bar.
|
||
|
|
|
||
|
|
Call this ONCE per scan event, BEFORE calling engine.step_bar().
|
||
|
|
Results are stored and retrievable via get_placement/get_signal/get_market(bar_idx).
|
||
|
|
"""
|
||
|
|
```
|
||
|
|
|
||
|
|
**Implementation steps inside `step_live()`**:
|
||
|
|
|
||
|
|
```python
|
||
|
|
def step_live(self, assets: List[str], bar_idx: int) -> None:
|
||
|
|
wall_ts = time.time()
|
||
|
|
asset_imbalances = []
|
||
|
|
asset_velocities = []
|
||
|
|
|
||
|
|
for asset in assets:
|
||
|
|
snap = self.provider.get_snapshot(asset, wall_ts)
|
||
|
|
if snap is None:
|
||
|
|
continue
|
||
|
|
|
||
|
|
# Initialise per-asset rolling histories on first call
|
||
|
|
if asset not in self._imbalance_history:
|
||
|
|
self._imbalance_history[asset] = deque(maxlen=self.IMBALANCE_LOOKBACK)
|
||
|
|
if asset not in self._depth_1pct_history:
|
||
|
|
self._depth_1pct_history[asset] = deque(maxlen=self.DEPTH_LOOKBACK)
|
||
|
|
|
||
|
|
# Incremental median_depth_ref via EMA (alpha=0.01 → ~100-bar half-life)
|
||
|
|
d1pct = compute_depth_1pct_nb(snap.bid_notional, snap.ask_notional)
|
||
|
|
if asset not in self._median_depth_ref:
|
||
|
|
self._median_depth_ref[asset] = d1pct
|
||
|
|
else:
|
||
|
|
self._median_depth_ref[asset] = (
|
||
|
|
0.99 * self._median_depth_ref[asset] + 0.01 * d1pct
|
||
|
|
)
|
||
|
|
|
||
|
|
# Feature kernels (same as preload_date inner loop)
|
||
|
|
imb = compute_imbalance_nb(snap.bid_notional, snap.ask_notional)
|
||
|
|
dq = compute_depth_quality_nb(d1pct, self._median_depth_ref[asset])
|
||
|
|
fp = compute_fill_probability_nb(dq)
|
||
|
|
sp = compute_spread_proxy_nb(snap.bid_notional, snap.ask_notional)
|
||
|
|
da = compute_depth_asymmetry_nb(snap.bid_notional, snap.ask_notional)
|
||
|
|
|
||
|
|
self._imbalance_history[asset].append(imb)
|
||
|
|
self._depth_1pct_history[asset].append(d1pct)
|
||
|
|
|
||
|
|
imb_arr = np.array(self._imbalance_history[asset], dtype=np.float64)
|
||
|
|
ma5_n = min(5, len(imb_arr))
|
||
|
|
imb_ma5 = float(np.mean(imb_arr[-ma5_n:])) if ma5_n > 0 else imb
|
||
|
|
persist = compute_imbalance_persistence_nb(imb_arr, self.IMBALANCE_LOOKBACK)
|
||
|
|
dep_arr = np.array(self._depth_1pct_history[asset], dtype=np.float64)
|
||
|
|
velocity = compute_withdrawal_velocity_nb(
|
||
|
|
dep_arr, min(self.DEPTH_LOOKBACK, len(dep_arr) - 1)
|
||
|
|
)
|
||
|
|
|
||
|
|
# Store in live caches
|
||
|
|
if asset not in self._live_placement:
|
||
|
|
self._live_placement[asset] = {}
|
||
|
|
if asset not in self._live_signal:
|
||
|
|
self._live_signal[asset] = {}
|
||
|
|
|
||
|
|
self._live_placement[asset][bar_idx] = OBPlacementFeatures(
|
||
|
|
depth_1pct_usd=d1pct, depth_quality=dq,
|
||
|
|
fill_probability=fp, spread_proxy_bps=sp,
|
||
|
|
)
|
||
|
|
self._live_signal[asset][bar_idx] = OBSignalFeatures(
|
||
|
|
imbalance=imb, imbalance_ma5=imb_ma5,
|
||
|
|
imbalance_persistence=persist, depth_asymmetry=da,
|
||
|
|
withdrawal_velocity=velocity,
|
||
|
|
)
|
||
|
|
|
||
|
|
asset_imbalances.append(imb)
|
||
|
|
asset_velocities.append(velocity)
|
||
|
|
|
||
|
|
# Cross-asset macro (Sub-3 + Sub-4)
|
||
|
|
if asset_imbalances:
|
||
|
|
imb_arr_cross = np.array(asset_imbalances, dtype=np.float64)
|
||
|
|
vel_arr_cross = np.array(asset_velocities, dtype=np.float64)
|
||
|
|
n = len(asset_imbalances)
|
||
|
|
med_imb, agreement = compute_market_agreement_nb(imb_arr_cross, n)
|
||
|
|
cascade = compute_cascade_signal_nb(vel_arr_cross, n, self.CASCADE_THRESHOLD)
|
||
|
|
|
||
|
|
# Update macro depth history
|
||
|
|
if not hasattr(self, '_live_macro_depth_hist'):
|
||
|
|
self._live_macro_depth_hist = deque(maxlen=self.DEPTH_LOOKBACK)
|
||
|
|
agg_depth = float(np.mean([
|
||
|
|
self._median_depth_ref.get(a, 0.0) for a in assets
|
||
|
|
]))
|
||
|
|
self._live_macro_depth_hist.append(agg_depth)
|
||
|
|
macro_dep_arr = np.array(self._live_macro_depth_hist, dtype=np.float64)
|
||
|
|
depth_vel = compute_withdrawal_velocity_nb(
|
||
|
|
macro_dep_arr, min(self.DEPTH_LOOKBACK, len(macro_dep_arr) - 1)
|
||
|
|
)
|
||
|
|
# acceleration: simple first-difference of velocity
|
||
|
|
if not hasattr(self, '_live_macro_vel_prev'):
|
||
|
|
self._live_macro_vel_prev = depth_vel
|
||
|
|
accel = depth_vel - self._live_macro_vel_prev
|
||
|
|
self._live_macro_vel_prev = depth_vel
|
||
|
|
|
||
|
|
if not hasattr(self, '_live_macro'):
|
||
|
|
self._live_macro = {}
|
||
|
|
self._live_macro[bar_idx] = OBMacroFeatures(
|
||
|
|
median_imbalance=med_imb, agreement_pct=agreement,
|
||
|
|
depth_pressure=float(np.sum(imb_arr_cross)),
|
||
|
|
cascade_regime=cascade,
|
||
|
|
depth_velocity=depth_vel, acceleration=accel,
|
||
|
|
)
|
||
|
|
|
||
|
|
self._live_mode = True
|
||
|
|
self._live_bar_idx = bar_idx
|
||
|
|
```
|
||
|
|
|
||
|
|
**New instance variables to initialise in `__init__`** (add after existing init):
|
||
|
|
```python
|
||
|
|
self._live_placement: Dict[str, Dict[int, OBPlacementFeatures]] = {}
|
||
|
|
self._live_signal: Dict[str, Dict[int, OBSignalFeatures]] = {}
|
||
|
|
self._live_macro: Dict[int, OBMacroFeatures] = {}
|
||
|
|
self._live_mode: bool = False
|
||
|
|
self._live_bar_idx: int = -1
|
||
|
|
self._live_macro_depth_hist: deque = deque(maxlen=self.DEPTH_LOOKBACK)
|
||
|
|
self._live_macro_vel_prev: float = 0.0
|
||
|
|
```
|
||
|
|
|
||
|
|
### 3.2 Modify `_resolve_idx()` to handle live bar lookups
|
||
|
|
|
||
|
|
In `_resolve_idx()` (currently line 549), add a live-mode branch **before** the existing logic:
|
||
|
|
|
||
|
|
```python
|
||
|
|
def _resolve_idx(self, asset: str, timestamp_or_idx: float) -> Optional[int]:
|
||
|
|
# Live mode: bar_idx is the key directly (small integers, no ts_to_idx lookup)
|
||
|
|
if self._live_mode:
|
||
|
|
bar = int(timestamp_or_idx)
|
||
|
|
if asset in self._live_placement and bar in self._live_placement[asset]:
|
||
|
|
return bar
|
||
|
|
# Fall back to latest known bar (graceful degradation)
|
||
|
|
if asset in self._live_placement and self._live_placement[asset]:
|
||
|
|
return max(self._live_placement[asset].keys())
|
||
|
|
return None
|
||
|
|
# ... existing preload logic unchanged below ...
|
||
|
|
```
|
||
|
|
|
||
|
|
### 3.3 Modify `get_placement()`, `get_signal()`, `get_market()`, `get_macro()` to use live caches
|
||
|
|
|
||
|
|
Each method currently reads from `_preloaded_placement[asset][idx]`. Add a live-mode branch:
|
||
|
|
|
||
|
|
```python
|
||
|
|
def get_placement(self, asset: str, timestamp_or_idx: float) -> OBPlacementFeatures:
|
||
|
|
idx = self._resolve_idx(asset, timestamp_or_idx)
|
||
|
|
if idx is None:
|
||
|
|
return OBPlacementFeatures(...) # defaults (same as today)
|
||
|
|
if self._live_mode:
|
||
|
|
return self._live_placement.get(asset, {}).get(idx, OBPlacementFeatures(...))
|
||
|
|
return self._preloaded_placement.get(asset, {}).get(idx, OBPlacementFeatures(...))
|
||
|
|
```
|
||
|
|
|
||
|
|
Apply same pattern to `get_signal()`, `get_market()`, `get_macro()`.
|
||
|
|
|
||
|
|
### 3.4 Update `nautilus_event_trader.py` — `_wire_obf()`
|
||
|
|
|
||
|
|
Replace `MockOBProvider` with `HZOBProvider`:
|
||
|
|
|
||
|
|
```python
|
||
|
|
def _wire_obf(self, assets):
|
||
|
|
if not assets or self.ob_assets:
|
||
|
|
return
|
||
|
|
self.ob_assets = assets
|
||
|
|
from nautilus_dolphin.nautilus.hz_ob_provider import HZOBProvider
|
||
|
|
live_ob = HZOBProvider(
|
||
|
|
hz_cluster=HZ_CLUSTER,
|
||
|
|
hz_host=HZ_HOST,
|
||
|
|
assets=assets,
|
||
|
|
)
|
||
|
|
self.ob_eng = OBFeatureEngine(live_ob)
|
||
|
|
# No preload_date() call — live mode uses step_live() per scan
|
||
|
|
self.eng.set_ob_engine(self.ob_eng)
|
||
|
|
log(f" OBF wired: HZOBProvider, {len(assets)} assets (LIVE mode)")
|
||
|
|
```
|
||
|
|
|
||
|
|
Store `self.ob_eng` on `DolphinLiveTrader` so it can be called from `on_scan`.
|
||
|
|
|
||
|
|
### 3.5 Call `step_live()` in `on_scan()` before `step_bar()`
|
||
|
|
|
||
|
|
In `DolphinLiveTrader.on_scan()`, after `self._rollover_day()` and `_wire_obf()`, add:
|
||
|
|
|
||
|
|
```python
|
||
|
|
# Feed live OB data into OBF engine for this bar
|
||
|
|
if self.ob_eng is not None and self.ob_assets:
|
||
|
|
self.ob_eng.step_live(self.ob_assets, self.bar_idx)
|
||
|
|
```
|
||
|
|
|
||
|
|
This must happen **before** the `eng.step_bar()` call so OBF features are fresh for this bar.
|
||
|
|
|
||
|
|
---
|
||
|
|
|
||
|
|
## 4. Live Cache Eviction (Memory Management)
|
||
|
|
|
||
|
|
`_live_placement/signal/macro` grow unboundedly as dicts. Add LRU eviction — keep only the last `N=500` bar_idx entries:
|
||
|
|
|
||
|
|
```python
|
||
|
|
# At end of step_live(), after storing:
|
||
|
|
MAX_LIVE_CACHE = 500
|
||
|
|
for asset in list(self._live_placement.keys()):
|
||
|
|
if len(self._live_placement[asset]) > MAX_LIVE_CACHE:
|
||
|
|
oldest = sorted(self._live_placement[asset].keys())[:-MAX_LIVE_CACHE]
|
||
|
|
for k in oldest:
|
||
|
|
del self._live_placement[asset][k]
|
||
|
|
# Same for _live_signal, _live_macro
|
||
|
|
```
|
||
|
|
|
||
|
|
---
|
||
|
|
|
||
|
|
## 5. Staleness Guard
|
||
|
|
|
||
|
|
If `obf_prefect_flow.py` is down, `HZOBProvider.get_snapshot()` returns `None` for all assets (graceful). `step_live()` skips assets with no snapshot. The engine falls back to `ob_engine is None` behaviour (random 40% pass at `ob_confirm_rate`).
|
||
|
|
|
||
|
|
Add a staleness warning log in `step_live()` if 0 snapshots were fetched for more than 3 consecutive bars:
|
||
|
|
|
||
|
|
```python
|
||
|
|
if fetched_count == 0:
|
||
|
|
self._live_stale_count = getattr(self, '_live_stale_count', 0) + 1
|
||
|
|
if self._live_stale_count >= 3:
|
||
|
|
logger.warning("OBF step_live: no snapshots for %d bars — OBF gate degraded to random", self._live_stale_count)
|
||
|
|
else:
|
||
|
|
self._live_stale_count = 0
|
||
|
|
```
|
||
|
|
|
||
|
|
---
|
||
|
|
|
||
|
|
## 6. Files to Modify
|
||
|
|
|
||
|
|
| File | Full Path | Change |
|
||
|
|
|---|---|---|
|
||
|
|
| `ob_features.py` | `/mnt/dolphinng5_predict/nautilus_dolphin/nautilus_dolphin/nautilus/ob_features.py` | Add `step_live()`, live caches in `__init__`, live branch in `_resolve_idx/get_*` |
|
||
|
|
| `nautilus_event_trader.py` | `/mnt/dolphinng5_predict/prod/nautilus_event_trader.py` | `_wire_obf()` → `HZOBProvider`; add `self.ob_eng`; call `ob_eng.step_live()` in `on_scan` |
|
||
|
|
| `hz_ob_provider.py` | `/mnt/dolphinng5_predict/nautilus_dolphin/nautilus_dolphin/nautilus/hz_ob_provider.py` | Timestamp ISO→float normalization (DONE 2026-03-26) |
|
||
|
|
|
||
|
|
**Do NOT modify**:
|
||
|
|
- `/mnt/dolphinng5_predict/nautilus_dolphin/nautilus_dolphin/nautilus/alpha_orchestrator.py` — `set_ob_engine()` / `get_placement()` calls unchanged
|
||
|
|
- `/mnt/dolphinng5_predict/prod/obf_prefect_flow.py` — already writing correct format
|
||
|
|
- `/mnt/dolphinng5_predict/nautilus_dolphin/nautilus_dolphin/nautilus/dolphin_actor.py` — paper mode uses `preload_date()` which stays as-is
|
||
|
|
|
||
|
|
---
|
||
|
|
|
||
|
|
## 7. Tests to Write
|
||
|
|
|
||
|
|
In `/mnt/dolphinng5_predict/nautilus_dolphin/tests/test_hz_ob_provider_live.py`:
|
||
|
|
|
||
|
|
```
|
||
|
|
test_step_live_fetches_snapshots — mock HZOBProvider returns valid OBSnapshot
|
||
|
|
test_step_live_populates_placement_cache — after step_live(bar_idx=5), get_placement(asset, 5.0) returns non-default
|
||
|
|
test_step_live_populates_signal_cache — imbalance, persistence populated
|
||
|
|
test_step_live_market_features — agreement_pct and cascade computed
|
||
|
|
test_step_live_none_snapshot_skipped — provider returns None → asset skipped gracefully
|
||
|
|
test_step_live_stale_warning — 3 consecutive empty → warning logged
|
||
|
|
test_step_live_cache_eviction — after 501 bars, oldest entries deleted
|
||
|
|
test_resolve_idx_live_mode — live mode returns bar_idx directly
|
||
|
|
test_resolve_idx_live_fallback — unknown bar_idx → latest bar returned
|
||
|
|
test_median_depth_ema — _median_depth_ref converges via EMA
|
||
|
|
test_hz_ob_provider_timestamp_iso — ISO string timestamp normalised to float
|
||
|
|
test_hz_ob_provider_timestamp_float — float timestamp passes through unchanged
|
||
|
|
```
|
||
|
|
|
||
|
|
---
|
||
|
|
|
||
|
|
## 8. Verification After Implementation
|
||
|
|
|
||
|
|
1. Start `obf_prefect_flow.py` (confirm running via supervisorctl)
|
||
|
|
2. Check HZ: `DOLPHIN_FEATURES["asset_BTCUSDT_ob"]` has fresh data (< 10s old)
|
||
|
|
3. Run `nautilus_event_trader.py` — look for `OBF wired: HZOBProvider` in log
|
||
|
|
4. On first scan, look for no errors in `step_live()`
|
||
|
|
5. After 10 scans: `get_placement("BTCUSDT", bar_idx)` should return non-zero `fill_probability`
|
||
|
|
6. Compare ob_edge decisions vs Mock run — expect variance (live book reacts to market)
|
||
|
|
|
||
|
|
---
|
||
|
|
|
||
|
|
## 9. Data Quality Caveat (preserved from assessment 2026-03-26)
|
||
|
|
|
||
|
|
> **IMPORTANT**: Until this spec is implemented, OBF runs on `MockOBProvider` with static per-asset imbalance biases (BTC=-0.086, ETH=-0.092, BNB=+0.05, SOL=+0.05). All four OBF functional dimensions compute and produce real outputs feeding the alpha gate — but with frozen, market-unresponsive inputs. The OB cascade regime will always be CALM (no depth drain in mock data). This is acceptable for paper trading; it is NOT acceptable for live capital deployment.
|
||
|
|
|
||
|
|
---
|
||
|
|
|
||
|
|
*Created: 2026-03-26*
|
||
|
|
*Author: Claude (session Final_ND-Trader_Check)*
|