VIOLET OA: add venue OB provider seam

Co-authored-by: Codex <codex@openai.com>

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

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)

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
+