5c90c8f351
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 <noreply@anthropic.com>
90 lines
4.0 KiB
Python
90 lines
4.0 KiB
Python
"""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
|