PythagorasGoal/tests/test_wave_0726.py

"""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 pandas as pd
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.

    ⚠️ RISCRITTO il 2026-08-26 (era il rosso noto §5.9, indagato PRIMA di toccarlo —
    `r0826_skh_band_drift.py`). La versione precedente confrontava una banda scritta
    sulla finestra del 02/07 con un numero calcolato sul feed VIVO, che il cron allunga
    ogni notte: l'hold-out e' passato da 1,64 a 1,93 senza che nulla si fosse rotto —
    e la SOLA settimana 15-22/08 (rally ETH con SKH01 long) vale +0,35 di Sharpe
    hold-out. Una banda che insegue il feed non riproduce niente: qui il feed viene
    TAGLIATO alla data dell'audit, cosi' il test torna a essere una riproduzione
    (N11) — si rompe se cambiano il CODICE o il DATO sulla finestra congelata, e per
    nessun altro motivo. La deriva del vivo non e' piu' affare suo: per quella ci sono
    i monitor."""
    import r0702_anchor_skh01 as r02
    import r0726_skh_onbook as ob
    CUT = int(pd.Timestamp("2026-07-02", tz="UTC").value // 10**6)
    get5m_vero = r02.get5m

    def _tagliato(asset: str):
        df = get5m_vero(asset)
        return df[df["timestamp"] < CUT].reset_index(drop=True)

    r02.get5m = _tagliato
    r02._CACHE.clear()                     # run_asset memoizza in un dict, non in lru_cache
    r02.skh_port.cache_clear()
    try:
        s, _ = ob.skh_series(0, "canonical")
        full, ins, hold, dd = ob.sh3(s)
    finally:
        r02.get5m = get5m_vero
        r02._CACHE.clear()                 # le cache tagliate non devono avvelenare altri test
        r02.skh_port.cache_clear()
    # i numeri misurati da r0826_skh_band_drift al taglio 02/07 (== audit 02/07)
    assert abs(full - 1.463) < 0.02, full
    assert abs(ins - 1.424) < 0.02, ins
    assert abs(hold - 1.6376) < 0.02, hold
    assert abs(dd - 0.181) < 0.01, dd


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