From 12b768bc4fc2d0521235d1d8036b909902852127 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 16 Jun 2026 23:01:17 +0200 Subject: [PATCH] VIOLET OA: add venue OB provider seam Co-authored-by: Codex --- .../violet/test_violet_venue_ob_provider.py | 132 +++++++++++++ prod/clean_arch/violet/venue_ob_provider.py | 178 ++++++++++++++++++ .../VIOLET_SPEC__MULTI_EXCHANGE_OB_SEAM.md | 47 +++++ 3 files changed, 357 insertions(+) create mode 100644 prod/clean_arch/violet/test_violet_venue_ob_provider.py create mode 100644 prod/clean_arch/violet/venue_ob_provider.py create mode 100644 prod/docs/VIOLET_SPEC__MULTI_EXCHANGE_OB_SEAM.md diff --git a/prod/clean_arch/violet/test_violet_venue_ob_provider.py b/prod/clean_arch/violet/test_violet_venue_ob_provider.py new file mode 100644 index 00000000..249c8289 --- /dev/null +++ b/prod/clean_arch/violet/test_violet_venue_ob_provider.py @@ -0,0 +1,132 @@ +"""Tests for the Violet venue-agnostic OB provider seam.""" + +from __future__ import annotations + +import math +import sys +import time + +import pytest +from pydantic import ValidationError + +sys.path.insert(0, "/mnt/dolphinng5_predict") +sys.path.insert(0, "/mnt/dolphinng5_predict/nautilus_dolphin") + +from nautilus_dolphin.nautilus.ob_features import OBFeatureEngine + +from prod.clean_arch.violet.venue_ob_provider import ( + MockTickVenueOBProvider, + VenueOBTick, + VioletVenueOBProvider, +) + + +def _records(base_ts: float = 1_700_000_000.0): + assets = ("BTCUSDT", "ETHUSDT") + rows = [] + for asset_idx, asset in enumerate(assets): + for snap_idx in range(3): + ts = base_ts + 30.0 * snap_idx + bias = 0.12 if asset_idx == 0 else -0.04 + bid = tuple(float((1 + bias) * (i + 1) * 1000.0) for i in range(5)) + ask = tuple(float((1 - bias) * (i + 1) * 1000.0) for i in range(5)) + depth = tuple(float(v / (100.0 + asset_idx)) for v in bid) + ask_depth = tuple(float(v / (100.0 + asset_idx)) for v in ask) + rows.append( + VenueOBTick( + timestamp=ts, + asset=asset, + bid_notional_levels=bid, + ask_notional_levels=ask, + bid_depth_levels=depth, + ask_depth_levels=ask_depth, + ) + ) + return rows + + +def test_provider_conforms_and_returns_snapshots(): + provider = VioletVenueOBProvider(ticks=_records(), tolerance_s=60.0) + + assert provider.get_assets() == ["BTCUSDT", "ETHUSDT"] + assert provider.get_snapshot_count("BTCUSDT") == 3 + assert provider.get_all_timestamps("ETHUSDT").tolist() == [1_700_000_000.0, 1_700_000_030.0, 1_700_000_060.0] + + snap = provider.get_snapshot("BTCUSDT", 1_700_000_031.0) + assert snap is not None + assert snap.asset == "BTCUSDT" + assert snap.bid_notional.shape == (5,) + assert snap.ask_notional.shape == (5,) + assert all(math.isfinite(float(v)) for v in snap.bid_notional) + assert all(math.isfinite(float(v)) for v in snap.ask_notional) + + +def test_poison_rejected_at_construction(): + with pytest.raises(ValidationError): + VenueOBTick( + timestamp=-1.0, + asset="BTCUSDT", + bid_notional_levels=(1.0, 2.0, 3.0, 4.0, 5.0), + ask_notional_levels=(1.0, 2.0, 3.0, 4.0, 5.0), + bid_depth_levels=(1.0, 2.0, 3.0, 4.0, 5.0), + ask_depth_levels=(1.0, 2.0, 3.0, 4.0, 5.0), + ) + + with pytest.raises(ValidationError): + VenueOBTick( + timestamp=1.0, + asset="BTCUSDT", + bid_notional_levels=(1.0, 2.0, 3.0, 4.0), + ask_notional_levels=(1.0, 2.0, 3.0, 4.0, 5.0), + bid_depth_levels=(1.0, 2.0, 3.0, 4.0, 5.0), + ask_depth_levels=(1.0, 2.0, 3.0, 4.0, 5.0), + ) + + with pytest.raises(ValidationError): + VenueOBTick( + timestamp=1.0, + asset="BTCUSDT", + bid_notional_levels=(1.0, 2.0, 3.0, 4.0, float("nan")), + ask_notional_levels=(1.0, 2.0, 3.0, 4.0, 5.0), + bid_depth_levels=(1.0, 2.0, 3.0, 4.0, 5.0), + ask_depth_levels=(1.0, 2.0, 3.0, 4.0, 5.0), + ) + + +def test_engine_step_live_and_get_market(monkeypatch): + base_ts = 1_700_000_000.0 + provider = MockTickVenueOBProvider( + assets=("BTCUSDT", "ETHUSDT", "SOLUSDT"), + num_snapshots=4, + base_timestamp=base_ts, + interval_s=30.0, + imbalance_biases={"BTCUSDT": 0.14, "ETHUSDT": -0.06, "SOLUSDT": -0.02}, + ) + engine = OBFeatureEngine(provider) + + monkeypatch.setattr(time, "time", lambda: base_ts + 31.0) + engine.step_live(["BTCUSDT", "ETHUSDT", "SOLUSDT"], bar_idx=7) + + market = engine.get_market(7, ["BTCUSDT", "ETHUSDT", "SOLUSDT"]) + placement = engine.get_placement("BTCUSDT", 7) + signal = engine.get_signal("BTCUSDT", 7) + + for value in (market.median_imbalance, market.agreement_pct, market.depth_pressure): + assert math.isfinite(float(value)) + for value in (placement.depth_1pct_usd, placement.depth_quality, placement.fill_probability, placement.spread_proxy_bps): + assert math.isfinite(float(value)) + for value in (signal.imbalance, signal.imbalance_ma5, signal.imbalance_persistence, signal.depth_asymmetry, signal.withdrawal_velocity): + assert math.isfinite(float(value)) + + +def test_callable_refresh_loads_ticks(): + base_ts = 1_700_000_000.0 + + def source(asset: str): + for row in _records(base_ts): + if row.asset == asset: + yield row + + provider = VioletVenueOBProvider(tick_source=source, assets=("BTCUSDT", "ETHUSDT")) + assert provider.get_snapshot_count("BTCUSDT") == 3 + assert provider.get_snapshot("ETHUSDT", base_ts + 10.0) is not None diff --git a/prod/clean_arch/violet/venue_ob_provider.py b/prod/clean_arch/violet/venue_ob_provider.py new file mode 100644 index 00000000..b6fe80ff --- /dev/null +++ b/prod/clean_arch/violet/venue_ob_provider.py @@ -0,0 +1,178 @@ +"""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) diff --git a/prod/docs/VIOLET_SPEC__MULTI_EXCHANGE_OB_SEAM.md b/prod/docs/VIOLET_SPEC__MULTI_EXCHANGE_OB_SEAM.md new file mode 100644 index 00000000..412fbecf --- /dev/null +++ b/prod/docs/VIOLET_SPEC__MULTI_EXCHANGE_OB_SEAM.md @@ -0,0 +1,47 @@ +# VIOLET Multi-Exchange OB Seam + +Date: 2026-06-16 + +## Scope + +This note describes the new read-only OB provider seam added on the VIOLET side: + +- `prod/clean_arch/violet/venue_ob_provider.py` +- `prod/clean_arch/violet/test_violet_venue_ob_provider.py` + +The seam exists so VIOLET can later consume a venue-specific OB feed without +changing BLUE or wiring any live exchange connection into the current shadow +path. + +## What the seam does + +- normalizes venue ticks into BLUE-shaped `OBSnapshot` records +- keeps the provider read-only and in-memory +- validates poison values at ingress with V-TYPES +- supports either an injected callable tick source or a preloaded buffer +- exposes the exact `OBProvider` interface that `OBFeatureEngine` expects + +## What it does not do + +- no live BingX connection +- no Hazelcast writes +- no BLUE code changes +- no live wiring into `live_blue_source.py` or `shadow_live_factors.py` + +## Future adapter shape + +A venue-specific live adapter can later be layered on top of this seam by +feeding `VenueOBTick` records into `VioletVenueOBProvider.refresh()` from any +normalized feed source. The only contract is the canonical `OBSnapshot` shape +that `OBFeatureEngine` already consumes. + +## Validation + +The companion test file checks: + +- `VioletVenueOBProvider` implements the `OBProvider` surface +- malformed values are rejected at construction +- `OBFeatureEngine(provider)` can run `step_live()` and read back finite OB + features +- the callable-source path loads as expected +