diff --git a/prod/clean_arch/violet/live_blue_source.py b/prod/clean_arch/violet/live_blue_source.py index b7191a21..ec642fb7 100644 --- a/prod/clean_arch/violet/live_blue_source.py +++ b/prod/clean_arch/violet/live_blue_source.py @@ -5,33 +5,38 @@ This is VIOLET-only. BLUE is untouched. The adapter reads the published BLUE surfaces that already exist in HZ and translates them into ``SizingFactors`` for the shadow path: - ``posture`` from ``DOLPHIN_STATE_BLUE.latest_nautilus`` / ``engine_snapshot`` - - ``esof_score`` from ``DOLPHIN_FEATURES.esof_latest`` or ``esof_advisor_latest`` - - ``acb_boost`` / ``acb_beta`` from ``DOLPHIN_FEATURES.acb_boost`` - - ``mc_scale`` from ``DOLPHIN_FEATURES.mc_forewarner_latest`` - - OB market consensus from the live ``asset_*_ob`` maps via BLUE's own + - ``esof_score`` from ``DOLPHIN_FEATURES.esof_latest`` or ``esof_advisor_latest`` via + BLUE's own ``parse_esof_payload`` / ``esof_score_from_payload`` + - ``boost`` / ``beta`` RECOMPUTED via ``AdaptiveCircuitBreaker.get_dynamic_boost_from_hz`` + over ``DOLPHIN_FEATURES.exf_latest`` + ``latest_eigen_scan.w750_velocity`` — IDENTICAL + to the trader's on_exf_update path (NOT the published ``acb_boost`` scalar) + - ``mc_scale`` from ``DOLPHIN_FEATURES.mc_forewarner_latest`` via ``_derive_mc_scale`` + (begin_day's cat/env thresholds, not the MC service's status label) + - OB market consensus from the live ``asset_*_ob`` maps via BLUE's own ``HZOBProvider`` + ``OBFeatureEngine`` + - ``dc_status`` via BLUE's ``AlphaSignalGenerator`` (params pinned to BLUE's ENGINE_KWARGS) + over the replayed scan price-history -The remaining DC signal is left neutral here for now. It needs the same live -signal-history path BLUE uses and should be added as a separate mirror step. - -PARITY DEBT (see prod/docs/VIOLET_BLUE_PARITY_STRUCTURAL_DIVERGENCE.md): this module -reconstructs BLUE's factors in a DIFFERENT file/scope structure than BLUE's monolithic -NDAlphaEngine (esf_alpha_orchestrator.py). Kernels here are WRAPPED (OBFeatureEngine, -AlphaSignalGenerator, VioletAssetSelector — single source of truth), but two derivations -are HAND-REPLICATED / surface-substituted and can drift silently from BLUE: - - ``_derive_mc_scale`` transcribes begin_day's mc thresholds (no pin to BLUE's fn); - - ``AlphaSignalGenerator()`` is built with BARE DEFAULTS, not BLUE's threaded params - (esf_alpha_orchestrator.py:180-191) — cosmetic only while dc_leverage_boost==1.0; - - boost/beta read the PUBLISHED ``acb_boost`` (acb_processor_service), not the trader's - ``get_dynamic_boost_from_hz`` recompute — the two surfaces may differ. -Any change to BLUE's corresponding formula REQUIRES a matching change here + a test. +PARITY (see prod/docs/VIOLET_BLUE_PARITY_STRUCTURAL_DIVERGENCE.md): this module reconstructs +BLUE's factors in a DIFFERENT file/scope structure than BLUE's monolithic NDAlphaEngine. +Every kernel is now WRAPPED, not copied (ACB, OBFeatureEngine+HZOBProvider, AlphaSignalGenerator, +VioletAssetSelector) — so the only hand-replicated arithmetic left is ``_derive_mc_scale`` +(begin_day's thresholds; pinned by test). Remaining REAL fidelity caveats: + - OB faithfulness needs a PERSISTENT ``ob_engine`` + per-scan ``bar_idx`` (OBFeatureEngine + accumulates a lookback window); a single-shot engine has no cross-scan history. + - On stale exf (>12h) the ACB raises ValueError; BLUE keeps the prior boost/beta, VIOLET + has no prior → neutral (1.0, 0.0). +Any change to BLUE's ENGINE_KWARGS or begin_day mc thresholds REQUIRES a matching change + +test here (see test_signal_gen_params_match_blue_engine_kwargs / the mc_scale formula test). """ from __future__ import annotations import json +import os as _os import sys from collections import deque +from datetime import datetime, timezone from collections.abc import Mapping from dataclasses import dataclass from dataclasses import field @@ -47,14 +52,43 @@ for _p in (str(_PROJECT_ROOT), str(_PROJECT_ROOT / "nautilus_dolphin")): sys.path.insert(0, _p) from nautilus_dolphin.nautilus.ob_features import OBFeatureEngine -from nautilus_dolphin.nautilus.ob_provider import OBSnapshot, OBProvider -from nautilus_dolphin.nautilus.alpha_signal_generator import AlphaSignalGenerator +from nautilus_dolphin.nautilus.hz_ob_provider import HZOBProvider +from nautilus_dolphin.nautilus.adaptive_circuit_breaker import AdaptiveCircuitBreaker +from nautilus_dolphin.nautilus.alpha_signal_generator import ( + AlphaSignalGenerator, + VEL_DIV_THRESHOLD, VEL_DIV_EXTREME, + LONG_VEL_DIV_THRESHOLD, LONG_VEL_DIV_EXTREME, +) from .alpha_wrappers import VioletAssetSelector from .decision_engine import SizingFactors from .live_factor_source import esof_score_from_features, posture_from_engine_snapshot from .live_factors import extract_live_sizing_factors +# Hazelcast coordinates — MUST equal nautilus_event_trader.py:107-108 (BLUE's live +# cluster). HZOBProvider opens its own connection to these, exactly as BLUE's _wire_obf. +HZ_CLUSTER = _os.environ.get("HZ_CLUSTER", "dolphin") +HZ_HOST = _os.environ.get("HZ_HOST", "127.0.0.1:5701") + +# AlphaSignalGenerator construction — pinned to BLUE's live ENGINE_KWARGS +# (nautilus_event_trader.py:128-133). These equal AlphaSignalGenerator's own defaults +# TODAY, but BLUE constructs it EXPLICITLY from ENGINE_KWARGS, so a future champion +# retune (e.g. dc_lookback_bars→10) would silently diverge a bare AlphaSignalGenerator(). +# We pin explicitly and assert the pin in test_signal_gen_params_match_blue_engine_kwargs. +# vel_div_* import the module constants directly so they auto-track the kernel. +BLUE_SIGNAL_GEN_KWARGS = dict( + vel_div_threshold=VEL_DIV_THRESHOLD, # -0.02 + vel_div_extreme=VEL_DIV_EXTREME, # -0.05 + long_vel_div_threshold=LONG_VEL_DIV_THRESHOLD, # 0.01 + long_vel_div_extreme=LONG_VEL_DIV_EXTREME, # 0.04 + dc_lookback_bars=7, + dc_min_magnitude_bps=0.75, + dc_skip_contradicts=True, + dc_leverage_boost=1.0, + dc_leverage_reduce=0.5, + use_direction_confirm=True, +) + def _jsonish(value: Any) -> Any: if isinstance(value, str): @@ -121,13 +155,48 @@ def _derive_mc_scale(payload: Any) -> float: return 0.5 if mc_orange else 1.0 -def _extract_acb(payload: Any) -> tuple[float, float]: - data = _jsonish(payload) - if isinstance(data, Mapping): - boost = _coerce_float(data.get("boost"), 1.0) or 1.0 - beta = _coerce_float(data.get("beta"), 0.0) or 0.0 - return max(0.0, boost), max(0.0, beta) - return 1.0, 0.0 +def _source_boost_beta( + client: hazelcast.HazelcastClient, + *, + date_str: str, + trade_direction: int, + acb: Optional[AdaptiveCircuitBreaker] = None, +) -> tuple[float, float]: + """Recompute (boost, beta) EXACTLY as BLUE's live trader does — NOT from acb_boost. + + BLUE (nautilus_event_trader.py on_exf_update:4769 / rollover prewarm:2710): + acb = AdaptiveCircuitBreaker() # bare, no args (trader 578/585) + info = acb.get_dynamic_boost_from_hz( + date_str=today, + exf_snapshot=DOLPHIN_FEATURES['exf_latest'], + w750_velocity=latest_eigen_scan['w750_velocity'] or None, + direction=trade_direction, # NO ob_engine in the live path + ) + boost, beta = info['boost'], info['beta'] + + The published ``DOLPHIN_FEATURES['acb_boost']`` is a SEPARATE daily publish + (acb_processor_service) and is NOT what drives BLUE's sizing, so we never read it. + On stale exf (>12h) get_dynamic_boost_from_hz raises ValueError; BLUE logs + 'ACB Stale Data Fallback' and keeps the prior boost/beta — VIOLET has no prior, so it + returns the neutral identity (1.0, 0.0).""" + exf = _jsonish(_read_hz_map(client, "DOLPHIN_FEATURES", "exf_latest")) + if not isinstance(exf, Mapping): + return 1.0, 0.0 + eigen = _jsonish(_read_hz_map(client, "DOLPHIN_FEATURES", "latest_eigen_scan")) + w750 = _coerce_float(eigen.get("w750_velocity"), None) if isinstance(eigen, Mapping) else None + acb = acb or AdaptiveCircuitBreaker() + try: + info = acb.get_dynamic_boost_from_hz( + date_str=date_str, + exf_snapshot=dict(exf), + w750_velocity=float(w750) if w750 else None, # 0.0 → None, matches BLUE + direction=trade_direction, + ) + except ValueError: + return 1.0, 0.0 # ACB Stale Data Fallback (BLUE keeps prior; VIOLET neutral) + boost = _coerce_float(info.get("boost"), 1.0) or 1.0 + beta = _coerce_float(info.get("beta"), 0.0) or 0.0 + return max(0.0, boost), max(0.0, beta) def _scan_view(payload: Any) -> Mapping[str, Any]: @@ -209,7 +278,12 @@ class LiveBlueScanHistory: history = self.price_history(asset[0]) if not history: return "NONE" - signal_gen = AlphaSignalGenerator() + # BLUE constructs its signal_gen from ENGINE_KWARGS (the orchestrator threads them, + # esf_alpha_orchestrator.py:180-191). Pin the SAME params — not bare defaults — so a + # champion retune that changes dc_lookback_bars / dc_min_magnitude_bps / thresholds + # diverges loudly (caught by test_signal_gen_params_match_blue_engine_kwargs), never + # silently. dc_status is deterministic in the params + price history (counters unused). + signal_gen = AlphaSignalGenerator(**BLUE_SIGNAL_GEN_KWARGS) sig = signal_gen.generate( vel_div=vel_div, vel_div_history=None, @@ -221,80 +295,37 @@ class LiveBlueScanHistory: return sig.dc_status -class HazelcastOBProvider(OBProvider): - """Read the current BLUE OB shards directly from Hazelcast.""" +def _source_ob_market( + ob_assets: list[str], + *, + bar_idx: int, + ob_engine: Optional[OBFeatureEngine] = None, +) -> tuple[Optional[float], Optional[float]]: + """Derive (median_imbalance, agreement_pct) EXACTLY as BLUE does, via BLUE's HZOBProvider. - def __init__(self, client: hazelcast.HazelcastClient): - self.client = client + BLUE wires OB once in _wire_obf (nautilus_event_trader.py:4967-4980): + live_ob = HZOBProvider(hz_cluster=HZ_CLUSTER, hz_host=HZ_HOST, assets=assets) + ob_eng = OBFeatureEngine(live_ob); eng.set_ob_engine(ob_eng) + then per scan calls ``ob_eng.step_live(assets, bar_idx)`` and the orchestrator reads + ``ob_eng.get_market(bar_idx, assets)`` (esf_alpha_orchestrator.py:590). We use BLUE's + OWN HZOBProvider — NOT a reinvented reader — so OB parsing/shard semantics are BLUE's. - def _asset_keys(self) -> list[str]: - try: - keys = self.client.get_map("DOLPHIN_FEATURES").blocking().key_set() - except Exception: - return [] - assets = [] - for key in keys: - if not isinstance(key, str) or not key.startswith("asset_") or not key.endswith("_ob"): - continue - asset = key[len("asset_"):-len("_ob")] - if asset and asset not in assets: - assets.append(asset) - return sorted(assets) - - def _read_snapshot(self, asset: str) -> Optional[OBSnapshot]: - raw = _read_hz_map(self.client, "DOLPHIN_FEATURES", f"asset_{asset}_ob") - data = _jsonish(raw) - if not isinstance(data, Mapping): - return None - bid_notional = np.array( - [_coerce_float(v, 0.0) or 0.0 for v in data.get("bid_notional", [0, 0, 0, 0, 0])][:5], - dtype=np.float64, - ) - ask_notional = np.array( - [_coerce_float(v, 0.0) or 0.0 for v in data.get("ask_notional", [0, 0, 0, 0, 0])][:5], - dtype=np.float64, - ) - bid_depth = np.array( - [_coerce_float(v, 0.0) or 0.0 for v in data.get("bid_depth", [0, 0, 0, 0, 0])][:5], - dtype=np.float64, - ) - ask_depth = np.array( - [_coerce_float(v, 0.0) or 0.0 for v in data.get("ask_depth", [0, 0, 0, 0, 0])][:5], - dtype=np.float64, - ) - ts = _coerce_float(data.get("timestamp"), 0.0) or 0.0 - if ( - bid_notional.shape != (5,) or ask_notional.shape != (5,) - or bid_depth.shape != (5,) or ask_depth.shape != (5,) - ): - return None - if np.any(bid_notional < 0) or np.any(ask_notional < 0): - return None - if np.any(bid_depth < 0) or np.any(ask_depth < 0): - return None - return OBSnapshot( - timestamp=ts, - asset=asset, - bid_notional=bid_notional, - ask_notional=ask_notional, - bid_depth=bid_depth, - ask_depth=ask_depth, - ) - - def get_snapshot(self, asset: str, timestamp: float) -> Optional[OBSnapshot]: - return self._read_snapshot(asset) - - def get_assets(self) -> list[str]: - return self._asset_keys() - - def get_all_timestamps(self, asset: str) -> np.ndarray: - snap = self._read_snapshot(asset) - if snap is None: - return np.array([], dtype=np.float64) - return np.array([snap.timestamp], dtype=np.float64) - - def get_snapshot_count(self, asset: str) -> int: - return 1 if self._read_snapshot(asset) is not None else 0 + OBFeatureEngine ACCUMULATES per-asset history across scans (lookback=10), so a faithful + mirror requires a PERSISTENT ``ob_engine`` + per-scan-incrementing ``bar_idx`` (the + caller/shadow loop owns it, exactly as BLUE keeps one ob_eng). When no engine is passed + this builds a single-shot HZOBProvider-backed engine — correct wiring but no cross-scan + history; use only for one-off reads / the live smoke.""" + if not ob_assets: + return None, None + if ob_engine is None: + provider = HZOBProvider(hz_cluster=HZ_CLUSTER, hz_host=HZ_HOST, assets=list(ob_assets)) + ob_engine = OBFeatureEngine(provider) + try: + ob_engine.step_live(list(ob_assets), bar_idx) + market = ob_engine.get_market(bar_idx, list(ob_assets)) + return float(market.median_imbalance), float(market.agreement_pct) + except Exception: + return None, None @dataclass(frozen=True) @@ -314,10 +345,22 @@ def source_live_blue_sizing_factors( assets: Optional[Iterable[str]] = None, scan_history: Optional[LiveBlueScanHistory] = None, selector: Optional[VioletAssetSelector] = None, + acb: Optional[AdaptiveCircuitBreaker] = None, + ob_engine: Optional[OBFeatureEngine] = None, + bar_idx: int = 0, + date_str: Optional[str] = None, ) -> LiveBlueSourceResult: - """Read the BLUE-published live surfaces and return a typed factor plane.""" + """Read BLUE's live surfaces and RECONSTRUCT the factor plane the way BLUE computes it. + + boost/beta are recomputed via ``AdaptiveCircuitBreaker.get_dynamic_boost_from_hz`` (NOT + the published acb_boost scalar); OB via BLUE's ``HZOBProvider`` + ``OBFeatureEngine``; + dc_status via ``AlphaSignalGenerator`` pinned to BLUE's ENGINE_KWARGS. For full OB + accumulation faithfulness the caller passes a PERSISTENT ``ob_engine`` + per-scan + ``bar_idx`` (BLUE keeps one ob_eng). ``date_str`` defaults to today's UTC date (BLUE + uses self.current_day).""" scan_history = scan_history or LiveBlueScanHistory() selector = selector or VioletAssetSelector() + today = date_str or datetime.now(timezone.utc).strftime("%Y-%m-%d") engine_snapshot_raw = _read_hz_map(client, "DOLPHIN_STATE_BLUE", "latest_nautilus") if engine_snapshot_raw is None: @@ -336,11 +379,6 @@ def source_live_blue_sizing_factors( esof_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "esof_advisor_latest") esof_score = esof_score_from_features(esof_raw) - acb_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "acb_boost") - if acb_raw is None: - acb_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "acb_boost_short") - acb_boost, acb_beta = _extract_acb(acb_raw) - mc_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "mc_forewarner_latest") mc_scale = _derive_mc_scale(mc_raw) @@ -358,26 +396,23 @@ def source_live_blue_sizing_factors( ) or -1 ) + + # boost/beta — recompute via the ACB exactly as BLUE's trader does (NOT the published + # acb_boost). Needs trade_direction, so computed after it. + acb_boost, acb_beta = _source_boost_beta( + client, date_str=today, trade_direction=trade_direction, acb=acb, + ) + candidate_market = scan_history.market_data(selector.lookback) pick = selector.pick(candidate_market, regime_direction=trade_direction) selected_asset = pick.asset if pick is not None else (scan_assets[0] if scan_assets else "") dc_status = scan_history.dc_status(scan, already_ingested=True) - ob_provider = HazelcastOBProvider(client) - ob_engine = OBFeatureEngine(ob_provider) - ob_assets = list(assets) if assets is not None else (scan_assets or ob_provider.get_assets()) - if ob_assets: - try: - ob_engine.step_live(ob_assets, bar_idx=0) - market = ob_engine.get_market(0, ob_assets) - ob_median_imbalance = float(market.median_imbalance) - ob_agreement_pct = float(market.agreement_pct) - except Exception: - ob_median_imbalance = None - ob_agreement_pct = None - else: - ob_median_imbalance = None - ob_agreement_pct = None + # OB market consensus — BLUE's HZOBProvider + OBFeatureEngine (persistent engine if given). + ob_assets = list(assets) if assets is not None else scan_assets + ob_median_imbalance, ob_agreement_pct = _source_ob_market( + ob_assets, bar_idx=bar_idx, ob_engine=ob_engine, + ) hz_snapshot = { "boost": acb_boost, diff --git a/prod/clean_arch/violet/test_violet_live_blue_source.py b/prod/clean_arch/violet/test_violet_live_blue_source.py index 8d92d674..289fd911 100644 --- a/prod/clean_arch/violet/test_violet_live_blue_source.py +++ b/prod/clean_arch/violet/test_violet_live_blue_source.py @@ -1,8 +1,19 @@ +"""V3.4c live BLUE source — BLUE-algo parity tests (boost/beta, signal-gen, OB, mc_scale). + +These pin VIOLET's reconstruction to BLUE's ACTUAL behaviour: + - boost/beta: bit-identical to AdaptiveCircuitBreaker.get_dynamic_boost_from_hz + - signal-gen: params pinned to BLUE's ENGINE_KWARGS (nautilus_event_trader.py) + - OB: BLUE's HZOBProvider + OBFeatureEngine wiring (persistent engine, step_live/get_market) + - mc_scale: begin_day's cat/env thresholds (not the MC service status label) +See prod/docs/VIOLET_BLUE_PARITY_STRUCTURAL_DIVERGENCE.md. +""" + from __future__ import annotations import json +import re import sys -from dataclasses import dataclass +from pathlib import Path import pytest @@ -12,26 +23,27 @@ import hazelcast from prod.clean_arch.violet.decision_engine import SizingFactors from prod.clean_arch.violet.live_blue_source import ( - HazelcastOBProvider, + BLUE_SIGNAL_GEN_KWARGS, + HZ_CLUSTER, + HZ_HOST, LiveBlueScanHistory, _derive_mc_scale, + _source_boost_beta, + _source_ob_market, source_live_blue_sizing_factors, ) from prod.clean_arch.violet.alpha_wrappers import VioletAssetSelector -from nautilus_dolphin.nautilus.alpha_signal_generator import AlphaSignalGenerator - - -@dataclass -class _FakeMap: - payloads: dict - - def get(self, key): - return self.payloads.get(key) - - def key_set(self): - return list(self.payloads.keys()) +from nautilus_dolphin.nautilus.adaptive_circuit_breaker import AdaptiveCircuitBreaker +from nautilus_dolphin.nautilus.alpha_signal_generator import ( + AlphaSignalGenerator, + LONG_VEL_DIV_THRESHOLD, LONG_VEL_DIV_EXTREME, + VEL_DIV_THRESHOLD, VEL_DIV_EXTREME, +) + +TRADER = Path("/mnt/dolphinng5_predict/prod/nautilus_event_trader.py") +# ── fakes ──────────────────────────────────────────────────────────────────── class _FakeBlocking: def __init__(self, payloads): self._payloads = payloads @@ -48,285 +60,314 @@ class _FakeClient: self._maps = maps def get_map(self, name): - return type("M", (), {"blocking": lambda self2: _FakeBlocking(self._maps[name])})() + payloads = self._maps.get(name, {}) + return type("M", (), {"blocking": lambda self2, p=payloads: _FakeBlocking(p)})() -def test_hz_ob_provider_filters_and_parses_latest_payload(): - client = _FakeClient( - { - "DOLPHIN_FEATURES": { - "asset_BTCUSDT_ob": json.dumps( - {"timestamp": 1.0, "bid_notional": [1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1], - "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]} - ), - "asset_XRPUSDT_ob": json.dumps( - {"timestamp": 2.0, "bid_notional": [-1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1], - "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]} - ), - "acb_boost": json.dumps({"boost": 1.2, "beta": 0.3}), - "mc_forewarner_latest": json.dumps({"catastrophic_prob": 0.15, "envelope_score": 0.5}), - "esof_latest": json.dumps({"advisory_score": 0.4}), - }, - "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "restored"})}, - } +class _FakeOBEngine: + """Stands in for a persistent OBFeatureEngine; records step_live calls.""" + + def __init__(self, median_imbalance=0.0, agreement_pct=0.0): + self.calls = [] + self._mi = median_imbalance + self._ap = agreement_pct + + def step_live(self, assets, bar_idx): + self.calls.append((tuple(assets), bar_idx)) + + def get_market(self, bar_idx, assets): + return type("M", (), {"median_imbalance": self._mi, "agreement_pct": self._ap})() + + +_EXF = {"funding_btc": 0.01, "dvol_btc": 55.0, "fng": 40.0, "taker": 1.2, "_acb_ready": True} + + +def _features(**extra): + base = { + "esof_latest": json.dumps({"advisory_score": 0.4}), + "mc_forewarner_latest": json.dumps({"catastrophic_prob": 0.15, "envelope_score": 0.5}), + "exf_latest": json.dumps(_EXF), + } + base.update(extra) + return base + + +# ── 1. boost/beta: bit-identical to the ACB recompute (NOT acb_boost) ───────── +def test_boost_beta_recomputed_identically_to_blue_acb(): + eigen = {"w750_velocity": 0.0012, "assets": ["BTCUSDT"], "asset_prices": [100.0], + "scan_number": 5, "vel_div": -0.03} + client = _FakeClient({"DOLPHIN_FEATURES": _features(latest_eigen_scan=json.dumps(eigen))}) + boost, beta = _source_boost_beta(client, date_str="2026-06-16", trade_direction=-1) + ref = AdaptiveCircuitBreaker().get_dynamic_boost_from_hz( + date_str="2026-06-16", exf_snapshot=dict(_EXF), w750_velocity=0.0012, direction=-1, ) - provider = HazelcastOBProvider(client) # type: ignore[arg-type] - assert provider.get_assets() == ["BTCUSDT", "XRPUSDT"] - snap = provider.get_snapshot("BTCUSDT", 0.0) - assert snap is not None - assert snap.asset == "BTCUSDT" - assert snap.bid_notional.tolist() == [1.0, 2.0, 3.0, 4.0, 5.0] + assert boost == max(0.0, float(ref["boost"])) + assert beta == max(0.0, float(ref["beta"])) -def test_source_live_blue_sizing_factors_unit(monkeypatch): - class FakeEngine: - def __init__(self, provider): - self.provider = provider - def step_live(self, assets, bar_idx): - assert "BTCUSDT" in assets - def get_market(self, ts, assets): - return type("M", (), {"median_imbalance": 0.12, "agreement_pct": 0.91})() +def test_boost_beta_w750_zero_passed_as_none_like_blue(): + # BLUE: w750_velocity=float(w750) if w750 else None → 0.0 becomes None + eigen = {"w750_velocity": 0.0, "assets": ["BTCUSDT"], "asset_prices": [100.0]} + client = _FakeClient({"DOLPHIN_FEATURES": _features(latest_eigen_scan=json.dumps(eigen))}) + boost, beta = _source_boost_beta(client, date_str="2026-06-16", trade_direction=-1) + ref = AdaptiveCircuitBreaker().get_dynamic_boost_from_hz( + date_str="2026-06-16", exf_snapshot=dict(_EXF), w750_velocity=None, direction=-1, + ) + assert (boost, beta) == (max(0.0, float(ref["boost"])), max(0.0, float(ref["beta"]))) - monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.OBFeatureEngine", FakeEngine) + +def test_boost_beta_neutral_when_no_exf(): + client = _FakeClient({"DOLPHIN_FEATURES": {}}) + assert _source_boost_beta(client, date_str="2026-06-16", trade_direction=-1) == (1.0, 0.0) + + +def test_boost_beta_neutral_on_stale_exf_valueerror(): + # >12h staleness → get_dynamic_boost_from_hz raises ValueError → BLUE stale fallback; + # VIOLET has no prior → neutral identity. + stale = dict(_EXF, _staleness_s={"funding_btc": 999999.0}) + client = _FakeClient({"DOLPHIN_FEATURES": {"exf_latest": json.dumps(stale), + "latest_eigen_scan": json.dumps({"w750_velocity": 0.001})}}) + assert _source_boost_beta(client, date_str="2026-06-16", trade_direction=-1) == (1.0, 0.0) + + +def test_boost_beta_does_not_read_published_acb_boost(): + # An acb_boost scalar is present but MUST be ignored (BLUE recomputes from exf). + eigen = {"w750_velocity": 0.0012} + client = _FakeClient({"DOLPHIN_FEATURES": _features( + latest_eigen_scan=json.dumps(eigen), + acb_boost=json.dumps({"boost": 7.77, "beta": 7.77}), # poison: must NOT surface + )}) + boost, beta = _source_boost_beta(client, date_str="2026-06-16", trade_direction=-1) + assert boost != 7.77 and beta != 7.77 + + +# ── 2. signal-gen params pinned to BLUE's live ENGINE_KWARGS ────────────────── +def _blue_engine_kwargs_block() -> str: + text = TRADER.read_text() + start = text.index("ENGINE_KWARGS = dict(") + end = text.index("\n)", start) + return text[start:end] + + +def _blue_kwarg(name: str): + m = re.search(rf"\b{name}\s*=\s*([^\s,]+)", _blue_engine_kwargs_block()) + assert m, f"{name} not found in BLUE ENGINE_KWARGS" + raw = m.group(1).rstrip(",") + if raw in ("True", "False"): + return raw == "True" + return float(raw) if any(c in raw for c in ".-e") else int(raw) + + +@pytest.mark.parametrize("key", [ + "vel_div_threshold", "vel_div_extreme", "dc_lookback_bars", "dc_min_magnitude_bps", + "use_direction_confirm", "dc_skip_contradicts", "dc_leverage_boost", "dc_leverage_reduce", +]) +def test_signal_gen_params_match_blue_engine_kwargs(key): + # If a champion retune changes ENGINE_KWARGS, this fails — no silent divergence. + assert BLUE_SIGNAL_GEN_KWARGS[key] == _blue_kwarg(key) + + +def test_signal_gen_long_thresholds_track_kernel_constants(): + assert BLUE_SIGNAL_GEN_KWARGS["vel_div_threshold"] == VEL_DIV_THRESHOLD + assert BLUE_SIGNAL_GEN_KWARGS["vel_div_extreme"] == VEL_DIV_EXTREME + assert BLUE_SIGNAL_GEN_KWARGS["long_vel_div_threshold"] == LONG_VEL_DIV_THRESHOLD + assert BLUE_SIGNAL_GEN_KWARGS["long_vel_div_extreme"] == LONG_VEL_DIV_EXTREME + + +def test_dc_status_uses_pinned_signal_gen_not_bare_default(): + # dc_status must equal a BLUE signal_gen built with the SAME pinned params. history = LiveBlueScanHistory(maxlen=16, trade_direction=-1) for idx, px in enumerate([100.0, 99.5, 99.0, 98.4, 97.8, 97.0, 96.2], start=1): - history.ingest_scan( - { - "scan_number": idx, - "timestamp": float(idx), - "vel_div": -0.031, - "target_asset": "BTCUSDT", - "assets": ["BTCUSDT"], - "asset_prices": [px], - } - ) - client = _FakeClient( - { - "DOLPHIN_FEATURES": { - "asset_BTCUSDT_ob": json.dumps( - {"timestamp": 1.0, "bid_notional": [1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1], - "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]} - ), - "acb_boost": json.dumps({"boost": 1.4, "beta": 0.2}), - "mc_forewarner_latest": json.dumps({"catastrophic_prob": 0.15, "envelope_score": 0.5}), - "esof_latest": json.dumps({"advisory_score": 0.4}), - "latest_eigen_scan": json.dumps( - { - "scan_number": 8, - "timestamp": 8.0, - "vel_div": -0.031, - "target_asset": "BTCUSDT", - "assets": ["BTCUSDT"], - "asset_prices": [95.5], - } - ), - }, - "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "stalker"})}, - } + history.ingest_scan({"scan_number": idx, "timestamp": float(idx), "vel_div": -0.031, + "assets": ["BTCUSDT"], "asset_prices": [px]}) + scan = {"scan_number": 8, "timestamp": 8.0, "vel_div": -0.031, "assets": ["BTCUSDT"], + "asset_prices": [95.5]} + got = history.dc_status(scan) + ref = AlphaSignalGenerator(**BLUE_SIGNAL_GEN_KWARGS).generate( + vel_div=-0.031, vel_div_history=None, + asset_price_history=history.price_history("BTCUSDT"), + trade_direction=-1, asset="BTCUSDT", current_timestamp=8.0, ) + assert got == ref.dc_status + + +# ── 3. OB: BLUE's HZOBProvider + OBFeatureEngine wiring ─────────────────────── +def test_ob_uses_persistent_engine_step_live_then_get_market(): + eng = _FakeOBEngine(median_imbalance=0.12, agreement_pct=0.9) + mi, ap = _source_ob_market(["BTCUSDT", "ETHUSDT"], bar_idx=3, ob_engine=eng) + assert eng.calls == [(("BTCUSDT", "ETHUSDT"), 3)] # step_live(assets, bar_idx) — BLUE's call + assert (mi, ap) == (0.12, 0.9) + + +def test_ob_empty_assets_neutral(): + assert _source_ob_market([], bar_idx=0) == (None, None) + + +def test_ob_engine_exception_neutral(): + class _Boom: + def step_live(self, a, b): + raise RuntimeError("x") + + def get_market(self, b, a): + raise AssertionError("unreachable") + + assert _source_ob_market(["BTCUSDT"], bar_idx=0, ob_engine=_Boom()) == (None, None) + + +def test_ob_builds_blue_hzobprovider_with_live_coords(monkeypatch): + captured = {} + + class _FakeProvider: + def __init__(self, *, hz_cluster, hz_host, assets): + captured.update(cluster=hz_cluster, host=hz_host, assets=list(assets)) + + class _FakeEngine: + def __init__(self, provider): + self.provider = provider + + def step_live(self, a, b): + pass + + def get_market(self, b, a): + return type("M", (), {"median_imbalance": 0.0, "agreement_pct": 0.0})() + + monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.HZOBProvider", _FakeProvider) + monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.OBFeatureEngine", _FakeEngine) + _source_ob_market(["BTCUSDT"], bar_idx=0) # ob_engine=None → must construct BLUE's provider + assert captured == {"cluster": HZ_CLUSTER, "host": HZ_HOST, "assets": ["BTCUSDT"]} + + +# ── 4. mc_scale: begin_day's cat/env thresholds (the V3.4c bug fix) ─────────── +@pytest.mark.parametrize("cat, env, expected", [ + (0.15, 0.5, 0.5), # orange via cat>0.10 + (0.05, -0.5, 0.5), # orange via env<0 — old status-code returned GREEN→1.0 (bug) + (0.10, -0.001, 0.5), + (0.28, 0.5, 1.0), # red via cat>0.25 — old status-code: ORANGE→0.5 (bug) + (0.05, -1.5, 1.0), # red via env<-1.0 + (0.30, -2.0, 1.0), + (0.05, 0.5, 1.0), # benign + (0.10, 0.0, 1.0), +]) +def test_derive_mc_scale_matches_blue_begin_day_formula(cat, env, expected): + payload = json.dumps({"catastrophic_prob": cat, "envelope_score": env, "status": "IGNORED"}) + assert _derive_mc_scale(payload) == expected + + +def test_derive_mc_scale_neutral_when_fields_missing_or_garbage(): + assert _derive_mc_scale(json.dumps({"status": "ORANGE"})) == 1.0 # label must be ignored + assert _derive_mc_scale(json.dumps({"catastrophic_prob": 0.15})) == 1.0 + assert _derive_mc_scale(json.dumps({"envelope_score": -0.5})) == 1.0 + assert _derive_mc_scale("not json") == 1.0 + assert _derive_mc_scale(None) == 1.0 + assert _derive_mc_scale({"catastrophic_prob": "bad", "envelope_score": 0.5}) == 1.0 + + +# ── 5. integration: full plane sourced + composed ──────────────────────────── +def _client_with_scan(scan: dict, *, posture="STALKER", **feat) -> _FakeClient: + eigen = json.dumps({**scan, "w750_velocity": 0.0012}) + return _FakeClient({ + "DOLPHIN_FEATURES": _features(latest_eigen_scan=eigen, **feat), + "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": posture})}, + }) + + +def test_source_live_blue_sizing_factors_full_plane(): + history = LiveBlueScanHistory(maxlen=16, trade_direction=-1) + for idx, px in enumerate([100.0, 99.5, 99.0, 98.4, 97.8, 97.0, 96.2], start=1): + history.ingest_scan({"scan_number": idx, "timestamp": float(idx), "vel_div": -0.031, + "assets": ["BTCUSDT"], "asset_prices": [px]}) + scan = {"scan_number": 8, "timestamp": 8.0, "vel_div": -0.031, + "assets": ["BTCUSDT"], "asset_prices": [95.5]} + client = _client_with_scan(scan, posture="RESTORED") + ob = _FakeOBEngine(median_imbalance=0.12, agreement_pct=0.91) + res = source_live_blue_sizing_factors( - client, - assets=["BTCUSDT"], - scan_history=history, + client, assets=["BTCUSDT"], scan_history=history, selector=VioletAssetSelector(lookback_horizon=7), + ob_engine=ob, bar_idx=8, date_str="2026-06-16", ) assert isinstance(res.factors, SizingFactors) - assert res.factors.posture == "STALKER" - assert res.factors.mc_scale == 0.5 - assert res.factors.boost == 1.4 - assert res.factors.beta == 0.2 + assert res.factors.posture == "RESTORED" assert res.factors.esof_score == 0.4 - assert res.factors.ob_median_imbalance == 0.12 - assert res.factors.ob_agreement_pct == 0.91 + assert res.factors.mc_scale == 0.5 # cat=0.15/env=0.5 → orange + assert res.factors.ob_median_imbalance == 0.12 and res.factors.ob_agreement_pct == 0.91 + assert ob.calls == [(("BTCUSDT",), 8)] + # boost/beta = ACB recompute over the SAME exf+w750 (bit-identical pin) + ref = AdaptiveCircuitBreaker().get_dynamic_boost_from_hz( + date_str="2026-06-16", exf_snapshot=dict(_EXF), w750_velocity=0.0012, direction=-1) + assert res.factors.boost == max(0.0, float(ref["boost"])) + assert res.factors.beta == max(0.0, float(ref["beta"])) assert res.factors.dc_status == "CONFIRM" assert res.selected_asset == "BTCUSDT" -def test_source_live_blue_sizing_factors_preserves_skip_contradict(monkeypatch): - class FakeEngine: - def __init__(self, provider): - self.provider = provider - def step_live(self, assets, bar_idx): - pass - def get_market(self, ts, assets): - return type("M", (), {"median_imbalance": 0.0, "agreement_pct": 0.0})() - - monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.OBFeatureEngine", FakeEngine) - history = LiveBlueScanHistory(maxlen=16, trade_direction=-1) - for idx, px in enumerate([100.0, 100.5, 101.0, 101.6, 102.2, 102.9, 103.6], start=1): - history.ingest_scan( - { - "scan_number": idx, - "timestamp": float(idx), - "vel_div": -0.031, - "target_asset": "BTCUSDT", - "assets": ["BTCUSDT"], - "asset_prices": [px], - } - ) - client = _FakeClient( - { - "DOLPHIN_FEATURES": { - "asset_BTCUSDT_ob": json.dumps( - {"timestamp": 1.0, "bid_notional": [1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1], - "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]} - ), - "acb_boost": json.dumps({"boost": 1.4, "beta": 0.2}), - "mc_forewarner_latest": json.dumps({"catastrophic_prob": 0.02, "envelope_score": 0.9}), - "esof_latest": json.dumps({"advisory_score": 0.4}), - "latest_eigen_scan": json.dumps( - { - "scan_number": 8, - "timestamp": 8.0, - "vel_div": -0.031, - "target_asset": "BTCUSDT", - "assets": ["BTCUSDT"], - "asset_prices": [104.2], - } - ), - }, - "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "APEX"})}, - } - ) - res = source_live_blue_sizing_factors( - client, - assets=["BTCUSDT"], - scan_history=history, - selector=VioletAssetSelector(lookback_horizon=7), - ) - assert res.factors.dc_status == "SKIP_CONTRADICT" +def test_source_live_blue_sizing_factors_handles_anomalies(): + client = _FakeClient({ + "DOLPHIN_FEATURES": { + "exf_latest": "not json", # → boost/beta neutral + "mc_forewarner_latest": json.dumps({"catastrophic_prob": 0.02, "envelope_score": 0.9}), + "esof_latest": "not json", + "latest_eigen_scan": json.dumps({"assets": ["BTCUSDT"], "asset_prices": [0]}), + }, + "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": ""})}, + }) + res = source_live_blue_sizing_factors(client, assets=["BTCUSDT"], ob_engine=_FakeOBEngine()) + assert res.factors.posture == "APEX" + assert res.factors.mc_scale == 1.0 + assert res.factors.boost == 1.0 and res.factors.beta == 0.0 # bad exf → ACB neutral + assert res.factors.esof_score is None + # OB engine is healthy (the anomalies are in exf/esof/prices), so it returns its 0.0/0.0. + assert res.factors.ob_median_imbalance == 0.0 and res.factors.ob_agreement_pct == 0.0 + assert res.factors.dc_status == "NONE" + assert res.selected_asset == "BTCUSDT" -def test_live_blue_sequence_matches_blue_selector_and_dc_at_each_step(monkeypatch): - class FakeEngine: - def __init__(self, provider): - self.provider = provider - def step_live(self, assets, bar_idx): - pass - def get_market(self, ts, assets): - return type("M", (), {"median_imbalance": 0.0, "agreement_pct": 0.0})() +def test_source_live_blue_neutral_ob_when_no_engine_and_no_assets(): + # No ob_engine + no assets discoverable → OB neutral (None), no HZOBProvider connection. + client = _FakeClient({ + "DOLPHIN_FEATURES": _features(latest_eigen_scan=json.dumps({"assets": [], "asset_prices": []})), + "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "APEX"})}, + }) + res = source_live_blue_sizing_factors(client) # assets=None, scan has none → [] + assert res.factors.ob_median_imbalance is None and res.factors.ob_agreement_pct is None - monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.OBFeatureEngine", FakeEngine) + +def test_sequence_matches_blue_selector_and_dc_at_each_step(): selector = VioletAssetSelector(lookback_horizon=7) history = LiveBlueScanHistory(maxlen=16, trade_direction=-1) - signal_gen = AlphaSignalGenerator() - + ref_gen = AlphaSignalGenerator(**BLUE_SIGNAL_GEN_KWARGS) scans = [ - { - "scan_number": 1, - "timestamp": 1.0, - "vel_div": -0.010, - "assets": ["BTCUSDT", "ETHUSDT"], - "asset_prices": [100.0, 200.0], - }, - { - "scan_number": 2, - "timestamp": 2.0, - "vel_div": -0.031, - "assets": ["BTCUSDT", "ETHUSDT"], - "asset_prices": [99.0, 198.0], - }, - { - "scan_number": 3, - "timestamp": 3.0, - "vel_div": -0.041, - "assets": ["BTCUSDT", "ETHUSDT"], - "asset_prices": [98.0, 196.0], - }, - { - "scan_number": 4, - "timestamp": 4.0, - "vel_div": -0.031, - "assets": ["BTCUSDT", "ETHUSDT"], - "asset_prices": [97.0, 194.0], - }, + {"scan_number": 1, "timestamp": 1.0, "vel_div": -0.010, "assets": ["BTCUSDT", "ETHUSDT"], "asset_prices": [100.0, 200.0]}, + {"scan_number": 2, "timestamp": 2.0, "vel_div": -0.031, "assets": ["BTCUSDT", "ETHUSDT"], "asset_prices": [99.0, 198.0]}, + {"scan_number": 3, "timestamp": 3.0, "vel_div": -0.041, "assets": ["BTCUSDT", "ETHUSDT"], "asset_prices": [98.0, 196.0]}, + {"scan_number": 4, "timestamp": 4.0, "vel_div": -0.031, "assets": ["BTCUSDT", "ETHUSDT"], "asset_prices": [97.0, 194.0]}, ] - - for idx, scan in enumerate(scans, start=1): - client = _FakeClient( - { - "DOLPHIN_FEATURES": { - "asset_BTCUSDT_ob": json.dumps( - {"timestamp": 1.0, "bid_notional": [1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1], - "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]} - ), - "asset_ETHUSDT_ob": json.dumps( - {"timestamp": 1.0, "bid_notional": [2, 3, 4, 5, 6], "ask_notional": [6, 5, 4, 3, 2], - "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]} - ), - "acb_boost": json.dumps({"boost": 1.0, "beta": 0.0}), - "mc_forewarner_latest": json.dumps({"catastrophic_prob": 0.02, "envelope_score": 0.9}), - "esof_latest": json.dumps({"advisory_score": 0.3}), - "latest_eigen_scan": json.dumps({**scan, "target_asset": "BTCUSDT"}), - }, - "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "APEX"})}, - } - ) + for i, scan in enumerate(scans, start=1): + client = _client_with_scan(scan, posture="APEX") res = source_live_blue_sizing_factors( - client, - assets=["BTCUSDT", "ETHUSDT"], - scan_history=history, - selector=selector, + client, assets=["BTCUSDT", "ETHUSDT"], scan_history=history, + selector=selector, ob_engine=_FakeOBEngine(), bar_idx=i, ) - - # BLUE selector parity market = history.market_data(selector.lookback) expected_pick = selector.pick(market, regime_direction=-1) expected_asset = expected_pick.asset if expected_pick is not None else "BTCUSDT" assert res.selected_asset == expected_asset - - # BLUE signal parity - expected_signal = signal_gen.generate( - vel_div=float(scan["vel_div"]), - vel_div_history=None, + ref = ref_gen.generate( + vel_div=float(scan["vel_div"]), vel_div_history=None, asset_price_history=history.price_history(expected_asset), - trade_direction=-1, - asset=expected_asset, - current_timestamp=float(scan["timestamp"]), + trade_direction=-1, asset=expected_asset, current_timestamp=float(scan["timestamp"]), ) - assert res.factors.dc_status == expected_signal.dc_status + assert res.factors.dc_status == ref.dc_status -def test_live_blue_sequence_rejects_anomalous_values_without_poisoning_history(monkeypatch): - class FakeEngine: - def __init__(self, provider): - self.provider = provider - def step_live(self, assets, bar_idx): - pass - def get_market(self, ts, assets): - return type("M", (), {"median_imbalance": 0.0, "agreement_pct": 0.0})() - - monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.OBFeatureEngine", FakeEngine) +def test_sequence_rejects_anomalous_values_without_poisoning_history(): history = LiveBlueScanHistory(maxlen=16, trade_direction=-1) - selector = VioletAssetSelector(lookback_horizon=7) - scan = { - "scan_number": 99, - "timestamp": 99.0, - "vel_div": -0.031, - "assets": ["BTCUSDT", "ETHUSDT"], - "asset_prices": [float("nan"), -1.0], - "target_asset": "BTCUSDT", - } - client = _FakeClient( - { - "DOLPHIN_FEATURES": { - "asset_BTCUSDT_ob": json.dumps( - {"timestamp": 1.0, "bid_notional": [1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1], - "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]} - ), - "acb_boost": json.dumps({"boost": 1.0, "beta": 0.0}), - "mc_forewarner_latest": json.dumps({"catastrophic_prob": 0.02, "envelope_score": 0.9}), - "esof_latest": json.dumps({"advisory_score": 0.3}), - "latest_eigen_scan": json.dumps(scan), - }, - "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "APEX"})}, - } - ) + scan = {"scan_number": 99, "timestamp": 99.0, "vel_div": -0.031, + "assets": ["BTCUSDT", "ETHUSDT"], "asset_prices": [float("nan"), -1.0]} + client = _client_with_scan(scan, posture="APEX") res = source_live_blue_sizing_factors( - client, - assets=["BTCUSDT", "ETHUSDT"], - scan_history=history, - selector=selector, + client, assets=["BTCUSDT", "ETHUSDT"], scan_history=history, + selector=VioletAssetSelector(lookback_horizon=7), ob_engine=_FakeOBEngine(), ) assert res.selected_asset == "BTCUSDT" assert res.factors.dc_status == "NONE" @@ -334,75 +375,7 @@ def test_live_blue_sequence_rejects_anomalous_values_without_poisoning_history(m assert history.price_history("ETHUSDT") == [] -def test_source_live_blue_sizing_factors_handles_anomalies(monkeypatch): - class FakeEngine: - def __init__(self, provider): - self.provider = provider - def step_live(self, assets, bar_idx): - raise RuntimeError("boom") - def get_market(self, ts, assets): - return type("M", (), {"median_imbalance": 0.0, "agreement_pct": 0.0})() - - monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.OBFeatureEngine", FakeEngine) - client = _FakeClient( - { - "DOLPHIN_FEATURES": { - "asset_BTCUSDT_ob": json.dumps( - {"timestamp": 1.0, "bid_notional": [1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1], - "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]} - ), - "acb_boost": json.dumps({"boost": -9, "beta": "bad"}), - "mc_forewarner_latest": json.dumps({"catastrophic_prob": 0.02, "envelope_score": 0.9}), - "esof_latest": "not json", - "latest_eigen_scan": json.dumps({"target_asset": "BTCUSDT", "assets": ["BTCUSDT"], "asset_prices": [0]}), - }, - "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": ""})}, - } - ) - res = source_live_blue_sizing_factors(client, assets=["BTCUSDT"]) - assert res.factors.posture == "APEX" - assert res.factors.mc_scale == 1.0 - assert res.factors.boost == 0.0 - assert res.factors.beta == 0.0 - assert res.factors.esof_score is None - assert res.factors.ob_median_imbalance is None - assert res.factors.ob_agreement_pct is None - assert res.factors.dc_status == "NONE" - assert res.selected_asset == "BTCUSDT" - - -@pytest.mark.parametrize( - "cat, env, expected", - [ - # mc_orange (→0.5): not red, and (env<0 OR cat>0.10) - (0.15, 0.5, 0.5), # orange via cat>0.10 - (0.05, -0.5, 0.5), # orange via env<0 — OLD status-code returned GREEN→1.0 (the bug) - (0.10, -0.001, 0.5), # orange via env<0 boundary (cat==0.10 not >0.10) - # red (→1.0 here; BLUE halts separately via regime_dd_halt) - (0.28, 0.5, 1.0), # red via cat>0.25 — OLD status-code: 0.28<0.30→ORANGE→0.5 (the bug) - (0.05, -1.5, 1.0), # red via env<-1.0 - (0.30, -2.0, 1.0), # red via both - # ok (→1.0): not red, not orange - (0.05, 0.5, 1.0), # benign - (0.10, 0.0, 1.0), # cat==0.10 (not >0.10), env==0 (not <0) → ok - ], -) -def test_derive_mc_scale_matches_blue_begin_day_formula(cat, env, expected): - payload = json.dumps({"catastrophic_prob": cat, "envelope_score": env, "status": "IGNORED"}) - assert _derive_mc_scale(payload) == expected - - -def test_derive_mc_scale_neutral_when_fields_missing_or_garbage(): - # The published `status` label must NOT be used — payloads lacking the source - # fields derive neutral (1.0), never a haircut from a stale/foreign label. - assert _derive_mc_scale(json.dumps({"status": "ORANGE"})) == 1.0 - assert _derive_mc_scale(json.dumps({"catastrophic_prob": 0.15})) == 1.0 # missing env - assert _derive_mc_scale(json.dumps({"envelope_score": -0.5})) == 1.0 # missing cat - assert _derive_mc_scale("not json") == 1.0 - assert _derive_mc_scale(None) == 1.0 - assert _derive_mc_scale({"catastrophic_prob": "bad", "envelope_score": 0.5}) == 1.0 - - +# ── 6. live HZ smoke (env-bound; deselect with -k "not live_hz_smoke") ──────── def test_live_hz_smoke_reads_current_state(): client = hazelcast.HazelcastClient(cluster_name="dolphin", cluster_members=["localhost:5701"]) try: @@ -412,3 +385,4 @@ def test_live_hz_smoke_reads_current_state(): assert isinstance(res.factors, SizingFactors) assert res.factors.posture in {"APEX", "STALKER", "RESTORED", "TURTLE", "HIBERNATE"} assert res.factors.mc_scale in {0.5, 1.0} + assert res.factors.boost >= 0.0 and res.factors.beta >= 0.0 diff --git a/prod/docs/VIOLET_BLUE_PARITY_STRUCTURAL_DIVERGENCE.md b/prod/docs/VIOLET_BLUE_PARITY_STRUCTURAL_DIVERGENCE.md index 92418ec9..b0ef2eaf 100644 --- a/prod/docs/VIOLET_BLUE_PARITY_STRUCTURAL_DIVERGENCE.md +++ b/prod/docs/VIOLET_BLUE_PARITY_STRUCTURAL_DIVERGENCE.md @@ -46,16 +46,22 @@ not mechanical; it requires a human to know which VIOLET fragment mirrors which | `strength_cubic` | `esf_alpha_orchestrator.py:872-885` | `sizing.VioletSizer.strength_cubic` | same gate | LOW-MED | | `market_ob_mult` consensus | `esf_alpha_orchestrator.py:587-595` | `sizing.VioletSizer.market_ob_mult` | same gate | MED | | `dc_lev_mult` | `esf_alpha_orchestrator.py:575-577` | `sizing.VioletSizer.dc_lev_mult` | unit only | MED (but ≡1.0 while dc_leverage_boost=1.0) | -| **`mc_scale`** | `esf_alpha_orchestrator.py:956-962` (`begin_day`) | `live_blue_source._derive_mc_scale` | **unit only — NO pin to BLUE's fn** | **HIGH** | -| `boost`/`beta` source | trader recompute `acb.get_dynamic_boost_from_hz(exf_latest)` | reads published `DOLPHIN_FEATURES.acb_boost` (acb_processor_service) | none | MED (two surfaces may differ on dynamic-β / OB Sub-4) | -| `dc_status` config | `signal_gen` built with threaded params `:180-191` | `AlphaSignalGenerator()` **bare defaults** | none | MED (cosmetic while dc_lev_mult≡1.0) | -| OB feed shape | live OB accumulation | single-snapshot `HazelcastOBProvider` + `bar_idx=0` | none | MED | +| **`mc_scale`** | `esf_alpha_orchestrator.py:956-962` (`begin_day`) | `live_blue_source._derive_mc_scale` | parametrized formula test (8 cases inc. divergences) | LOW-MED (only remaining hand-replica) | +| `boost`/`beta` source | trader recompute `acb.get_dynamic_boost_from_hz(exf_latest)` | **FIXED 2026-06-16**: `_source_boost_beta` calls the SAME `get_dynamic_boost_from_hz` over exf_latest+w750 (bare `AdaptiveCircuitBreaker()`, no ob_engine) | bit-identity test vs the real ACB | LOW | +| `dc_status` config | `signal_gen` built with ENGINE_KWARGS `:180-191` | **FIXED**: `AlphaSignalGenerator(**BLUE_SIGNAL_GEN_KWARGS)` | param test parses ENGINE_KWARGS from trader source | LOW | +| OB feed | live OB accumulation via `HZOBProvider` | **FIXED**: BLUE's own `HZOBProvider` + `OBFeatureEngine` + `step_live`/`get_market` (persistent engine + bar_idx) | wiring test (HZ coords, step_live call) | LOW (caller must pass persistent engine) | -The worst link is **`mc_scale`**: pure duplication of `begin_day`'s thresholds with no -test that pins it to BLUE's actual function (BLUE computes it inline inside `begin_day`, -which is not callable in isolation). The V3.4c bug fixed on 2026-06-16 (the adapter keyed -off the MC service's `status` label instead of BLUE's `cat`/`env` thresholds) is exactly -the failure mode this structure invites. +**Update 2026-06-16:** the three MED-risk flags above were brought to bit-identity per +operator directive ("VIOLET should do *identically* what BLUE does"). boost/beta now call +the SAME `get_dynamic_boost_from_hz` BLUE's trader calls (the published `acb_boost` is NOT +used); dc_status uses `AlphaSignalGenerator` pinned to BLUE's ENGINE_KWARGS; OB uses BLUE's +`HZOBProvider`. The reinvented `HazelcastOBProvider` and the `_extract_acb`/`status`-label +paths were deleted. The ONLY remaining hand-replicated arithmetic is `_derive_mc_scale` +(begin_day computes it inline inside an un-callable method), pinned by a formula test. + +OPEN follow-up: the launcher (`shadow_decision_step`) must pass a PERSISTENT `ob_engine` + +per-scan-incrementing `bar_idx` into `source_live_blue_sizing_factors` so OB accumulation +matches BLUE across scans; today it single-shots, which is wired-correctly but historyless. ## Why we accept it (for now)