VIOLET V3.4d: launcher OB-persistence + boost/beta prior; OB-feed/HZ-bridge/coordination doc
Launcher shadow path now keeps the persistent state BLUE keeps across scans (faithful, not per-scan single-shot): ONE OBFeatureEngine wired lazily on first assets and kept for the service lifetime, reading BLUE's EXTANT published OBF feed via HZOBProvider (read-only HZ entry-listener cache — NO new OB storage; provider is the swap seam for a future direct BingX/3rd-party OB stream); per-scan ob_bar_idx into step_live/get_market; and prior_boost_beta carried across scans (stale exf keeps the prior). shadow_decision_step threads ob_engine + bar_idx + prior into source_live_blue_sizing_factors; build_shadow_live_source seeds the state + a default ob_engine_factory (injectable for tests). TODO_HZBRIDGE markers added at all 3 VIOLET->Hazelcast touch points: per operator the upcoming dolphinng5_predict/hzbridge must become the sanctioned HZ connection (silent client-death / lockup / dropout mitigation) — refactor ASAP once it ships. Doc VIOLET_OB_FEED_AND_AGENT_COORDINATION.md: OB-feed sourcing doctrine, HZ-bridge TODO, and the multi-agent worktree + doctrine/release/status scheme (prompted by the shared-index incidents where agents' staged files cross-contaminate commits). 6 launcher tests (+wired-once/bar_idx-increments, +prior-carries). violet-only; partial commit (only my 4 paths) to avoid sweeping a concurrent agent's staged files. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Codex
@@ -70,6 +70,9 @@ LOGGER = logging.getLogger("violet.live_blue_source")
|
||||
|
||||
# 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.
|
||||
# TODO_HZBRIDGE: all VIOLET HZ access (the caller's client + HZOBProvider) must move to
|
||||
# dolphinng5_predict/hzbridge when it ships — raw HazelcastClient is the silent-death/lockup
|
||||
# surface (see prod/docs/VIOLET_OB_FEED_AND_AGENT_COORDINATION.md + hz_client_death memory).
|
||||
HZ_CLUSTER = _os.environ.get("HZ_CLUSTER", "dolphin")
|
||||
HZ_HOST = _os.environ.get("HZ_HOST", "127.0.0.1:5701")
|
||||
|
||||
|
||||
@@ -2,6 +2,17 @@
|
||||
|
||||
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
|
||||
@@ -9,7 +20,25 @@ from __future__ import annotations
|
||||
import logging
|
||||
import os
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
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(
|
||||
@@ -18,14 +47,18 @@ def build_shadow_live_source(
|
||||
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."""
|
||||
"""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")],
|
||||
@@ -40,9 +73,35 @@ def build_shadow_live_source(
|
||||
"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,
|
||||
@@ -52,17 +111,28 @@ def shadow_decision_step(
|
||||
vel_div: float,
|
||||
vol_ok: bool,
|
||||
) -> bool:
|
||||
"""Run one shadow decision against the live BLUE factor plane."""
|
||||
"""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
|
||||
|
||||
@@ -92,7 +92,8 @@ def test_shadow_decision_step_uses_live_factors_and_journals():
|
||||
"client": object(),
|
||||
"scan_history": object(),
|
||||
"selector": object(),
|
||||
"live_source": lambda client, scan_history, selector: SimpleNamespace(
|
||||
"live_source": lambda client, scan_history=None, selector=None, ob_engine=None,
|
||||
bar_idx=0, prior_boost_beta=None: SimpleNamespace(
|
||||
factors=SizingFactors(
|
||||
boost=1.4,
|
||||
beta=0.2,
|
||||
@@ -104,6 +105,8 @@ def test_shadow_decision_step_uses_live_factors_and_journals():
|
||||
posture="APEX",
|
||||
),
|
||||
selected_asset="BTCUSDT",
|
||||
acb_boost=1.4,
|
||||
acb_beta=0.2,
|
||||
),
|
||||
"live_decisions": 0,
|
||||
"last_live_source": None,
|
||||
@@ -153,3 +156,69 @@ def test_shadow_decision_step_skips_without_live_factor_plane():
|
||||
vol_ok=True,
|
||||
)
|
||||
assert ok is False
|
||||
|
||||
|
||||
class _NoDecisionEngine:
|
||||
def observe(self, payload, scan_number):
|
||||
pass
|
||||
|
||||
def decide(self, **kwargs):
|
||||
return None # step returns False after the live-source call (which is what we probe)
|
||||
|
||||
|
||||
def test_shadow_step_lazily_wires_ob_engine_once_and_increments_bar_idx():
|
||||
from prod.clean_arch.violet import shadow_live_factors as slf
|
||||
|
||||
built, seen = [], []
|
||||
|
||||
def ob_factory(assets):
|
||||
eng = SimpleNamespace(assets=list(assets))
|
||||
built.append(eng)
|
||||
return eng
|
||||
|
||||
def live_source(client, *, scan_history, selector, ob_engine, bar_idx, prior_boost_beta):
|
||||
seen.append((ob_engine, bar_idx))
|
||||
return SimpleNamespace(factors=SizingFactors(posture="APEX"), selected_asset="BTCUSDT",
|
||||
acb_boost=1.0, acb_beta=0.0)
|
||||
|
||||
shadow = slf.build_shadow_live_source(
|
||||
client_factory=lambda: object(), selector_factory=lambda: object(),
|
||||
source_factory=live_source, scan_history_factory=lambda: object(),
|
||||
ob_engine_factory=ob_factory,
|
||||
)
|
||||
shadow["engine"] = _NoDecisionEngine()
|
||||
shadow["capital"] = 1.0
|
||||
payload = {"assets": ["BTCUSDT", "ETHUSDT"], "vel_div": -0.03}
|
||||
for i in range(3):
|
||||
slf.shadow_decision_step(shadow, payload, scan_number=i, now_ns=i, vel_div=-0.03, vol_ok=True)
|
||||
|
||||
assert len(built) == 1 # OB engine wired exactly ONCE
|
||||
assert all(s[0] is built[0] for s in seen) # same persistent engine reused
|
||||
assert [s[1] for s in seen] == [0, 1, 2] # bar_idx increments per scan
|
||||
|
||||
|
||||
def test_shadow_step_carries_boost_beta_prior_across_scans():
|
||||
from prod.clean_arch.violet import shadow_live_factors as slf
|
||||
|
||||
priors_seen, counter = [], {"n": 0}
|
||||
|
||||
def live_source(client, *, scan_history, selector, ob_engine, bar_idx, prior_boost_beta):
|
||||
priors_seen.append(prior_boost_beta)
|
||||
counter["n"] += 1
|
||||
b = float(counter["n"])
|
||||
return SimpleNamespace(factors=SizingFactors(posture="APEX"), selected_asset="X",
|
||||
acb_boost=b, acb_beta=b / 10.0)
|
||||
|
||||
shadow = slf.build_shadow_live_source(
|
||||
client_factory=lambda: object(), selector_factory=lambda: object(),
|
||||
source_factory=live_source, scan_history_factory=lambda: object(),
|
||||
ob_engine_factory=lambda assets: object(),
|
||||
)
|
||||
shadow["engine"] = _NoDecisionEngine()
|
||||
shadow["capital"] = 1.0
|
||||
payload = {"assets": ["X"], "vel_div": -0.03}
|
||||
for i in range(3):
|
||||
slf.shadow_decision_step(shadow, payload, scan_number=i, now_ns=i, vel_div=-0.03, vol_ok=True)
|
||||
|
||||
# scan0: no prior yet; scan1: prior = scan0's (1.0, 0.1); scan2: prior = scan1's (2.0, 0.2)
|
||||
assert priors_seen == [None, (1.0, 0.1), (2.0, 0.2)]
|
||||
|
||||
Reference in New Issue
Block a user