"""VIOLET venue-agnostic OB provider seam. This is a read-only, in-memory seam for future non-BLUE order book sources. It produces the exact ``OBSnapshot`` shape expected by BLUE's ``OBFeatureEngine`` without opening any live exchange connection. """ from __future__ import annotations from bisect import bisect_left try: from beartype.typing import Annotated, Callable, Iterable, Optional except ImportError: # pragma: no cover - beartype always present in prod from typing import Annotated, Callable, Iterable, Optional import numpy as np from pydantic import Field from .domain import StrictModel, Symbol, typed try: from nautilus_dolphin.nautilus.ob_provider import OBProvider, OBSnapshot except ImportError: # pragma: no cover - import path fallback for direct runs from nautilus_dolphin.nautilus.ob_provider import OBProvider, OBSnapshot # type: ignore OBLevel = Annotated[float, Field(ge=0.0, allow_inf_nan=False)] Level5 = tuple[OBLevel, OBLevel, OBLevel, OBLevel, OBLevel] class VenueOBTick(StrictModel): """Normalized OB tick for one asset at one point in time.""" timestamp: Annotated[float, Field(ge=0.0, allow_inf_nan=False)] asset: Symbol bid_notional_levels: Level5 ask_notional_levels: Level5 bid_depth_levels: Level5 ask_depth_levels: Level5 class VioletVenueOBProvider(OBProvider): """Venue-agnostic, in-memory OB provider. Input comes from either an injected callable or a flat iterable of ``VenueOBTick`` records. No exchange transport lives here. """ def __init__( self, *, ticks: Optional[Iterable[VenueOBTick]] = None, tick_source: Optional[Callable[[str], Iterable[VenueOBTick]]] = None, assets: Optional[Iterable[str]] = None, tolerance_s: float = 30.0, ) -> None: self.tolerance_s = float(tolerance_s) self._tick_source = tick_source self._assets = sorted({str(a) for a in assets or [] if str(a).strip()}) self._ticks: dict[str, list[VenueOBTick]] = {} self._timestamps: dict[str, np.ndarray] = {} if ticks is not None: self.load_ticks(ticks) if tick_source is not None and self._assets: self.refresh() @typed def load_ticks(self, ticks: Iterable[VenueOBTick]) -> None: grouped: dict[str, list[VenueOBTick]] = {} for tick in ticks: grouped.setdefault(tick.asset, []).append(tick) for asset, rows in grouped.items(): rows.sort(key=lambda t: t.timestamp) self._ticks[asset] = rows self._timestamps[asset] = np.array([t.timestamp for t in rows], dtype=np.float64) if asset not in self._assets: self._assets.append(asset) self._assets.sort() @typed def refresh(self) -> None: if self._tick_source is None: return for asset in self._assets: self.load_ticks(self._tick_source(asset)) def _to_snapshot(self, tick: VenueOBTick) -> OBSnapshot: return OBSnapshot( timestamp=float(tick.timestamp), asset=tick.asset, bid_notional=np.array(tick.bid_notional_levels, dtype=np.float64), ask_notional=np.array(tick.ask_notional_levels, dtype=np.float64), bid_depth=np.array(tick.bid_depth_levels, dtype=np.float64), ask_depth=np.array(tick.ask_depth_levels, dtype=np.float64), ) def get_snapshot(self, asset: str, timestamp: float) -> Optional[OBSnapshot]: rows = self._ticks.get(asset) if not rows: return None ts_arr = self._timestamps.get(asset) if ts_arr is None or len(ts_arr) == 0: return None idx = bisect_left(ts_arr, float(timestamp)) candidates = [max(0, idx - 1), min(idx, len(ts_arr) - 1)] best_idx = candidates[0] best_dist = abs(ts_arr[best_idx] - timestamp) for cand in candidates[1:]: dist = abs(ts_arr[cand] - timestamp) if dist < best_dist: best_dist = dist best_idx = cand if best_dist > self.tolerance_s: return None return self._to_snapshot(rows[best_idx]) def get_assets(self) -> list[str]: return list(self._assets) def get_all_timestamps(self, asset: str) -> np.ndarray: ts = self._timestamps.get(asset) return np.array([], dtype=np.float64) if ts is None else ts.copy() def get_snapshot_count(self, asset: str) -> int: rows = self._ticks.get(asset) return len(rows) if rows is not None else 0 def get_snapshot_by_index(self, asset: str, idx: int) -> Optional[OBSnapshot]: rows = self._ticks.get(asset) if rows is None or idx < 0 or idx >= len(rows): return None return self._to_snapshot(rows[idx]) class MockTickVenueOBProvider(VioletVenueOBProvider): """Deterministic synthetic tick source for tests.""" def __init__( self, *, assets: Optional[Iterable[str]] = None, num_snapshots: int = 8, base_timestamp: float = 1_700_000_000.0, interval_s: float = 30.0, base_notional: float = 100_000.0, depth_scale: float = 1.0, imbalance_biases: Optional[dict[str, float]] = None, tolerance_s: float = 60.0, ) -> None: assets = list(assets or ("BTCUSDT", "ETHUSDT")) ticks: list[VenueOBTick] = [] level_weights = (1.0, 2.0, 3.0, 4.0, 5.0) for asset_idx, asset in enumerate(assets): bias = (imbalance_biases or {}).get(asset, 0.08 if asset_idx % 2 == 0 else -0.08) for snap_idx in range(num_snapshots): ts = base_timestamp + snap_idx * interval_s drift = 1.0 + 0.01 * snap_idx bid_mult = 1.0 + bias ask_mult = 1.0 - bias bid_not = tuple( float(base_notional * depth_scale * drift * w * bid_mult) for w in level_weights ) ask_not = tuple( float(base_notional * depth_scale * drift * w * ask_mult) for w in level_weights ) approx_price = 100.0 + 5.0 * asset_idx bid_dep = tuple(float(v / approx_price) for v in bid_not) ask_dep = tuple(float(v / approx_price) for v in ask_not) ticks.append( VenueOBTick( timestamp=float(ts), asset=asset, bid_notional_levels=bid_not, ask_notional_levels=ask_not, bid_depth_levels=bid_dep, ask_depth_levels=ask_dep, ) ) super().__init__(ticks=ticks, assets=assets, tolerance_s=tolerance_s)