VIOLET OA: add sizing parity pin tests

Co-authored-by: Codex <codex@openai.com>
This commit is contained in:
Codex
2026-06-16 22:51:07 +02:00
parent f0a73491bd
commit 53bdd908f4

View File

@@ -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