"""VIOLET launcher shadow helpers for live BLUE factor sourcing. These helpers stay separate from the launcher module so they can be unit-tested without importing the full launcher import chain. They hold the PERSISTENT state BLUE keeps across scans so the shadow plane is faithful to BLUE's live factor stream rather than a per-scan single-shot: - ONE OBFeatureEngine, wired lazily on first assets and kept for the service lifetime (OBFeatureEngine accumulates a lookback window; a fresh engine per scan has none). It reads BLUE's EXTANT published OBF feed via HZOBProvider (a read-only HZ entry- listener cache — NO new OB storage). The provider is the swap seam for a future direct BingX / 3rd-party OB stream. See VIOLET_OB_FEED_AND_AGENT_COORDINATION.md. - a per-scan-incrementing ``ob_bar_idx`` (BLUE steps one bar_idx into step_live). - the last good ``prior_boost_beta`` so a stale-exf scan keeps the prior (BLUE keeps _day_base_boost/_day_beta). """ from __future__ import annotations import logging import os LOGGER = logging.getLogger("violet.shadow_live_factors") def _default_ob_engine_factory(assets): """Wire BLUE's EXTANT OBF feed: HZOBProvider (read-only listener) → OBFeatureEngine. Exactly BLUE's _wire_obf (nautilus_event_trader.py:4967-4980). No OB storage, no new exchange WS — it consumes the asset_*_ob shards BLUE already publishes to Hazelcast.""" from nautilus_dolphin.nautilus.ob_features import OBFeatureEngine from nautilus_dolphin.nautilus.hz_ob_provider import HZOBProvider from .live_blue_source import HZ_CLUSTER, HZ_HOST # TODO_HZBRIDGE: HZOBProvider opens its OWN raw HazelcastClient. Once dolphinng5_predict/ # hzbridge ships, this connection MUST route through the bridge (silent HZ client death / # lockup mitigation — see [[hz_client_death_investigation]]). One of 3 VIOLET HZ touch # points to refactor (also build_shadow_live_source's client_factory + the live smoke). provider = HZOBProvider(hz_cluster=HZ_CLUSTER, hz_host=HZ_HOST, assets=list(assets)) return OBFeatureEngine(provider) def build_shadow_live_source( *, client_factory=None, selector_factory=None, source_factory=None, scan_history_factory=None, ob_engine_factory=None, ): """Create the read-only BLUE live-factor mirror for the shadow path, with the persistent OB engine / bar_idx / boost-beta-prior state BLUE keeps across scans.""" if client_factory is None or selector_factory is None or source_factory is None or scan_history_factory is None: import hazelcast from .alpha_wrappers import VioletAssetSelector from .live_blue_source import LiveBlueScanHistory, source_live_blue_sizing_factors # TODO_HZBRIDGE: raw HazelcastClient — route through dolphinng5_predict/hzbridge once # it ships, to avoid the silent-death/lockup class. See [[hz_client_death_investigation]]. client_factory = client_factory or (lambda: hazelcast.HazelcastClient( cluster_name=os.environ.get("HZ_CLUSTER", "dolphin"), cluster_members=[os.environ.get("HZ_HOST", "localhost:5701")], )) selector_factory = selector_factory or VioletAssetSelector source_factory = source_factory or source_live_blue_sizing_factors scan_history_factory = scan_history_factory or LiveBlueScanHistory client = client_factory() return { "client": client, "scan_history": scan_history_factory(), "selector": selector_factory(), "live_source": source_factory, # persistent state (BLUE keeps these across scans) "ob_engine": None, "ob_engine_factory": ob_engine_factory or _default_ob_engine_factory, "ob_bar_idx": 0, "prior_boost_beta": None, "last_live_source": None, } def _ensure_ob_engine(shadow, payload): """Lazily wire ONE OB engine on the first scan that has an asset universe — like BLUE's _wire_obf (`if not assets or self.ob_assets: return`). Kept for the service lifetime.""" if shadow.get("ob_engine") is not None: return shadow["ob_engine"] factory = shadow.get("ob_engine_factory") if factory is None: return None from .live_blue_source import _scan_assets, _scan_view assets = _scan_assets(_scan_view(payload)) if not assets: return None eng = factory(assets) shadow["ob_engine"] = eng shadow["ob_assets"] = assets LOGGER.info("shadow OB wired to BLUE's extant feed for %d assets", len(assets)) return eng def shadow_decision_step( shadow: dict, payload: dict, *, scan_number: int, now_ns: int, vel_div: float, vol_ok: bool, ) -> bool: """Run one muted shadow decision against the live BLUE factor plane, carrying the persistent OB engine, bar_idx, and boost/beta prior across scans (BLUE-faithful).""" shadow["engine"].observe(payload, scan_number) live_source = shadow.get("live_source") factors = None if live_source is not None: ob_engine = _ensure_ob_engine(shadow, payload) live_result = live_source( shadow["client"], scan_history=shadow["scan_history"], selector=shadow["selector"], ob_engine=ob_engine, bar_idx=shadow.get("ob_bar_idx", 0), prior_boost_beta=shadow.get("prior_boost_beta"), ) shadow["last_live_source"] = live_result shadow["ob_bar_idx"] = shadow.get("ob_bar_idx", 0) + 1 # persist last good boost/beta as next scan's prior (BLUE keeps _day_base_boost/_day_beta). ab = getattr(live_result, "acb_boost", None) bb = getattr(live_result, "acb_beta", None) if ab is not None and bb is not None: shadow["prior_boost_beta"] = (ab, bb) factors = live_result.factors if factors is None: return False decision = shadow["engine"].decide( now_ns=now_ns, scan_number=scan_number, capital=shadow["capital"], vel_div=vel_div, vol_ok=vol_ok, factors=factors, ) if decision is None: return False return shadow["journal"].journal(decision, mono_ns=now_ns)