10 Commits

Author SHA1 Message Date
Codex
520d911722 DOCS: expand recent Violet touched files inventory 2026-06-16 15:55:09 +02:00
Codex
fe56ef522e DOCS: add recent Violet touched files list 2026-06-16 15:47:35 +02:00
Codex
47da295ffe DOCS: rename RECENT VIOLET 34E to hash-based name 2026-06-16 15:40:45 +02:00
Codex
2d37ef0ae0 VIOLET V3.4c: trade-slot comparison harness 2026-06-16 15:38:25 +02:00
Codex
d6e967f120 DOCS: rename RECENT VIOLET 34D to hash-based name 2026-06-16 15:17:53 +02:00
Codex
fb344318aa VIOLET V3.4b: launcher shadow live-factor wiring 2026-06-16 15:14:57 +02:00
Codex
16add44326 DOCS: add RECENT VIOLET 34C detail note 2026-06-16 14:36:55 +02:00
Codex
1ac3f627df VIOLET V3.4c: read-only BLUE live source parity 2026-06-16 14:34:49 +02:00
Codex
a632c595ba VIOLET V3.4b: validate live-factor field paths + HZ sourcing adapter
#1 (validation) — VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md documents the
ground truth from BLUE's own code (esf_alpha_orchestrator / adaptive_circuit_breaker
/ nautilus_event_trader): of the 8 sizing inputs, ONLY `posture` is a flat HZ key.
`esof_score` is in HZ but as a payload to parse. The other five — boost, beta,
mc_scale, ob_median_imbalance, ob_agreement_pct, dc_status — are BLUE-organ outputs
(ACB over DOLPHIN_FEATURES.exf_latest, MC flag derivation, OBFeatureEngine, the
per-asset signal generator) and are NOT present as scalars in any HZ map. This
inverts live_factors.py's flat-snapshot premise; its speculative alternate paths
(acb_boost / s_acb_boost / ("acb","boost") / …) match nothing real. The full live
sourcing of the five is a multi-organ sprint (V3.4c), not a HZ scrape.

#2 (sourcing adapter) — live_factor_source.py sources what BLUE actually publishes,
read-only and pure (callers pass already-fetched HZ blobs; no client, no I/O):
  - posture     ← engine_snapshot['posture'] (DOLPHIN_STATE_BLUE), default APEX
  - esof_score  ← DOLPHIN_FEATURES['esof_latest'] via BLUE's OWN parse_esof_payload
                  + esof_score_from_payload (wrap, don't reimplement; staleness gate
                  honored when max_age_s supplied)
  - boost/beta/mc_scale/ob_*/dc_status ← BLUE's neutral sentinels (1.0/0.0/1.0/None/
                  None/"NONE") until V3.4c — explicit, never silently faked.
Flows through the validated extract_live_sizing_factors normalizer. ORGAN_DERIVED_
FACTORS names the six deferred to V3.4c so the journal can mark them NEUTRAL.

9 new tests green (posture default/upper, esof dict+raw-JSON+staleness, neutral
integration, all-neutral-when-empty). violet-only; no shared-file edits; no soak.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-16 13:22:59 +02:00
Codex
722fd9f054 VIOLET V3.4b/V3e: journal the full-sizing breakdown (DDL + row guard)
ShadowDecision already carries the V3.4 5-factor breakdown (base_leverage,
dc_lev_mult, regime_size_mult, market_ob_mult, esof_size_mult), but the journal
row and CH DDL dropped it — so a DARK soak would record decisions WITHOUT the
factor decomposition that V3.4 exists to expose. Thread it through:

- 22_violet_decisions.sql: 5 additive Nullable(Float64) breakdown columns. NULL
  on the legacy base-only path; populated once the launcher feeds live factor
  planes. Note added: on a pre-existing table use ALTER ... ADD COLUMN instead of
  the CREATE IF NOT EXISTS (no live table yet — VIOLET is DARK, never soaked).
- shadow_journal.py: DecisionRow gains the 5 Optional[float] fields (ge=0.0,
  finite-guarded); journal() populates them via getattr(..., None) so the
  base-only path and duck-typed reject tests stay NULL/rejected rather than
  raising on attribute access.
- test_violet_shadow_journal.py: breakdown round-trips on the full path; NULL on
  base-only; a negative multiplier is rejected at the row guard. The existing
  DecisionRow-fields == DDL-columns parity test still holds with the new columns.

violet-only; no shared-file edits; no soak. 29 violet tests green
(7 journal + 22 engine/DDL-apply).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-16 13:04:30 +02:00
17 changed files with 2201 additions and 9 deletions

View File

