DOCS: expand recent Violet touched files inventory

prod/clean_arch/violet/shadow_live_factors.py

@@ -0,0 +1,79 @@
+"""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.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+
+LOGGER = logging.getLogger(__name__)
+
+
+def build_shadow_live_source(
+    *,
+    client_factory=None,
+    selector_factory=None,
+    source_factory=None,
+    scan_history_factory=None,
+):
+    """Create the read-only BLUE live-factor mirror for the shadow path."""
+    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
+
+        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,
+    }
+
+
+def shadow_decision_step(
+    shadow: dict,
+    payload: dict,
+    *,
+    scan_number: int,
+    now_ns: int,
+    vel_div: float,
+    vol_ok: bool,
+) -> bool:
+    """Run one shadow decision against the live BLUE factor plane."""
+    shadow["engine"].observe(payload, scan_number)
+    live_source = shadow.get("live_source")
+    factors = None
+    if live_source is not None:
+        live_result = live_source(
+            shadow["client"],
+            scan_history=shadow["scan_history"],
+            selector=shadow["selector"],
+        )
+        shadow["last_live_source"] = live_result
+        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)

prod/clean_arch/violet/test_violet_launcher_shadow_live_factors.py

@@ -0,0 +1,155 @@
+"""V3.4b launcher shadow wiring — live BLUE factor plane is mandatory."""
+
+from __future__ import annotations
+
+import sys
+from types import SimpleNamespace
+
+sys.path.insert(0, "/mnt/dolphinng5_predict")
+
+from prod.clean_arch.violet.decision_engine import ShadowDecision, SizingFactors
+from prod.clean_arch.violet.shadow_journal import VioletDecisionJournal
+
+
+def test_build_shadow_includes_live_factor_source():
+    from prod.clean_arch.violet import shadow_live_factors as slf
+
+    class FakeClient:
+        pass
+
+    shadow = slf.build_shadow_live_source(
+        client_factory=lambda: FakeClient(),
+        selector_factory=lambda: object(),
+        source_factory=lambda client, scan_history, selector: object(),
+        scan_history_factory=lambda: object(),
+    )
+    assert isinstance(shadow["client"], FakeClient)
+    assert shadow["live_source"] is not None
+    assert shadow["scan_history"] is not None
+    assert shadow["selector"] is not None
+
+
+def test_build_shadow_propagates_client_factory_failure():
+    from prod.clean_arch.violet import shadow_live_factors as slf
+
+    try:
+        slf.build_shadow_live_source(
+            client_factory=lambda: (_ for _ in ()).throw(RuntimeError("hz down")),
+            selector_factory=lambda: object(),
+            source_factory=lambda client, scan_history, selector: object(),
+            scan_history_factory=lambda: object(),
+        )
+    except RuntimeError as exc:
+        assert "hz down" in str(exc)
+    else:
+        raise AssertionError("expected live source failure")
+
+
+def test_shadow_decision_step_uses_live_factors_and_journals():
+    from prod.clean_arch.violet import shadow_live_factors as slf
+
+    observed = []
+    decided = []
+    journal_rows = []
+
+    class FakeEngine:
+        def observe(self, payload, scan_number):
+            observed.append((scan_number, payload["vel_div"]))
+
+        def decide(self, **kwargs):
+            decided.append(kwargs)
+            factors = kwargs["factors"]
+            assert isinstance(factors, SizingFactors)
+            assert factors.posture == "APEX"
+            return ShadowDecision(
+                ts_ns=kwargs["now_ns"],
+                scan_number=kwargs["scan_number"],
+                asset="BTCUSDT",
+                side="SHORT",
+                vel_div=kwargs["vel_div"],
+                fraction=0.2,
+                conviction_leverage=3.0,
+                notional_fraction=0.6,
+                target_exposure=41400.0,
+                ars_score=1.23,
+                bucket_idx=1,
+                actuated=True,
+                base_leverage=1.0,
+                dc_lev_mult=1.0,
+                regime_size_mult=1.0,
+                market_ob_mult=1.0,
+                esof_size_mult=1.0,
+            )
+
+    shadow = {
+        "engine": FakeEngine(),
+        "journal": VioletDecisionJournal(
+            sink=lambda table, row: journal_rows.append((table, row)),
+            session_id="sess",
+        ),
+        "capital": 69_000.0,
+        "mono_ns": lambda: 123,
+        "client": object(),
+        "scan_history": object(),
+        "selector": object(),
+        "live_source": lambda client, scan_history, selector: SimpleNamespace(
+            factors=SizingFactors(
+                boost=1.4,
+                beta=0.2,
+                mc_scale=0.5,
+                esof_score=0.42,
+                ob_median_imbalance=0.12,
+                ob_agreement_pct=0.91,
+                dc_status="CONFIRM",
+                posture="APEX",
+            ),
+            selected_asset="BTCUSDT",
+        ),
+        "live_decisions": 0,
+        "last_live_source": None,
+    }
+    payload = {"vel_div": -0.031, "vol_ok": True}
+    ok = slf.shadow_decision_step(
+        shadow,
+        payload,
+        scan_number=7,
+        now_ns=123,
+        vel_div=-0.031,
+        vol_ok=True,
+    )
+    assert ok is True
+    assert observed == [(7, -0.031)]
+    assert decided and decided[0]["factors"].dc_status == "CONFIRM"
+    assert shadow["last_live_source"].selected_asset == "BTCUSDT"
+    assert journal_rows and journal_rows[0][0] == "violet_decisions"
+
+
+def test_shadow_decision_step_skips_without_live_factor_plane():
+    from prod.clean_arch.violet import shadow_live_factors as slf
+
+    class FailEngine:
+        def observe(self, payload, scan_number):
+            pass
+
+        def decide(self, **kwargs):
+            raise AssertionError("must not fall back to base-only")
+
+    shadow = {
+        "engine": FailEngine(),
+        "journal": VioletDecisionJournal(sink=lambda table, row: None, session_id="sess"),
+        "capital": 69_000.0,
+        "mono_ns": lambda: 123,
+        "client": object(),
+        "scan_history": object(),
+        "selector": object(),
+        "live_source": None,
+    }
+    ok = slf.shadow_decision_step(
+        shadow,
+        {"vel_div": -0.031, "vol_ok": True},
+        scan_number=7,
+        now_ns=123,
+        vel_div=-0.031,
+        vol_ok=True,
+    )
+    assert ok is False

prod/clean_arch/violet/test_violet_trade_slot_compare.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+from prod.clean_arch.violet.trade_slot_compare import (
+    compare_trade_slot_granularity,
+)
+
+
+def test_compare_trade_slot_granularity_collapses_and_matches():
+    decisions = [
+        {
+            "asset": "BTCUSDT",
+            "side": "SHORT",
+            "scan_number": 10,
+            "ts": 1_000,
+            "actuated": True,
+            "conviction_leverage": 3.0,
+            "target_exposure": 30.0,
+        },
+        {
+            "asset": "BTCUSDT",
+            "side": "SHORT",
+            "scan_number": 11,
+            "ts": 2_000,
+            "actuated": True,
+            "conviction_leverage": 4.0,
+            "target_exposure": 40.0,
+        },
+        {
+            "asset": "BTCUSDT",
+            "side": "LONG",
+            "scan_number": 16,
+            "ts": 4_000,
+            "actuated": True,
+            "conviction_leverage": 2.0,
+            "target_exposure": 20.0,
+        },
+    ]
+    trades = [
+        {
+            "trade_id": "t-1",
+            "asset": "BTCUSDT",
+            "side": "SHORT",
+            "ts": 900,
+            "bars_held": 1,
+            "net_pnl": 1.5,
+            "reason": "OPEN",
+        },
+        {
+            "trade_id": "t-1",
+            "asset": "BTCUSDT",
+            "side": "SHORT",
+            "ts": 2_500,
+            "bars_held": 2,
+            "net_pnl": 4.5,
+            "reason": "EXIT",
+        },
+        {
+            "trade_id": "t-2",
+            "asset": "BTCUSDT",
+            "side": "LONG",
+            "ts": 3_900,
+            "bars_held": 1,
+            "net_pnl": -0.25,
+            "reason": "EXIT",
+        },
+    ]
+
+    result = compare_trade_slot_granularity(decisions, trades)
+
+    assert len(result.decision_episodes) == 2
+    assert len(result.trade_episodes) == 2
+    assert len(result.matches) == 2
+    assert not result.decision_only
+    assert not result.trade_only
+
+    short_match = result.matches[0]
+    assert short_match.asset == "BTCUSDT"
+    assert short_match.side == "SHORT"
+    assert short_match.decision_episode.first_scan_number == 10
+    assert short_match.decision_episode.last_scan_number == 11
+    assert short_match.decision_episode.row_count == 2
+    assert short_match.trade_episode.trade_id == "t-1"
+    assert short_match.trade_episode.row_count == 2
+    assert short_match.trade_episode.terminal_reason == "EXIT"
+    assert short_match.trade_episode.net_pnl == 4.5
+    assert short_match.start_gap_ms == 100
+    assert short_match.end_gap_ms == 500
+    assert short_match.row_gap == 0
+    assert short_match.bars_gap == 0
+
+
+def test_compare_trade_slot_granularity_splits_on_scan_gap_and_ignores_bad_rows():
+    decisions = [
+        {
+            "asset": "ETHUSDT",
+            "side": "SHORT",
+            "scan_number": 1,
+            "ts": 10,
+            "actuated": True,
+            "conviction_leverage": 1.0,
+            "target_exposure": 10.0,
+        },
+        {
+            "asset": "ETHUSDT",
+            "side": "SHORT",
+            "scan_number": 2,
+            "ts": 20,
+            "actuated": True,
+            "conviction_leverage": 1.1,
+            "target_exposure": 11.0,
+        },
+        {
+            "asset": "ETHUSDT",
+            "side": "SHORT",
+            "scan_number": 8,
+            "ts": 80,
+            "actuated": True,
+            "conviction_leverage": 1.2,
+            "target_exposure": 12.0,
+        },
+        {
+            "asset": None,
+            "side": "SHORT",
+            "scan_number": "bad",
+            "ts": 90,
+            "actuated": True,
+        },
+    ]
+    trades = [
+        {"trade_id": "x-1", "asset": "ETHUSDT", "side": "SHORT", "ts": 15, "reason": "EXIT"},
+        {"trade_id": "x-2", "asset": "ETHUSDT", "side": "SHORT", "ts": 99, "reason": "EXIT"},
+    ]
+
+    result = compare_trade_slot_granularity(decisions, trades)
+
+    assert len(result.decision_episodes) == 2
+    assert [ep.row_count for ep in result.decision_episodes] == [2, 1]
+    assert len(result.trade_episodes) == 2
+    assert len(result.matches) == 2
+    assert [m.trade_episode.trade_id for m in result.matches] == ["x-1", "x-2"]
+    assert not result.decision_only
+    assert not result.trade_only
+
+
+def test_compare_trade_slot_granularity_handles_empty_input():
+    result = compare_trade_slot_granularity([], [])
+    assert result.decision_episodes == []
+    assert result.trade_episodes == []
+    assert result.matches == []
+    assert result.decision_only == []
+    assert result.trade_only == []

prod/clean_arch/violet/trade_slot_compare.py

@@ -0,0 +1,309 @@
+"""VIOLET trade/slot comparison harness.
+
+This is the missing V3 comparison unit from the main spec: collapse shadow
+decisions into episode-sized runs, collapse raw trade rows into terminal trade
+episodes, and compare them without requiring live execution wiring.
+
+VIOLET-only. BLUE is untouched.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from collections.abc import Iterable, Mapping
+from typing import Any, Optional
+
+from pydantic import Field
+
+from .domain import StrictModel, Symbol, typed
+
+
+def _coerce_int(value: Any, default: Optional[int] = None) -> Optional[int]:
+    try:
+        if value is None:
+            return default
+        out = int(value)
+        return out if out >= 0 else default
+    except (TypeError, ValueError):
+        return default
+
+
+def _coerce_float(value: Any, default: Optional[float] = None) -> Optional[float]:
+    try:
+        if value is None:
+            return default
+        out = float(value)
+        return out if out == out and out not in (float("inf"), float("-inf")) else default
+    except (TypeError, ValueError):
+        return default
+
+
+def _coerce_text(value: Any, default: str = "") -> str:
+    if value is None:
+        return default
+    text = str(value).strip()
+    return text if text else default
+
+
+def _row_asset(row: Mapping[str, Any]) -> str:
+    return _coerce_text(row.get("asset") or row.get("symbol") or row.get("instrument")).upper()
+
+
+def _row_side(row: Mapping[str, Any]) -> str:
+    return _coerce_text(row.get("side") or row.get("direction") or row.get("trade_side")).upper()
+
+
+def _row_scan_number(row: Mapping[str, Any]) -> Optional[int]:
+    return _coerce_int(row.get("scan_number") or row.get("scan") or row.get("scan_idx"))
+
+
+def _row_ts_ms(row: Mapping[str, Any]) -> Optional[int]:
+    candidates = (
+        row.get("ts"),
+        row.get("ts_ms"),
+        row.get("timestamp"),
+        row.get("exit_ts"),
+        row.get("entry_ts"),
+        row.get("mono_ns"),
+    )
+    for value in candidates:
+        ts = _coerce_int(value)
+        if ts is not None:
+            return ts if value is None or value != row.get("mono_ns") else ts // 1_000_000
+    return None
+
+
+def _row_trade_id(row: Mapping[str, Any]) -> str:
+    return _coerce_text(
+        row.get("trade_id") or row.get("id") or row.get("slot_id") or row.get("episode_id"),
+    )
+
+
+class DecisionEpisode(StrictModel):
+    asset: Symbol
+    side: str = Field(min_length=1, max_length=16)
+    first_scan_number: int = Field(ge=0)
+    last_scan_number: int = Field(ge=0)
+    first_ts_ms: int = Field(ge=0)
+    last_ts_ms: int = Field(ge=0)
+    row_count: int = Field(ge=1)
+    actuated_count: int = Field(ge=0)
+    max_conviction_leverage: float = Field(ge=0.0, allow_inf_nan=False)
+    last_target_exposure: float = Field(ge=0.0, allow_inf_nan=False)
+
+
+class TradeEpisode(StrictModel):
+    trade_id: str = Field(min_length=1)
+    asset: Symbol
+    side: str = Field(min_length=1, max_length=16)
+    entry_ts_ms: int = Field(ge=0)
+    exit_ts_ms: int = Field(ge=0)
+    row_count: int = Field(ge=1)
+    bars_held: Optional[int] = Field(default=None, ge=0)
+    net_pnl: Optional[float] = Field(default=None, allow_inf_nan=False)
+    terminal_reason: Optional[str] = None
+
+
+class EpisodeMatch(StrictModel):
+    asset: Symbol
+    side: str = Field(min_length=1, max_length=16)
+    decision_episode: DecisionEpisode
+    trade_episode: TradeEpisode
+    start_gap_ms: int = Field(ge=0)
+    end_gap_ms: int = Field(ge=0)
+    row_gap: int = Field(ge=0)
+    bars_gap: Optional[int] = Field(default=None, ge=0)
+
+
+class TradeSlotComparison(StrictModel):
+    decision_episodes: list[DecisionEpisode]
+    trade_episodes: list[TradeEpisode]
+    matches: list[EpisodeMatch]
+    decision_only: list[DecisionEpisode]
+    trade_only: list[TradeEpisode]
+
+
+def _collapse_decision_rows(
+    rows: Iterable[Mapping[str, Any]],
+    *,
+    max_scan_gap: int = 1,
+) -> list[DecisionEpisode]:
+    grouped: dict[tuple[str, str], list[Mapping[str, Any]]] = defaultdict(list)
+    for row in rows:
+        asset = _row_asset(row)
+        side = _row_side(row)
+        scan_number = _row_scan_number(row)
+        if not asset or not side or scan_number is None:
+            continue
+        grouped[(asset, side)].append(row)
+
+    episodes: list[DecisionEpisode] = []
+    for (asset, side), bucket in grouped.items():
+        bucket = sorted(
+            bucket,
+            key=lambda row: (
+                _row_scan_number(row) or 0,
+                _row_ts_ms(row) or 0,
+            ),
+        )
+        current: list[Mapping[str, Any]] = []
+        prev_scan: Optional[int] = None
+        for row in bucket:
+            scan_number = _row_scan_number(row)
+            if scan_number is None:
+                continue
+            if current and prev_scan is not None and scan_number > prev_scan + max_scan_gap:
+                episodes.append(_decision_episode_from_rows(asset, side, current))
+                current = []
+            current.append(row)
+            prev_scan = scan_number
+        if current:
+            episodes.append(_decision_episode_from_rows(asset, side, current))
+
+    return sorted(episodes, key=lambda ep: (ep.first_ts_ms, ep.asset, ep.side, ep.first_scan_number))
+
+
+def _decision_episode_from_rows(
+    asset: str,
+    side: str,
+    rows: list[Mapping[str, Any]],
+) -> DecisionEpisode:
+    scans = [sn for sn in (_row_scan_number(r) for r in rows) if sn is not None]
+    times = [ts for ts in (_row_ts_ms(r) for r in rows) if ts is not None]
+    conv = [
+        value for value in (_coerce_float(r.get("conviction_leverage"), None) for r in rows)
+        if value is not None
+    ]
+    exposure = [
+        value for value in (_coerce_float(r.get("target_exposure"), None) for r in rows)
+        if value is not None
+    ]
+    actuated = sum(1 for r in rows if bool(r.get("actuated")))
+    return DecisionEpisode(
+        asset=asset,
+        side=side,
+        first_scan_number=min(scans),
+        last_scan_number=max(scans),
+        first_ts_ms=min(times) if times else 0,
+        last_ts_ms=max(times) if times else 0,
+        row_count=len(rows),
+        actuated_count=actuated,
+        max_conviction_leverage=max(conv) if conv else 0.0,
+        last_target_exposure=exposure[-1] if exposure else 0.0,
+    )
+
+
+def _collapse_trade_rows(rows: Iterable[Mapping[str, Any]]) -> list[TradeEpisode]:
+    grouped: dict[str, list[Mapping[str, Any]]] = defaultdict(list)
+    for row in rows:
+        trade_id = _row_trade_id(row)
+        asset = _row_asset(row)
+        side = _row_side(row)
+        if not trade_id or not asset or not side:
+            continue
+        grouped[trade_id].append(row)
+
+    episodes: list[TradeEpisode] = []
+    for trade_id, bucket in grouped.items():
+        bucket = sorted(bucket, key=lambda row: (_row_ts_ms(row) or 0, _coerce_int(row.get("scan_number")) or 0))
+        first = bucket[0]
+        last = bucket[-1]
+        entry_ts = _row_ts_ms(first) or 0
+        exit_ts = _row_ts_ms(last) or entry_ts
+        bars = None
+        for row in reversed(bucket):
+            bars = _coerce_int(row.get("bars_held"))
+            if bars is not None:
+                break
+        net_pnl = None
+        for row in reversed(bucket):
+            net_pnl = _coerce_float(row.get("net_pnl") or row.get("pnl"), None)
+            if net_pnl is not None:
+                break
+        reason = None
+        for row in reversed(bucket):
+            reason = _coerce_text(row.get("reason") or row.get("exit_reason"), "")
+            if reason:
+                break
+        episodes.append(
+            TradeEpisode(
+                trade_id=trade_id,
+                asset=_row_asset(first),
+                side=_row_side(first),
+                entry_ts_ms=entry_ts,
+                exit_ts_ms=exit_ts,
+                row_count=len(bucket),
+                bars_held=bars,
+                net_pnl=net_pnl,
+                terminal_reason=reason or None,
+            )
+        )
+
+    return sorted(episodes, key=lambda ep: (ep.entry_ts_ms, ep.asset, ep.side, ep.trade_id))
+
+
+def _match_episodes(
+    decisions: list[DecisionEpisode],
+    trades: list[TradeEpisode],
+) -> tuple[list[EpisodeMatch], list[DecisionEpisode], list[TradeEpisode]]:
+    by_key: dict[tuple[str, str], list[TradeEpisode]] = defaultdict(list)
+    for trade in trades:
+        by_key[(trade.asset, trade.side)].append(trade)
+    for bucket in by_key.values():
+        bucket.sort(key=lambda ep: (ep.entry_ts_ms, ep.exit_ts_ms, ep.trade_id))
+
+    matches: list[EpisodeMatch] = []
+    decision_only: list[DecisionEpisode] = []
+    used_trade_ids: set[str] = set()
+
+    for decision in decisions:
+        bucket = by_key.get((decision.asset, decision.side), [])
+        candidate = None
+        for trade in bucket:
+            if trade.trade_id in used_trade_ids:
+                continue
+            candidate = trade
+            break
+        if candidate is None:
+            decision_only.append(decision)
+            continue
+        used_trade_ids.add(candidate.trade_id)
+        matches.append(
+            EpisodeMatch(
+                asset=decision.asset,
+                side=decision.side,
+                decision_episode=decision,
+                trade_episode=candidate,
+                start_gap_ms=abs(decision.first_ts_ms - candidate.entry_ts_ms),
+                end_gap_ms=abs(decision.last_ts_ms - candidate.exit_ts_ms),
+                row_gap=abs(decision.row_count - candidate.row_count),
+                bars_gap=(
+                    abs(decision.row_count - candidate.bars_held)
+                    if candidate.bars_held is not None
+                    else None
+                ),
+            )
+        )
+
+    trade_only = [trade for trade in trades if trade.trade_id not in used_trade_ids]
+    return matches, decision_only, trade_only
+
+
+@typed
+def compare_trade_slot_granularity(
+    decision_rows: Iterable[Mapping[str, Any]],
+    trade_rows: Iterable[Mapping[str, Any]],
+    *,
+    max_scan_gap: int = 1,
+) -> TradeSlotComparison:
+    """Collapse both surfaces to episodes and compare them at slot granularity."""
+    decisions = _collapse_decision_rows(decision_rows, max_scan_gap=max_scan_gap)
+    trades = _collapse_trade_rows(trade_rows)
+    matches, decision_only, trade_only = _match_episodes(decisions, trades)
+    return TradeSlotComparison(
+        decision_episodes=decisions,
+        trade_episodes=trades,
+        matches=matches,
+        decision_only=decision_only,
+        trade_only=trade_only,
+    )

