"""Test dell'ondata 2026-07-26 (T1 esecuzione SKH01, T2 DVOLSPREAD, T3 XSR01 su equity). Blindano le proprieta' da cui dipendono le conclusioni — non i numeri (che si muovono col feed), ma le INVARIANTI che, se si rompessero, renderebbero i numeri privi di significato. """ from __future__ import annotations import sys from pathlib import Path import numpy as np import pytest ROOT = Path(__file__).resolve().parents[1] for p in (ROOT, ROOT / "scripts" / "research", ROOT / "scripts" / "research" / "alt", ROOT / "scripts" / "research" / "ortho"): sys.path.insert(0, str(p)) pytest.importorskip("pandas") # ============================================================ T1 — esecuzione SKH01 def test_t1_le_modalita_condividono_le_stesse_trade_boundaries(): """Il confronto fra modalita' d'esecuzione ha senso SOLO se i trade sono gli stessi: cambia il prezzo di fill, non quali trade si fanno. Se le boundaries divergessero staremmo confrontando due strategie diverse e non due esecuzioni della stessa.""" import r0702_anchor_skh01 as r02 import r0726_skh_onbook as ob _, _, ltf, ent = r02.run_asset("BTC", 0) df5 = r02.get5m("BTC") args = (df5["timestamp"].values.astype(np.int64) + 300_000, df5["open"].values.astype(float), df5["high"].values.astype(float), df5["low"].values.astype(float), df5["close"].values.astype(float)) stats = {} for m in ("canonical", "hourly", "onbook", "onbook_tp"): _, st = ob.sim_equity_exec(ltf, ent, m, *args) stats[m] = (st["sl"], st["tp"], st["time"]) assert len(set(stats.values())) == 1, stats def test_t1_canonical_riproduce_il_backtest_ufficiale(): """La modalita' 'canonical' deve coincidere col backtest dello sleeve: e' l'ancora di tutto il confronto. Tolleranza larga (il feed cresce ogni giorno).""" import r0726_skh_onbook as ob s, _ = ob.skh_series(0, "canonical") full, _, hold, dd = ob.sh3(s) assert 1.2 < full < 1.7, full # audit 02/07: ~1.46 assert 1.3 < hold < 1.9, hold # audit 02/07: ~1.64 assert 0.12 < dd < 0.24, dd # audit 02/07: ~18.1% def test_t1_il_limit_tp_non_puo_riempire_peggio_del_livello(): """Un limit resting o riempie AL livello o non riempie. Se il modello desse un prezzo peggiore del livello sul lato TP, starebbe modellando un ordine a mercato.""" import r0702_anchor_skh01 as r02 import r0726_skh_onbook as ob _, _, ltf, ent = r02.run_asset("BTC", 0) df5 = r02.get5m("BTC") eq_tp, _ = ob.sim_equity_exec( ltf, ent, "onbook_tp", df5["timestamp"].values.astype(np.int64) + 300_000, df5["open"].values.astype(float), df5["high"].values.astype(float), df5["low"].values.astype(float), df5["close"].values.astype(float)) assert np.all(np.isfinite(eq_tp)) and eq_tp[-1] > 0 def test_t1_slippage_piu_alta_non_migliora_lo_stop(): """Monotonia elementare: piu' slippage sullo stop-market non puo' dare un book migliore. Se si rompe, il segno dello slippage e' invertito da qualche parte.""" import r0726_skh_onbook as ob keep = ob.SL_SLIP try: ob.SL_SLIP = 0.0 lo = ob.sh3(ob.book(ob.skh_series(0, "onbook")[0], 0.25))[0] ob.SL_SLIP = 0.02 hi = ob.sh3(ob.book(ob.skh_series(0, "onbook")[0], 0.25))[0] finally: ob.SL_SLIP = keep assert lo > hi, (lo, hi) # ============================================================ T2 — DVOLSPREAD def test_t2_il_book_e_market_neutral_per_costruzione(): """w_eth == -w_btc: e' l'unica ragione per cui il lead e' a bassa correlazione con TP01. Se si rompesse, la bassa correlazione sarebbe un caso e non una proprieta'.""" import r0726_dvolspread_gate as g ev = g.evaluate(**g.CHOSEN) assert abs(ev["net_beta"]) < 0.02, ev["net_beta"] def test_t2_book_flat_prima_che_esista_il_dvol(): """Prima del 2021-03 non c'e' DVOL: il book DEVE essere flat, altrimenti sta usando un valore inventato (e il campione 'attivo' sarebbe una bugia).""" import pandas as pd import r0726_dvolspread_gate as g r = g.daily_returns(**g.CHOSEN) pre = r[r.index < pd.Timestamp("2021-03-01", tz="UTC")] assert len(pre) > 100 assert float(np.abs(pre.values).max()) < 1e-12 def test_t2_eseguibile_al_capitale_reale(): """Il lead vale solo se sta dentro il cap di $300/asset a $600 di capitale.""" import r0726_dvolspread_gate as g ev = g.evaluate(**g.CHOSEN) assert ev["executable"], (ev["max_leg_frac"], ev["gross_lev"]) # ============================================================ T3 — XSR01 su equity def test_t3_usa_il_meccanismo_congelato_non_una_copia(): """Il test di falsificazione vale solo se gira LO STESSO meccanismo del candidato: se qualcuno cambiasse W o il segno, il test smetterebbe di dire qualcosa su XSR01.""" from r0725_statarb_multi import SGN, W assert (W, SGN) == (45, +1) def test_t3_il_demean_azzera_la_posizione_media(): """Il demeaning cross-sezionale deve rendere la posizione media giornaliera nulla: e' cio' che elimina algebricamente la gamba comune. Senza, non e' XSR01.""" import r0726_xsr_equity as xe from eqlib import SECTORS_CLASSIC P, _ = xe.build(SECTORS_CLASSIC) Q = P.sub(P.mean(axis=1), axis=0) assert float(np.abs(np.nan_to_num(Q.mean(axis=1).values)).max()) < 1e-12 def test_t3_lo_split_e_riparato_in_lettura(): """IWM ha uno split non aggiustato il 2005-06-09 (difetto trovato il 25/07). Se il pannello non lo riparasse, il test su 26 anni sarebbe avvelenato.""" import pandas as pd from eqlib import load_eq d = load_eq("IWM") r = d["close"].pct_change() around = r[(r.index >= pd.Timestamp("2005-06-01", tz="UTC")) & (r.index <= pd.Timestamp("2005-06-20", tz="UTC"))] assert float(around.min()) > -0.20, float(around.min()) def test_t3_annualizzazione_su_giorni_di_borsa(): """Lezione 25/07: una serie su ~252 barre/anno annualizzata a 365 gonfia lo Sharpe di 1.2x.""" import r0726_xsr_equity as xe assert abs(xe.ANN - np.sqrt(252.0)) < 1e-9