@@ -0,0 +1,361 @@
"""VIOLET V3.4c: read BLUE-published live organs from Hazelcast.
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
``OBFeatureEngine``
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.
"""
from __future__ import annotations
import json
import sys
from collections import deque
from collections.abc import Mapping
from dataclasses import dataclass
from dataclasses import field
from pathlib import Path
from typing import Any, Deque, Iterable, Optional
import hazelcast
import numpy as np
_PROJECT_ROOT = Path(__file__).resolve().parents[3]
for _p in (str(_PROJECT_ROOT), str(_PROJECT_ROOT / "nautilus_dolphin")):
if _p not in sys.path:
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 .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
def _jsonish(value: Any) -> Any:
if isinstance(value, str):
try:
return json.loads(value)
except Exception:
return value
return value
def _coerce_float(value: Any, default: Optional[float] = None) -> Optional[float]:
try:
if value is None:
return default
out = float(value)
if not np.isfinite(out):
return default
return out
except (TypeError, ValueError):
return default
def _read_hz_map(client: hazelcast.HazelcastClient, map_name: str, key: str) -> Any:
try:
return client.get_map(map_name).blocking().get(key)
except Exception:
return None
def _map_status_to_mc_scale(payload: Any) -> float:
data = _jsonish(payload)
if isinstance(data, Mapping):
status = str(data.get("status", "")).upper()
else:
status = str(data).upper()
return 0.5 if status == "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 _scan_view(payload: Any) -> Mapping[str, Any]:
"""Normalize NG7 nested or NG8 flat scan payloads into one mapping."""
data = _jsonish(payload)
if not isinstance(data, Mapping):
return {}
view: dict[str, Any] = dict(data)
result = data.get("result")
if isinstance(result, Mapping):
view.update(result)
return view
def _scan_assets(scan: Mapping[str, Any]) -> list[str]:
assets = scan.get("assets")
if isinstance(assets, list) and assets:
return [str(a).upper() for a in assets if a is not None]
target = scan.get("target_asset") or scan.get("asset")
return [str(target).upper()] if target else []
def _scan_prices(scan: Mapping[str, Any]) -> list[float]:
prices = scan.get("asset_prices") or scan.get("prices")
if not isinstance(prices, list):
return []
out: list[float] = []
for value in prices:
px = _coerce_float(value, None)
if px is None or px <= 0:
continue
out.append(px)
return out
@dataclass
class LiveBlueScanHistory:
"""Stateful scan history mirror for DC confirmation.
BLUE computes `dc_status` from the active asset price history. This helper keeps
a read-only replay of the published scan stream so VIOLET can reproduce that
state without touching BLUE or inventing a new schema.
"""
maxlen: int = 32
trade_direction: int = -1
_prices: dict[str, Deque[float]] = field(default_factory=dict)
last_scan_number: Optional[int] = None
def ingest_scan(self, scan_payload: Any) -> Mapping[str, Any]:
scan = _scan_view(scan_payload)
assets = _scan_assets(scan)
prices = _scan_prices(scan)
for asset, price in zip(assets, prices):
if price <= 0:
continue
hist = self._prices.setdefault(asset, deque(maxlen=self.maxlen))
hist.append(float(price))
sn = _coerce_float(scan.get("scan_number"), None)
if sn is not None:
self.last_scan_number = int(sn)
return scan
def price_history(self, asset: str) -> list[float]:
return list(self._prices.get(str(asset).upper(), ()))
def market_data(self, lookback: int) -> dict[str, list[float]]:
need = max(1, int(lookback) + 1)
return {asset: list(hist) for asset, hist in self._prices.items() if len(hist) >= need}
def dc_status(self, scan_payload: Any, *, already_ingested: bool = False) -> str:
scan = _scan_view(scan_payload) if already_ingested else self.ingest_scan(scan_payload)
asset = _scan_assets(scan)
if not asset:
return "NONE"
vel_div = _coerce_float(scan.get("vel_div"), None)
if vel_div is None:
return "NONE"
history = self.price_history(asset[0])
if not history:
return "NONE"
signal_gen = AlphaSignalGenerator()
sig = signal_gen.generate(
vel_div=vel_div,
vel_div_history=None,
asset_price_history=history,
trade_direction=self.trade_direction,
asset=asset[0],
current_timestamp=_coerce_float(scan.get("timestamp"), 0.0) or 0.0,
)
return sig.dc_status
class HazelcastOBProvider(OBProvider):
"""Read the current BLUE OB shards directly from Hazelcast."""
def __init__(self, client: hazelcast.HazelcastClient):
self.client = client
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
@dataclass(frozen=True)
class LiveBlueSourceResult:
factors: SizingFactors
acb_boost: float
acb_beta: float
mc_scale: float
posture: str
dc_status: str
selected_asset: str
def source_live_blue_sizing_factors(
client: hazelcast.HazelcastClient,
*,
assets: Optional[Iterable[str]] = None,
scan_history: Optional[LiveBlueScanHistory] = None,
selector: Optional[VioletAssetSelector] = None,
) -> LiveBlueSourceResult:
"""Read the BLUE-published live surfaces and return a typed factor plane."""
scan_history = scan_history or LiveBlueScanHistory()
selector = selector or VioletAssetSelector()
engine_snapshot_raw = _read_hz_map(client, "DOLPHIN_STATE_BLUE", "latest_nautilus")
if engine_snapshot_raw is None:
engine_snapshot_raw = _read_hz_map(client, "DOLPHIN_STATE_BLUE", "engine_snapshot")
engine_snapshot = _jsonish(engine_snapshot_raw)
if isinstance(engine_snapshot, str):
try:
engine_snapshot = json.loads(engine_snapshot)
except Exception:
engine_snapshot = {}
posture = posture_from_engine_snapshot(engine_snapshot if isinstance(engine_snapshot, Mapping) else None)
esof_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "esof_latest")
if esof_raw is None:
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 = _map_status_to_mc_scale(mc_raw)
scan_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "latest_eigen_scan")
scan = scan_history.ingest_scan(scan_raw)
scan_assets = _scan_assets(scan)
trade_direction = int(
_coerce_float(
engine_snapshot.get("trade_direction_runtime") if isinstance(engine_snapshot, Mapping) else None,
None,
)
or _coerce_float(
engine_snapshot.get("trade_direction_base") if isinstance(engine_snapshot, Mapping) else None,
None,
)
or -1
)
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
hz_snapshot = {
"boost": acb_boost,
"beta": acb_beta,
"mc_scale": mc_scale,
"esof_score": esof_score,
"ob_median_imbalance": ob_median_imbalance,
"ob_agreement_pct": ob_agreement_pct,
"dc_status": dc_status,
"posture": posture,
}
factors = extract_live_sizing_factors(hz_snapshot=hz_snapshot)
return LiveBlueSourceResult(
factors=factors,
acb_boost=acb_boost,
acb_beta=acb_beta,
mc_scale=mc_scale,
posture=posture,
dc_status=dc_status,
selected_asset=selected_asset,
)