prod/docs/RECENT_VIOLET_34D_fb34431.md

@@ -0,0 +1,102 @@
+# RECENT_VIOLET_34D_fb34431
+
+## What This Work Was
+
+This pass continued the VIOLET plan after `V3.4c` by wiring the launcher-side
+shadow path to the live BLUE factor plane.
+
+The goal stayed read-only. BLUE code, BLUE schemas, and BLUE data structures
+were not modified. The change was entirely on the VIOLET side.
+
+## Scope
+
+This pass added a thin shadow-side live-factor attachment and a focused test
+surface:
+
+- `prod/clean_arch/violet/shadow_live_factors.py`
+- `prod/clean_arch/violet/test_violet_launcher_shadow_live_factors.py`
+- `prod/launch_dolphin_violet.py`
+
+It reused the existing V3.4c live BLUE source adapter:
+
+- `prod/clean_arch/violet/live_blue_source.py`
+- `prod/clean_arch/violet/live_factor_source.py`
+- `prod/clean_arch/violet/live_factors.py`
+
+## Why This Was Needed
+
+Before this pass, the Violet shadow launcher still had a base-only decision
+path. That meant the `VioletDecisionEngine` could produce a muted decision, but
+it did not yet receive the full live factor plane from BLUE’s published
+surfaces inside the launcher path.
+
+The missing piece was the launcher-side wiring: source live BLUE factors,
+thread them into `decide(...)`, and keep the journaled breakdown faithful to the
+same factor plane BLUE would have seen at that scan.
+
+## What Was Added
+
+### 1. Shadow live-factor helper
+
+`shadow_live_factors.py` now provides two small helpers:
+
+- `build_shadow_live_source(...)`
+- `shadow_decision_step(...)`
+
+`build_shadow_live_source(...)` assembles the read-only BLUE live factor mirror
+for the shadow path. The helper keeps imports lazy so it can be unit-tested
+without dragging in the full launcher import chain.
+
+`shadow_decision_step(...)` runs one muted shadow decision using the live BLUE
+factor plane, then journals the result when the decision is actuated.
+
+### 2. Launcher wiring
+
+`launch_dolphin_violet.py` now:
+
+- builds the live-factor shadow source when shadow mode is enabled
+- passes the live `SizingFactors` into `VioletDecisionEngine.decide(...)`
+- skips the shadow decision instead of silently falling back to base-only when
+  the live factor plane is missing
+- keeps the existing journal path intact
+
+This preserves the existing muted-shadow architecture while making the shadow
+decision reflect the live BLUE factor plane rather than a reduced fallback.
+
+### 3. Focused tests
+
+`test_violet_launcher_shadow_live_factors.py` covers:
+
+- the live-factor helper contract
+- failure propagation from the client factory
+- the shadow decision step with live factors and journaling
+- the no-live-factor skip path
+
+Because the mount was slow under `pytest`, I verified the helper path directly
+with a small execution script instead of waiting on long file-system waits.
+
+## Exactness Rules Followed
+
+The pass stayed conservative:
+
+- no BLUE edits
+- no schema edits
+- no live fallback to a fake factor plane
+- no execution path changes outside the muted shadow branch
+- no silent loss of the live factor breakdown
+
+## Verification
+
+Direct runtime check:
+
+- the new helper built successfully with injected factories
+- the shadow decision step accepted the live factor plane
+- the decision was journaled with the expected Violet journal table
+
+Observed direct result:
+
+- `ok`
+
+`pytest` on this mount was slow and repeatedly stalled in netfs waits, so I did
+not treat that as a code failure.
+

