PythagorasGoal/scripts/research/intraday_regime_analysis.py

"""intraday_regime_analysis.py — ANALISI DI ROBUSTEZZA del LEAD ERM (filone B) — 2026-06-29.

Il lead di B (ERM 8h L=2.0 thr=0.35 L/S) fa earns_slot=True, ma con 2 caveat NON quantificati
dallo script di scoperta `intraday_regime.py`:
  (1) il VINCITORE e' selezionato per min_hold MASSIMO su ~60 celle  -> selezione-sull'hold-out;
  (2) il plateau hold-out e' a UNA SOLA RIGA (positivo solo a L~2.0; L>=2.5 va negativo sull'hold).
Insieme = rischio multiple-testing / overfit della finestra recente, mai deflazionato (a
differenza del filone C che ha il deflated-Sharpe).

Questo script attacca esattamente quei nodi, SENZA toccare il live (read-only, branch separato):

  A) DEFLATED SHARPE (Bailey & Lopez de Prado) del vincitore vs TUTTI i trial realmente
     cercati (ERM+VEM+VBR+TOD, tutte le celle/TF). Se DSR << 0.95 lo Sharpe non e' significativo
     dopo la correzione per multiple-testing.

  B) SELEZIONE IN-SAMPLE-ONLY: ri-scelgo la cella ERM usando SOLO lo Sharpe PRE-2025 (mai
     l'hold-out), poi ne valuto earns_slot sull'intera storia. Se una cella scelta SENZA vedere
     l'hold-out continua ad ADDS, l'edge non e' hold-out-mined.

  C) ENSEMBLE DEL PLATEAU: invece della singola cella migliore, media i pesi su tutto il
     plateau ERM 8h (L x thr) -> un candidato unico "non-cherry-picked" -> earns_slot. Se la
     famiglia regge senza scegliere L, il caveat (1)+(2) si attenua.

  D) DOVE VIVE L'EDGE: Sharpe per-anno standalone + uplift per-anno del blend 3-way
     (TP01+SKH+ERM) vs 2-way (TP01+SKH), e corr(ERM,SKH) per-anno (e' un hedge-di-SKH?).

Esecuzione:  uv run python scripts/research/intraday_regime_analysis.py
Idempotente, niente scritture su disco (solo report a stdout).
"""
from __future__ import annotations

import sys

import numpy as np
import pandas as pd

sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al  # noqa: E402
from intraday_regime import make_erm, make_vem, make_vbr, make_tod, _skh_daily  # noqa: E402

DPY = 365.25
HOLD = al.HOLDOUT
ASSETS = ("BTC", "ETH")
SCREEN_TFS = ("1h", "4h", "6h", "8h", "12h")

WIN = dict(tf="8h", L_days=2.0, thr=0.35, long_flat=False)   # il lead di B

deflated_sharpe = al.deflated_sharpe   # gate canonico (codificato in altlib da questo filone)


# ===========================================================================
def all_trials():
    """Ricostruisce la griglia COMPLETA realmente cercata in intraday_regime.py, ritorna una
    lista di (tag, factory, tf, params) per pesare il multiple-testing onestamente."""
    out = []
    erm_grid = [dict(L_days=L, thr=t, long_flat=lf)
                for L in (1.0, 2.0, 3.0) for t in (0.35, 0.50) for lf in (False, True)]
    for tf in SCREEN_TFS:
        for p in erm_grid:
            out.append(("ERM", make_erm, tf, p))
    # + il plateau fine a 8h (L x thr, lf=False) — anche quelle sono celle testate
    for L in (1.5, 2.0, 2.5, 3.0):
        for t in (0.30, 0.35, 0.40, 0.45, 0.50):
            out.append(("ERM-plat", make_erm, "8h", dict(L_days=L, thr=t, long_flat=False)))
    vem_grid = [dict(Lmom_days=lm, Lshort_days=2.0, Llong_days=10.0, long_flat=lf)
                for lm in (1.0, 3.0) for lf in (False, True)]
    for tf in ("4h", "6h", "8h", "12h"):
        for p in vem_grid:
            out.append(("VEM", make_vem, tf, p))
    vbr_grid = [dict(k=k, atr_win=14, long_flat=lf) for k in (0.5, 1.0, 1.5) for lf in (False, True)]
    for tf in ("4h", "6h", "8h", "12h"):
        for p in vbr_grid:
            out.append(("VBR", make_vbr, tf, p))
    for lf in (False, True):
        out.append(("TOD", make_tod, "1h", dict(long_flat=lf)))
    return out