View File

@@ -0,0 +1,123 @@
"""VIOLET V3.4b: source live ``SizingFactors`` from BLUE's published HZ planes.
The field-path validation (``prod/docs/VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md``)
established that of the eight sizing inputs, only ``posture`` and ``esof_score`` are
present in maps live BLUE publishes to Hazelcast:
- ``posture`` ← ``DOLPHIN_STATE_BLUE`` ``engine_snapshot['posture']``
- ``esof_score`` ← ``DOLPHIN_FEATURES['esof_latest'|'esof_advisor_latest']``,
parsed by BLUE's OWN ``parse_esof_payload`` /
``esof_score_from_payload`` (wrap, don't reimplement).
The remaining five (``boost``, ``beta``, ``mc_scale``, ``ob_median_imbalance``,
``ob_agreement_pct``, ``dc_status``) are BLUE-organ outputs — the ACB over
``DOLPHIN_FEATURES['exf_latest']``, the MC flag→scale derivation, ``OBFeatureEngine``,
and the per-asset signal generator — and are NOT present as scalars in any HZ map.
Sourcing them live is the V3.4c organ-wiring sprint.
Until then this adapter sources the two HZ-available factors faithfully and supplies
BLUE's OWN neutral sentinels for the organ-derived five (``boost=1.0``, ``beta=0.0``,
``mc_scale=1.0``, ``ob_*=None``, ``dc_status="NONE"``). The result flows through the
validated ``extract_live_sizing_factors`` normalizer, so the V3.4 shadow breakdown
records posture+esof LIVE and the rest NEUTRAL — explicit, never silently faked.
Pure boundary: callers pass already-fetched HZ blobs (the DARK service reads the maps
and hands the dicts in). No Hazelcast client here, no I/O, no launcher coupling —
mirrors ``live_factors.py``'s philosophy and keeps the adapter fully unit-testable.
"""
from __future__ import annotations
import sys
from collections.abc import Mapping
from pathlib import Path
from typing import Any, Optional
from .decision_engine import SizingFactors
from .domain import typed
from .live_factors import extract_live_sizing_factors
_PROJECT_ROOT = Path(__file__).resolve().parents[3]
# The organ-derived factors this adapter cannot source from HZ yet (V3.4c). Listed so
# the journal/diagnostics can mark them NEUTRAL rather than mistaking them for live.
ORGAN_DERIVED_FACTORS = (
"boost", "beta", "mc_scale",
"ob_median_imbalance", "ob_agreement_pct", "dc_status",
)
def _import_esof_gate() -> Any:
"""Import BLUE's ``esof_size_gate`` (same root-injection as ``sizing.py``)."""
try:
from nautilus_dolphin.nautilus import esof_size_gate # type: ignore
except ImportError:
for p in (str(_PROJECT_ROOT / "nautilus_dolphin"), str(_PROJECT_ROOT)):
if p not in sys.path:
sys.path.insert(0, p)
sys.modules.pop("nautilus_dolphin", None)
from nautilus_dolphin.nautilus import esof_size_gate # type: ignore
return esof_size_gate
def posture_from_engine_snapshot(snapshot: Optional[Mapping[str, Any]]) -> str:
"""BLUE's ``engine_snapshot['posture']`` (DOLPHIN_STATE_BLUE), defaulting to APEX.
Mirrors BLUE's own default (``getattr(self, '_day_posture', 'APEX')``,
esf_alpha_orchestrator.py:365/613). Upper-cased for the SizingFactors contract.
"""
if not isinstance(snapshot, Mapping):
return "APEX"
raw = snapshot.get("posture")
text = str(raw).strip() if raw is not None else ""
return text.upper() if text else "APEX"
def esof_score_from_features(
esof_raw: Any,
*,
max_age_s: Optional[float] = None,
) -> Optional[float]:
"""Extract the EsoF advisory score from a raw HZ ``esof_latest`` value.
Mirrors BLUE's ``_read_esof_payload`` two-step exactly: ``parse_esof_payload(raw)``
(the HZ value is a raw JSON blob) then ``esof_score_from_payload`` — both BLUE's
OWN functions (nautilus_event_trader.py:716,729), no reimplementation. ``max_age_s``
mirrors BLUE's freshness gate (``ESOF_FRESHNESS_S``); ``None`` skips the staleness
check. Returns ``None`` when the value is missing/unparseable/stale — SizingFactors
then leaves ``esof_score`` unset (BLUE's ``esof_size_mult_from_score(None)`` neutral
path).
"""
if esof_raw is None:
return None
gate = _import_esof_gate()
payload = gate.parse_esof_payload(esof_raw)
if not payload:
return None
score = gate.esof_score_from_payload(payload, max_age_s=max_age_s)
return None if score is None else float(score)
@typed
def source_live_sizing_factors(
*,
engine_snapshot: Optional[Mapping[str, Any]] = None,
esof_payload: Any = None,
esof_max_age_s: Optional[float] = None,
) -> SizingFactors:
"""Build live ``SizingFactors`` from BLUE's published HZ blobs.
``posture`` and ``esof_score`` are sourced LIVE from ``engine_snapshot`` and the
``esof_latest`` payload; the six organ-derived factors fall to BLUE's neutral
sentinels via the ``extract_live_sizing_factors`` defaults. The DARK service is
expected to fetch ``DOLPHIN_STATE_BLUE['engine_snapshot']`` and
``DOLPHIN_FEATURES['esof_latest']`` and pass them here.
"""
posture = posture_from_engine_snapshot(engine_snapshot)
esof_score = esof_score_from_features(esof_payload, max_age_s=esof_max_age_s)
hz_snapshot: dict[str, Any] = {"posture": posture}
if esof_score is not None:
hz_snapshot["esof_score"] = esof_score
return extract_live_sizing_factors(hz_snapshot=hz_snapshot)

