siloqy/prod/clean_arch/violet/live_factors.py

"""VIOLET V3.4b helper: normalize live factor planes into ``SizingFactors``.

This is a standalone boundary helper for the launcher-side V3.4b work item.
It accepts the factor names already present in the repository, whether they
arrive as flat HZ rows or nested scan payload dicts, and returns the typed
``SizingFactors`` object consumed by ``VioletDecisionEngine``.

The helper is intentionally boring:
- no I/O
- no launcher coupling
- no live client assumptions
- strict coercion for the few scalar types we actually need

Precedence is explicit: later sources override earlier ones, and the helper
prefers Hazelcast-style factor snapshots over scan payload fields when both are
present.
"""

from __future__ import annotations

from collections.abc import Mapping, Sequence
from typing import Any

from pydantic import Field

from .decision_engine import SizingFactors
from .domain import StrictModel, typed


class LiveFactorPlane(StrictModel):
    """Raw live-factor inputs before they are handed to ``SizingFactors``."""

    boost: float = Field(default=1.0, ge=0.0, allow_inf_nan=False)
    beta: float = Field(default=0.0, ge=0.0, allow_inf_nan=False)
    mc_scale: float = Field(default=1.0, ge=0.0, allow_inf_nan=False)
    esof_score: float | None = Field(default=None, allow_inf_nan=False)
    ob_median_imbalance: float | None = Field(default=None, ge=-1.0, le=1.0, allow_inf_nan=False)
    ob_agreement_pct: float | None = Field(default=None, ge=0.0, le=1.0, allow_inf_nan=False)
    dc_status: str = "NONE"
    posture: str = "APEX"

    def to_sizing_factors(self) -> SizingFactors:
        return SizingFactors.model_validate(self.model_dump())


def _walk(source: Mapping[str, Any] | None, path: Sequence[str]) -> Any | None:
    cur: Any = source
    for key in path:
        if not isinstance(cur, Mapping) or key not in cur:
            return None
        cur = cur[key]
    return cur


def _first_value(sources: Sequence[Mapping[str, Any] | None], *paths: Sequence[str]) -> Any | None:
    for source in sources:
        if source is None:
            continue
        for path in paths:
            value = _walk(source, path)
            if value is not None and value != "":
                return value
    return None


def _coerce_float(value: Any, default: float | None) -> float | None:
    if value is None:
        return default
    if isinstance(value, bool):
        return float(value)
    try:
        return float(value)
    except (TypeError, ValueError):
        return default


def _coerce_str(value: Any, default: str) -> str:
    if value is None:
        return default
    text = str(value).strip()
    return text or default


@typed
def extract_live_factor_plane(
    *,
    scan_payload: Mapping[str, Any] | None = None,
    hz_snapshot: Mapping[str, Any] | None = None,
) -> LiveFactorPlane:
    """Normalize the live factor plane from the current scan and HZ snapshot.

    Resolution order is:
    1. defaults
    2. scan payload
    3. Hazelcast snapshot

    The implementation searches Hazelcast first, then the scan payload, so the
    HZ plane wins on conflicts.
    """
    sources = (hz_snapshot, scan_payload)

    boost = _coerce_float(
        _first_value(
            sources,
            ("boost",),
            ("acb_boost",),
            ("s_acb_boost",),
            ("acb", "boost"),
        ),
        1.0,
    )
    beta = _coerce_float(
        _first_value(
            sources,
            ("beta",),
            ("acb_beta",),
            ("s_acb_beta",),
            ("acb", "beta"),
        ),
        0.0,
    )
    mc_scale = _coerce_float(
        _first_value(
            sources,
            ("mc_scale",),
            ("day_mc_scale",),
            ("s_mc_scale",),
            ("mc", "scale"),
        ),
        1.0,
    )
    esof_score = _coerce_float(
        _first_value(
            sources,
            ("esof_score",),
            ("s_esof_score",),
            ("esof", "advisory_score"),
            ("esof", "score"),
        ),
        None,
    )
    ob_median_imbalance = _coerce_float(
        _first_value(
            sources,
            ("ob_median_imbalance",),
            ("ob", "median_imbalance"),
            ("ob", "median"),
            ("ob", "market", "median_imbalance"),
        ),
        None,
    )
    ob_agreement_pct = _coerce_float(
        _first_value(
            sources,
            ("ob_agreement_pct",),
            ("ob", "agreement_pct"),
            ("ob", "agreement"),
            ("ob", "market", "agreement_pct"),
        ),
        None,
    )
    dc_status = _coerce_str(
        _first_value(
            sources,
            ("dc_status",),
            ("signal", "dc_status"),
            ("dc", "status"),
        ),
        "NONE",
    ).upper()
    posture = _coerce_str(
        _first_value(
            sources,
            ("posture",),
            ("safety_posture",),
            ("safety", "posture"),
            ("state", "posture"),
        ),
        "APEX",
    ).upper()

    return LiveFactorPlane(
        boost=boost if boost is not None else 1.0,
        beta=beta if beta is not None else 0.0,
        mc_scale=mc_scale if mc_scale is not None else 1.0,
        esof_score=esof_score,
        ob_median_imbalance=ob_median_imbalance,
        ob_agreement_pct=ob_agreement_pct,
        dc_status=dc_status,
        posture=posture,
    )


@typed
def extract_live_sizing_factors(
    *,
    scan_payload: Mapping[str, Any] | None = None,
    hz_snapshot: Mapping[str, Any] | None = None,
) -> SizingFactors:
    """Return the typed ``SizingFactors`` used by the V3.4 shadow path."""
    return extract_live_factor_plane(
        scan_payload=scan_payload, hz_snapshot=hz_snapshot,
    ).to_sizing_factors()