"""VIOLET V3.3 — full sizing parity: BLUE's complete conviction-leverage composition. V3a (alpha_wrappers.VioletBetSizer) reproduces the BASE cubic-convex curve. V3.2 (modulation.VioletSizeModulation) folds the EsoF haircut. This layer composes the REMAINDER of BLUE's full sizing path so that VIOLET's conviction leverage is bit-identical to live BLUE's ``esf_alpha_orchestrator.NDAlphaEngine._try_entry``. The authoritative composition (esf_alpha_orchestrator.py:600-619, transcribed verbatim below) multiplies five factors and applies two caps: raw_leverage = base_leverage # AlphaBetSizer cubic conviction * dc_lev_mult # dc CONFIRM boost, else 1.0 * regime_size_mult # ACB base_boost × (1+β·s³) × mc_scale * market_ob_mult # cross-asset OB consensus [0.85,1.20] * _esof_size_mult # EsoF haircut (esof_size_mult_from_score) clamped_max = min(base_max_leverage × regime × ob × esof, abs_max_leverage) if posture==STALKER: clamped_max = min(clamped_max, 2.0) leverage = max(min_leverage, min(raw_leverage, clamped_max)) notional = capital × fraction × leverage WRAP, DON'T REIMPLEMENT — every factor is produced by BLUE's REAL kernel: - base_leverage / fraction : ``AlphaBetSizer.calculate_size`` (via VioletBetSizer) - _esof_size_mult : ``esof_size_mult_from_score`` (esof_size_gate.py) - regime_size_mult : the ACB day-state × the orchestrator's own ``_strength_cubic`` + ``_update_regime_size_mult`` formula (3-scale: base_boost·(1+β·s³)·mc_scale) - market_ob_mult : the orchestrator's OB consensus formula (:587-595) over ``OBFeatureEngine.get_market`` outputs - dc_lev_mult : signal_gen.dc_leverage_boost iff dc_status=="CONFIRM" The only thing replicated is the ~8-line arithmetic composition (trivial deterministic float math — bit-identical when operation order is preserved, which the @gate Monte-Carlo proves against the REAL orchestrator). Gold-spec caps (FROZEN_ALGO_SPEC_GOLD_REFERENCE.md §4): base_max_leverage=8.0 (soft, the boost lifts toward abs), abs_max_leverage=9.0 (hard). Exchange-agnostic (L1): ``notional_fraction = fraction × conviction_leverage`` is the conviction side of the dual-leverage; the exchange-leverage mapping is L3. VIOLET stays DARK — this layer emits a sizing decision, never an order. """ from __future__ import annotations import sys from pathlib import Path from typing import Annotated, Any, Literal, Optional, Tuple from pydantic import Field from .alpha_wrappers import ConvictionLeverage, Fraction, SizeDecision, VioletBetSizer from .domain import StrictModel, typed _PROJECT_ROOT = Path(__file__).resolve().parents[3] # ── refined scalars / posture ───────────────────────────────────────────────── Posture = Literal["APEX", "STALKER", "RESTORED", "TURTLE", "HIBERNATE"] # A size multiplier in the composition (regime / ob / esof / dc). NO upper cap — # BLUE imposes none; an arbitrary ceiling could reject a valid extreme BLUE value # (review 2026-06-15: removed the le=64/le=4 liberty — "no hygiene BLUE lacks"). # Only guards are V-TYPES poison-rejection (non-negative + finite), which can never # reject a real BLUE factor: all are products of non-negative finite kernel outputs. SizeMult = Annotated[float, Field(ge=0.0, allow_inf_nan=False)] Boost = Annotated[float, Field(ge=0.0, allow_inf_nan=False)] Beta = Annotated[float, Field(ge=0.0, allow_inf_nan=False)] McScale = Annotated[float, Field(ge=0.0, allow_inf_nan=False)] Strength = Annotated[float, Field(ge=0.0, le=1.0, allow_inf_nan=False)] # math range [0,1] # OB market consensus inputs — faithful domains (not liberties). Imbalance = Annotated[float, Field(ge=-1.0, le=1.0, allow_inf_nan=False)] Agreement = Annotated[float, Field(ge=0.0, le=1.0, allow_inf_nan=False)] def _import_esof_gate() -> Any: """Import BLUE's ``esof_size_gate`` (same root-injection as 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 # ── the full multiplier breakdown (for the gate report + diagnostics) ────────── class SizingBreakdown(StrictModel): """Every factor that entered the composition, for traceability/replay. Mirrors the orchestrator's own intermediate state at :600-619 so the @gate can assert bit-identity factor-by-factor, not just on the final leverage. """ base_leverage: ConvictionLeverage base_fraction: Fraction dc_lev_mult: SizeMult regime_size_mult: SizeMult market_ob_mult: SizeMult esof_size_mult: SizeMult strength_cubic: Strength raw_leverage: float = Field(allow_inf_nan=False) clamped_max_leverage: float = Field(allow_inf_nan=False) posture: str min_leverage: float = Field(ge=0.0, allow_inf_nan=False) base_max_leverage: float = Field(gt=0.0, allow_inf_nan=False) abs_max_leverage: float = Field(gt=0.0, allow_inf_nan=False) class FullSizeDecision(StrictModel): """The composed sizing decision + its factor breakdown.""" decision: SizeDecision breakdown: SizingBreakdown # ── the sizer ────────────────────────────────────────────────────────────────── class VioletSizer: """Composes BLUE's full 5-multiplier conviction leverage + caps. This is a SIZING-MATH layer: it composes factors that the caller supplies (from the live ACB / OB engine / EsoF payload / signal generator). Each factor-producing method (``regime_size_mult``, ``esof_size_mult``, ``market_ob_mult``, ``dc_lev_mult``) WRAPS BLUE's real kernel or replicates its pure-arithmetic formula verbatim; ``compose`` applies the authoritative 8-line composition (orchestrator :600-619) bit-for-bit. Gold-spec defaults: ``base_max_leverage=8.0`` (soft; the multipliers lift the base cubic *toward* the abs cap), ``abs_max_leverage=9.0`` (hard). The base bet-sizer is constructed with ``max_leverage=base_max_leverage`` so its own clamp matches the orchestrator's ``bet_sizer.max_leverage``. """ def __init__( self, *, base_fraction: float = 0.20, min_leverage: float = 0.5, base_max_leverage: float = 8.0, abs_max_leverage: float = 9.0, vel_div_threshold: float = -0.02, vel_div_extreme: float = -0.05, long_vel_div_threshold: float = 0.01, long_vel_div_extreme: float = 0.04, leverage_convexity: float = 3.0, dc_leverage_boost: float = 1.0, use_dynamic_leverage: bool = True, use_alpha_layers: bool = True, ): if base_max_leverage > abs_max_leverage: raise ValueError( f"base_max_leverage ({base_max_leverage}) must not exceed " f"abs_max_leverage ({abs_max_leverage})" ) self.base_max_leverage = float(base_max_leverage) self.abs_max_leverage = float(abs_max_leverage) self.min_leverage = float(min_leverage) self.vel_div_threshold = float(vel_div_threshold) self.vel_div_extreme = float(vel_div_extreme) self.long_vel_div_threshold = float(long_vel_div_threshold) self.long_vel_div_extreme = float(long_vel_div_extreme) self.leverage_convexity = float(leverage_convexity) self.dc_leverage_boost = float(dc_leverage_boost) # The base sizer's own clamp == orchestrator bet_sizer.max_leverage. self._bet_sizer = VioletBetSizer( base_fraction=base_fraction, min_leverage=min_leverage, max_leverage=base_max_leverage, leverage_convexity=leverage_convexity, vel_div_threshold=vel_div_threshold, vel_div_extreme=vel_div_extreme, use_dynamic_leverage=use_dynamic_leverage, use_alpha_layers=use_alpha_layers, ) self._esof_gate = _import_esof_gate() # ── factor producers (each WRAPS BLUE's real kernel / pure formula) ──────── @typed def base_size( self, *, capital: float, vel_div: float, vel_div_trend: float = 0.0, trade_direction: int = -1, ) -> SizeDecision: """BLUE's ``AlphaBetSizer.calculate_size`` (cubic conviction + fraction).""" return self._bet_sizer.calculate( capital=capital, vel_div=vel_div, vel_div_trend=vel_div_trend, trade_direction=trade_direction, ) @typed def strength_cubic(self, vel_div: float, *, trade_direction: int = -1) -> Strength: """The orchestrator's ``_strength_cubic`` (esf_alpha_orchestrator.py:872-885). Normalised signal strength in [0,1]^convexity for the active side. Replicated verbatim — it is the SAME knobs the orchestrator feeds its own ``_update_regime_size_mult``; bit-identity requires the identical formula. """ if trade_direction == 1: if vel_div <= self.long_vel_div_threshold: return 0.0 denom = self.long_vel_div_extreme - self.long_vel_div_threshold raw = (vel_div - self.long_vel_div_threshold) / denom if denom != 0.0 else 0.0 else: if vel_div >= self.vel_div_threshold: return 0.0 denom = self.vel_div_threshold - self.vel_div_extreme raw = (self.vel_div_threshold - vel_div) / denom if denom != 0.0 else 0.0 return min(1.0, max(0.0, raw)) ** self.leverage_convexity @typed def regime_size_mult( self, vel_div: float, *, boost: Boost, beta: Beta, mc_scale: McScale, trade_direction: int = -1, ) -> SizeMult: """The orchestrator's ``_update_regime_size_mult`` (esf_alpha_orchestrator.py:898-909). 3-scale formula: base_boost × (1 + β × strength³) × mc_scale. β>0 gate is doctrinal: applies whenever eigenvalue-velocity regime is active. The boost / beta come from the live ``AdaptiveCircuitBreaker``; mc_scale from the MC-Forewarner — the caller supplies them (sizing-math layer owns no I/O). """ if beta > 0: ss = self.strength_cubic(vel_div, trade_direction=trade_direction) return boost * (1.0 + beta * ss) * mc_scale return boost * mc_scale @typed def esof_size_mult(self, score: Any) -> SizeMult: """BLUE's ``esof_size_mult_from_score`` (orchestrator :857, RAW — no clamp). The orchestrator stores ``float(esof_size_mult_from_score(score))`` with no [0,1] clamp and no rounding; the function's own range is [0.30, 1.0]. Mirrored exactly so the composition sees the identical float. """ return float(self._esof_gate.esof_size_mult_from_score(score)) @typed def market_ob_mult( self, median_imbalance: Imbalance, agreement_pct: Agreement, *, trade_direction: int = -1, ) -> SizeMult: """The orchestrator's OB market consensus (esf_alpha_orchestrator.py:587-595). ``OBFeatureEngine.get_market`` → (median_imbalance, agreement_pct); the orchestrator flips sign for SHORT, then boosts (up to +20%) on aligned consensus or haircuts (down to 0.85) on adverse consensus — both gated on agreement_pct > 0.70. Transcribed verbatim. """ eff_imb = -median_imbalance if trade_direction == -1 else median_imbalance if eff_imb > 0.08 and agreement_pct > 0.70: return 1.0 + min(0.20, eff_imb * agreement_pct * 0.5) if eff_imb < -0.08 and agreement_pct > 0.70: return max(0.85, 1.0 - abs(eff_imb) * agreement_pct * 0.3) return 1.0 @typed def dc_lev_mult(self, dc_status: str) -> SizeMult: """dc_leverage_boost iff dc_status=="CONFIRM", else 1.0 (orchestrator :575-577).""" return self.dc_leverage_boost if dc_status == "CONFIRM" else 1.0 # ── the authoritative composition (orchestrator :600-619, VERBATIM) ───────── @typed def compose( self, base: SizeDecision, *, dc_lev_mult: SizeMult, regime_size_mult: SizeMult, market_ob_mult: SizeMult, esof_size_mult: SizeMult, posture: Posture = "APEX", strength_cubic: Optional[float] = None, ) -> SizeDecision: """Apply BLUE's full composition (:600-619) to a base SizeDecision. Operation order is load-bearing for float bit-identity — left-to-right multiply, then the two-stage clamp (soft/abs ceiling, STALKER 2.0, then the min_leverage floor). ``base.fraction`` is carried through UNCHANGED (the multipliers scale leverage, never the fraction). """ base_leverage = base.conviction_leverage # :600-603 — the soft×regime×ob×esof ceiling, floored by the hard abs cap. clamped_max_leverage = min( self.base_max_leverage * regime_size_mult * market_ob_mult * esof_size_mult, self.abs_max_leverage, ) # :604-610 — raw conviction = base × dc × regime × ob × esof. raw_leverage = ( base_leverage * dc_lev_mult * regime_size_mult * market_ob_mult * esof_size_mult ) # :612-614 — STALKER structural ceiling. if posture == "STALKER": clamped_max_leverage = min(clamped_max_leverage, 2.0) # :616-617 — cap then floor. leverage = min(raw_leverage, clamped_max_leverage) leverage = max(self.min_leverage, leverage) return SizeDecision( fraction=base.fraction, conviction_leverage=leverage, notional_fraction=base.fraction * leverage, bucket_idx=base.bucket_idx, strength_score=base.strength_score, signal_bucket=base.signal_bucket, ) # ── end-to-end: produce every factor from raw inputs, then compose ───────── @typed def size( self, *, capital: float, vel_div: float, boost: Boost = 1.0, beta: Beta = 0.0, mc_scale: McScale = 1.0, esof_score: Any = None, ob_median_imbalance: Optional[float] = None, ob_agreement_pct: Optional[float] = None, dc_status: str = "NONE", posture: Posture = "APEX", vel_div_trend: float = 0.0, trade_direction: int = -1, ) -> FullSizeDecision: """Full sizing path: wrapped kernels produce each factor, then compose. ``boost``/``beta`` are the live ACB day-state (get_dynamic_boost_for_date); ``mc_scale`` the MC-Forewarner scale; ``esof_score`` the advisory score; ``ob_*`` the OBFeatureEngine.get_market outputs (None → no OB engine → 1.0). Returns the composed SizeDecision + a full factor breakdown. """ base = self.base_size( capital=capital, vel_div=vel_div, vel_div_trend=vel_div_trend, trade_direction=trade_direction, ) dcm = self.dc_lev_mult(dc_status) rsm = self.regime_size_mult( vel_div, boost=boost, beta=beta, mc_scale=mc_scale, trade_direction=trade_direction, ) if ob_median_imbalance is not None and ob_agreement_pct is not None: obm = self.market_ob_mult( ob_median_imbalance, ob_agreement_pct, trade_direction=trade_direction, ) else: obm = 1.0 esm = self.esof_size_mult(esof_score) ss = self.strength_cubic(vel_div, trade_direction=trade_direction) decision = self.compose( base, dc_lev_mult=dcm, regime_size_mult=rsm, market_ob_mult=obm, esof_size_mult=esm, posture=posture, strength_cubic=ss, ) base_lev = base.conviction_leverage clamped = min( self.base_max_leverage * rsm * obm * esm, self.abs_max_leverage, ) if posture == "STALKER": clamped = min(clamped, 2.0) raw = base_lev * dcm * rsm * obm * esm breakdown = SizingBreakdown( base_leverage=base_lev, base_fraction=base.fraction, dc_lev_mult=dcm, regime_size_mult=rsm, market_ob_mult=obm, esof_size_mult=esm, strength_cubic=ss, raw_leverage=raw, clamped_max_leverage=clamped, posture=posture, min_leverage=self.min_leverage, base_max_leverage=self.base_max_leverage, abs_max_leverage=self.abs_max_leverage, ) return FullSizeDecision(decision=decision, breakdown=breakdown)