View File

@@ -40,6 +40,14 @@ class DecisionRow(StrictModel):
ars_score: float = Field(allow_inf_nan=False) ars_score: float = Field(allow_inf_nan=False)
bucket_idx: int = Field(ge=0, le=255) # UInt8 bucket_idx: int = Field(ge=0, le=255) # UInt8
actuated: int = Field(ge=0, le=1) actuated: int = Field(ge=0, le=1)
# V3.4 full-sizing breakdown (Nullable(Float64) in DDL): NULL on the base-only
# path, populated when live SizingFactors drive the decision. ge=0.0 — leverage
# and all four multipliers are non-negative by BLUE's construction.
base_leverage: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
dc_lev_mult: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
regime_size_mult: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
market_ob_mult: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
esof_size_mult: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
class VioletDecisionJournal: class VioletDecisionJournal:
@@ -70,6 +78,13 @@ class VioletDecisionJournal:
ars_score=float(decision.ars_score), ars_score=float(decision.ars_score),
bucket_idx=int(decision.bucket_idx), bucket_idx=int(decision.bucket_idx),
actuated=1 if decision.actuated else 0, actuated=1 if decision.actuated else 0,
# getattr-with-None: duck-typed decisions (and the base-only path)
# may omit the breakdown; it stays NULL rather than raising here.
base_leverage=getattr(decision, "base_leverage", None),
dc_lev_mult=getattr(decision, "dc_lev_mult", None),
regime_size_mult=getattr(decision, "regime_size_mult", None),
market_ob_mult=getattr(decision, "market_ob_mult", None),
esof_size_mult=getattr(decision, "esof_size_mult", None),
) )
except ValidationError as exc: except ValidationError as exc:
self.rows_rejected += 1 self.rows_rejected += 1

View File

@@ -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)

View File

@@ -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

View File

