PythagorasGoal/scripts/research/r0826_skh_band_drift.py

"""r0826_skh_band_drift.py — si guarda SOTTO il test rosso §5.9 prima di toccare la banda.

IL FATTO. `test_t1_canonical_riproduce_il_backtest_ufficiale` fallisce dal ~24-25/08:
Sharpe hold-out SKH01 `canonical` = 1,9223 contro la banda cablata `1.3 < h < 1.9`
(scritta il 25/07 sull'audit del 02/07: ~1,64). Il codice non e' cambiato; `data/raw/`
e' gitignored e il cron lo ricostruisce ogni notte -> la finestra si allunga e il numero
deriva. Lezione 07/08: un test che fallisce senza che il codice sia cambiato sta
segnalando QUALCOSA — si misura cosa, non si allarga la banda a occhio.

LA MISURA. Stesso codice di produzione (`r0726_skh_onbook.skh_series`), stesso `sh3`,
ma col feed TAGLIATO a date crescenti: se il numero alla data dell'audit riproduce
~1,64 e la curva sale in modo regolare con le barre aggiunte, la deriva e' CRESCITA
DELLA FINESTRA (hold-out 2025+ che si allunga con settimane favorevoli), non un dato
corrotto ne' un codice cambiato. Se invece il taglio all'epoca NON riproduce il numero
dell'epoca, il problema e' nel DATO RICOSTRUITO e la banda e' l'ultima delle questioni.

COSA MI ASPETTO PRIMA DI MISURARE (M12): riproduzione al taglio 02/07 dentro ±0,05;
curva monotona o quasi verso 1,92 (ago 2026 e' stato un rally forte, BTC +21%/30g,
e SKH01 era long ETH da meta' agosto).

    uv run python scripts/research/r0826_skh_band_drift.py
"""
from __future__ import annotations

import sys
from pathlib import Path

import pandas as pd

ROOT = Path(__file__).resolve().parents[2]
for p in (ROOT, ROOT / "scripts" / "research", ROOT / "scripts" / "research" / "alt"):
    sys.path.insert(0, str(p))

import r0702_anchor_skh01 as r02          # noqa: E402
import r0726_skh_onbook as ob             # noqa: E402

TAGLI = ("2026-07-02",   # la data dell'audit che ha scritto ~1,64
         "2026-07-25",   # il giorno in cui la banda e' stata cablata nel test
         "2026-08-01", "2026-08-08", "2026-08-15", "2026-08-22",
         None)           # oggi: deve riprodurre l'1,9223 del test rosso

_GET5M_VERO = r02.get5m.__wrapped__       # la funzione nuda, senza lru_cache


def _taglia(cut_ms: int | None):
    """Rimpiazza get5m con una versione tagliata e svuota TUTTE le cache a valle."""
    if cut_ms is None:
        def g(asset: str):
            return _GET5M_VERO(asset)
    else:
        def g(asset: str):
            df = _GET5M_VERO(asset)
            return df[df["timestamp"] < cut_ms].reset_index(drop=True)
    r02.get5m = g
    # ⚠️ run_asset non usa lru_cache: memoizza in un DICT di modulo (`_CACHE`) — svuotare
    # solo le lru_cache serviva a niente e la prima stesura di questo script ha misurato
    # SETTE volte la stessa finestra (tutte le righe identiche). Si svuota il dict.
    r02._CACHE.clear()
    r02.skh_port.cache_clear()


def main() -> None:
    print("=" * 96)
    print("  SKH01 canonical — Sharpe (full / in-sample / HOLD-OUT) in funzione della finestra dati")
    print("  hold-out = barre >= 2025-01-01: si ALLUNGA ogni notte col cron")
    print("=" * 96)
    print(f"  {'taglio':>12} {'barre HO (g)':>13} {'full':>7} {'in-s':>7} {'HOLD-OUT':>9} "
          f"{'maxDD':>8}   nota")
    print("  " + "-" * 90)
    esiti = {}
    for cut in TAGLI:
        cut_ms = None if cut is None else int(pd.Timestamp(cut, tz="UTC").value // 10**6)
        _taglia(cut_ms)
        s, _ = ob.skh_series(0, "canonical")
        full, is_, hold, dd = ob.sh3(s)
        n_ho = int((s.index >= r02.HOLDOUT).sum())
        nota = ""
        if cut == "2026-07-02":
            nota = "<- deve riprodurre ~1,64 (audit 02/07)"
        elif cut == "2026-07-25":
            nota = "<- il giorno della banda 1.3<h<1.9"
        elif cut is None:
            nota = "<- deve riprodurre l'1,9223 del test rosso"
        lab = cut or "oggi"
        esiti[lab] = hold
        print(f"  {lab:>12} {n_ho:>13} {full:>7.3f} {is_:>7.3f} {hold:>9.4f} {dd:>7.1%}   {nota}")

    print("\n" + "=" * 96)
    print("  VERDETTO (runtime)")
    print("=" * 96)
    rip_audit = abs(esiti["2026-07-02"] - 1.64) < 0.05
    rip_oggi = abs(esiti["oggi"] - 1.9223) < 0.01
    serie = [esiti[k] for k in esiti]
    mono = all(b >= a - 0.03 for a, b in zip(serie, serie[1:]))
    print(f"  taglio 02/07 riproduce l'audit (~1,64)?      {'SI' if rip_audit else 'NO'} "
          f"({esiti['2026-07-02']:.4f})")
    print(f"  nessun taglio riproduce -> dato cambiato:    "
          f"{'no, il dato regge' if rip_audit else 'SI - INDAGARE IL FEED'}")
    print(f"  oggi riproduce il numero del test rosso?     {'SI' if rip_oggi else 'NO'} "
          f"({esiti['oggi']:.4f})")
    print(f"  la curva sale con la finestra (quasi-monot.)? {'SI' if mono else 'NO'}")
    if rip_audit and rip_oggi:
        print("\n  => La deriva e' CRESCITA DELLA FINESTRA, non corruzione ne' codice: il test")
        print("     confronta un numero che si muove col feed contro una banda scritta su una")
        print("     finestra congelata. La riparazione giusta e' ancorare il test alla finestra")
        print("     su cui la banda fu scritta (taglio fisso), non allargare la banda — una")
        print("     banda che insegue il feed non riproduce niente (N11).")


if __name__ == "__main__":
    main()