PythagorasGoal/scripts/research/r0726_edge_death.py

#!/usr/bin/env python
"""r0726_edge_death.py — "come faccio a capire se l'edge e' morto?"

IL BUCO. Il progetto ha gate di kill PRE-REGISTRATI per i candidati (DVOLSPREAD kill 24/10,
XSR01 23/10, STATARB 27/09) e **nessuno per il book che gira con soldi veri**. TP01 e SKH01 non
hanno una condizione dichiarata sotto la quale si direbbe "ha smesso di funzionare": finora la
risposta implicita e' "si vedra'", che e' esattamente cio' che il progetto non accetta dai
candidati.

Stessa disciplina del tripwire di venue costruito oggi: **si tara sul NULLO** (edge intatto -> non
deve scattare) e **si valida su un CONTROLLO POSITIVO** (edge morto -> deve scattare, e si misura
quanto ci mette).

IL PROBLEMA STATISTICO. Con Sharpe ~1.7 e vol ~8%, un anno di risultati non distingue niente:
SE(Sharpe) su N anni ≈ sqrt((1+S²/2)/N), che a N=1 vale ~1.1 — piu' grande dell'edge stesso. Un
criterio di kill onesto deve quindi (a) accettare di essere LENTO, oppure (b) accettare falsi kill.
Qui si misura il compromesso invece di sceglierlo a occhio.

⚠️ E I DUE SLEEVE NON SI GIUDICANO ALLO STESSO MODO. Il leave-one-out del 26/07 ha misurato che
il contributo hold-out di TP01 e' **negativo nel 99.1% delle configurazioni d'ancora**: e' la firma
dell'assicurazione, che paga premio negli anni senza incendio. **Per uno sleeve difensivo "non ha
guadagnato" NON e' evidenza di morte** — la sua morte e' *non proteggere quando serve*. Due
criteri diversi, non uno.

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

import sys
from pathlib import Path

import numpy as np
import pandas as pd

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

import r0725_capcurve as CC        # noqa: E402

N_PATHS = 4000
BLOCK = 20
SEED = 20260726
YEARS = 10
# Finestre di valutazione in mesi. Sotto i 12 mesi non c'e' potenza, sopra i 36 il kill arriva
# quando il danno e' fatto: la griglia serve a mostrare proprio questo.
WINDOWS_M = (6, 12, 18, 24, 36)


def rolling_sharpe(eq_ret: np.ndarray, win_days: int) -> np.ndarray:
    """Sharpe annualizzato su finestra mobile, per ogni path (path x tempo)."""
    n_paths, n = eq_ret.shape
    out = np.full((n_paths, n), np.nan)
    c1 = np.cumsum(eq_ret, axis=1)
    c2 = np.cumsum(eq_ret ** 2, axis=1)
    for t in range(win_days, n):
        s1 = c1[:, t] - c1[:, t - win_days]
        s2 = c2[:, t] - c2[:, t - win_days]
        m = s1 / win_days
        v = np.maximum(s2 / win_days - m ** 2, 1e-18)
        out[:, t] = m / np.sqrt(v) * np.sqrt(365)
    return out


def min_rolling(paths: np.ndarray, win_days: int) -> np.ndarray:
    """Il PEGGIOR Sharpe rolling visto lungo ciascun path — la statistica che un criterio di kill
    guarda davvero (basta che scenda una volta sotto soglia per far scattare la regola)."""
    rs = rolling_sharpe(paths, win_days)
    return np.nanmin(rs, axis=1)


def main() -> None:
    print("=" * 100)
    print('  "come faccio a capire se l\'edge e\' morto?"')
    print("=" * 100)

    book = CC.book_series("hourly")
    r = book.values.astype(float)
    mu = float(r.mean())
    print(f"  book live TP01+SKH01, {len(r):,} giorni, Sharpe storico "
          f"{mu/r.std()*np.sqrt(365):.2f}, vol {r.std()*np.sqrt(365):.1%}")

    rng = np.random.default_rng(SEED)
    n_days = 365 * YEARS
    alive = CC._boot_paths(r, N_PATHS, n_days, BLOCK, rng)      # NULLO: edge intatto
    dead = alive - mu                                            # CONTROPROVA: drift a zero, stessa vol

    # ------------------------------------------------------------------ 1. potenza
    print("\n" + "=" * 100)
    print("  1. QUANTO SERVE PER DISTINGUERE — la statistica prima delle regole")
    print("=" * 100)
    print(f"\n  {'finestra':>9}{'Sharpe se VIVO (p5..p50..p95)':>36}{'se MORTO (p5..p50..p95)':>32}"
          f"{'sovrapposizione':>17}")
    for m in WINDOWS_M:
        w = int(m * 30.44)
        a = rolling_sharpe(alive, w)[:, -1]
        d = rolling_sharpe(dead, w)[:, -1]
        # quota di casi "morto" che stanno sopra il 5° pctl dei "vivo" = indistinguibili
        thr = np.nanpercentile(a, 5)
        overlap = float(np.nanmean(d > thr))
        print(f"  {m:>7}m  {np.nanpercentile(a,5):>10.2f}{np.nanpercentile(a,50):>11.2f}"
              f"{np.nanpercentile(a,95):>11.2f}   {np.nanpercentile(d,5):>10.2f}"
              f"{np.nanpercentile(d,50):>10.2f}{np.nanpercentile(d,95):>10.2f}{overlap:>16.1%}")
    print("\n    'sovrapposizione' = quota di casi MORTI che un test al 5% NON riconoscerebbe.")
    print("    A 6-12 mesi e' altissima: un anno brutto non e' informazione, e' rumore.")

    # ------------------------------------------------------------------ 2. falsi kill
    print("\n" + "=" * 100)
    print("  2. TARATURA SUL NULLO — quanto spesso un edge VIVO verrebbe ucciso per sbaglio")
    print("=" * 100)
    print(f"\n  Regola: 'Sharpe rolling a N mesi scende sotto S almeno una volta in {YEARS} anni'")
    print(f"\n  {'finestra':>9}" + "".join(f"{f'S<{s:+.1f}':>12}" for s in (-0.5, 0.0, 0.5)))
    fp = {}
    for m in WINDOWS_M:
        w = int(m * 30.44)
        worst = min_rolling(alive, w)
        cells = []
        for s in (-0.5, 0.0, 0.5):
            fp[(m, s)] = float(np.mean(worst < s))
            cells.append(f"{fp[(m,s)]:>11.1%}")
        print(f"  {m:>7}m " + "".join(cells))
    print(f"\n    Questi sono FALSI KILL: l'edge c'e' e la regola lo spegne. Una regola a")
    print(f"    6 mesi e' inutilizzabile; a 24-36 mesi diventa gestibile.")

    # ------------------------------------------------------------------ 3. controllo positivo
    print("\n" + "=" * 100)
    print("  3. CONTROLLO POSITIVO — se l'edge E' morto, quanto ci mette la regola ad accorgersene")
    print("=" * 100)
    print(f"\n  {'finestra':>9}{'soglia':>9}{'falso kill':>13}{'lo prende':>12}"
          f"{'mediana rilevamento':>22}")
    best = None
    for m in WINDOWS_M:
        w = int(m * 30.44)
        rs_dead = rolling_sharpe(dead, w)
        for s in (-0.5, 0.0, 0.5):
            below = rs_dead < s
            first = np.where(below.any(axis=1), below.argmax(axis=1), -1)
            det = first[first >= 0] / 365.0
            power = float((first >= 0).mean())
            med = float(np.median(det)) if len(det) else np.nan
            if fp[(m, s)] <= 0.05 and power >= 0.90:
                if best is None or med < best[3]:
                    best = (m, s, power, med)
            print(f"  {m:>7}m{s:>9.1f}{fp[(m,s)]:>12.1%}{power:>11.1%}"
                  f"{med:>19.1f} anni" if not np.isnan(med) else
                  f"  {m:>7}m{s:>9.1f}{fp[(m,s)]:>12.1%}{power:>11.1%}{'mai':>22}")

    if best:
        m, s, power, med = best
        print(f"\n  --> REGOLA SCELTA: Sharpe rolling a {m} mesi sotto {s:+.1f}")
        print(f"      falso kill {fp[(m,s)]:.1%} in {YEARS} anni · lo riconosce nel {power:.0%} dei casi"
              f" · rilevamento mediano {med:.1f} anni")
    else:
        print(f"\n  ✋ Nessuna coppia soddisfa insieme falso-kill<=5% e potenza>=90%.")

    # ------------------------------------------------------------------ 4. lo sleeve difensivo
    print("\n" + "=" * 100)
    print("  4. ⚠️ TP01 NON SI GIUDICA COSI' — uno sleeve difensivo si valuta sul SINISTRO")
    print("=" * 100)
    from src.portfolio.sleeves import tp01_sleeve

    tp = tp01_sleeve(weight=1.0).daily()
    from src.data.downloader import load_data
    from src.strategies.trend_portfolio import resample_1d, simple_returns
    bh = {}
    for a in ("BTC", "ETH"):
        df = resample_1d(load_data(a, "1h"))
        bh[a] = pd.Series(simple_returns(df["close"].values.astype(float)),
                          index=pd.to_datetime(df["datetime"]))
    J = pd.concat(bh, axis=1, join="inner").fillna(0.0)
    hold = pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index)
    common = tp.index.intersection(hold.index)
    tp, hold = tp.loc[common], hold.loc[common]

    def dd(s):
        e = (1 + s).cumprod()
        return float((1 - e / e.cummax()).max())

    print(f"\n  Il LOO del 26/07 ha misurato: contributo hold-out di TP01 NEGATIVO nel 99.1% delle")
    print(f"  configurazioni d'ancora. E' la firma dell'assicurazione — paga premio negli anni")
    print(f"  senza incendio. Quindi 'non ha guadagnato' NON e' evidenza di morte.")
    print(f"\n  Cio' che TP01 deve fare, per anno (maxDD suo vs buy&hold 50/50):")
    print(f"\n  {'anno':>6}{'DD buy&hold':>14}{'DD TP01':>11}{'protezione':>13}")
    fails = 0
    yrs_tested = 0
    for y, g in hold.groupby(hold.index.year):
        t = tp.loc[g.index]
        d_bh, d_tp = dd(g), dd(t)
        if d_bh < 0.10:
            continue                      # anno senza sinistro: non dice niente sull'assicurazione
        yrs_tested += 1
        ratio = d_tp / d_bh if d_bh > 0 else np.nan
        bad = ratio > 0.75
        fails += bad
        print(f"  {y:>6}{d_bh:>13.1%}{d_tp:>10.1%}{1/ratio:>11.1f}x"
              f"{'   <- NON HA PROTETTO' if bad else ''}")
    print(f"\n    Anni con sinistro (DD buy&hold > 10%): {yrs_tested}. Fallimenti: {fails}.")
    print(f"\n  --> CRITERIO PER TP01 (difensivo): non 'quanto rende' ma **in un anno con DD")
    print(f"      buy&hold > 10%, il DD di TP01 deve restare sotto il 75% di quello**. Due anni")
    print(f"      di sinistro consecutivi falliti = l'assicurazione non assicura piu'.")
    print(f"      NB: negli anni senza sinistro il criterio NON si valuta — non c'e' informazione.")


if __name__ == "__main__":
    main()