@@ -0,0 +1,381 @@
from __future__ import annotations
import json
import sys
from dataclasses import dataclass
import pytest
sys.path.insert(0, "/mnt/dolphinng5_predict")
import hazelcast
from prod.clean_arch.violet.decision_engine import SizingFactors
from prod.clean_arch.violet.live_blue_source import (
HazelcastOBProvider,
LiveBlueScanHistory,
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())
class _FakeBlocking:
def __init__(self, payloads):
self._payloads = payloads
def get(self, key):
return self._payloads.get(key)
def key_set(self):
return list(self._payloads.keys())
class _FakeClient:
def __init__(self, maps):
self._maps = maps
def get_map(self, name):
return type("M", (), {"blocking": lambda self2: _FakeBlocking(self._maps[name])})()
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({"status": "ORANGE"}),
"esof_latest": json.dumps({"advisory_score": 0.4}),
},
"DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "restored"})},
}
)
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]
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})()
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, 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({"status": "ORANGE"}),
"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"})},
}
)
res = source_live_blue_sizing_factors(
client,
assets=["BTCUSDT"],
scan_history=history,
selector=VioletAssetSelector(lookback_horizon=7),
)
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.esof_score == 0.4
assert res.factors.ob_median_imbalance == 0.12
assert res.factors.ob_agreement_pct == 0.91
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({"status": "GREEN"}),
"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_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})()
monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.OBFeatureEngine", FakeEngine)
selector = VioletAssetSelector(lookback_horizon=7)
history = LiveBlueScanHistory(maxlen=16, trade_direction=-1)
signal_gen = AlphaSignalGenerator()
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],
},
]
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({"status": "GREEN"}),
"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"})},
}
)
res = source_live_blue_sizing_factors(
client,
assets=["BTCUSDT", "ETHUSDT"],
scan_history=history,
selector=selector,
)
# 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,
asset_price_history=history.price_history(expected_asset),
trade_direction=-1,
asset=expected_asset,
current_timestamp=float(scan["timestamp"]),
)
assert res.factors.dc_status == expected_signal.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)
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({"status": "GREEN"}),
"esof_latest": json.dumps({"advisory_score": 0.3}),
"latest_eigen_scan": json.dumps(scan),
},
"DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "APEX"})},
}
)
res = source_live_blue_sizing_factors(
client,
assets=["BTCUSDT", "ETHUSDT"],
scan_history=history,
selector=selector,
)
assert res.selected_asset == "BTCUSDT"
assert res.factors.dc_status == "NONE"
assert history.price_history("BTCUSDT") == []
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({"status": "GREEN"}),
"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"
def test_live_hz_smoke_reads_current_state():
client = hazelcast.HazelcastClient(cluster_name="dolphin", cluster_members=["localhost:5701"])
try:
res = source_live_blue_sizing_factors(client, assets=["BTCUSDT", "ETHUSDT", "XRPUSDT"])
finally:
client.shutdown()
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}

View File

@@ -0,0 +1,85 @@
"""V3.4b: live_factor_source — SizingFactors from BLUE's published HZ blobs.
Validates the field-path findings in
prod/docs/VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md: posture + esof_score are
sourced LIVE from engine_snapshot / the esof_latest payload, the six organ-derived
factors fall to BLUE's neutral sentinels.
"""
from __future__ import annotations
import sys
import pytest
sys.path.insert(0, "/mnt/dolphinng5_predict")
from prod.clean_arch.violet.decision_engine import SizingFactors
from prod.clean_arch.violet.live_factor_source import (
ORGAN_DERIVED_FACTORS,
esof_score_from_features,
posture_from_engine_snapshot,
source_live_sizing_factors,
)
def test_posture_sourced_and_upper_cased():
assert posture_from_engine_snapshot({"posture": "STALKER"}) == "STALKER"
assert posture_from_engine_snapshot({"posture": "restored"}) == "RESTORED"
def test_posture_defaults_to_apex_like_blue():
assert posture_from_engine_snapshot(None) == "APEX"
assert posture_from_engine_snapshot({}) == "APEX"
assert posture_from_engine_snapshot({"posture": ""}) == "APEX"
assert posture_from_engine_snapshot({"posture": None}) == "APEX"
def test_esof_score_parsed_from_dict_payload():
# max_age_s=None skips staleness; advisory_score preferred over score.
assert esof_score_from_features({"advisory_score": 0.5}) == 0.5
assert esof_score_from_features({"score": -0.1}) == pytest.approx(-0.1)
def test_esof_score_parsed_from_raw_json_string():
# The HZ value is a raw JSON blob — BLUE's parse_esof_payload handles it.
assert esof_score_from_features('{"advisory_score": 0.25}') == 0.25
def test_esof_score_none_when_missing_or_unparseable():
assert esof_score_from_features(None) is None
assert esof_score_from_features("not json") is None
assert esof_score_from_features({}) is None # no advisory_score/score key
def test_esof_staleness_gate_honored_when_max_age_supplied():
# A payload with an ancient timestamp is stale → None when max_age_s is set.
stale = {"advisory_score": 0.5, "unix": 0.0} # 1970 → very old
assert esof_score_from_features(stale, max_age_s=30.0) is None
# …but with no freshness gate (None) the score still comes through.
assert esof_score_from_features(stale, max_age_s=None) == 0.5
def test_source_live_factors_posture_and_esof_live_rest_neutral():
factors = source_live_sizing_factors(
engine_snapshot={"posture": "RESTORED", "capital": 71591.1},
esof_payload={"advisory_score": 0.42},
)
assert isinstance(factors, SizingFactors)
assert factors.posture == "RESTORED"
assert factors.esof_score == 0.42
# the six organ-derived factors at BLUE's own neutral sentinels (V3.4c will source)
assert factors.boost == 1.0 and factors.beta == 0.0 and factors.mc_scale == 1.0
assert factors.ob_median_imbalance is None and factors.ob_agreement_pct is None
assert factors.dc_status == "NONE"
def test_source_live_factors_all_neutral_when_no_blobs():
assert source_live_sizing_factors() == SizingFactors()
def test_organ_derived_factor_set_is_the_documented_six():
assert set(ORGAN_DERIVED_FACTORS) == {
"boost", "beta", "mc_scale",
"ob_median_imbalance", "ob_agreement_pct", "dc_status",
}

