81 lines
2.9 KiB
Python
81 lines
2.9 KiB
Python
|
"""V3d: base-sizer median-curve parity vs recorded BLUE (unit + @gate)."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
sys.path.insert(0, "/mnt/dolphinng5_predict")
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from prod.clean_arch.violet.parity_harness import (
|
||
|
|
ParityReport, base_curve_parity, live_blue_sizer, load_recorded_samples_from_ch,
|
||
|
|
)
|
||
|
|
|
||
|
|
REPORTS_DIR = Path("/mnt/dolphinng5_predict/prod/VIOLET_dev/reports")
|
||
|
|
|
||
|
|
|
||
|
|
def _base_samples(per_bin: int = 12):
|
||
|
|
"""Synthetic samples drawn exactly from the base curve (perfect parity)."""
|
||
|
|
s = live_blue_sizer()
|
||
|
|
out = []
|
||
|
|
for k in range(2, 13): # vd = -0.02 .. -0.12, all on 0.01 bin centers
|
||
|
|
vd = -k / 100.0
|
||
|
|
base = s.calculate(capital=1.0, vel_div=vd, trade_direction=-1).conviction_leverage
|
||
|
|
out.extend([(vd, base)] * per_bin)
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
def test_perfect_base_samples_pass():
|
||
|
|
rep = base_curve_parity(_base_samples(), live_blue_sizer())
|
||
|
|
assert rep.passed
|
||
|
|
assert rep.max_abs_err == pytest.approx(0.0, abs=1e-6)
|
||
|
|
assert rep.pearson_r == pytest.approx(1.0, abs=1e-6)
|
||
|
|
|
||
|
|
|
||
|
|
def test_median_tracks_base_under_minority_haircuts():
|
||
|
|
# majority at base, minority haircut down -> median still tracks base -> passes.
|
||
|
|
s = live_blue_sizer()
|
||
|
|
samples = []
|
||
|
|
for k in range(2, 13): # vd on 0.01 bin centers
|
||
|
|
vd = -k / 100.0
|
||
|
|
base = s.calculate(capital=1.0, vel_div=vd, trade_direction=-1).conviction_leverage
|
||
|
|
samples.extend([(vd, base)] * 8) # 8 at base
|
||
|
|
samples.extend([(vd, base * 0.3)] * 3) # 3 haircut (minority)
|
||
|
|
rep = base_curve_parity(samples, s)
|
||
|
|
assert rep.passed
|
||
|
|
|
||
|
|
|
||
|
|
def test_decorrelated_noise_fails_gate():
|
||
|
|
# leverage independent of vel_div -> low correlation -> gate fails (guard works).
|
||
|
|
import random
|
||
|
|
rng = random.Random(7)
|
||
|
|
samples = [(vd / 1000.0, rng.uniform(0.5, 9.0))
|
||
|
|
for vd in range(-60, -1) for _ in range(12)]
|
||
|
|
rep = base_curve_parity(samples, live_blue_sizer())
|
||
|
|
assert not rep.passed
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.gate
|
||
|
|
def test_base_curve_parity_vs_recorded_blue():
|
||
|
|
"""GATE (prod host): the V3a base sizer reproduces BLUE's recorded median curve."""
|
||
|
|
samples = load_recorded_samples_from_ch(limit=5000)
|
||
|
|
assert len(samples) >= 500, f"too few recorded samples: {len(samples)}"
|
||
|
|
rep = base_curve_parity(samples, live_blue_sizer())
|
||
|
|
|
||
|
|
REPORTS_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
|
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
|
||
|
|
(REPORTS_DIR / f"violet_v3_parity_{ts}.json").write_text(
|
||
|
|
json.dumps(rep.model_dump(), indent=2, default=str))
|
||
|
|
|
||
|
|
assert isinstance(rep, ParityReport)
|
||
|
|
assert rep.n_bins >= 5
|
||
|
|
assert rep.passed, (
|
||
|
|
f"base-curve parity FAILED: max_abs_err={rep.max_abs_err:.3f} "
|
||
|
|
f"(budget {rep.max_abs_err_budget}), pearson={rep.pearson_r:.3f} "
|
||
|
|
f"(budget {rep.pearson_budget})"
|
||
|
|
)
|