def cand_full_is_sharpe(factory, tf, params):
    """(daily, full Sharpe, in-sample<2025 Sharpe) del candidato 50/50 di quella cella."""
    fn = factory(tf=tf, **params)
    daily = al.candidate_daily(fn, tf=tf)
    full = al._sh(daily)
    ins = al._sh(daily[daily.index < HOLD]) if (daily.index < HOLD).sum() > 60 else float("nan")
    return daily, full, ins


# ===========================================================================
def part_A_deflated():
    print("=" * 78)
    print("A) DEFLATED SHARPE del vincitore vs TUTTI i trial cercati (multiple-testing)")
    print("=" * 78)
    trials = all_trials()
    sr_all, sr_no_tod, sr_erm = [], [], []
    win_daily = win_full = None
    for tag, factory, tf, params in trials:
        try:
            daily, full, _ = cand_full_is_sharpe(factory, tf, params)
        except Exception:
            continue
        sr_all.append(full)
        if tag != "TOD":
            sr_no_tod.append(full)
        if tag.startswith("ERM"):
            sr_erm.append(full)
        if tag == "ERM" and tf == WIN["tf"] and params.get("L_days") == WIN["L_days"] \
                and params.get("thr") == WIN["thr"] and params.get("long_flat") is False:
            win_daily, win_full = daily, full
    sr_arr = np.array([s for s in sr_all if np.isfinite(s)])
    print(f"  N trial finiti          : {len(sr_arr)}")
    print(f"  Sharpe winner (50/50)   : {win_full:+.3f}")
    print(f"  Sharpe trial: mean {sr_arr.mean():+.2f}  std {sr_arr.std():.2f}  "
          f"max {sr_arr.max():+.2f}  >0: {int((sr_arr > 0).sum())}/{len(sr_arr)}")
    dsr = None
    for label, pool in (("TUTTI 122", sr_all), ("no-TOD", sr_no_tod), ("solo-ERM", sr_erm)):
        d, sr0 = deflated_sharpe(win_full, pool, win_daily)
        n = int(np.isfinite(np.array(pool)).sum())
        print(f"  DSR [{label:>9} N={n:>3}]: {d:.3f}  (Sh-max null {sr0:+.2f})  -> "
              f"{'PASS' if d >= 0.95 else 'FAIL'}")
        if label == "TUTTI 122":
            dsr = d
    return dsr, win_full, None


# ===========================================================================
def part_B_insample_pick():
    print("\n" + "=" * 78)
    print("B) SELEZIONE IN-SAMPLE-ONLY (scelgo la cella ERM solo con Sharpe PRE-2025)")
    print("=" * 78)
    erm_grid = [dict(L_days=L, thr=t, long_flat=lf)
                for L in (1.0, 1.5, 2.0, 2.5, 3.0) for t in (0.30, 0.35, 0.40, 0.50)
                for lf in (False, True)]
    rows = []
    for tf in SCREEN_TFS:
        for p in erm_grid:
            try:
                _, full, ins = cand_full_is_sharpe(make_erm, tf, p)
            except Exception:
                continue
            rows.append((ins, full, tf, p))
    rows = [r for r in rows if np.isfinite(r[0])]
    rows.sort(key=lambda r: r[0], reverse=True)   # ordina per Sharpe IN-SAMPLE (no hold-out)
    print(f"  Top 5 celle per Sharpe IN-SAMPLE (<2025):")
    for ins, full, tf, p in rows[:5]:
        print(f"    IS {ins:+.2f}  FULL {full:+.2f}  tf={tf:>3}  {p}")
    ins, full, tf, p = rows[0]
    print(f"\n  -> cella scelta SENZA vedere l'hold-out: tf={tf} {p}")
    sm = al.study_marginal(f"ERM-ISpick {p}", make_erm(tf=tf, **p), tf=tf)
    m = sm["marginal"]
    print(f"     earns_slot={sm['earns_slot']}  marginal={m['marginal_verdict']}  "
          f"abs={sm['absolute']['verdict']['grade']}")
    print(f"     corr->TP01 {m['corr_full']}  has_insample_edge={m['has_insample_edge']}  "
          f"is_hedge={m['is_hedge']}  robust_oos={m['robust_oos']}")
    print(f"     blend w25: full uplift {m['blends']['w25']['uplift_full']:+.3f}  "
          f"hold uplift {m['blends']['w25']['uplift_hold']:+.3f}")
    same = (tf == WIN["tf"] and abs(p["L_days"] - WIN["L_days"]) < 1e-9
            and abs(p["thr"] - WIN["thr"]) < 1e-9 and p["long_flat"] is False)
    print(f"     coincide col vincitore max-hold? {same}")
    return sm["earns_slot"], (tf, p)