View File

@@ -71,3 +71,47 @@ def test_negative_exposure_rejected():
) )
assert j.journal(bad, mono_ns=1) is False assert j.journal(bad, mono_ns=1) is False
assert j.rows_rejected == 1 assert j.rows_rejected == 1
_BREAKDOWN_COLS = (
"base_leverage", "dc_lev_mult", "regime_size_mult",
"market_ob_mult", "esof_size_mult",
)
def test_full_factor_breakdown_round_trips_into_row():
captured = []
j = VioletDecisionJournal(sink=lambda t, r: captured.append(r), session_id="s")
dec = _decision(
base_leverage=4.0, dc_lev_mult=1.5, regime_size_mult=1.2,
market_ob_mult=1.4, esof_size_mult=0.95,
)
assert j.journal(dec, mono_ns=1) and j.rows_emitted == 1
row = captured[0]
assert set(row.keys()) == _ddl_columns() # parity holds with new columns
assert row["base_leverage"] == 4.0 and row["dc_lev_mult"] == 1.5
assert row["regime_size_mult"] == 1.2 and row["market_ob_mult"] == 1.4
assert row["esof_size_mult"] == 0.95
def test_base_only_path_leaves_breakdown_null():
captured = []
j = VioletDecisionJournal(sink=lambda t, r: captured.append(r), session_id="s")
assert j.journal(_decision(), mono_ns=1) # no breakdown supplied
row = captured[0]
assert all(row[c] is None for c in _BREAKDOWN_COLS)
def test_negative_breakdown_multiplier_rejected_at_source():
j = VioletDecisionJournal(sink=lambda t, r: None, session_id="s")
# ShadowDecision itself guards ge=0.0, so build a duck-typed decision that
# smuggles a negative multiplier past the engine to prove the row guard catches it.
bad = SimpleNamespace(
scan_number=1, asset="BTCUSDT", side="SHORT", vel_div=-0.2,
fraction=0.2, conviction_leverage=9.0, notional_fraction=1.8,
target_exposure=1.0, ars_score=1.0, bucket_idx=1, actuated=True,
base_leverage=4.0, dc_lev_mult=-0.5, regime_size_mult=1.0,
market_ob_mult=1.0, esof_size_mult=1.0,
)
assert j.journal(bad, mono_ns=1) is False
assert j.rows_rejected == 1

View File

@@ -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 == []

View File

@@ -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,
)

View File

@@ -19,7 +19,16 @@ CREATE TABLE IF NOT EXISTS dolphin_violet.violet_decisions
`target_exposure` Float64, `target_exposure` Float64,
`ars_score` Float64, `ars_score` Float64,
`bucket_idx` UInt8, `bucket_idx` UInt8,
`actuated` UInt8 `actuated` UInt8,
-- V3.4 full-sizing breakdown: conviction = base × dc × regime(ACB) × ob × esof,
-- capped @9. NULL on the legacy base-only path (no live SizingFactors); populated
-- when the launcher feeds live factor planes. Additive columns — on a pre-existing
-- table apply the matching `ALTER TABLE ... ADD COLUMN IF NOT EXISTS` instead.
`base_leverage` Nullable(Float64),
`dc_lev_mult` Nullable(Float64),
`regime_size_mult` Nullable(Float64),
`market_ob_mult` Nullable(Float64),
`esof_size_mult` Nullable(Float64)
) )
ENGINE = MergeTree ENGINE = MergeTree
ORDER BY (asset, ts) ORDER BY (asset, ts)

View File

