diff --git a/prod/clean_arch/violet/alpha_wrappers.py b/prod/clean_arch/violet/alpha_wrappers.py new file mode 100644 index 0000000..09fe535 --- /dev/null +++ b/prod/clean_arch/violet/alpha_wrappers.py @@ -0,0 +1,214 @@ +"""VIOLET V3a: V-TYPES-bounded wrappers over BLUE's LIVE alpha kernels. + +L1 (PURE ALPHA) of the V3 three-layer architecture: decide + size *literally as +BLUE* by WRAPPING — never reimplementing — the same production kernels that +``prod/nautilus_event_trader.py`` (``DolphinLiveTrader``) runs: + + - ``AlphaAssetSelector`` (alpha_asset_selector.py) — IRP selection, BIBLE §5 + - ``AlphaBetSizer`` (alpha_bet_sizer.py) — cubic-convex conviction + leverage + alpha fraction §6 + - ``AlphaExitEngineV7`` (alpha_exit_v7_engine.py) — live multi-leg exit (as-is) + +``prod/clean_arch/dita_v2/blue_parity.py`` (PinkAssetPicker/PinkAlphaSizer) is the +REFERENCE for the wrap discipline (observe/dedupe-on-scan), NOT authoritative: +it caps ``max_leverage`` at 8.0 while the live kernel *default* is 5.0 and the +recorded conviction distribution reaches 9.0. Therefore ``max_leverage`` (and the +other sizer knobs) are REQUIRED explicit parameters here — the V3d parity harness +pins them from the recorded data; nothing in VIOLET hardcodes a value the way the +drifted wrapper did. + +DUAL-LEVERAGE: the conviction leverage produced here SIZES THE QUANTITY (the +internal/conviction side). The exchange-leverage mapping (``prod/bingx/leverage.py``, +max-3x cubic) is L3 (tradeability), applied downstream — NEVER in this module. +This module is exchange-agnostic by construction. +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import Annotated, Any, Dict, List, Literal, Optional, Tuple + +from pydantic import Field + +from .domain import StrictModel, Symbol, typed + +_PROJECT_ROOT = Path(__file__).resolve().parents[3] + +# ── refined alpha scalars (L1, exchange-agnostic) ───────────────────────────── +# ConvictionLeverage is the bet-sizer's internal leverage — it sizes the +# quantity. It is NOT exchange leverage. Range mirrors the kernel's own +# [min_leverage, max_leverage] envelope; we keep a generous finite ceiling and +# let the kernel's clamp be authoritative (the parity harness verifies it). +ConvictionLeverage = Annotated[float, Field(ge=0.0, le=64.0, allow_inf_nan=False)] +Fraction = Annotated[float, Field(ge=0.0, le=1.0, allow_inf_nan=False)] +VelDiv = Annotated[float, Field(allow_inf_nan=False)] +Side = Literal["LONG", "SHORT"] + + +def _import_blue_alpha() -> Tuple[Any, Any, Any]: + """Import BLUE's live selector/sizer/exit-v7 kernels. + + Mirrors blue_parity._import_blue_modules: the real package lives at + /nautilus_dolphin/nautilus_dolphin/; BLUE runs with + /nautilus_dolphin on sys.path. + """ + try: + from nautilus_dolphin.nautilus import ( # type: ignore + alpha_asset_selector, alpha_bet_sizer, alpha_exit_v7_engine, + ) + 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 ( # type: ignore + alpha_asset_selector, alpha_bet_sizer, alpha_exit_v7_engine, + ) + return alpha_asset_selector, alpha_bet_sizer, alpha_exit_v7_engine + + +# ── L1 value objects (the pure-alpha output, exchange-agnostic) ──────────────── + +class AssetPick(StrictModel): + """One IRP selection result — the asset the alpha would trade this scan.""" + + asset: Symbol + side: Side + ars_score: float + alignment: float + + +class SizeDecision(StrictModel): + """Bet-sizer output. ``notional_fraction = fraction * conviction_leverage`` + is the realized notional/capital (== the recorded ``our_leverage``); it is + the conviction side of the dual-leverage and is exchange-agnostic.""" + + fraction: Fraction + conviction_leverage: ConvictionLeverage + notional_fraction: float = Field(ge=0.0, allow_inf_nan=False) + bucket_idx: int = Field(ge=0, le=3) + strength_score: float = Field(ge=0.0, allow_inf_nan=False) + signal_bucket: str + + +# ── wrappers ─────────────────────────────────────────────────────────────────── + +class VioletAssetSelector: + """Wraps BLUE's ``AlphaAssetSelector`` — IRP ranking over the scan universe. + + Faithful to blue_parity.PinkAssetPicker semantics: per-asset price history + accumulated from scan payloads, ranked at signal time; no-fallback (§5.5). + """ + + def __init__(self, lookback_horizon: int = 0, min_alignment: float = 0.0): + sel_mod, _, _ = _import_blue_alpha() + self._mod = sel_mod + self.lookback = int(lookback_horizon) if lookback_horizon > 0 else int(sel_mod.IRP_LOOKBACK) + self.min_alignment = float(min_alignment) + self._selector = sel_mod.AlphaAssetSelector(lookback_horizon=self.lookback) + + @typed + def pick(self, market_data: dict[str, list[float]], regime_direction: int) -> Optional[AssetPick]: + """Rank the warm universe; return the first survivor as a typed pick. + + ``market_data``: per-asset price history (>= lookback+1 prices). + ``regime_direction``: -1 short regime, +1 long. + """ + if not market_data: + return None + rankings = self._selector.rank_assets(market_data, regime_direction=int(regime_direction)) + expected = "SHORT" if int(regime_direction) == -1 else "LONG" + for r in rankings: + if self.min_alignment > 0 and r.metrics.alignment < self.min_alignment: + continue + if r.action != expected: + continue + return AssetPick( + asset=str(r.asset).upper(), + side=r.action, + ars_score=float(r.ars_score), + alignment=float(r.metrics.alignment), + ) + return None + + +class VioletBetSizer: + """Wraps BLUE's ``AlphaBetSizer`` — cubic-convex conviction leverage + fraction. + + All sizer knobs are EXPLICIT (no hidden defaults): the live kernel default + ``max_leverage`` is 5.0, blue_parity passes 8.0, and recorded conviction + reaches 9.0 — so the caller (and the parity harness) must pass the value + pinned from live BLUE, never inherit a drifted default. + """ + + def __init__( + self, + *, + base_fraction: float, + min_leverage: float, + max_leverage: float, + leverage_convexity: float = 3.0, + vel_div_threshold: float = -0.02, + vel_div_extreme: float = -0.05, + use_dynamic_leverage: bool = True, + use_alpha_layers: bool = True, + ): + _, sizer_mod, _ = _import_blue_alpha() + self._sizer = sizer_mod.AlphaBetSizer( + base_fraction=base_fraction, + min_leverage=min_leverage, + max_leverage=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.max_leverage = float(max_leverage) + + @typed + def calculate( + self, *, capital: float, vel_div: float, + vel_div_trend: float = 0.0, trade_direction: int = -1, + ) -> SizeDecision: + raw = self._sizer.calculate_size( + capital=float(capital), vel_div=float(vel_div), + vel_div_trend=float(vel_div_trend), trade_direction=int(trade_direction), + ) + fraction = float(raw["fraction"]) + leverage = float(raw["leverage"]) + breakdown = raw.get("breakdown", {}) + return SizeDecision( + fraction=fraction, + conviction_leverage=leverage, + notional_fraction=fraction * leverage, # == our_leverage + bucket_idx=int(raw.get("bucket_idx", 3)), + strength_score=float(breakdown.get("strength_score", 0.0)), + signal_bucket=str(breakdown.get("signal_bucket", "")), + ) + + def record_trade(self, bucket_idx: int, pnl: float) -> None: + """Feed realized PnL back into the alpha layers (bucket boost/streak).""" + self._sizer.record_trade(int(bucket_idx), float(pnl)) + + +class VioletExitEngine: + """Wraps BLUE's live ``AlphaExitEngineV7`` (multi-leg exit). Run AS-IS first + (operator mandate); refactor/promote/demote later. Exchange-agnostic: emits + an exit decision dict from the kernel, not a venue order.""" + + def __init__(self, *, bar_duration_sec: float = 11.0, **engine_kwargs: Any): + _, _, exit_mod = _import_blue_alpha() + self._mod = exit_mod + self._engine = exit_mod.AlphaExitEngineV7( + bar_duration_sec=bar_duration_sec, **engine_kwargs, + ) + + def make_context(self, *args: Any, **kwargs: Any) -> Any: + return self._engine.make_context(*args, **kwargs) + + @typed + def evaluate(self, context: Any) -> dict: + """Return the kernel's exit decision dict verbatim (parity-faithful).""" + return self._engine.evaluate(context) diff --git a/prod/clean_arch/violet/test_violet_alpha_wrappers.py b/prod/clean_arch/violet/test_violet_alpha_wrappers.py new file mode 100644 index 0000000..699cccd --- /dev/null +++ b/prod/clean_arch/violet/test_violet_alpha_wrappers.py @@ -0,0 +1,87 @@ +"""V3a: V-TYPES wrappers over BLUE's live alpha kernels — parity + drift guards.""" + +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 pydantic import ValidationError + +from prod.clean_arch.violet.alpha_wrappers import ( + AssetPick, SizeDecision, VioletAssetSelector, VioletBetSizer, +) + + +def _sizer(max_leverage: float = 9.0) -> VioletBetSizer: + return VioletBetSizer(base_fraction=0.20, min_leverage=0.5, max_leverage=max_leverage) + + +def _universe(n: int = 4, bars: int = 60) -> dict: + return { + f"A{i}USDT": [100.0 * (1 + 0.001 * math.sin(0.1 * t) + 0.0005 * i * t) for t in range(bars)] + for i in range(n) + } + + +# ── sizing reproduces the recorded model ────────────────────────────────────── + +def test_notional_fraction_is_fraction_times_conviction(): + d = _sizer().calculate(capital=69_000.0, vel_div=-0.05, trade_direction=-1) + assert d.notional_fraction == pytest.approx(d.fraction * d.conviction_leverage) + + +def test_extreme_short_signal_saturates_at_max_leverage_9(): + # vel_div at/under extreme drives conviction to the cap; with max_leverage=9 + # and base_fraction 0.20, notional_fraction == 1.8 — the recorded our_leverage max. + d = _sizer(9.0).calculate(capital=69_000.0, vel_div=-0.20, trade_direction=-1) + assert d.conviction_leverage == pytest.approx(9.0) + assert d.notional_fraction == pytest.approx(1.8) + + +def test_max_leverage_parameterization_is_live_not_a_default(): + # The drift guard: the kernel default is 5.0, blue_parity passes 8.0, live is + # 9.0. Passing different caps MUST change the saturated conviction — proving + # VIOLET never inherits a hardcoded/drifted value. + cap5 = _sizer(5.0).calculate(capital=69_000.0, vel_div=-0.20, trade_direction=-1) + cap9 = _sizer(9.0).calculate(capital=69_000.0, vel_div=-0.20, trade_direction=-1) + assert cap5.conviction_leverage == pytest.approx(5.0) + assert cap9.conviction_leverage == pytest.approx(9.0) + assert cap5.conviction_leverage != cap9.conviction_leverage + + +def test_size_decision_is_typed_and_frozen(): + d = _sizer().calculate(capital=50_000.0, vel_div=-0.04, trade_direction=-1) + assert isinstance(d, SizeDecision) + with pytest.raises(ValidationError): + d.fraction = 0.5 # frozen + + +# ── selection returns a typed pick or None ──────────────────────────────────── + +def test_selector_returns_typed_short_pick(): + pick = VioletAssetSelector(lookback_horizon=50).pick(_universe(), regime_direction=-1) + if pick is not None: + assert isinstance(pick, AssetPick) + assert pick.side == "SHORT" + + +def test_selector_empty_universe_is_none(): + assert VioletAssetSelector(lookback_horizon=50).pick({}, regime_direction=-1) is None + + +# ── property: conviction always within the kernel envelope; output finite ───── + +@given( + vel_div=st.floats(min_value=-0.5, max_value=0.5, allow_nan=False, allow_infinity=False), + cap=st.floats(min_value=1.0, max_value=1e7, allow_nan=False, allow_infinity=False), +) +@settings(max_examples=60, deadline=None) +def test_conviction_within_envelope_and_finite(vel_div, cap): + d = _sizer(9.0).calculate(capital=cap, vel_div=vel_div, trade_direction=-1) + assert 0.0 <= d.conviction_leverage <= 9.0 + 1e-9 + assert math.isfinite(d.notional_fraction) and d.notional_fraction >= 0.0 + assert 0 <= d.bucket_idx <= 3