"""VIOLET V3.2: EsoF size-modulation layer — folds BLUE's SC/EsoF haircut. The base sizer (alpha_wrappers.VioletBetSizer) reproduces BLUE's cubic-convex conviction curve. But recorded BLUE `leverage` = base_sizer(vel_div) x EsoF_mult (the SC size gate) — the per-trade scatter the parity harness measured (V3d / the modulation-vs-underutilization finding). This module folds in that EXACT mult. FAITHFULNESS — replicates BLUE's `_apply_sc_entry_size_multiplier` (nautilus_event_trader.py:3307) step-for-step, in order, with the same atomicity: 1. mult = esof_size_mult_from_score(score) (WRAP BLUE's canonical fn) 2. mult = max(0.0, min(1.0, mult)) (:3316 — HAIRCUT ONLY, [0,1]) 3. if mult >= 0.999: return base UNCHANGED (:3318 — near-1 = no-op) 4. effective_leverage = round(base_leverage * mult, 6) (:3338) effective_notional = round(base_notional * mult, 12) (:3329) The score itself comes from the EsoF payload via esof_score_from_payload (WRAP); stale/missing → esof_size_mult_from_score(None) = the defensive fallback mult. Exchange-agnostic (L1): operates on conviction/notional fractions only. """ from __future__ import annotations import sys from pathlib import Path from typing import Any, Optional, Tuple from .alpha_wrappers import SizeDecision from .domain import typed _PROJECT_ROOT = Path(__file__).resolve().parents[3] def _import_esof_gate() -> Any: """Import BLUE's esof_size_gate (same root-injection as blue_parity/alpha_wrappers).""" 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 class VioletSizeModulation: """BLUE's EsoF/SC size haircut, folded onto the base SizeDecision. Wraps BLUE's canonical ``esof_size_mult_from_score`` / ``esof_score_from_payload`` (no reimplementation → exact ESOF_* constants + band mapping).""" def __init__(self) -> None: self._gate = _import_esof_gate() # ── the EsoF score → mult (BLUE canonical fn + the [0,1] haircut clamp) ────── def mult_for(self, score: Any) -> float: """esof_size_mult_from_score(score) clamped to [0,1] (BLUE :3316). ``score`` is accepted loosely (Any) to mirror BLUE's fn, which robustly coerces/guards any score incl. None/stale → defensive fallback mult.""" mult = float(self._gate.esof_size_mult_from_score(score)) if mult != mult or mult in (float("inf"), float("-inf")): # non-finite guard (:3314) mult = 1.0 return max(0.0, min(1.0, mult)) def score_from_payload(self, payload: Optional[dict], **kw: Any) -> Optional[float]: """Wrap esof_score_from_payload (HZ EsoF payload → fresh advisory score).""" return self._gate.esof_score_from_payload(payload, **kw) # ── apply to the base size, step-for-step as BLUE :3318-3338 ───────────────── @typed def apply(self, size: SizeDecision, score: Any) -> Tuple[SizeDecision, float]: """Return (modulated_size, mult). Near-1 mult ⇒ base returned UNCHANGED.""" mult = self.mult_for(score) if mult >= 0.999: # BLUE :3318 — no-op return size, mult eff_leverage = round(size.conviction_leverage * mult, 6) # :3338 eff_notional_fraction = round(size.notional_fraction * mult, 12) # :3329 grain modulated = SizeDecision( fraction=size.fraction, conviction_leverage=eff_leverage, notional_fraction=eff_notional_fraction, bucket_idx=size.bucket_idx, strength_score=size.strength_score, signal_bucket=size.signal_bucket, ) return modulated, mult