@@ -0,0 +1,165 @@
# RECENT_VIOLET_34C_a632c59
## What This Work Was
This change completed the first V3.4c VIOLET-side mirror for BLUE live-factor
inputs.
The goal was not to modify BLUE. The goal was to make VIOLET read the same live,
published BLUE surfaces and reconstruct the same intermediate factors BLUE would
see, without changing BLUE code, schemas, or state layout.
## Scope
This work stayed inside VIOLET and added a read-only live-source adapter plus
tests:
- `prod/clean_arch/violet/live_blue_source.py`
- `prod/clean_arch/violet/test_violet_live_blue_source.py`
It also reused the existing V3.4b live-factor helpers:
- `prod/clean_arch/violet/live_factor_source.py`
- `prod/clean_arch/violet/live_factors.py`
- `prod/clean_arch/violet/alpha_wrappers.py`
## Why This Was Needed
The earlier V3.4b adapter could source only the BLUE-published pieces that were
already directly available in Hazelcast:
- `posture`
- `esof_score`
The remaining sizing inputs were not flat HZ scalars. They had to be mirrored
from the same BLUE inputs and kernels that generate them:
- `boost` / `beta` from the ACB output
- `mc_scale` from MC-Forewarner status
- `ob_median_imbalance` / `ob_agreement_pct` from live OB data
- `dc_status` from the signal generator
The V3.4c step is the read-only, VIOLET-side reconstruction of those live
factors.
## What Was Added
### 1. Live BLUE source adapter
`live_blue_source.py` now:
- reads `DOLPHIN_STATE_BLUE.latest_nautilus` / `engine_snapshot`
- reads `DOLPHIN_FEATURES.esof_latest` / `esof_advisor_latest`
- reads `DOLPHIN_FEATURES.acb_boost`
- reads `DOLPHIN_FEATURES.mc_forewarner_latest`
- reads live OB shard maps from `DOLPHIN_FEATURES.asset_*_ob`
- reconstructs `dc_status` from the published scan stream
The module stays read-only. It does not write Hazelcast. It does not call BLUE
internals for mutation. It only mirrors what BLUE already published.
### 2. Stateful scan replay for DC
`LiveBlueScanHistory` was added to keep a per-asset price history on the VIOLET
side. This is needed because BLUEs `dc_status` is derived from the live scan
sequence and a short price history, not from a single HZ scalar.
The adapter now:
- ingests each `latest_eigen_scan`
- keeps the asset histories in memory
- uses BLUEs own `AlphaSignalGenerator`
- produces the same DC status labels that BLUE would emit
### 3. Read-only OB mirror
`HazelcastOBProvider` was added so VIOLET can feed BLUEs own
`OBFeatureEngine` from the live `asset_*_ob` entries already published in HZ.
That lets the VIOLET path derive:
- `ob_median_imbalance`
- `ob_agreement_pct`
without inventing a new OB schema or mutating BLUE.
### 4. Asset selection parity
The adapter now uses `VioletAssetSelector` on the replayed scan history so the
selected asset is not guessed from the current scan payload alone.
That matters because the exact factor sequence must track the same information
BLUE would have at that point in the scan stream.
## Exactness Rules Followed
The implementation was kept conservative:
- no BLUE file edits
- no BLUE schema changes
- no new HZ writers
- no invented factor names
- no silent fallback to fake live values when the live source exists
If input is malformed, the adapter rejects or neutralizes it instead of
poisoning the history.
Examples of handled anomalies:
- missing `assets`
- non-finite prices
- negative prices
- malformed JSON payloads
- missing `posture`
- missing `esof` payloads
- broken OB payloads
## Tests Added
The new test file covers three layers:
### Unit tests
- OB shard parsing from HZ payloads
- neutral handling for malformed ACB / ESOF / MC payloads
- scan replay ingestion
- DC preservation for `CONFIRM`
- DC preservation for `SKIP_CONTRADICT`
### Sequence parity tests
The new sequence test walks multiple scan events and checks that VIOLET tracks:
- the BLUE asset selector output
- the BLUE signal-generator `dc_status`
at each step in the replayed scan history.
### End-to-end smoke
A live Hazelcast smoke test reads the current cluster state and verifies the
adapter can build a typed `SizingFactors` object from the live BLUE surfaces.
## Result
The V3.4c mirror now reconstructs the full live factor plane on the VIOLET
side, read-only, with parity-style coverage around the intermediate factor
computation.
## Verification
Commit:
- `a632c59``VIOLET V3.4c: read-only BLUE live source parity`
Tests run:
- `prod/clean_arch/violet/test_violet_live_blue_source.py`
- `prod/clean_arch/violet/test_violet_live_factor_source.py`
- `prod/clean_arch/violet/test_violet_live_factors.py`
Observed result:
- V3.4c source tests passed
- existing V3.4b live-factor tests passed

View File

@@ -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 BLUEs 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.

View File

@@ -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.

View File

@@ -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)

View File

