siloqy/prod/clean_arch/violet/venue_ob_provider.py

"""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)