PythagorasGoal/scripts/analysis/xs01_tranche_research.py

"""XS01 phase-tranching — PRE-REGISTRATO: il roll non-sovrapposto di XS01 (entry a i,
exit a i+hold, re-entry) ha una FASE arbitraria (dipende dal primo indice valido).
Se l'esito dipende dalla fase, c'e' timing-luck: dividere il book in K sub-book
sfasati di hold/K barre (capitale 1/K ciascuno) e' un ensemble di fase che riduce
la varianza SENZA parametri fittati (K=2 e K=3 riportati entrambi, nessun best-pick).

Test (griglia fissata qui, completa):
  [1] SENSIBILITA' DI FASE: xsec_sim base con offset di partenza 0..hold-1
      -> dispersione di Sharpe/ret/DD (FULL e OOS). Se la dispersione e' piccola,
      il tranching non serve (verdetto onesto, fine).
  [2] TRANCHING: K in {2, 3} sub-book sfasati equal-capital -> Sharpe/DD/ret
      FULL e OOS vs la MEDIA e il WORST delle fasi singole.
Criterio (tutti necessari): il tranched riduce la dispersione (per costruzione) e
il suo Sharpe OOS >= worst-fase OOS con DD <= mediana fasi; fee identiche (il
tranching NON cambia il turnover per unita' di capitale).

    uv run python scripts/analysis/xs01_tranche_research.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.strategies.XS01_cross_sectional import (
    aligned_panel, UNIVERSE, LB, HOLD, FEE_RT, LEV, POS, OOS_FRAC)


def xsec_trades(phase: int = 0, lb: int = LB, hold: int = HOLD, fee_rt: float = FEE_RT,
                split_frac: float = 0.0, M=None):
    """Replica ESATTA della logica di xsec_sim ma parte da max(lb, split)+phase e
    ritorna la lista trade (i, j, net) — stessa matematica, fee = 2*fee_rt."""
    C = M[UNIVERSE].values
    n = len(C)
    logC = np.log(C)
    split = int(n * split_frac)
    fee = 2 * fee_rt
    out = []
    last = -1
    i = max(lb, split) + phase
    while i < n - hold:
        if i <= last:
            i += 1
            continue
        dm = logC[i] - logC[i - lb]
        dm = dm - dm.mean()
        gw = np.sum(np.abs(dm))
        if gw < 1e-9:
            i += 1
            continue
        w = -dm / gw
        book = float(np.sum(w * (logC[i + hold] - logC[i])))
        out.append((i, i + hold, book - fee))
        last = i + hold
        i += 1
    return out


def equity_from(trades, ts, pos=POS, lev=LEV, weight=1.0):
    """Equity compounding a punti-exit (convenzione xsec_sim), peso per tranche."""
    cap = 1000.0
    eq_ts, eq_v = [], []
    for i, j, net in sorted(trades, key=lambda t: t[1]):
        cap = max(cap + cap * pos * lev * net * weight, 10.0)
        eq_ts.append(ts[j])
        eq_v.append(cap)
    return pd.Series(eq_v, index=pd.DatetimeIndex(eq_ts))


def metrics(trades, ts, yrs_span):
    rets = [t[2] * POS for t in trades]
    n = len(rets)
    sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(n / yrs_span)) if n > 1 and np.std(rets) > 0 else 0.0
    eq = equity_from(trades, ts)
    pk = eq.cummax()
    dd = float(((pk - eq) / pk).max() * 100) if len(eq) else 0.0
    ret = float((eq.iloc[-1] / 1000 - 1) * 100) if len(eq) else 0.0
    return dict(n=n, ret=ret, sharpe=sharpe, dd=dd)


def combined_metrics(branches, ts, yrs_span):
    """K tranche -> un'unica equity: pesa ogni trade 1/K sul capitale comune."""
    K = len(branches)
    allt = sorted([t for b in branches for t in b], key=lambda t: t[1])
    cap = 1000.0
    eq_ts, eq_v, rets = [], [], []
    for i, j, net in allt:
        rets.append(net * POS / K)
        cap = max(cap + cap * POS * LEV * net / K, 10.0)
        eq_ts.append(ts[j])
        eq_v.append(cap)
    eq = pd.Series(eq_v, index=pd.DatetimeIndex(eq_ts))
    pk = eq.cummax()
    n = len(rets)
    sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(n / yrs_span)) if n > 1 and np.std(rets) > 0 else 0.0
    return dict(n=n, ret=float((eq.iloc[-1] / 1000 - 1) * 100),
                sharpe=sharpe, dd=float(((pk - eq) / pk).max() * 100))


