diff --git a/prod/clean_arch/violet/live_blue_source.py b/prod/clean_arch/violet/live_blue_source.py index 261fcc1b..0eafbb89 100644 --- a/prod/clean_arch/violet/live_blue_source.py +++ b/prod/clean_arch/violet/live_blue_source.py @@ -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") diff --git a/prod/clean_arch/violet/shadow_live_factors.py b/prod/clean_arch/violet/shadow_live_factors.py index 78896228..7b52432e 100644 --- a/prod/clean_arch/violet/shadow_live_factors.py +++ b/prod/clean_arch/violet/shadow_live_factors.py @@ -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 diff --git a/prod/clean_arch/violet/test_violet_launcher_shadow_live_factors.py b/prod/clean_arch/violet/test_violet_launcher_shadow_live_factors.py index 620176ff..afe9247c 100644 --- a/prod/clean_arch/violet/test_violet_launcher_shadow_live_factors.py +++ b/prod/clean_arch/violet/test_violet_launcher_shadow_live_factors.py @@ -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)] diff --git a/prod/docs/VIOLET_OB_FEED_AND_AGENT_COORDINATION.md b/prod/docs/VIOLET_OB_FEED_AND_AGENT_COORDINATION.md new file mode 100644 index 00000000..fc69bb81 --- /dev/null +++ b/prod/docs/VIOLET_OB_FEED_AND_AGENT_COORDINATION.md @@ -0,0 +1,87 @@ +# VIOLET — OB feed sourcing, HZ Bridge, and multi-agent coordination + +Date: 2026-06-16. Operator-driven decisions captured during the V3.4c/d review. + +## 1. OB feed sourcing doctrine + +**VIOLET consumes BLUE's EXTANT published OBF feed — it does NOT store OB data.** + +- `live_blue_source._source_ob_market` and `shadow_live_factors._default_ob_engine_factory` + use BLUE's OWN `HZOBProvider` (`nautilus_dolphin/.../hz_ob_provider.py`) + `OBFeatureEngine`, + wired exactly as BLUE's `_wire_obf` (nautilus_event_trader.py:4967-4980). +- `HZOBProvider` is a **read-only Hazelcast entry-listener cache**: it subscribes to the + `asset_*_ob` shards BLUE already publishes and keeps only the latest snapshot per asset in + memory. Verified: **no writes, no persistence, no new exchange WS**. OB raw data is the + heaviest disk load in the system; VIOLET adds ZERO to it by reading the published shards. +- `OBFeatureEngine` keeps an in-memory lookback window only (no disk). The launcher holds + ONE engine for the service lifetime (`shadow["ob_engine"]`), stepping a per-scan `bar_idx`, + so OB accumulation matches BLUE across scans (a fresh per-scan engine has no history). + +**Swap seam for future multi-exchange OB.** The `OBProvider` behind `OBFeatureEngine` is the +single swap point. To run a SEPARATE OB stream (e.g. BingX testnet / a third-party venue, +which has genuinely different OB than the Binance reference), replace `HZOBProvider` with a +direct-WS provider feeding the SAME `OBFeatureEngine` — no other code changes. This is +specced/reasonable and the design already accommodates it. + +**Cadence compliance (VIOLET's whole point — faster-than-OBF, event-driven).** Two regimes: +- NOW (shadow / parity): match BLUE, which samples OB at the OBF cadence (~1s native). You + cannot be *more* faithful than BLUE's own OB sampling, so reading the published shards is + correct for parity. `HZOBProvider` is itself a PUSH entry-listener (not polling), which is + already event-driven and aligned with the reactor model. +- LATER (V5/V6 sub-second): when VIOLET wants OB faster than BLUE's OBF, it swaps in a direct + WS OB provider (the seam above) on the reactor clock. That is a DELIBERATE VIOLET feature / + divergence, gated separately — not a parity break of the current shadow stage. + +## 2. HZ Bridge — TODO_HZBRIDGE (refactor ASAP when it ships) + +An upcoming **Hazelcast Bridge** (`dolphinng5_predict/hzbridge`) will be the sanctioned way to +connect to Hazelcast, mitigating the silent client-death / lockup / dropout class (see the +`hz_client_death_investigation` memory + black-box dump work in BLUE). **All VIOLET raw +`HazelcastClient` / `HZOBProvider` connections must route through the bridge once available.** +In-code `TODO_HZBRIDGE` markers flag the three touch points: +1. `shadow_live_factors.build_shadow_live_source` — the launcher's `client_factory`. +2. `shadow_live_factors._default_ob_engine_factory` — `HZOBProvider`'s own connection. +3. `live_blue_source` HZ_CLUSTER/HZ_HOST + `_source_ob_market` provider construction; the + live-HZ smoke test. +Refactor priority: ASAP after the bridge lands (or accelerate the bridge). Until then, VIOLET +uses raw clients, accepting the known fragility (it is DARK, so a dropout loses shadow rows, +never orders). + +## 3. Multi-agent coordination (worktrees + doctrine/status) + +Agents on this box: **Claude, CommandCode, Codex, Crush.** The 2026-06-16 incident — one +agent's staged files swept into another's commit (a doc landed in the forbidden `dita_v2/`) — +was caused by **multiple agents sharing ONE working tree + ONE `.git/index`**. `index.lock` +serializes plumbing ops; it does NOT isolate logical work, and `git commit` commits the WHOLE +index regardless of which files you `git add`. Careful add does not protect you — only +isolation does. + +**Target model (industry standard):** +- **One `git worktree` per agent** over the shared object DB: each gets its own working dir + + index + HEAD, so collisions are impossible. + ``` + git worktree add ../vp-claude -b agent/claude + git worktree add ../vp-commandcode -b agent/commandcode + git worktree add ../vp-codex -b agent/codex + git worktree add ../vp-crush -b agent/crush + ``` + Each agent works ONLY in its own tree. (This is what the Claude Code harness's + `isolation: "worktree"` already does for subagents.) +- **Branch-per-agent → integrate via Gitea PRs.** Push agent branches to the Gitea remote + (`hjnormey/siloqy`); merge to a canonical branch via review. `main`/`release` protected. +- **Doctrine / release = a protected canonical branch + tags.** The LIVE working tree (the one + BLUE/PINK actually run from) tracks the canonical branch only; tag doctrinal snapshots + (`git tag release-YYYYMMDD`). Anything not on the canonical branch is WIP by definition — + that answers "which files are doctrinal vs in-work". +- **Agent work-status board = Gitea branch + PR list.** Each open PR / agent branch (with its + ahead/behind + last-commit author) IS the status dashboard. Optionally a top-level + `AGENT_WORKLOG.md` or the existing `.beads/` tracker for human-readable status. + +**Critical secondary point:** this working tree is ALSO the live deployment path (BLUE runs +`prod/nautilus_event_trader.py` from here). Agents editing it directly means WIP code sits in +the live path — a stray restart could load half-finished edits. Worktrees fix this too: +agents edit isolated trees; deployment becomes an explicit checkout/merge of the canonical +branch onto the live tree. + +Setup is operator-gated (it reorganizes how all four agents work + touches the live tree), so +it is documented here for greenlight rather than executed unilaterally.