@@ -0,0 +1,61 @@
# VIOLET V3.4b — live-factor field-path validation
**Date:** 2026-06-16
**Task:** validate `prod/clean_arch/violet/live_factors.py`'s candidate field paths
against how live BLUE actually sources the five sizing multipliers, *before* wiring
the launcher-sourcing half of V3.4b.
## TL;DR
`live_factors.py` assumes the eight sizing inputs arrive as flat/nested keys in a
single `hz_snapshot` dict. **That premise holds for only one of them (`posture`).**
`esof_score` is present in HZ but as a *payload to parse*, not a flat key. The other
five (`boost`, `beta`, `mc_scale`, `ob_median_imbalance`, `ob_agreement_pct`,
`dc_status`) are **BLUE-organ outputs that are not published to any HZ map** — they
live in the live `NDAlphaEngine`'s process memory / are recomputed per scan.
The speculative alternate paths in `live_factors.py` (`acb_boost`, `s_acb_boost`,
`("acb","boost")`, `day_mc_scale`, `("esof","advisory_score")`, `("ob","market",…)`,
`("signal","dc_status")`, `safety_posture`, …) **correspond to nothing in live BLUE.**
They are harmless (first-match-wins, flat canonical key is tried first) but dead.
## Per-factor validated sourcing
Source of truth: `esf_alpha_orchestrator.py` (the composition), `adaptive_circuit_breaker.py`
(ACB), `nautilus_event_trader.py` (the HZ reads/publishes).
| Factor | How live BLUE gets it | In a VIOLET-readable HZ map? |
|---|---|---|
| `posture` | `_day_posture`, set in `begin_day(posture=…)`; published in `engine_snapshot['posture']` (`nautilus_event_trader.py:5097`, map `DOLPHIN_STATE_BLUE`) and `DOLPHIN_SAFETY.latest.posture` | ✅ **flat key `posture`** in engine_snapshot |
| `esof_score` | `_read_esof_payload()``DOLPHIN_FEATURES['esof_latest'\|'esof_advisor_latest']``parse_esof_payload``esof_score_from_payload(..., max_age_s=ESOF_FRESHNESS_S)` (`nautilus_event_trader.py:707-719`) | ✅ but a **payload parse**, not a flat `esof_score` |
| `boost` / `beta` | `acb.get_dynamic_boost_from_hz(date)``acb_info['boost'\|'beta']`; the ACB **computes** them from `DOLPHIN_FEATURES['exf_latest']` (funding_btc/dvol_btc/fng/taker — `adaptive_circuit_breaker.py:511,528-533`). Applied via `begin_day` / `update_acb_boost` (`esf_alpha_orchestrator.py:764,946`). | ❌ raw inputs are in HZ; the **scalar requires running the ACB** |
| `mc_scale` | `_day_mc_scale`, **derived** in `begin_day` from MC-Forewarner `mc_orange`/`mc_red` flags (`esf_alpha_orchestrator.py:962-964`: orange→0.5, red/TURTLE/HIBERNATE→…) | ❌ not a HZ scalar |
| `ob_median_imbalance` / `ob_agreement_pct` | `ob_engine.get_market(bar_idx, symbols)` over the live OB feed, **per asset** (`esf_alpha_orchestrator.py:590-595`) | ❌ computed live; not in HZ |
| `dc_status` | per-asset `signal.dc_status` from the signal generator (`esf_alpha_orchestrator.py:576`) | ❌ computed per-scan; not in HZ |
`engine_snapshot` payload (the map BLUE publishes for consumers) was inspected in full
(`nautilus_event_trader.py:5092-5126`): it carries `posture`, `last_vel_div`, `vol_ok`,
`last_scan_number`, `capital`, leverage caps, position list — **and none of the five
organ-derived multipliers.**
## Consequence for V3.4b sourcing (#2)
A faithful, *complete* live-factor source is NOT a HZ scrape — it requires VIOLET to run
the same organs BLUE does:
- an **ACB** over `DOLPHIN_FEATURES['exf_latest']` → boost/beta,
- the **MC** flag→`mc_scale` derivation,
- an **OBFeatureEngine** over the OB feed → ob_*,
- a **signal generator** → dc_status.
That is a multi-organ sprint (call it **V3.4c**), not a quick wiring.
What IS sourceable now, from maps BLUE already publishes, read-only:
- **`posture`** ← `engine_snapshot['posture']`
- **`esof_score`** ← `DOLPHIN_FEATURES['esof_latest']` via BLUE's own `esof_score_from_payload`
So the honest V3.4b increment (this PR) is a **pure adapter**
`live_factor_source.py` — that sources those two faithfully and supplies BLUE's own
neutral sentinels for the organ-derived five (`boost=1.0`, `beta=0.0`, `mc_scale=1.0`,
`ob_*=None`, `dc_status="NONE"`), feeding `extract_live_sizing_factors`. The shadow
journal's V3.4 breakdown then records posture+esof live and the rest neutral — explicit,
not silently faked. The organ wiring (boost/beta/mc/ob/dc live) is deferred to V3.4c.

View File

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