PythagorasGoal/scripts/research/r0725_statarb_basket_gate.py

"""STATARB-BASKET — il paniere multi-coppia e' uno SLEEVE? Giudizio coi gate veri del progetto.

DA DOVE VIENE. `r0725_statarb_multi.py` ha applicato il meccanismo CONGELATO (W=45, sgn=+1, preso
da un altro studio: qui non e' stato cercato nulla) alle 50 coppie alt/BTC di Hyperliquid e ha
trovato un paniere EW con Sharpe 0.82, maxDD -6.0%, corr a XS01 solo 0.207, 0/50 coppie degeneri.
Quello script pero' si fermava alla statistica descrittiva. Questo lo porta davanti ai gate che il
progetto usa per ammettere uno sleeve:

    study/marginal_vs_tp01  -> ADDS / HEDGE / NOISE / REDUNDANT / DILUTES / NEUTRAL
                               (include multi-cut, noise-null a corr-zero, hedge-vs-alpha)
    deflated_sharpe         -> PASS >= 0.95, coi trial DAVVERO fatti
    weights_tilt_null       -> ogni proposta di peso vs il null dei tilt casuali
    corr per-sleeve         -> ridondanza contro i 5 sleeve attivi

PRECEDENTE CHE RENDE LA DOMANDA LEGITTIMA: XS01 e' nel book al 15% pur essendo STAT-MODE (19 gambe,
non eseguibili a $600). Un paniere di coppie a 51 gambe sarebbe ammissibile alle STESSE condizioni,
se supera i gate. Non e' quindi l'eseguibilita' a decidere qui, e' l'edge.

VARIANTI PRE-REGISTRATE (3, contate nel deflated-Sharpe; nessuna scelta guardando i risultati):
  V1 ALL50   — tutte le coppie alt/BTC valutabili. Zero liberta' di selezione.
  V2 MAJ19   — solo i 19 major liquidi, cioe' l'universo XS_UNIVERSE definito il 2026-06-19 per
               ragioni di LIQUIDITA' in un altro studio: sottoinsieme a priori, non scelto oggi.
  V3 DEMEAN  — ALL50 con le posizioni demeanate cross-sezionalmente ogni giorno. Motivo strutturale
               dichiarato prima di guardare: le 50 coppie condividono la gamba BTC, per questo
               l'ampiezza effettiva era 4.5 invece di 50; togliere la componente comune dovrebbe
               alzarla. E' una costruzione standard (neutralizzare il fattore comune), non un fit.

    uv run python scripts/research/r0725_statarb_basket_gate.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"))
sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt"))

from r0725_statarb_multi import (BASE, BLOCK, MIN_BARS, SEED, load_hl, pnl, signal, universe,
                                 _dd, _sh)

ANN = np.sqrt(365.0)


def pair_frames() -> tuple[pd.DataFrame, pd.DataFrame]:
    """Matrici [data x coppia] di posizione e ritorno-spread, per tutte le coppie valutabili."""
    base_px = load_hl(BASE)
    pos_cols, spr_cols = {}, {}
    for sym in universe():
        try:
            tgt = load_hl(sym)
        except FileNotFoundError:
            continue
        ix = base_px.index.intersection(tgt.index)
        if len(ix) < MIN_BARS:
            continue
        p, s = signal(base_px[ix], tgt[ix])
        pos_cols[sym] = pd.Series(p, index=ix)
        spr_cols[sym] = pd.Series(s, index=ix)
    P = pd.concat(pos_cols, axis=1, sort=True).sort_index()
    S = pd.concat(spr_cols, axis=1, sort=True).sort_index()
    return P, S


def basket_from_positions(P: pd.DataFrame, S: pd.DataFrame, cols=None,
                          demean: bool = False) -> pd.Series:
    """Rendimento EW del paniere, ricalcolando fee su ogni gamba (2 per coppia)."""
    if cols is not None:
        cols = [c for c in cols if c in P.columns]
        P, S = P[cols], S[cols]
    if demean:
        P = P.sub(P.mean(axis=1), axis=0)
    out = {}
    for c in P.columns:
        p = P[c].to_numpy(float)
        s = S[c].to_numpy(float)
        out[c] = pd.Series(pnl(p, s), index=P.index)
    return pd.concat(out, axis=1, sort=True).mean(axis=1, skipna=True).dropna()


def eff_breadth(P: pd.DataFrame, S: pd.DataFrame, cols=None, demean=False) -> float:
    if cols is not None:
        cols = [c for c in cols if c in P.columns]
        P, S = P[cols], S[cols]
    if demean:
        P = P.sub(P.mean(axis=1), axis=0)
    R = {}
    for c in P.columns:
        R[c] = pd.Series(pnl(P[c].to_numpy(float), S[c].to_numpy(float)), index=P.index)
    M = pd.concat(R, axis=1, sort=True)
    C = M.corr().values
    off = C[~np.eye(len(C), dtype=bool)]
    rbar = float(np.nanmean(off))
    return len(C) / (1.0 + (len(C) - 1) * rbar) if rbar > -1 / (len(C) - 1) else float(len(C))


def main() -> None:
    print("=" * 100)
    print("  STATARB-BASKET — il paniere multi-coppia supera i gate di ammissione a sleeve?")
    print("=" * 100)

    from src.portfolio.sleeves import XS_UNIVERSE

    P, S = pair_frames()
    print(f"\n  coppie valutabili: {P.shape[1]}   barre: {len(P)}   "
          f"da {P.index[0].date()} a {P.index[-1].date()}")

    maj = [s for s in XS_UNIVERSE if s != BASE]
    variants = {
        "V1 ALL50":  dict(cols=None, demean=False),
        "V2 MAJ19":  dict(cols=maj, demean=False),
        "V3 DEMEAN": dict(cols=None, demean=True),
    }
    series, srs = {}, []
    print("\n" + "-" * 100)
    print("  (1) LE TRE VARIANTI PRE-REGISTRATE")
    print("-" * 100)
    print(f"  {'variante':<12}{'gambe':>7}{'Sharpe':>9}{'maxDD':>9}{'ret tot':>10}{'ampiezza eff':>15}")
    for nm, kw in variants.items():
        r = basket_from_positions(P, S, **kw)
        series[nm] = r
        srs.append(_sh(r.values))
        nlegs = (P.shape[1] if kw["cols"] is None else len([c for c in kw["cols"] if c in P.columns]))
        print(f"  {nm:<12}{nlegs:>7}{_sh(r.values):>9.2f}{_dd(r.values)*100:>8.1f}%"
              f"{(np.prod(1+r.values)-1)*100:>9.1f}%{eff_breadth(P, S, **kw):>15.1f}")

    # ---------------- gate del progetto su ciascuna variante
    from altlib import deflated_sharpe, marginal_vs_tp01

    print("\n" + "-" * 100)
    print("  (2) GATE MARGINALE vs TP01 (il gate che decide, non lo Sharpe assoluto)")
    print("-" * 100)
    reports = {}
    for nm, r in series.items():
        rr = r.copy()
        rr.index = pd.to_datetime(rr.index, utc=True)
        m = marginal_vs_tp01(rr)
        reports[nm] = m
        print(f"\n  --- {nm} ---")
        print(f"  verdetto      : {m.get('marginal_verdict')}")
        for k in ("corr", "robust_oos", "beats_noise_null", "is_hedge", "has_insample_edge",
                  "tp01_beta", "alpha_ann"):
            if k in m:
                v = m[k]
                print(f"  {k:<15}: {v if not isinstance(v, float) else round(v, 4)}")
        bl = m.get("blends", {})
        for w, d in (bl.items() if isinstance(bl, dict) else []):
            if isinstance(d, dict):
                print(f"  blend w={w}: " + "  ".join(
                    f"{k}={round(v,3) if isinstance(v,(int,float)) else v}" for k, v in d.items()))

    # ---------------- deflated Sharpe coi trial veri
    print("\n" + "-" * 100)
    print("  (3) DEFLATED SHARPE — trial reali = 3 varianti (la config W=45/sgn=+1 viene da un")
    print("      altro studio: su QUESTI dati non e' stata cercata)")
    print("-" * 100)
    for nm, r in series.items():
        dsr, null_max = deflated_sharpe(_sh(r.values), srs, r.values, dpy=365.0)
        print(f"  {nm:<12} Sharpe {_sh(r.values):>5.2f}   DSR {dsr:>6.3f}   "
              f"(max atteso sotto il null {null_max:>5.2f})   {'PASS' if dsr >= 0.95 else 'sotto 0.95'}")

    # ---------------- correlazione ai 5 sleeve attivi + tilt-null
    print("\n" + "-" * 100)
    print("  (4) RIDONDANZA vs I 5 SLEEVE ATTIVI  +  weights_tilt_null sul peso proposto")
    print("-" * 100)
    from src.portfolio.portfolio import Sleeve, StrategyPortfolio, weights_tilt_null
    from src.portfolio.sleeves import active_sleeves

    sl = active_sleeves()
    best = max(series.items(), key=lambda kv: _sh(kv[1].values))
    nm_best, r_best = best
    rb = r_best.copy()
    rb.index = pd.to_datetime(rb.index, utc=True)
    print(f"  variante col miglior Sharpe: {nm_best}")
    for s in sl:
        j = pd.concat({"a": s.daily(), "b": rb}, axis=1, sort=True).dropna()
        if len(j) > 60:
            print(f"    corr -> {s.name:<20} {j['a'].corr(j['b']):>6.3f}  ({len(j)} giorni comuni)")

    daily_cols = {s.name: s.daily() for s in sl}
    daily_cols["STATARB_BASKET"] = rb
    w_cur = {s.name: s.weight for s in sl}
    w_cur["STATARB_BASKET"] = 0.0
    for w_new in (0.10, 0.15):
        k = 1.0 - w_new
        w_prop = {s.name: s.weight * k for s in sl}
        w_prop["STATARB_BASKET"] = w_new
        try:
            res = weights_tilt_null(daily_cols, w_cur, w_prop)
            print(f"\n  peso {w_new:.0%} (i 5 scalati x{k:.2f}):")
            for kk, vv in res.items():
                if isinstance(vv, (int, float, bool, str)):
                    print(f"    {kk:<22} {round(vv,4) if isinstance(vv,float) else vv}")
        except Exception as e:
            print(f"  [tilt-null non calcolabile a w={w_new}: {e.__class__.__name__}: {e}]")

    # ---------------- book con/senza
    print("\n" + "-" * 100)
    print("  (5) BOOK CON E SENZA — Sharpe/DD FULL e HOLD-OUT")
    print("-" * 100)
    b0 = StrategyPortfolio(sl).backtest()
    for w_new in (0.10, 0.15):
        k = 1.0 - w_new
        sl2 = [Sleeve(s.name, s.weight * k, s.daily_fn, s.pos_fn) for s in sl]
        sl2.append(Sleeve("STATARB_BASKET", w_new, lambda _r=rb: _r))
        b1 = StrategyPortfolio(sl2).backtest()
        print(f"  w={w_new:.0%}   FULL {b0['full']['sharpe']:.2f} -> {b1['full']['sharpe']:.2f}   "
              f"HOLD {b0['holdout']['sharpe']:.2f} -> {b1['holdout']['sharpe']:.2f}   "
              f"DD {b0['full']['maxdd']*100:.1f}% -> {b1['full']['maxdd']*100:.1f}%")

    print("\n" + "=" * 100)


if __name__ == "__main__":
    main()