prod/docs/RECENT_VIOLET_34E_2d37ef0.md

@@ -0,0 +1,83 @@
+# RECENT VIOLET 34E — trade/slot-granularity comparison harness
+
+This pass implemented the next numbered comparison item from the main Violet spec:
+the trade/slot-granularity comparison that was still deferred after the V3.4c
+live-source work and shadow journal wiring.
+
+## What existed before
+
+Before this pass, Violet already had:
+
+- `VioletDecisionEngine` producing shadow decisions
+- `VioletDecisionJournal` persisting actuated decisions to `violet_decisions`
+- live-source adapters for BLUE-published factors
+- full sizing parity through `VioletSizer`
+
+What was still missing was a dedicated comparison unit that could collapse those
+journaled decisions into episode-sized runs and compare them against the terminal
+trade surface at the same granularity.
+
+## What was added
+
+I added a new Violet-only comparator:
+
+- [prod/clean_arch/violet/trade_slot_compare.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/trade_slot_compare.py)
+- [prod/clean_arch/violet/test_violet_trade_slot_compare.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_trade_slot_compare.py)
+
+The new module provides:
+
+- `DecisionEpisode`: a collapsed run of contiguous shadow decisions for one
+  asset/side
+- `TradeEpisode`: a collapsed terminal trade record grouped by `trade_id`
+- `EpisodeMatch`: a paired decision/trade episode with timing and row-count gaps
+- `TradeSlotComparison`: the full comparison result
+- `compare_trade_slot_granularity(...)`: the top-level collapse-and-compare API
+
+## How it works
+
+The comparator is deliberately narrow:
+
+- it groups decision rows by `asset` + `side`
+- it splits them into episodes when the scan number gap exceeds the configured
+  `max_scan_gap`
+- it groups trade rows by `trade_id`
+- it ignores malformed rows rather than failing on them
+- it matches episodes by asset/side and preserves unmatched decision/trade rows
+
+This is downstream of the existing shadow journal. It does not reimplement ACB,
+EsoF, OB, or sizing math. It only compares the surfaces that already exist.
+
+## Anomaly handling
+
+The comparator rejects or skips bad inputs instead of letting them pollute the
+episode view:
+
+- missing asset or side
+- missing or invalid scan number
+- malformed or missing trade id
+- non-finite numeric fields
+
+That keeps the harness usable against noisy journal extracts and historical trade
+rows that may contain replay artifacts.
+
+## Tests
+
+The new tests cover:
+
+- episode collapse across contiguous decision rows
+- terminal trade dedupe through `trade_id`
+- scan-gap splitting
+- malformed row handling
+- empty-input behavior
+
+Verification on this pass:
+
+- `PYTHONPATH=/mnt/dolphinng5_predict rtk python -m pytest -q /mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_trade_slot_compare.py`
+- result: `3 passed`
+
+## Why this is the right next step
+
+The main Violet plan explicitly defers trade/slot-granularity comparison until the
+shadow side has a comparable execution surface. That condition is now met well
+enough to build the comparison harness without touching BLUE or the live executor.
+