def regression_lock(M=None):
    """xsec_trades(phase=0) DEVE riprodurre xsec_sim (n trade e ret totale):
    la copia e' load-bearing (la usano gate e validate_xsec_worker) — senza
    lock eseguito un fix all'engine canonico non si propagherebbe e la
    validazione live certificherebbe un engine stantio (code-review 2026-06-11)."""
    from scripts.strategies.XS01_cross_sectional import xsec_sim
    if M is None:
        M = aligned_panel()
    ts = pd.to_datetime(M.index, unit="ms", utc=True)
    tr = xsec_trades(phase=0, M=M)
    eq = equity_from(tr, ts)
    ref = xsec_sim()
    n_ok = len(tr) == ref["trades"]
    ret = (eq.iloc[-1] / 1000 - 1) * 100 if len(eq) else 0.0
    ret_ok = abs(ret - ref["ret"]) < 1e-6 * max(1.0, abs(ref["ret"]))
    if not (n_ok and ret_ok):
        raise AssertionError(f"REGRESSION-LOCK FALLITO: trades {len(tr)} vs {ref['trades']}, "
                             f"ret {ret:.6f} vs {ref['ret']:.6f}")
    print(f"  regression-lock OK: {len(tr)} trade, ret {ret:+.2f}% == xsec_sim")


def run():
    M = aligned_panel()
    ts = pd.to_datetime(M.index, unit="ms", utc=True)
    n = len(M)
    print("=" * 96)
    print("  XS01 phase-tranching — sensibilita' di fase + ensemble K=2/3 | griglia pre-registrata")
    print("=" * 96)
    regression_lock(M)
    for tag, split in (("FULL", 0.0), ("OOS", 1 - OOS_FRAC)):
        yrs_span = (ts[-1] - ts[max(int(n * split), LB)]).days / 365.25 or 1
        rows = []
        for ph in range(HOLD):
            tr = xsec_trades(phase=ph, split_frac=split, M=M)
            rows.append(metrics(tr, ts, yrs_span))
        sh = np.array([r["sharpe"] for r in rows])
        dd = np.array([r["dd"] for r in rows])
        rt = np.array([r["ret"] for r in rows])
        print(f"\n  [{tag}] sensibilita' di fase (12 fasi):")
        print(f"    Sharpe: min {sh.min():.2f} | med {np.median(sh):.2f} | max {sh.max():.2f} | std {sh.std():.2f}")
        print(f"    ret%:   min {rt.min():+.0f} | med {np.median(rt):+.0f} | max {rt.max():+.0f}")
        print(f"    DD%:    min {dd.min():.1f} | med {np.median(dd):.1f} | max {dd.max():.1f}")
        for K in (2, 3):
            branches = [xsec_trades(phase=int(p), split_frac=split, M=M)
                        for p in np.linspace(0, HOLD, K, endpoint=False)]
            m = combined_metrics(branches, ts, yrs_span)
            print(f"    K={K} tranched: Sharpe {m['sharpe']:.2f} | ret {m['ret']:+.0f}% | "
                  f"DD {m['dd']:.1f}% | trade {m['n']}")
    print("\n  criterio: tranched-Sharpe OOS >= worst-fase OOS e DD <= mediana fasi; "
          "se la dispersione di fase e' gia' piccola -> NON SERVE (verdetto onesto).")


if __name__ == "__main__":
    run()