# ===========================================================================
def ensemble_target(tf, cells):
    """Media (equal-weight) dei pesi vol-targeted su piu' celle ERM -> un unico stream per asset.
    Ritorna un target_fn(df) che ricostruisce l'ensemble per quel df."""
    def fn(df):
        ws = [make_erm(tf=tf, **c)(df) for c in cells]
        return np.nanmean(np.vstack(ws), axis=0)
    return fn


def part_C_plateau_ensemble():
    print("\n" + "=" * 78)
    print("C) ENSEMBLE DEL PLATEAU ERM 8h (media celle L x thr, NIENTE cherry-pick)")
    print("=" * 78)
    cells = [dict(L_days=L, thr=t, long_flat=False)
             for L in (1.5, 2.0, 2.5, 3.0) for t in (0.30, 0.35, 0.40, 0.45, 0.50)]
    fn = ensemble_target("8h", cells)
    print(f"  celle nell'ensemble: {len(cells)} (L 1.5-3.0 x thr 0.30-0.50, lf=False)")
    caus = al.causality_ok(fn, tf="8h")
    print(f"  causale: ok={caus['ok']} max_tail_diff={caus['max_tail_diff']}")
    sm = al.study_marginal("ERM-plateau-ens", fn, tf="8h")
    m = sm["marginal"]
    print(f"  earns_slot={sm['earns_slot']}  marginal={m['marginal_verdict']}  "
          f"abs={sm['absolute']['verdict']['grade']}")
    print(f"  standalone full {m['cand_full_sharpe']}  hold {m['cand_hold_sharpe']}  "
          f"in-sample {m.get('cand_insample_sharpe')}")
    print(f"  corr->TP01 {m['corr_full']}  has_insample_edge={m['has_insample_edge']}  "
          f"is_hedge={m['is_hedge']}  robust_oos={m['robust_oos']}")
    print(f"  blend w25: full {m['blends']['w25']['full']} (uplift "
          f"{m['blends']['w25']['uplift_full']:+.3f})  hold {m['blends']['w25']['hold']} "
          f"(uplift {m['blends']['w25']['uplift_hold']:+.3f})")
    print(f"  multicut: {m['multicut_uplift']}")
    return sm["earns_slot"]