prod/docs/RECENT_VIOLET_TOUCHED_FILES.md

@@ -0,0 +1,48 @@
+# RECENT VIOLET Touched Files
+
+This inventory covers the recent VIOLET 3.4-series commits on this branch.
+It includes all files touched across the series, grouped by commit, so the
+surface is explicit rather than inferred.
+
+## 722fd9f — VIOLET V3.4b/V3e: journal the full-sizing breakdown
+
+- [prod/clean_arch/violet/shadow_journal.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/shadow_journal.py)
+- [prod/clean_arch/violet/test_violet_shadow_journal.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_shadow_journal.py)
+- [prod/clickhouse/violet/22_violet_decisions.sql](/mnt/dolphinng5_predict/prod/clickhouse/violet/22_violet_decisions.sql)
+
+## a632c59 — VIOLET V3.4b: validate live-factor field paths + HZ sourcing adapter
+
+- [prod/clean_arch/violet/live_factor_source.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/live_factor_source.py)
+- [prod/clean_arch/violet/test_violet_live_factor_source.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_live_factor_source.py)
+- [prod/docs/VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md](/mnt/dolphinng5_predict/prod/docs/VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md)
+
+## 1ac3f62 — VIOLET V3.4c: read-only BLUE live source parity
+
+- [prod/clean_arch/violet/live_blue_source.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/live_blue_source.py)
+- [prod/clean_arch/violet/test_violet_live_blue_source.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_live_blue_source.py)
+
+## 16add44 — DOCS: add RECENT VIOLET 34C detail note
+
+- [prod/docs/RECENT_VIOLET_34C_a632c59.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34C_a632c59.md)
+
+## fb34431 — VIOLET V3.4b: launcher shadow live-factor wiring
+
+- [prod/clean_arch/violet/shadow_live_factors.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/shadow_live_factors.py)
+- [prod/clean_arch/violet/test_violet_launcher_shadow_live_factors.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_launcher_shadow_live_factors.py)
+- [prod/docs/RECENT_VIOLET_34D_pending.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34D_pending.md)
+- [prod/launch_dolphin_violet.py](/mnt/dolphinng5_predict/prod/launch_dolphin_violet.py)
+
+## 2d37ef0 — VIOLET V3.4c: trade-slot comparison harness
+
+- [prod/clean_arch/violet/test_violet_trade_slot_compare.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_trade_slot_compare.py)
+- [prod/clean_arch/violet/trade_slot_compare.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/trade_slot_compare.py)
+- [prod/docs/RECENT_VIOLET_34E_pending.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34E_pending.md)
+
+## Rename / note follow-ups in the same series
+
+- [prod/docs/RECENT_VIOLET_34D_fb34431.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34D_fb34431.md)
+- [prod/docs/RECENT_VIOLET_34E_2d37ef0.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34E_2d37ef0.md)
+- [prod/docs/RECENT_VIOLET_TOUCHED_FILES.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_TOUCHED_FILES.md)
+- [prod/docs/RECENT_VIOLET_34C_a632c59.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34C_a632c59.md)
+- [prod/docs/RECENT_VIOLET_34D_pending.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34D_pending.md)
+- [prod/docs/RECENT_VIOLET_34E_pending.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34E_pending.md)

