From 5c90c8f351fb3c147557bf329395163aa384ac30 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 15 Jun 2026 07:53:08 +0200 Subject: [PATCH] VIOLET V3.2: EsoF size-modulation fold (BLUE SC haircut, exact) modulation.py: VioletSizeModulation wraps BLUE's canonical esof_size_mult_from_score + esof_score_from_payload (exact ESOF_* constants), applies the SC haircut step-for-step as _apply_sc_entry_size_multiplier (nautilus_event_trader.py:3307): mult clamped [0,1] HAIRCUT-ONLY (:3316), near-1 no-op (:3318), round(lev*mult,6)/ round(notional*mult,12). 8 tests pass. Empirical mult-recovery on recorded BLUE: median 1.000, EsoF haircut bands (0.65/0.8/0.9/0.3) visible. NOTE: 28% upward tail (recorded>base) = NEXT parity step (base mid-range param OR gold/gauge up-mult); EsoF is haircut-only by design. Not yet wired into decision_engine (needs EsoF HZ score plane + restart, held). Co-Authored-By: Claude Opus 4.8 --- prod/clean_arch/violet/modulation.py | 89 +++++++++++++++++ .../violet/test_violet_modulation.py | 97 +++++++++++++++++++ 2 files changed, 186 insertions(+) create mode 100644 prod/clean_arch/violet/modulation.py create mode 100644 prod/clean_arch/violet/test_violet_modulation.py diff --git a/prod/clean_arch/violet/modulation.py b/prod/clean_arch/violet/modulation.py new file mode 100644 index 0000000..7055da6 --- /dev/null +++ b/prod/clean_arch/violet/modulation.py @@ -0,0 +1,89 @@ +"""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 diff --git a/prod/clean_arch/violet/test_violet_modulation.py b/prod/clean_arch/violet/test_violet_modulation.py new file mode 100644 index 0000000..eeda37e --- /dev/null +++ b/prod/clean_arch/violet/test_violet_modulation.py @@ -0,0 +1,97 @@ +"""V3.2: EsoF size-modulation — exact replication of BLUE's SC haircut (profuse).""" + +from __future__ import annotations + +import math +import sys + +sys.path.insert(0, "/mnt/dolphinng5_predict") + +import pytest +from hypothesis import given, settings, strategies as st + +from prod.clean_arch.violet.alpha_wrappers import SizeDecision, VioletBetSizer +from prod.clean_arch.violet.modulation import VioletSizeModulation + +MOD = VioletSizeModulation() +G = MOD._gate + + +def _base(vel_div: float = -0.20) -> SizeDecision: + return VioletBetSizer(base_fraction=0.20, min_leverage=0.5, + max_leverage=9.0).calculate(capital=69_000.0, vel_div=vel_div) + + +# ── band mapping reproduces BLUE exactly (drift-guarded against live constants) ─ + +def test_band_values_match_blue_constants(): + assert MOD.mult_for(0.0) == pytest.approx(G.ESOF_NEUTRAL_CORE_MULT) # neutral core + assert MOD.mult_for(-0.5) == pytest.approx(G.ESOF_UNFAVORABLE_CORE_MULT) # deepest cut + assert MOD.mult_for(None) == pytest.approx( + min(1.0, max(0.0, G.ESOF_STALE_FALLBACK_MULT))) # stale fallback + assert MOD.mult_for(0.5) == pytest.approx(1.0) # full size + assert MOD.mult_for(-0.15) == pytest.approx(1.0) # mild-neg full + + +def test_mult_equals_wrapped_blue_fn_clamped(): + # mult_for == clamp(esof_size_mult_from_score, 0, 1) for a grid of scores. + for sc in [0.3, 0.05, 0.0, -0.03, -0.07, -0.2, -0.25, -0.3, -1.0, None]: + expect = max(0.0, min(1.0, float(G.esof_size_mult_from_score(sc)))) + assert MOD.mult_for(sc) == pytest.approx(expect) + + +# ── haircut-only [0,1] clamp (BLUE :3316) ────────────────────────────────────── + +@given(score=st.one_of(st.none(), + st.floats(min_value=-5.0, max_value=5.0, allow_nan=False, + allow_infinity=False))) +@settings(max_examples=120, deadline=None) +def test_mult_always_in_unit_interval(score): + assert 0.0 <= MOD.mult_for(score) <= 1.0 + + +def test_nonfinite_score_is_safe(): + for bad in (float("nan"), float("inf"), float("-inf")): + assert 0.0 <= MOD.mult_for(bad) <= 1.0 + + +# ── apply: near-1 = no-op; haircut scales with exact rounding (BLUE :3318-3338) ─ + +def test_near_one_mult_returns_base_unchanged(): + base = _base() + mod, mult = MOD.apply(base, 0.5) # full-size band -> mult 1.0 + assert mult >= 0.999 + assert mod is base or mod.conviction_leverage == base.conviction_leverage + + +def test_haircut_scales_with_blue_rounding(): + base = _base() # conviction 9.0 + mod, mult = MOD.apply(base, 0.0) # neutral core -> mult 0.8 + assert mult == pytest.approx(0.8) + assert mod.conviction_leverage == round(base.conviction_leverage * mult, 6) + assert mod.notional_fraction == round(base.notional_fraction * mult, 12) + # haircut ⇒ strictly smaller size + assert mod.conviction_leverage < base.conviction_leverage + assert mod.notional_fraction < base.notional_fraction + + +def test_apply_preserves_fraction_bucket_signal(): + base = _base() + mod, _ = MOD.apply(base, -0.5) # deep haircut + assert mod.fraction == base.fraction # only size/leverage scale, not fraction + assert mod.bucket_idx == base.bucket_idx + assert mod.signal_bucket == base.signal_bucket + assert isinstance(mod, SizeDecision) + + +# ── property: modulated size never exceeds base (haircut-only) and stays finite ─ + +@given(vel_div=st.floats(min_value=-0.5, max_value=-0.021, allow_nan=False), + score=st.one_of(st.none(), st.floats(min_value=-1.0, max_value=1.0, allow_nan=False))) +@settings(max_examples=80, deadline=None) +def test_modulated_never_exceeds_base(vel_div, score): + base = _base(vel_div) + mod, mult = MOD.apply(base, score) + assert mod.conviction_leverage <= base.conviction_leverage + 1e-9 + assert mod.notional_fraction <= base.notional_fraction + 1e-9 + assert math.isfinite(mod.conviction_leverage) and mod.conviction_leverage >= 0.0