262cada664
Per operator: VIOLET must keep a "prior" on stale ExF rather than collapse to neutral, and the ACB's role must be commented-in + logged. - KEEP-PRIOR on stale/missing exf: BLUE keeps _day_base_boost/_day_beta on the stale branch (update_acb_boost isn't called). _source_boost_beta now takes `prior` (boost,beta) and returns it on ValueError / no-exf, seeded by the first successful compute. Threaded as `prior_boost_beta` through source_live_blue_sizing_factors (caller persists last good). - LOGGING: violet.live_blue_source logger emits live boost/beta + the full sourced plane. - SEMANTICS: documents the ACB in effect (inverse v6) and CONFIRMS ACB IS in the sizing layer via _day_base_boost/_day_beta → _update_regime_size_mult (esf_alpha_orchestrator.py:771-772, 898-909). OB Sub-4 beta modulation dormant in live — per operator found NON-PERFORMANT and deliberately bypassed; TODO_SOMEDAY in-code to find documentary confirmation. 37 tests. violet-only; no shared-file edits. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
478 lines
21 KiB
Python
478 lines
21 KiB
Python
"""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`` via
|
|
BLUE's own ``parse_esof_payload`` / ``esof_score_from_payload``
|
|
- ``boost`` / ``beta`` RECOMPUTED via ``AdaptiveCircuitBreaker.get_dynamic_boost_from_hz``
|
|
over ``DOLPHIN_FEATURES.exf_latest`` + ``latest_eigen_scan.w750_velocity`` — IDENTICAL
|
|
to the trader's on_exf_update path (NOT the published ``acb_boost`` scalar)
|
|
- ``mc_scale`` from ``DOLPHIN_FEATURES.mc_forewarner_latest`` via ``_derive_mc_scale``
|
|
(begin_day's cat/env thresholds, not the MC service's status label)
|
|
- OB market consensus from the live ``asset_*_ob`` maps via BLUE's own ``HZOBProvider`` +
|
|
``OBFeatureEngine``
|
|
- ``dc_status`` via BLUE's ``AlphaSignalGenerator`` (params pinned to BLUE's ENGINE_KWARGS)
|
|
over the replayed scan price-history
|
|
|
|
PARITY (see prod/docs/VIOLET_BLUE_PARITY_STRUCTURAL_DIVERGENCE.md): this module reconstructs
|
|
BLUE's factors in a DIFFERENT file/scope structure than BLUE's monolithic NDAlphaEngine.
|
|
Every kernel is now WRAPPED, not copied (ACB, OBFeatureEngine+HZOBProvider, AlphaSignalGenerator,
|
|
VioletAssetSelector) — so the only hand-replicated arithmetic left is ``_derive_mc_scale``
|
|
(begin_day's thresholds; pinned by test). Remaining REAL fidelity caveats:
|
|
- OB faithfulness needs a PERSISTENT ``ob_engine`` + per-scan ``bar_idx`` (OBFeatureEngine
|
|
accumulates a lookback window); a single-shot engine has no cross-scan history.
|
|
- On stale exf (>12h) the ACB raises ValueError; BLUE keeps the prior boost/beta, VIOLET
|
|
has no prior → neutral (1.0, 0.0).
|
|
Any change to BLUE's ENGINE_KWARGS or begin_day mc thresholds REQUIRES a matching change +
|
|
test here (see test_signal_gen_params_match_blue_engine_kwargs / the mc_scale formula test).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os as _os
|
|
import sys
|
|
from collections import deque
|
|
from datetime import datetime, timezone
|
|
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.hz_ob_provider import HZOBProvider
|
|
from nautilus_dolphin.nautilus.adaptive_circuit_breaker import AdaptiveCircuitBreaker
|
|
from nautilus_dolphin.nautilus.alpha_signal_generator import (
|
|
AlphaSignalGenerator,
|
|
VEL_DIV_THRESHOLD, VEL_DIV_EXTREME,
|
|
LONG_VEL_DIV_THRESHOLD, LONG_VEL_DIV_EXTREME,
|
|
)
|
|
|
|
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
|
|
|
|
LOGGER = logging.getLogger("violet.live_blue_source")
|
|
|
|
# Hazelcast coordinates — MUST equal nautilus_event_trader.py:107-108 (BLUE's live
|
|
# cluster). HZOBProvider opens its own connection to these, exactly as BLUE's _wire_obf.
|
|
HZ_CLUSTER = _os.environ.get("HZ_CLUSTER", "dolphin")
|
|
HZ_HOST = _os.environ.get("HZ_HOST", "127.0.0.1:5701")
|
|
|
|
# AlphaSignalGenerator construction — pinned to BLUE's live ENGINE_KWARGS
|
|
# (nautilus_event_trader.py:128-133). These equal AlphaSignalGenerator's own defaults
|
|
# TODAY, but BLUE constructs it EXPLICITLY from ENGINE_KWARGS, so a future champion
|
|
# retune (e.g. dc_lookback_bars→10) would silently diverge a bare AlphaSignalGenerator().
|
|
# We pin explicitly and assert the pin in test_signal_gen_params_match_blue_engine_kwargs.
|
|
# vel_div_* import the module constants directly so they auto-track the kernel.
|
|
BLUE_SIGNAL_GEN_KWARGS = dict(
|
|
vel_div_threshold=VEL_DIV_THRESHOLD, # -0.02
|
|
vel_div_extreme=VEL_DIV_EXTREME, # -0.05
|
|
long_vel_div_threshold=LONG_VEL_DIV_THRESHOLD, # 0.01
|
|
long_vel_div_extreme=LONG_VEL_DIV_EXTREME, # 0.04
|
|
dc_lookback_bars=7,
|
|
dc_min_magnitude_bps=0.75,
|
|
dc_skip_contradicts=True,
|
|
dc_leverage_boost=1.0,
|
|
dc_leverage_reduce=0.5,
|
|
use_direction_confirm=True,
|
|
)
|
|
|
|
|
|
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 _derive_mc_scale(payload: Any) -> float:
|
|
"""Mirror BLUE ``begin_day``'s ``mc_scale`` EXACTLY (esf_alpha_orchestrator.py:956-962).
|
|
|
|
BLUE does NOT use the MC service's published ``status`` label for sizing — that label
|
|
(mc_forewarner_flow.py: GREEN<0.10 / ORANGE<0.30 / RED) is observability-only and uses
|
|
DIFFERENT thresholds from the engine. The live trader re-derives the size haircut from
|
|
the SAME published fields ``catastrophic_prob`` + ``envelope_score`` with the engine's
|
|
own thresholds:
|
|
|
|
mc_red = catastrophic_prob > 0.25 or envelope_score < -1.0
|
|
mc_orange = (not mc_red) and (envelope_score < 0 or catastrophic_prob > 0.10)
|
|
mc_scale = 0.5 if mc_orange else 1.0 # RED → 1.0 here; BLUE halts via regime_dd_halt
|
|
|
|
Reading ``status`` instead diverges (e.g. cat=0.05/env=-0.5 → publisher GREEN→1.0 but
|
|
BLUE orange→0.5).
|
|
|
|
AMBIGUITY (flagged 2026-06-16): the MC service's ``status`` and the engine's begin_day
|
|
thresholds genuinely disagree, and from OUTSIDE BLUE there is no way to know which is the
|
|
"nominal" intent — they are two independent threshold sets over the same numbers. We go
|
|
with the SOURCE FIELDS + begin_day formula because that is the path that actually drives
|
|
BLUE's live sizing (the published ``status`` is consumed only by the TUI/observability).
|
|
If BLUE's begin_day thresholds change, THIS must change with them. Operator confirmation
|
|
of the canonical surface is still desirable.
|
|
|
|
Missing/unparseable fields → neutral 1.0 (no haircut)."""
|
|
data = _jsonish(payload)
|
|
if not isinstance(data, Mapping):
|
|
return 1.0
|
|
cat = _coerce_float(data.get("catastrophic_prob"), None)
|
|
env = _coerce_float(data.get("envelope_score"), None)
|
|
if cat is None or env is None:
|
|
return 1.0
|
|
mc_red = cat > 0.25 or env < -1.0
|
|
mc_orange = (not mc_red) and (env < 0.0 or cat > 0.10)
|
|
return 0.5 if mc_orange else 1.0
|
|
|
|
|
|
# Inverse-ACB neutral identity — used ONLY at cold start (no prior yet AND no fresh exf).
|
|
# In continuous operation exf is warmed, so the first call seeds the prior and this is
|
|
# never the steady-state value.
|
|
_BOOST_BETA_NEUTRAL = (1.0, 0.0)
|
|
|
|
|
|
def _source_boost_beta(
|
|
client: hazelcast.HazelcastClient,
|
|
*,
|
|
date_str: str,
|
|
trade_direction: int,
|
|
acb: Optional[AdaptiveCircuitBreaker] = None,
|
|
prior: Optional[tuple[float, float]] = None,
|
|
) -> tuple[float, float]:
|
|
"""Recompute (boost, beta) EXACTLY as BLUE's live trader does — NOT from acb_boost.
|
|
|
|
WHAT THE ACB IS DOING, IN EFFECT (AdaptiveCircuitBreaker v6, "inverse" mode):
|
|
- ``boost`` = a DAILY position-size governor >= 1.0 driven by EXTERNAL FACTORS (ExF:
|
|
funding_btc/dvol_btc/fng/taker from DOLPHIN_FEATURES.exf_latest). When stress
|
|
signals fire, ``boost = 1 + 0.5*ln(1+signals)``; else 1.0. INVERSE = it leans size
|
|
UP under stress, it does not cut (adaptive_circuit_breaker.py:591-593).
|
|
- ``beta`` = regime sensitivity in {BETA_HIGH=0.8, BETA_LOW=0.2}, keyed on the w750
|
|
eigenvalue-velocity (>= threshold → HIGH). It sets how strongly per-bar signal
|
|
strength amplifies size: regime_size_mult = base_boost*(1 + beta*strength^3)*mc_scale.
|
|
- Both become the engine's _day_base_boost / _day_beta via update_acb_boost
|
|
(esf_alpha_orchestrator.py:771-772), which then feed BLUE's SIZING directly:
|
|
_update_regime_size_mult (:898-909) = _day_base_boost * (1 + _day_beta*strength^3)
|
|
* _day_mc_scale. So the ACB IS in the sizing layer — confirmed, not incidental.
|
|
|
|
BLUE's LIVE path passes NO ob_engine (on_exf_update:4769 / rollover prewarm:2710), so the
|
|
ACB's OB Sub-4 macro-regime beta modulation (x1.25 stress / x0.85 calm,
|
|
adaptive_circuit_breaker.py:613-628) is DORMANT in live — only the NPZ backtest path
|
|
get_dynamic_boost_for_date applies it. Per operator recollection (2026-06-16), OB Sub-4
|
|
beta modulation was found NON-PERFORMANT and deliberately removed/bypassed — so passing
|
|
no ob_engine is INTENTIONAL design, not an oversight. We pass no ob_engine to match BLUE
|
|
bit-for-bit either way.
|
|
# TODO_SOMEDAY: find the documentary confirmation of the OB Sub-4 non-performant bypass
|
|
# (operator: likely SYSTEM_BIBLE v7 lineage or a deep code comment / the in-progress
|
|
# INDEX_DOLPHINNG5_PREDICT doc). Behaviour is already bit-identical to live BLUE.
|
|
|
|
The published DOLPHIN_FEATURES.acb_boost is acb_processor_service's SEPARATE daily value
|
|
and is never read here.
|
|
|
|
STALE / MISSING exf: BLUE keeps the prior _day_base_boost/_day_beta (update_acb_boost
|
|
simply isn't called on the stale branch), so VIOLET KEEPS ``prior`` too. ``prior`` is
|
|
seeded by the first successful compute — exf is warmed in continuous BLUE operation, so
|
|
the steady state always has a prior; only a cold start with no prior AND no fresh exf
|
|
falls back to the neutral identity."""
|
|
fallback = prior if prior is not None else _BOOST_BETA_NEUTRAL
|
|
exf = _jsonish(_read_hz_map(client, "DOLPHIN_FEATURES", "exf_latest"))
|
|
if not isinstance(exf, Mapping):
|
|
LOGGER.debug("ACB: no exf_latest — keeping prior boost/beta=%s", fallback)
|
|
return fallback
|
|
eigen = _jsonish(_read_hz_map(client, "DOLPHIN_FEATURES", "latest_eigen_scan"))
|
|
w750 = _coerce_float(eigen.get("w750_velocity"), None) if isinstance(eigen, Mapping) else None
|
|
acb = acb or AdaptiveCircuitBreaker()
|
|
try:
|
|
info = acb.get_dynamic_boost_from_hz(
|
|
date_str=date_str,
|
|
exf_snapshot=dict(exf),
|
|
w750_velocity=float(w750) if w750 else None, # 0.0 → None, matches BLUE
|
|
direction=trade_direction, # NO ob_engine — matches BLUE live
|
|
)
|
|
except ValueError as exc:
|
|
# BLUE logs "ACB Stale Data Fallback" and keeps the prior boost/beta.
|
|
LOGGER.info("ACB Stale Data Fallback (%s) — keeping prior boost/beta=%s", exc, fallback)
|
|
return fallback
|
|
boost = max(0.0, _coerce_float(info.get("boost"), 1.0) or 1.0)
|
|
beta = max(0.0, _coerce_float(info.get("beta"), 0.0) or 0.0)
|
|
LOGGER.debug("ACB live boost=%.6f beta=%.6f (signals=%s w750=%s dir=%s)",
|
|
boost, beta, info.get("signals"), w750, trade_direction)
|
|
return boost, beta
|
|
|
|
|
|
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"
|
|
# BLUE constructs its signal_gen from ENGINE_KWARGS (the orchestrator threads them,
|
|
# esf_alpha_orchestrator.py:180-191). Pin the SAME params — not bare defaults — so a
|
|
# champion retune that changes dc_lookback_bars / dc_min_magnitude_bps / thresholds
|
|
# diverges loudly (caught by test_signal_gen_params_match_blue_engine_kwargs), never
|
|
# silently. dc_status is deterministic in the params + price history (counters unused).
|
|
signal_gen = AlphaSignalGenerator(**BLUE_SIGNAL_GEN_KWARGS)
|
|
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
|
|
|
|
|
|
def _source_ob_market(
|
|
ob_assets: list[str],
|
|
*,
|
|
bar_idx: int,
|
|
ob_engine: Optional[OBFeatureEngine] = None,
|
|
) -> tuple[Optional[float], Optional[float]]:
|
|
"""Derive (median_imbalance, agreement_pct) EXACTLY as BLUE does, via BLUE's HZOBProvider.
|
|
|
|
BLUE wires OB once in _wire_obf (nautilus_event_trader.py:4967-4980):
|
|
live_ob = HZOBProvider(hz_cluster=HZ_CLUSTER, hz_host=HZ_HOST, assets=assets)
|
|
ob_eng = OBFeatureEngine(live_ob); eng.set_ob_engine(ob_eng)
|
|
then per scan calls ``ob_eng.step_live(assets, bar_idx)`` and the orchestrator reads
|
|
``ob_eng.get_market(bar_idx, assets)`` (esf_alpha_orchestrator.py:590). We use BLUE's
|
|
OWN HZOBProvider — NOT a reinvented reader — so OB parsing/shard semantics are BLUE's.
|
|
|
|
OBFeatureEngine ACCUMULATES per-asset history across scans (lookback=10), so a faithful
|
|
mirror requires a PERSISTENT ``ob_engine`` + per-scan-incrementing ``bar_idx`` (the
|
|
caller/shadow loop owns it, exactly as BLUE keeps one ob_eng). When no engine is passed
|
|
this builds a single-shot HZOBProvider-backed engine — correct wiring but no cross-scan
|
|
history; use only for one-off reads / the live smoke."""
|
|
if not ob_assets:
|
|
return None, None
|
|
if ob_engine is None:
|
|
provider = HZOBProvider(hz_cluster=HZ_CLUSTER, hz_host=HZ_HOST, assets=list(ob_assets))
|
|
ob_engine = OBFeatureEngine(provider)
|
|
try:
|
|
ob_engine.step_live(list(ob_assets), bar_idx)
|
|
market = ob_engine.get_market(bar_idx, list(ob_assets))
|
|
return float(market.median_imbalance), float(market.agreement_pct)
|
|
except Exception:
|
|
return None, None
|
|
|
|
|
|
@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,
|
|
acb: Optional[AdaptiveCircuitBreaker] = None,
|
|
ob_engine: Optional[OBFeatureEngine] = None,
|
|
bar_idx: int = 0,
|
|
date_str: Optional[str] = None,
|
|
prior_boost_beta: Optional[tuple[float, float]] = None,
|
|
) -> LiveBlueSourceResult:
|
|
"""Read BLUE's live surfaces and RECONSTRUCT the factor plane the way BLUE computes it.
|
|
|
|
boost/beta are recomputed via ``AdaptiveCircuitBreaker.get_dynamic_boost_from_hz`` (NOT
|
|
the published acb_boost scalar); OB via BLUE's ``HZOBProvider`` + ``OBFeatureEngine``;
|
|
dc_status via ``AlphaSignalGenerator`` pinned to BLUE's ENGINE_KWARGS. For full OB
|
|
accumulation faithfulness the caller passes a PERSISTENT ``ob_engine`` + per-scan
|
|
``bar_idx`` (BLUE keeps one ob_eng). ``date_str`` defaults to today's UTC date (BLUE
|
|
uses self.current_day)."""
|
|
scan_history = scan_history or LiveBlueScanHistory()
|
|
selector = selector or VioletAssetSelector()
|
|
today = date_str or datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
|
|
|
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)
|
|
|
|
mc_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "mc_forewarner_latest")
|
|
mc_scale = _derive_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
|
|
)
|
|
|
|
# boost/beta — recompute via the ACB exactly as BLUE's trader does (NOT the published
|
|
# acb_boost). Needs trade_direction, so computed after it.
|
|
acb_boost, acb_beta = _source_boost_beta(
|
|
client, date_str=today, trade_direction=trade_direction, acb=acb,
|
|
prior=prior_boost_beta,
|
|
)
|
|
|
|
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 market consensus — BLUE's HZOBProvider + OBFeatureEngine (persistent engine if given).
|
|
ob_assets = list(assets) if assets is not None else scan_assets
|
|
ob_median_imbalance, ob_agreement_pct = _source_ob_market(
|
|
ob_assets, bar_idx=bar_idx, ob_engine=ob_engine,
|
|
)
|
|
|
|
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)
|
|
LOGGER.debug(
|
|
"live BLUE plane: asset=%s posture=%s boost=%.4f beta=%.4f mc_scale=%.2f "
|
|
"esof=%s ob=(%s,%s) dc=%s dir=%s",
|
|
selected_asset, posture, acb_boost, acb_beta, mc_scale, esof_score,
|
|
ob_median_imbalance, ob_agreement_pct, dc_status, trade_direction,
|
|
)
|
|
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,
|
|
)
|