PythagorasGoal/scripts/analysis/xs01_tranche_gate.py

"""GATE — XS01 phase-tranching (K sub-book sfasati di hold/K, capitale comune).

`xs01_tranche_research.py` ha misurato timing-luck di fase reale (FULL Sharpe
1.87-2.87, DD 13.8-33.1% a seconda della fase di partenza, che live e' arbitraria).
Qui il confronto ONESTO su equity GIORNALIERA (stessa convenzione dei gate del
progetto: combine_portfolio.metrics) + gate PORT06 swap-sleeve:
  [1] standalone: base (fase canonica) vs K=2 vs K=3, FULL e OOS, daily Sharpe/DD;
      in piu' il range delle 12 fasi singole (il rischio che il tranching elimina).
  [2] PORT06: members canonici con XS01 sostituito dalla variante tranched.
Criterio (tutti): OOS Sharpe portafoglio non peggiora (>-0.02) E DD non sale;
K=2 e K=3 devono essere ENTRAMBI >= base (plateau, non best-pick di K).

    uv run python scripts/analysis/xs01_tranche_gate.py
"""
from __future__ import annotations

import sys
from pathlib import Path

import numpy as np
import pandas as pd

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

from scripts.analysis.combine_portfolio import port_returns, metrics, SPLIT, OOS_DATE
from scripts.analysis.report_families import daily_from
from scripts.analysis.xs01_tranche_research import xsec_trades
from scripts.strategies.XS01_cross_sectional import aligned_panel, HOLD, POS, LEV
from scripts.portfolios._defs import PORTFOLIOS
from src.portfolio.sleeves import all_sleeve_equities
from src.portfolio import weighting as W


def daily_equity_for(phases, M, ts):
    """Trade di tutte le fasi su capitale comune (peso 1/K) -> equity daily."""
    K = len(phases)
    allt = sorted([t for ph in phases for t in xsec_trades(phase=ph, M=M)],
                  key=lambda t: t[1])
    cap = 1000.0
    eq_ts, eq_v = [], []
    for i, j, net in allt:
        cap = max(cap + cap * POS * LEV * net / K, 10.0)
        eq_ts.append(ts[j])
        eq_v.append(cap)
    return daily_from(eq_ts, eq_v)


def port_metrics(members, ids, clusters, caps):
    dr = pd.DataFrame({i: members[i].pct_change().fillna(0.0) for i in ids})
    w = W.weight_vector("cap", ids, dr, caps=caps, clusters=clusters)
    drp = port_returns({i: members[i] for i in ids}, w)
    return metrics(drp), metrics(drp, lo=SPLIT)


def main():
    M = aligned_panel()
    ts = pd.to_datetime(M.index, unit="ms", utc=True)
    p = PORTFOLIOS["PORT06"]
    print("=" * 92)
    print("  GATE — XS01 phase-tranching | equity daily, OOS da", OOS_DATE)
    print("=" * 92)

    variants = {"base (fase 0)": [0],
                "K=2 (fasi 0,6)": [0, 6],
                "K=3 (fasi 0,4,8)": [0, 4, 8]}
    eqs = {k: daily_equity_for(v, M, ts) for k, v in variants.items()}

    print(f"\n  [1] STANDALONE daily — {'config':<18}{'FULL Sh':>8}{'FULL DD%':>9}{'OOS Sh':>8}{'OOS DD%':>8}")
    for k, e in eqs.items():
        r = e.pct_change().fillna(0.0)
        f, o = metrics(r), metrics(r, lo=SPLIT)
        print(f"      {k:<22}{f['sharpe']:>8.2f}{f['dd']:>9.2f}{o['sharpe']:>8.2f}{o['dd']:>8.2f}")
    # range delle 12 fasi singole (daily): il rischio di fase che il tranching elimina
    fs, os_ = [], []
    for ph in range(HOLD):
        r = daily_equity_for([ph], M, ts).pct_change().fillna(0.0)
        fs.append(metrics(r)["sharpe"])
        os_.append(metrics(r, lo=SPLIT)["sharpe"])
    print(f"      12 fasi singole: FULL Sh {min(fs):.2f}-{max(fs):.2f} | OOS Sh {min(os_):.2f}-{max(os_):.2f}")

    eq_base = dict(all_sleeve_equities())
    ids, cl, caps = list(p.sleeve_ids), p.clusters, p.caps
    print(f"\n  [2] PORT06 swap-sleeve — {'config':<18}{'FULL Sh':>8}{'FULL DD%':>9}{'OOS Sh':>8}{'OOS DD%':>8}")
    f0, o0 = port_metrics(eq_base, ids, cl, caps)
    print(f"      {'ATTUALE (base)':<22}{f0['sharpe']:>8.2f}{f0['dd']:>9.2f}{o0['sharpe']:>8.2f}{o0['dd']:>8.2f}")
    verdicts = []
    for k in ("K=2 (fasi 0,6)", "K=3 (fasi 0,4,8)"):
        mem = dict(eq_base)
        mem["XS01"] = eqs[k]
        f1, o1 = port_metrics(mem, ids, cl, caps)
        ok = (o1["sharpe"] >= o0["sharpe"] - 0.02 and o1["dd"] <= o0["dd"] + 1e-9
              and f1["sharpe"] >= f0["sharpe"] - 0.02 and f1["dd"] <= f0["dd"] + 1e-9)
        verdicts.append(ok)
        print(f"      {k:<22}{f1['sharpe']:>8.2f}{f1['dd']:>9.2f}{o1['sharpe']:>8.2f}{o1['dd']:>8.2f}"
              f"  {'OK' if ok else 'NO'}")
    print(f"\n  => {'PROMOSSO (plateau K=2 e K=3)' if all(verdicts) else 'NON promosso (serve plateau su entrambi i K)'}")


if __name__ == "__main__":
    main()