From 53bdd908f4bd417eea92a9006183325db54c9ebb Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 16 Jun 2026 22:51:07 +0200 Subject: [PATCH] VIOLET OA: add sizing parity pin tests Co-authored-by: Codex --- .../violet/test_violet_sizing_parity_pin.py | 153 ++++++++++++++++++ 1 file changed, 153 insertions(+) create mode 100644 prod/clean_arch/violet/test_violet_sizing_parity_pin.py diff --git a/prod/clean_arch/violet/test_violet_sizing_parity_pin.py b/prod/clean_arch/violet/test_violet_sizing_parity_pin.py new file mode 100644 index 00000000..c1b1795f --- /dev/null +++ b/prod/clean_arch/violet/test_violet_sizing_parity_pin.py @@ -0,0 +1,153 @@ +"""V3.4 parity pin tests for BLUE sizing intermediates. + +This is a direct regression pin, not a broad behavioral suite. It compares the +VIOLET replica against BLUE's live formulas for each sizing intermediate on a +deterministic grid, with exact equality. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np + +sys.path.insert(0, "/mnt/dolphinng5_predict") +sys.path.insert(0, "/mnt/dolphinng5_predict/nautilus_dolphin") + +from prod.clean_arch.violet.sizing import VioletSizer + + +def _blue_engine(*, dc_leverage_boost: float = 1.0): + from nautilus_dolphin.nautilus.esf_alpha_orchestrator import NDAlphaEngine + + eng = NDAlphaEngine( + initial_capital=69_000.0, + max_leverage=8.0, + abs_max_leverage=9.0, + min_leverage=0.5, + fraction=0.20, + use_asset_selection=False, + use_direction_confirm=True, + dc_leverage_boost=dc_leverage_boost, + ) + eng.begin_day("2026-06-16", posture="APEX", direction=-1) + return eng + + +def _blue_market_ob_mult(median_imbalance: float, agreement_pct: float, trade_direction: int) -> float: + 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 + + +def _ob_grid() -> list[tuple[float, float]]: + imbalances = np.linspace(-1.0, 1.0, 20, dtype=np.float64) + agreements = np.linspace(0.0, 1.0, 10, dtype=np.float64) + return [(float(i), float(a)) for i in imbalances for a in agreements] + + +def _dc_prices(status: str, lookback: int = 7) -> list[float]: + if status == "CONFIRM": + return [100.0 - 0.2 * i for i in range(lookback + 1)] + if status == "CONTRADICT": + return [100.0 + 0.2 * i for i in range(lookback + 1)] + return [100.0 for _ in range(lookback + 1)] + + +def test_strength_cubic_exact_parity_grid(): + eng = _blue_engine() + sz = VioletSizer() + + cases = [] + for trade_direction in (-1, 1): + for vel_div in np.linspace(-0.06, 0.04, 200, dtype=np.float64): + cases.append((float(vel_div), trade_direction)) + + assert len(cases) >= 200 + for vel_div, trade_direction in cases: + blue = eng._strength_cubic(vel_div, direction=trade_direction) + violet = sz.strength_cubic(vel_div, trade_direction=trade_direction) + assert violet == blue + + +def test_regime_size_mult_exact_parity_grid(): + eng = _blue_engine() + sz = VioletSizer() + + vel_divs = np.linspace(-0.06, 0.0, 50, dtype=np.float64) + boosts = (1.0, 1.25, 1.75, 2.5) + betas = (0.0, 0.2, 0.8) + mc_scales = (0.5, 1.0) + directions = (-1, 1) + + n = 0 + for trade_direction in directions: + eng.regime_direction = trade_direction + eng._day_posture = "APEX" + for vel_div in vel_divs: + for boost in boosts: + for beta in betas: + for mc_scale in mc_scales: + eng._day_base_boost = boost + eng._day_beta = beta + eng._day_mc_scale = mc_scale + eng._update_regime_size_mult(float(vel_div)) + violet = sz.regime_size_mult( + float(vel_div), + boost=float(boost), + beta=float(beta), + mc_scale=float(mc_scale), + trade_direction=trade_direction, + ) + assert violet == eng.regime_size_mult + n += 1 + assert n >= 200 + + +def test_market_ob_mult_exact_parity_grid(): + sz = VioletSizer() + + cases = _ob_grid() + assert len(cases) == 200 + for trade_direction in (-1, 1): + for median_imbalance, agreement_pct in cases: + blue = _blue_market_ob_mult(median_imbalance, agreement_pct, trade_direction) + violet = sz.market_ob_mult( + median_imbalance, + agreement_pct, + trade_direction=trade_direction, + ) + assert violet == blue + + +def test_dc_leverage_multiplier_exact_parity_grid(): + sz = VioletSizer(dc_leverage_boost=1.25) + eng = _blue_engine(dc_leverage_boost=1.25) + sig = eng.signal_gen + + short_vel_divs = np.linspace(-0.06, -0.021, 100, dtype=np.float64) + long_vel_divs = np.linspace(0.011, 0.04, 100, dtype=np.float64) + statuses = ("CONFIRM", "NEUTRAL", "SKIP_CONTRADICT") + + n = 0 + for trade_direction, vel_divs in ((-1, short_vel_divs), (1, long_vel_divs)): + for vel_div in vel_divs: + for status in statuses: + prices = _dc_prices(status) + result = sig.generate( + vel_div=float(vel_div), + vel_div_history=[float(vel_div)] * 10, + asset_price_history=prices, + trade_direction=trade_direction, + asset=None, + current_timestamp=0.0, + ) + blue = sig.dc_leverage_boost if result.dc_status == "CONFIRM" else 1.0 + violet = sz.dc_lev_mult(result.dc_status) + assert violet == blue + n += 1 + assert n >= 200