prod/launch_dolphin_violet.py

@@ -50,6 +50,10 @@ from prod.launch_dolphin_pink import (  # noqa: E402
     _resolve_bingx_exchange_leverage_cap,
     _resolve_bingx_recv_window_ms,
 )
+from prod.clean_arch.violet.shadow_live_factors import (  # noqa: E402
+    build_shadow_live_source,
+    shadow_decision_step,
+)
 
 logging.basicConfig(
     level=logging.INFO,
@@ -260,17 +264,18 @@ async def _divergence_driver(divergence, data_feed, poll_s: float, shadow=None)
                 if shadow is not None and started:
                     try:
                         sn = int(payload.get("scan_number") or 0)
-                        shadow["engine"].observe(payload, sn)
                         vd = payload.get("vel_div")
                         if vd is not None:
                             now_ns = shadow["mono_ns"]()
-                            d = shadow["engine"].decide(
-                                now_ns=now_ns, scan_number=sn,
-                                capital=shadow["capital"], vel_div=float(vd),
+                            if shadow_decision_step(
+                                shadow,
+                                payload,
+                                scan_number=sn,
+                                now_ns=now_ns,
+                                vel_div=float(vd),
                                 vol_ok=bool(payload.get("vol_ok", True)),
-                            )
-                            if d is not None:
-                                shadow["journal"].journal(d, mono_ns=now_ns)
+                            ):
+                                shadow["live_decisions"] += 1
                     except Exception as exc:  # noqa: BLE001 — shadow must never die
                         LOGGER.debug("shadow decision failed: %s", exc)
         except Exception as exc:  # noqa: BLE001 — sampling must never die
@@ -333,13 +338,29 @@ def _build_shadow():
     relaxed = abs(thr - (-0.02)) > 1e-9
     engine = VioletDecisionEngine(entry_vel_div_threshold=thr)
     journal = VioletDecisionJournal(sink=ch_put_violet, session_id=sess)
+    try:
+        live_source = build_shadow_live_source()
+    except Exception as exc:
+        LOGGER.warning(
+            "VIOLET shadow live-factor source unavailable (%s) — shadow DISABLED.",
+            exc,
+        )
+        return None
     LOGGER.warning(
         "VIOLET DECISION SHADOW ON (session=%s ref_capital=%.0f entry_thr=%.4f%s) — "
         "journaling muted decisions to dolphin_violet.violet_decisions; NO orders.",
         sess, capital, thr,
         " RELAXED:not-parity-faithful" if relaxed else "",
     )
-    return {"engine": engine, "journal": journal, "capital": capital, "mono_ns": mono_ns}
+    return {
+        "engine": engine,
+        "journal": journal,
+        "capital": capital,
+        "mono_ns": mono_ns,
+        **live_source,
+        "live_decisions": 0,
+        "last_live_source": None,
+    }
 
 
 async def run() -> None: