siloqy/prod/clean_arch/violet/live_blue_source.py

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

# 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


def _source_boost_beta(
    client: hazelcast.HazelcastClient,
    *,
    date_str: str,
    trade_direction: int,
    acb: Optional[AdaptiveCircuitBreaker] = None,
) -> tuple[float, float]:
    """Recompute (boost, beta) EXACTLY as BLUE's live trader does — NOT from acb_boost.

    BLUE (nautilus_event_trader.py on_exf_update:4769 / rollover prewarm:2710):
        acb = AdaptiveCircuitBreaker()                  # bare, no args (trader 578/585)
        info = acb.get_dynamic_boost_from_hz(
            date_str=today,
            exf_snapshot=DOLPHIN_FEATURES['exf_latest'],
            w750_velocity=latest_eigen_scan['w750_velocity'] or None,
            direction=trade_direction,                  # NO ob_engine in the live path
        )
        boost, beta = info['boost'], info['beta']

    The published ``DOLPHIN_FEATURES['acb_boost']`` is a SEPARATE daily publish
    (acb_processor_service) and is NOT what drives BLUE's sizing, so we never read it.
    On stale exf (>12h) get_dynamic_boost_from_hz raises ValueError; BLUE logs
    'ACB Stale Data Fallback' and keeps the prior boost/beta — VIOLET has no prior, so it
    returns the neutral identity (1.0, 0.0)."""
    exf = _jsonish(_read_hz_map(client, "DOLPHIN_FEATURES", "exf_latest"))
    if not isinstance(exf, Mapping):
        return 1.0, 0.0
    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,
        )
    except ValueError:
        return 1.0, 0.0   # ACB Stale Data Fallback (BLUE keeps prior; VIOLET neutral)
    boost = _coerce_float(info.get("boost"), 1.0) or 1.0
    beta = _coerce_float(info.get("beta"), 0.0) or 0.0
    return max(0.0, boost), max(0.0, 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,
) -> 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,
    )

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