PythagorasGoal/scripts/research/r0725_mat01_multiasset_trend.py

"""MAT01 — il trend difensivo GTAA su un universo MULTI-ASSET-CLASS (18 ETF) invece di 6.

DOMANDA (una sola, pre-registrata): il valore dello sleeve GTAA01 aumenta allargando l'universo
dalle 6 gambe attuali (SPY QQQ IWM TLT GLD HYG — tutte US, 4 su 6 azionarie) a un paniere che
copre le classi d'attivo vere (azioni US / internazionali / emergenti, bond governativi / credito
IG / HY, commodity, real estate)?

PERCHE' E' UN FILONE NUOVO E NON UN RE-TEST: il time-series momentum multi-asset-class e'
l'anomalia meglio documentata e replicata della finanza quantitativa (Moskowitz-Ooi-Pedersen 2012;
repliche su 100+ anni), e il suo motore dichiarato NON e' la previsione ma la DIVERSIFICAZIONE fra
classi con drawdown decorrelati. In questo progetto il trend e' stato testato solo dentro una
classe alla volta (crypto = TP01/trend-multiasset; equity US = GTAA01; settori = XS equity). I 12
ETF non-US / bond / commodity sono su disco CERTIFICATI dal 2026-06-22 ma sono stati usati SOLO
per il test lead-lag crypto->X (diario 2026-06-23-crossmarket-beyond-sp500), mai per un programma
di trend. Questo e' il vuoto.

DISCIPLINA (la parte che conta):
  * Il MECCANISMO E' CONGELATO. Si importa `_exposure` da src/portfolio/gtaa.py senza toccarlo:
    orizzonti (21,63,126,252), vol-target 12%, long-flat, fee 2bps/lato, EW sugli asset. Cambia
    UNA SOLA COSA: l'insieme degli asset. Zero gradi di liberta' sui parametri => niente
    ri-ottimizzazione mascherata da "espansione".
  * L'universo ALL18 e' definito A PRIORI per classe d'attivo (vedi CLASSES), non scelto per
    performance. A k=18 la liberta' di selezione e' ESATTAMENTE ZERO (esiste un solo sottoinsieme
    di 18 elementi su 18) => il confronto ALL18 vs GTAA6 non e' selezionabile.
  * Il rischio vero non e' l'universo scelto, e' che l'INCUMBENT sia stato cherry-picked. Quindi
    si misura anche il percentile di GTAA6 fra i sottoinsiemi casuali di dimensione 6 (test D).
  * Split IS/OOS = quello su cui GTAA01 fu validato (IS <=2014-12-31, OOS 2015+), indipendente
    dall'hold-out crypto.

TEST ESEGUITI
  A) Standalone per asset (solo diagnostica, MAI selezione): Sharpe del trend gated asset per asset.
  B) ALL18 vs GTAA6: FULL / IS / OOS su Sharpe, CAGR, volAnn, maxDD, corr fra i due.
  C) CURVA DI AMPIEZZA (il test di meccanismo): per k=1..18, N sorteggi di sottoinsiemi casuali ->
     distribuzione dello Sharpe OOS. Se il beneficio e' AMPIEZZA e non selezione, la mediana deve
     salire con k in modo monotono, e ALL18 deve stare vicino alla mediana di k=18 (che e' se stesso).
  D) L'incumbent e' cherry-picked? percentile di GTAA6 nella distribuzione dei 6-subset casuali.
  E) Marginale sul BOOK: sostituire GTAA6 con MAT18 al peso invariato 20% nel book 5-sleeve, e
     misurare Sharpe/DD FULL e HOLD-OUT del book. Nessun cambio di peso => `weights_tilt_null`
     non si applica (non e' un tilt, e' una sostituzione a peso costante), ma il confronto e' 1:1.
  F) Eseguibilita' a capitale piccolo: ticket minimo per gamba ai pesi correnti.

    uv run python scripts/research/r0725_mat01_multiasset_trend.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))

from src.portfolio.gtaa import _close, _exposure, EQ_UNIVERSE, FEE_SIDE  # meccanismo CONGELATO

ANN = np.sqrt(252.0)
OOS_START = pd.Timestamp("2015-01-01", tz="UTC")
SEED = 20260725
N_DRAWS = 400

# --- universo A PRIORI, per classe d'attivo (nessuna scelta guidata dai risultati) ---------------
CLASSES = {
    "equity_US":     ["SPY", "QQQ", "IWM", "DIA"],
    "equity_intl":   ["EFA", "EWJ", "FXI"],
    "equity_EM":     ["EEM"],
    "bond_gov":      ["TLT", "IEF", "AGG"],
    "bond_credit":   ["LQD", "HYG"],
    "commodity":     ["GLD", "SLV", "DBC", "USO"],
    "real_estate":   ["VNQ"],
}
ALL18 = [s for v in CLASSES.values() for s in v]
GTAA6 = list(EQ_UNIVERSE)


# ---------------------------------------------------------------------------- metriche
def _sh(r: pd.Series) -> float:
    r = r.dropna()
    return float(r.mean() / r.std() * ANN) if len(r) > 20 and r.std() > 0 else 0.0


def _dd(r: pd.Series) -> float:
    eq = (1.0 + r.fillna(0.0)).cumprod()
    return float((eq / eq.cummax() - 1.0).min())


def _cagr(r: pd.Series) -> float:
    r = r.dropna()
    if len(r) < 50:
        return 0.0
    eq = float((1.0 + r).prod())
    yrs = len(r) / 252.0
    return float(eq ** (1.0 / yrs) - 1.0) if eq > 0 else -1.0


def stats(r: pd.Series, label: str) -> dict:
    return {
        "label": label, "n": int(r.dropna().shape[0]),
        "sharpe": _sh(r), "cagr": _cagr(r), "vol": float(r.std() * ANN), "maxdd": _dd(r),
    }


def fmt(s: dict) -> str:
    return (f"{s['label']:<22} n={s['n']:>5}  Sh {s['sharpe']:>6.2f}  CAGR {s['cagr']*100:>6.2f}%  "
            f"vol {s['vol']*100:>5.2f}%  maxDD {s['maxdd']*100:>6.2f}%")


# ---------------------------------------------------------------------------- motore (congelato)
_CACHE: dict[str, pd.Series] = {}


def gated(sym: str) -> pd.Series:
    """Rendimenti netti daily del trend long-flat su `sym` — IDENTICO a gtaa._gated_returns."""
    if sym in _CACHE:
        return _CACHE[sym]
    close = _close(sym)
    ex = _exposure(close)
    ret = close.pct_change().fillna(0.0).values
    held = np.zeros(len(ex))
    held[1:] = ex.values[:-1]                    # causale: esposizione decisa a i-1, tenuta in i
    net = held * ret - FEE_SIDE * np.abs(np.diff(held, prepend=0.0))
    out = pd.Series(net, index=close.index)
    _CACHE[sym] = out
    return out


def basket(universe, end: pd.Timestamp | None = None) -> pd.Series:
    """EW dei rendimenti trend-gated sugli asset disponibili (outer, skipna) — come gtaa_returns."""
    cols = {a: gated(a) for a in universe}
    r = pd.concat(cols, axis=1, sort=True).sort_index().mean(axis=1, skipna=True).dropna()
    return r[r.index <= end] if end is not None else r


def split(r: pd.Series):
    return r[r.index < OOS_START], r[r.index >= OOS_START]


# ---------------------------------------------------------------------------- main
def main() -> None:
    print("=" * 100)
    print("  MAT01 — trend difensivo su universo MULTI-ASSET-CLASS (18 ETF) vs GTAA6 attuale")
    print("  meccanismo CONGELATO da src/portfolio/gtaa.py (21/63/126/252, vt 12%, long-flat, EW)")
    print("=" * 100)

    # end comune: i 12 ETF non-GTAA sono rinfrescati solo on-demand -> tronco al min dei last date
    ends = {s: _close(s).index[-1] for s in ALL18}
    end = min(ends.values())
    print(f"\n  finestra comune troncata a {end.date()} "
          f"(i 12 ETF fuori-GTAA sono rinfrescati on-demand, non nel cron giornaliero)")
    print(f"  universo a priori ({len(ALL18)}):")
    for cls, syms in CLASSES.items():
        print(f"    {cls:<14} {', '.join(syms)}")

    # ---------------- A) standalone per asset (DIAGNOSTICA, non selezione)
    print("\n" + "-" * 100)
    print("  (A) TREND STANDALONE PER ASSET — diagnostica pura, nessuna selezione su questi numeri")
    print("-" * 100)
    print(f"  {'sym':<6}{'classe':<14}{'da':<12}{'Sh FULL':>9}{'Sh IS':>9}{'Sh OOS':>9}{'CAGR':>9}{'maxDD':>9}")
    cls_of = {s: c for c, v in CLASSES.items() for s in v}
    rows = []
    for s in ALL18:
        r = gated(s)
        r = r[r.index <= end]
        ins, oos = split(r)
        rows.append((s, _sh(r), _sh(ins), _sh(oos)))
        print(f"  {s:<6}{cls_of[s]:<14}{str(r.index[0].date()):<12}{_sh(r):>9.2f}{_sh(ins):>9.2f}"
              f"{_sh(oos):>9.2f}{_cagr(r)*100:>8.1f}%{_dd(r)*100:>8.1f}%")
    n_pos_oos = sum(1 for _, _, _, so in rows if so > 0)
    print(f"\n  asset con Sharpe OOS > 0: {n_pos_oos}/{len(rows)}  "
          f"(mediana OOS {np.median([so for *_ , so in rows]):.2f})")

    # ---------------- B) ALL18 vs GTAA6
    print("\n" + "-" * 100)
    print("  (B) ALL18 vs GTAA6 — stesso meccanismo, stesso periodo, cambia solo l'universo")
    print("-" * 100)
    r6, r18 = basket(GTAA6, end), basket(ALL18, end)
    common = r6.index.intersection(r18.index)
    r6, r18 = r6[common], r18[common]
    for nm, r in (("GTAA6 (incumbent)", r6), ("MAT18 (candidato)", r18)):
        i, o = split(r)
        print(f"  {fmt(stats(r, nm + ' FULL'))}")
        print(f"  {fmt(stats(i, nm + ' IS<2015'))}")
        print(f"  {fmt(stats(o, nm + ' OOS2015+'))}")
        print()
    i6, o6 = split(r6)
    i18, o18 = split(r18)
    print(f"  corr(GTAA6, MAT18) FULL {r6.corr(r18):>5.2f}   OOS {o6.corr(o18):>5.2f}")
    print(f"  DELTA OOS   Sharpe {_sh(o18)-_sh(o6):+.2f}   maxDD {(_dd(o18)-_dd(o6))*100:+.1f}pp   "
          f"CAGR {(_cagr(o18)-_cagr(o6))*100:+.2f}pp")
    print(f"  DELTA IS    Sharpe {_sh(i18)-_sh(i6):+.2f}   maxDD {(_dd(i18)-_dd(i6))*100:+.1f}pp")

    # per anno OOS
    print("\n  Sharpe per anno (OOS 2015+):")
    yrs = sorted(set(o6.index.year))
    print("    anno  " + "".join(f"{y:>7}" for y in yrs))
    print("    GTAA6 " + "".join(f"{_sh(o6[o6.index.year == y]):>7.2f}" for y in yrs))
    print("    MAT18 " + "".join(f"{_sh(o18[o18.index.year == y]):>7.2f}" for y in yrs))
    wins = sum(1 for y in yrs if _sh(o18[o18.index.year == y]) > _sh(o6[o6.index.year == y]))
    print(f"    -> MAT18 batte GTAA6 in {wins}/{len(yrs)} anni OOS")

    # ---------------- C) curva di ampiezza
    print("\n" + "-" * 100)
    print(f"  (C) CURVA DI AMPIEZZA — {N_DRAWS} sottoinsiemi casuali per k, Sharpe OOS 2015+")
    print("      se il motore e' AMPIEZZA (diversificazione) la mediana sale con k; se e' selezione, no")
    print("-" * 100)
    rng = np.random.default_rng(SEED)
    print(f"  {'k':>3}{'mediana':>10}{'p10':>8}{'p90':>8}{'medDD':>9}{'frac Sh>0':>11}")
    curve = {}
    for k in range(1, len(ALL18) + 1):
        shs, dds = [], []
        n = 1 if k == len(ALL18) else N_DRAWS
        for _ in range(n):
            sub = list(rng.choice(ALL18, size=k, replace=False)) if k < len(ALL18) else ALL18
            r = basket(sub, end)
            _, o = split(r)
            shs.append(_sh(o))
            dds.append(_dd(o))
        curve[k] = (float(np.median(shs)), float(np.percentile(shs, 10)),
                    float(np.percentile(shs, 90)), float(np.median(dds)),
                    float(np.mean(np.array(shs) > 0)))
        m, p10, p90, mdd, fp = curve[k]
        print(f"  {k:>3}{m:>10.2f}{p10:>8.2f}{p90:>8.2f}{mdd*100:>8.1f}%{fp*100:>10.0f}%")
    ks = sorted(curve)
    med = [curve[k][0] for k in ks]
    mono = int(np.sum(np.diff(med) > 0))
    print(f"\n  monotonia: {mono}/{len(med)-1} passi di k con mediana in salita   "
          f"(k=1 {med[0]:.2f} -> k=18 {med[-1]:.2f}, delta {med[-1]-med[0]:+.2f})")
    print(f"  DD mediano: k=1 {curve[1][3]*100:.1f}%  ->  k=18 {curve[18][3]*100:.1f}%")

    # ---------------- D) l'incumbent e' cherry-picked?
    print("\n" + "-" * 100)
    print("  (D) GTAA6 e' un 6-subset SPECIALE? (se lo fosse, il confronto B sarebbe truccato a suo favore)")
    print("-" * 100)
    rng = np.random.default_rng(SEED + 1)
    sh6 = []
    for _ in range(N_DRAWS):
        sub = list(rng.choice(ALL18, size=6, replace=False))
        _, o = split(basket(sub, end))
        sh6.append(_sh(o))
    sh6 = np.array(sh6)
    pct = float((sh6 < _sh(o6)).mean())
    print(f"  Sharpe OOS GTAA6 = {_sh(o6):.2f}   percentile fra {N_DRAWS} 6-subset casuali = {pct*100:.0f}°")
    print(f"  distribuzione 6-subset: mediana {np.median(sh6):.2f}  p10 {np.percentile(sh6,10):.2f}  "
          f"p90 {np.percentile(sh6,90):.2f}")
    print("  lettura: >85° = incumbent fortunato (il confronto B lo favorisce); ~50° = confronto equo.")

    # ---------------- E) marginale sul book
    print("\n" + "-" * 100)
    print("  (E) MARGINALE SUL BOOK 5-SLEEVE — sostituzione a peso INVARIATO (20%), non un tilt")
    print("-" * 100)
    from src.portfolio.sleeves import active_sleeves, GTAA_BOOK_ACTIVATION
    from src.portfolio.portfolio import Sleeve, StrategyPortfolio

    def _book_shape(r: pd.Series) -> pd.Series:
        """Stesso trattamento che sleeves._gtaa_daily_returns da' al GTAA nel book:
        griglia daily piena (weekend/festivi = 0.0, capitale IB fermo) + attivazione all'era book."""
        s = r.dropna().sort_index()
        if s.index.tz is None:
            s.index = s.index.tz_localize("UTC")
        days = pd.date_range(s.index.min().normalize(), s.index.max().normalize(), freq="1D", tz="UTC")
        s = s.reindex(days).fillna(0.0)
        return s[s.index >= GTAA_BOOK_ACTIVATION]

    sl = active_sleeves()
    print(f"  sleeve attivi: {[(s.name, round(s.weight, 3)) for s in sl]}")
    swapped = [
        Sleeve(s.name + "→MAT18", s.weight, (lambda _x=_book_shape(r18): _x)) if "GTAA" in s.name.upper()
        else s
        for s in sl
    ]
    b_base = StrategyPortfolio(sl).backtest()
    b_swap = StrategyPortfolio(swapped).backtest()
    print(f"  {'book':<26}{'Sh FULL':>9}{'Sh HOLD':>9}{'DD FULL':>10}{'CAGR FULL':>11}")
    for nm, b in (("base (GTAA6)", b_base), ("con MAT18", b_swap)):
        print(f"  {nm:<26}{b['full']['sharpe']:>9.2f}{b['holdout']['sharpe']:>9.2f}"
              f"{b['full']['maxdd']*100:>9.1f}%{b['full']['cagr']*100:>10.1f}%")
    print(f"  {'DELTA':<26}{b_swap['full']['sharpe']-b_base['full']['sharpe']:>+9.2f}"
          f"{b_swap['holdout']['sharpe']-b_base['holdout']['sharpe']:>+9.2f}"
          f"{(b_swap['full']['maxdd']-b_base['full']['maxdd'])*100:>+9.1f}pp"
          f"{(b_swap['full']['cagr']-b_base['full']['cagr'])*100:>+10.1f}pp")

    # ---------------- F) eseguibilita'
    print("\n" + "-" * 100)
    print("  (F) ESEGUIBILITA' a capitale piccolo (IB, frazionabili)")
    print("-" * 100)
    ex_now = {s: float(_exposure(_close(s)).iloc[-1]) / len(ALL18) for s in ALL18}
    inv = {s: w for s, w in ex_now.items() if w > 1e-6}
    print(f"  gambe investite ora: {len(inv)}/{len(ALL18)}   cash {100*(1-sum(ex_now.values())):.1f}%")
    for cap in (600, 2000, 5000, 20000):
        sleeve_cap = cap * 0.20                    # GTAA pesa 20% del book
        tickets = [sleeve_cap * w for w in inv.values()]
        if not tickets:
            continue
        print(f"    capitale ${cap:>6}  sleeve ${sleeve_cap:>7.0f}  ticket min ${min(tickets):>6.2f}  "
              f"medio ${np.mean(tickets):>6.2f}  (< $1 = non eseguibile anche con frazioni)")

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


if __name__ == "__main__":
    main()