Compare commits
4 Commits
6d08e97e28
...
16add44326
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
16add44326 | ||
|
|
1ac3f627df | ||
|
|
a632c595ba | ||
|
|
722fd9f054 |
361
prod/clean_arch/violet/live_blue_source.py
Normal file
361
prod/clean_arch/violet/live_blue_source.py
Normal 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,
|
||||||
|
)
|
||||||
123
prod/clean_arch/violet/live_factor_source.py
Normal file
123
prod/clean_arch/violet/live_factor_source.py
Normal 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)
|
||||||
@@ -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
|
||||||
|
|||||||
381
prod/clean_arch/violet/test_violet_live_blue_source.py
Normal file
381
prod/clean_arch/violet/test_violet_live_blue_source.py
Normal 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}
|
||||||
85
prod/clean_arch/violet/test_violet_live_factor_source.py
Normal file
85
prod/clean_arch/violet/test_violet_live_factor_source.py
Normal 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",
|
||||||
|
}
|
||||||
@@ -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
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
165
prod/docs/RECENT_VIOLET_34C_a632c59.md
Normal file
165
prod/docs/RECENT_VIOLET_34C_a632c59.md
Normal 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 BLUE’s `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 BLUE’s own `AlphaSignalGenerator`
|
||||||
|
- produces the same DC status labels that BLUE would emit
|
||||||
|
|
||||||
|
### 3. Read-only OB mirror
|
||||||
|
|
||||||
|
`HazelcastOBProvider` was added so VIOLET can feed BLUE’s 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
|
||||||
|
|
||||||
61
prod/docs/VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md
Normal file
61
prod/docs/VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md
Normal 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.
|
||||||
Reference in New Issue
Block a user