# ===========================================================================
def part_D_where_edge():
    print("\n" + "=" * 78)
    print("D) DOVE VIVE L'EDGE — per-anno standalone + uplift 3-way vs 2-way + corr(ERM,SKH)")
    print("=" * 78)
    fn = make_erm(**WIN)
    cand = al.candidate_daily(fn, tf=WIN["tf"])
    tp = al.tp01_baseline_daily()
    skh = _skh_daily()
    J = pd.concat({"T": tp, "S": skh, "C": cand}, axis=1, join="inner").dropna()
    two = 0.75 * J["T"] + 0.25 * J["S"]
    three = 0.60 * J["T"] + 0.25 * J["S"] + 0.15 * J["C"]
    print(f"  {'anno':>5} {'ERM Sh':>7} {'TP+SKH':>7} {'+ERM':>7} {'Δuplift':>8} "
          f"{'corr(ERM,SKH)':>14} {'corr(ERM,TP)':>13}")
    for y in sorted(set(J.index.year)):
        sub = J[J.index.year == y]
        if len(sub) < 40:
            continue
        s2 = 0.75 * sub["T"] + 0.25 * sub["S"]
        s3 = 0.60 * sub["T"] + 0.25 * sub["S"] + 0.15 * sub["C"]
        print(f"  {y:>5} {al._sh(sub['C']):>+7.2f} {al._sh(s2):>+7.2f} {al._sh(s3):>+7.2f} "
              f"{al._sh(s3) - al._sh(s2):>+8.2f} {sub['C'].corr(sub['S']):>+14.2f} "
              f"{sub['C'].corr(sub['T']):>+13.2f}")
    print(f"  {'FULL':>5} {al._sh(J['C']):>+7.2f} {al._sh(two):>+7.2f} {al._sh(three):>+7.2f} "
          f"{al._sh(three) - al._sh(two):>+8.2f} {J['C'].corr(J['S']):>+14.2f} "
          f"{J['C'].corr(J['T']):>+13.2f}")
    JH = J[J.index >= HOLD]
    h2 = 0.75 * JH["T"] + 0.25 * JH["S"]
    h3 = 0.60 * JH["T"] + 0.25 * JH["S"] + 0.15 * JH["C"]
    print(f"  {'HOLD':>5} {al._sh(JH['C']):>+7.2f} {al._sh(h2):>+7.2f} {al._sh(h3):>+7.2f} "
          f"{al._sh(h3) - al._sh(h2):>+8.2f} {JH['C'].corr(JH['S']):>+14.2f} "
          f"{JH['C'].corr(JH['T']):>+13.2f}")


def part_E_codified_gate():
    """Validazione END-TO-END del gate appena codificato in altlib: study_family_honest sulla
    famiglia ERM deve dare earns_slot_honest=False (sceglie in-sample-only + deflaziona)."""
    print("\n" + "=" * 78)
    print("E) GATE CODIFICATO (al.study_family_honest) sulla famiglia ERM — deve bocciare")
    print("=" * 78)
    grid = [dict(L_days=L, thr=t, long_flat=lf)
            for L in (1.0, 1.5, 2.0, 2.5, 3.0) for t in (0.30, 0.35, 0.40, 0.50)
            for lf in (False, True)]
    rep = al.study_family_honest("ERM", make_erm, grid, SCREEN_TFS)
    ch = rep["chosen"]
    print(f"  n_celle={rep['n_cells']}  cella in-sample-best: tf={ch['tf']} {ch['params']}")
    print(f"  earns_slot_marginal={rep['earns_slot_marginal']}  "
          f"deflated_sharpe={rep['deflated_sharpe']} (dsr_pass={rep['dsr_pass']})")
    print(f"  => earns_slot_HONEST = {rep['earns_slot_honest']}  "
          f"(atteso False: slot bocciato dal gate)")
    return rep["earns_slot_honest"]


def main():
    print("ANALISI ROBUSTEZZA LEAD ERM (filone B) — read-only, nessun impatto live\n")
    dsr, win_full, sr0 = part_A_deflated()
    es_is, is_cell = part_B_insample_pick()
    es_ens = part_C_plateau_ensemble()
    part_D_where_edge()
    es_honest = part_E_codified_gate()
    print("\n" + "=" * 78)
    print("SINTESI ANALISI B")
    print("=" * 78)
    print(f"  A) deflated-Sharpe winner = {dsr:.3f}  ({'PASS' if dsr >= 0.95 else 'FAIL'} vs 0.95)")
    print(f"  B) cella scelta in-sample-only earns_slot = {es_is}  (cella {is_cell})")
    print(f"  C) ensemble del plateau earns_slot        = {es_ens}")
    print(f"  E) gate codificato earns_slot_honest      = {es_honest}")


if __name__ == "__main__":
    main()