PythagorasGoal/scripts/research/r0727_gtaa_band_gate.py

#!/usr/bin/env python
"""r0727_gtaa_band_gate.py — la banda al 25% della gamba passa i gate, o e' una cella scelta bene?

CONTESTO. Il 27/07 e' stata proposta una configurazione d'esecuzione nuova per GTAA01 — banda =
**25% della gamba** invece dei $50 assoluti, cadenza settimanale invariata — perche' la banda in
dollari assoluti degenera al variare del capitale (a $3.000 una banda da $400 e' l'80% della
gamba: 3 ordini l'anno e a mercato il 45% del tempo invece del 66%). La proposta e' stata
dichiarata esplicitamente **non validata**: "tarata su questa finestra, non passata per
`study_family_honest` ne' per un deflated-Sharpe".

Questo script fa quella validazione. NON e' un gate sui pesi (quello resta `weights_tilt_null`):
la banda e' un parametro di **esecuzione**, non di allocazione. Ma un parametro d'esecuzione
scelto guardando il risultato e' selezione come qualunque altra, e va trattato come tale.

TRE GATE, e uno e' specifico di questo tipo di parametro:
  (A) **SELEZIONE IN-SAMPLE** — la cella si sceglie con i soli dati PRE-2015 e si legge sul
      2015+, l'hold-out equity documentato di GTAA01 (validazione 2026-06-22) e indipendente
      dall'hold-out crypto. Se la cella scelta al buio non e' quella proposta, la proposta e'
      selection-on-holdout.
  (B) **DEFLATED SHARPE** su TUTTA la griglia cercata (Bailey & Lopez de Prado, PASS >= 0.95),
      con `dpy=252`: e' una serie su giorni di BORSA, e passarla a 365 gonfia lo Sharpe di
      ×1.20 (l'errore di annualizzazione del 25/07).
  (C) **NULL DE-LEVERING**, che qui indossa la veste "meno costi" invece di "meno drawdown":
      una banda larga puo' migliorare i numeri semplicemente smettendo di seguire il target.
      Il controllo non e' lo Sharpe ma quanto la serie bandata **assomiglia ancora** a quella
      non bandata (correlazione e rapporto di volatilita' contro il riferimento banda=0,
      cadenza=1). E' la 4a occorrenza dello stesso null in questo progetto.

⚠️ Il controllo (C) qui e' in coordinate diverse da quello del 27/07 (che misurava la quota di
tempo a mercato sulla finestra UCITS corta, 66% vs 45%): la API pubblica di `gtaa.py` non espone
le posizioni tenute, e la correlazione col riferimento misura la stessa cosa — se lo sleeve si
congela, smette di somigliare a se' stesso. Le due letture sono complementari, non alternative.

    uv run python scripts/research/r0727_gtaa_band_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" / "alt"))

from altlib import deflated_sharpe                                  # noqa: E402
from src.portfolio.gtaa import (                                    # noqa: E402
    EQ_UNIVERSE, GTAA_MIN_CAPITAL, REBAL_BAND_USD, REBAL_EVERY, gtaa_returns,
)

DPY = 252.0                       # giorni di BORSA: annualizzare a 365 gonfia Sharpe ×1.20
HOLDOUT = "2015-01-01"            # hold-out equity documentato di GTAA01 (validazione 22/06)
CAPITALS = (3_000.0, 10_000.0, 50_000.0)
CAP_REF = 10_000.0
EVERY_GRID = (1, 5, 10, 21, 42)
FRAC_GRID = (0.0, 0.05, 0.10, 0.25, 0.40, 0.60)
PROPOSTA = (5, 0.25)              # cadenza settimanale, banda = 25% della gamba
CANONICO = (REBAL_EVERY, REBAL_BAND_USD)
MIN_CORR = 0.95                   # sotto: la banda non filtra piu', congela


def met(s: pd.Series) -> dict:
    v = s.dropna().values.astype(float)
    if len(v) < 30 or v.std() == 0:
        return dict(sharpe=float("nan"), cagr=float("nan"), maxdd=float("nan"), vol=float("nan"))
    eq = (1.0 + v).cumprod()
    return dict(sharpe=float(v.mean() / v.std() * np.sqrt(DPY)),
                cagr=float(eq[-1] ** (DPY / len(v)) - 1.0),
                maxdd=float((1.0 - eq / np.maximum.accumulate(eq)).max()),
                vol=float(v.std() * np.sqrt(DPY)))


def run_cell(every: int, frac: float, capital: float) -> pd.Series:
    """La banda e' una FRAZIONE della gamba: cosi' il parametro non dipende dal capitale."""
    band = frac * capital / len(EQ_UNIVERSE)
    return gtaa_returns(capital=capital, band_usd=band, every=every)


def orders_per_year(s: pd.Series, capital: float, every: int, frac: float) -> float:
    """Ordini l'anno stimati: il conteggio esatto vive dentro `_gated_returns`, qui si usa il
    piano pubblico su una griglia di controlli — proporzionale e sufficiente al confronto."""
    from src.portfolio.gtaa import _close, _exposure
    band = frac * capital / len(EQ_UNIVERSE)
    n = 0
    for sym in EQ_UNIVERSE:
        e = _exposure(_close(sym)).dropna()
        held = 0.0
        for i, tgt in enumerate(e.values):
            if i % every:
                continue
            if abs(tgt - held) * (capital / len(EQ_UNIVERSE)) >= max(band, 1.0):
                held = float(tgt)
                n += 1
    span = max((s.index[-1] - s.index[0]).days / 365.25, 1e-9)
    return n / span


def main() -> None:
    print("=" * 104)
    print("  r0727 — GATE ONESTO SULLA BANDA GTAA01 (proposta: 25% della gamba, settimanale)")
    print("=" * 104)
    ref = run_cell(1, 0.0, CAP_REF)          # riferimento: nessuna banda, controllo giornaliero
    print(f"  storia {ref.index[0].date()} -> {ref.index[-1].date()} "
          f"({len(ref) / DPY:.1f} anni di borsa) · annualizzazione √{DPY:.0f} · "
          f"capitale di riferimento ${CAP_REF:,.0f}")
    print(f"  canonico in produzione: ogni {CANONICO[0]} barre, banda ${CANONICO[1]:.0f} "
          f"(= {CANONICO[1] / (CAP_REF / 6):.0%} della gamba a ${CAP_REF:,.0f})")

    # ---------------------------------------------------------------- griglia
    cells: dict[tuple[int, float], pd.Series] = {}
    rows = []
    for every in EVERY_GRID:
        for frac in FRAC_GRID:
            s = run_cell(every, frac, CAP_REF)
            cells[(every, frac)] = s
            full = met(s)
            ins = met(s.loc[: HOLDOUT])
            oos = met(s.loc[HOLDOUT:])
            j = pd.concat([s, ref], axis=1, join="inner").dropna()
            corr = float(j.iloc[:, 0].corr(j.iloc[:, 1]))
            rows.append(dict(every=every, frac=frac, sh_full=full["sharpe"], sh_in=ins["sharpe"],
                             sh_oos=oos["sharpe"], cagr=full["cagr"], vol=full["vol"],
                             corr=corr, volratio=full["vol"] / met(ref)["vol"]))
    G = pd.DataFrame(rows)

    print("\n" + "=" * 104)
    print("  GRIGLIA — Sharpe in-sample (pre-2015) / hold-out (2015+) / full, e somiglianza al riferimento")
    print("=" * 104)
    print(f"\n  {'cadenza':>8}{'banda%':>8}{'Sh IS':>8}{'Sh OOS':>9}{'Sh FULL':>9}"
          f"{'CAGR':>8}{'vol':>7}{'corr rif':>10}{'vol/rif':>9}")
    for _, r in G.iterrows():
        star = "  <- proposta" if (r['every'], r['frac']) == PROPOSTA else ""
        print(f"  {int(r['every']):>8}{r['frac']:>8.0%}{r['sh_in']:>8.2f}{r['sh_oos']:>9.2f}{r['sh_full']:>9.2f}"
              f"{r['cagr']:>8.2%}{r['vol']:>7.1%}{r['corr']:>10.3f}{r['volratio']:>9.2f}{star}")

    # ---------------------------------------------------------------- (A) selezione in-sample
    print("\n" + "=" * 104)
    print("  (A) SELEZIONE IN-SAMPLE — quale cella si sceglie guardando SOLO il pre-2015?")
    print("=" * 104)
    best = G.sort_values("sh_in", ascending=False).iloc[0]
    prop = G[(G["every"] == PROPOSTA[0]) & (G["frac"] == PROPOSTA[1])].iloc[0]
    print(f"\n    cella scelta al buio : cadenza {int(best['every'])}, banda {best['frac']:.0%} "
          f"-> IS {best['sh_in']:.2f} · OOS {best['sh_oos']:.2f} · FULL {best['sh_full']:.2f}")
    print(f"    cella proposta 27/07 : cadenza {PROPOSTA[0]}, banda {PROPOSTA[1]:.0%} "
          f"-> IS {prop['sh_in']:.2f} · OOS {prop['sh_oos']:.2f} · FULL {prop['sh_full']:.2f}")
    rank_in = int((G["sh_in"] > prop['sh_in']).sum()) + 1
    rank_oos = int((G["sh_oos"] > prop['sh_oos']).sum()) + 1
    print(f"    rango della proposta : {rank_in}/{len(G)} in-sample · {rank_oos}/{len(G)} sull'hold-out")
    print("       (⚠️ 2026-08-07: il confronto FRA I RANGHI e' stato ritirato come criterio — passa")
    print("        ~meta' delle celle per costruzione e sulla proposta si decideva su 0.001 di")
    print("        Sharpe. Resta stampato come descrizione. Vedi r0807_gtaa_gate_resolution.py.)")

    # criterio decidibile: la proposta e' una BANDA, la cadenza e' gia' quella di produzione
    ev = PROPOSTA[0]
    riga = G[G["every"] == ev]
    blind = riga.sort_values("sh_in", ascending=False).iloc[0]
    hold = riga.sort_values("sh_oos", ascending=False).iloc[0]
    secondo = riga[riga["frac"] != blind["frac"]]["sh_in"].max()
    print(f"\n    CRITERIO DECIDIBILE — a cadenza di produzione ({ev}), quale banda si sceglie?")
    print(f"      sui soli dati pre-2015 : {blind['frac']:.0%}  "
          f"(margine sulla 2ª: {blind['sh_in'] - secondo:+.4f} di Sharpe)")
    print(f"      guardando l'hold-out   : {hold['frac']:.0%}  "
          "<- controllo positivo: se coincidessero, il gate non avrebbe potenza")
    if blind["frac"] == PROPOSTA[1] and hold["frac"] != PROPOSTA[1]:
        print("      -> la proposta e' la cella scelta AL BUIO, e chi avesse guardato l'hold-out")
        print("         ne avrebbe scelta un'altra: e' il contrario di una selezione-sull'hold-out.")
    print(f"\n    Costo di scegliere al buio invece che a posteriori (hold-out): "
          f"{best['sh_oos'] - prop['sh_oos']:+.2f} di Sharpe")
    n_best = orders_per_year(cells[(int(best["every"]), float(best["frac"]))], CAP_REF,
                             int(best["every"]), float(best["frac"]))
    n_prop = orders_per_year(cells[PROPOSTA], CAP_REF, *PROPOSTA)
    print(f"\n    ⚠️ La cella scelta al buio controlla il target OGNI GIORNO ({n_best:.0f} ordini/anno)")
    print(f"       contro la cadenza settimanale della proposta ({n_prop:.0f} ordini/anno). Il conto")
    print(f"       di esecuzione (Degiro/Revolut) NON ha API: 250 controlli manuali l'anno non sono")
    print(f"       una configurazione, sono un'ipotesi. La proposta costa "
          f"{prop['sh_full'] - best['sh_full']:+.2f} di Sharpe FULL e")
    print("       compra una cadenza eseguibile a mano. E' un vincolo dichiarato, non una scoperta.")

    # ---------------------------------------------------------------- (B) deflated Sharpe
    print("\n" + "=" * 104)
    print(f"  (B) DEFLATED SHARPE sulla griglia intera ({len(G)} celle), dpy={DPY:.0f}")
    print("=" * 104)
    all_sr = list(G["sh_full"].values)
    for label, cell in (("proposta 27/07", PROPOSTA),
                        ("scelta in-sample", (int(best['every']), float(best['frac'])))):
        s = cells[cell]
        dsr, null_max = deflated_sharpe(met(s)["sharpe"], all_sr, s.dropna().values, dpy=DPY)
        print(f"\n    {label:<18} cadenza {cell[0]}, banda {cell[1]:.0%} · "
              f"Sharpe {met(s)['sharpe']:.2f} vs massimo atteso sotto il nullo {null_max:.2f}")
        print(f"    {'':<18} DSR = {dsr:.3f}  ->  {'PASS' if dsr >= 0.95 else 'FAIL'} (soglia 0.95)")

    # ---------------------------------------------------------------- (C) null de-levering
    print("\n" + "=" * 104)
    print("  (C) NULL DE-LEVERING — la banda filtra, o congela?")
    print("=" * 104)
    print(f"\n    Riferimento = banda 0%, controllo giornaliero. Una banda che FILTRA lascia la")
    print(f"    serie quasi identica (corr >= {MIN_CORR}); una che CONGELA la stacca.")
    print(f"\n  {'banda%':>8}{'corr rif':>11}{'vol/rif':>10}{'Sh FULL':>10}   verdetto")
    for frac in FRAC_GRID:
        r = G[(G["every"] == PROPOSTA[0]) & (G["frac"] == frac)].iloc[0]
        v = "OK" if r['corr'] >= MIN_CORR else "❌ FUORI TRACCIA"
        print(f"  {frac:>8.0%}{r['corr']:>11.3f}{r['volratio']:>10.2f}{r['sh_full']:>10.2f}   {v}")
    print("\n    ⚠️ Il modo di fallire NON e' quello atteso, e va detto: allargando la banda la")
    print("       volatilita' NON scende (vol/rif resta 0.99-1.10), quindi qui il null de-levering")
    print("       classico non morde. Cio' che si rompe e' il TRACKING: la serie smette di")
    print("       somigliare a se' stessa perche' tiene posizioni vecchie. E i due segnali")
    print("       CONCORDANO — oltre il 25% la correlazione cala E lo Sharpe smette di migliorare")
    print("       (0.70 -> 0.67 -> 0.59): non c'e' una zona in cui il numero premia il congelamento.")
    print(f"       Ma il 25% e' AL BORDO della regione (corr {G[(G['every'] == 5) & (G['frac'] == 0.25)].iloc[0]['corr']:.3f} "
          f"contro una soglia di {MIN_CORR}): non e' un plateau centrale.")

    # ---------------------------------------------------------------- invarianza
    print("\n" + "=" * 104)
    print("  INVARIANZA AL CAPITALE — il motivo per cui la proposta esiste")
    print("=" * 104)
    print(f"\n  {'capitale':>10}{'banda 25%':>12}{'ord/anno':>11}{'Sharpe':>9}{'corr rif':>10}"
          f"   |  {'banda $50 fissa':>16}{'ord/anno':>11}{'Sharpe':>9}")
    for cap in CAPITALS:
        s_f = run_cell(PROPOSTA[0], PROPOSTA[1], cap)
        s_a = gtaa_returns(capital=cap, band_usd=CANONICO[1], every=CANONICO[0])
        n_f = orders_per_year(s_f, cap, PROPOSTA[0], PROPOSTA[1])
        n_a = orders_per_year(s_a, cap, CANONICO[0], CANONICO[1] / (cap / 6))
        j = pd.concat([s_f, ref], axis=1, join="inner").dropna()
        print(f"  ${cap:>9,.0f}{PROPOSTA[1] * cap / 6:>11,.0f}${n_f:>11.0f}{met(s_f)['sharpe']:>9.2f}"
              f"{float(j.iloc[:, 0].corr(j.iloc[:, 1])):>10.3f}   |  "
              f"{CANONICO[1]:>15,.0f}${n_a:>11.0f}{met(s_a)['sharpe']:>9.2f}")
    print(f"\n    (sotto ${GTAA_MIN_CAPITAL:,.0f} lo sleeve non e' deployabile comunque: "
          "GTAA_MIN_CAPITAL)")

    # ---------------------------------------------------------------- impatto sul book
    print("\n" + "=" * 104)
    print("  IMPATTO SUL BOOK — il numero che decide se toccare la produzione")
    print("=" * 104)
    from src.portfolio.portfolio import combine_outer
    from src.portfolio.sleeves import GTAA_BOOK_ACTIVATION, active_sleeves

    def as_book(gtaa_series: pd.Series | None) -> pd.Series:
        cols, w = {}, {}
        for sl in active_sleeves():
            s = sl.daily()
            if sl.name == "GTAA01_eq_trend" and gtaa_series is not None:
                g = gtaa_series.dropna().sort_index()
                if g.index.tz is None:
                    g.index = g.index.tz_localize("UTC")
                days = pd.date_range(g.index.min().normalize(), g.index.max().normalize(),
                                     freq="1D", tz="UTC")
                s = g.reindex(days).fillna(0.0)
                s = s[s.index >= GTAA_BOOK_ACTIVATION]
            cols[sl.name], w[sl.name] = s, sl.weight
        return combine_outer(cols, w)

    def bk(s: pd.Series) -> dict:
        v = s.dropna().values.astype(float)
        eq = (1.0 + v).cumprod()
        return dict(sharpe=float(v.mean() / v.std() * np.sqrt(365.0)),
                    maxdd=float((1.0 - eq / np.maximum.accumulate(eq)).max()))

    b_can = bk(as_book(None))
    b_new = bk(as_book(run_cell(PROPOSTA[0], PROPOSTA[1], CAP_REF)))
    print(f"\n    book 5-sleeve, banda CANONICA $50   : Sharpe FULL {b_can['sharpe']:.3f} · "
          f"maxDD {b_can['maxdd']:.2%}")
    print(f"    book 5-sleeve, banda PROPOSTA 25%   : Sharpe FULL {b_new['sharpe']:.3f} · "
          f"maxDD {b_new['maxdd']:.2%}")
    print(f"    delta                               : {b_new['sharpe'] - b_can['sharpe']:+.3f} Sharpe · "
          f"{(b_new['maxdd'] - b_can['maxdd']) * 100:+.2f} pt di maxDD")
    print("\n    ⚠️ Il book si annualizza a 365 (griglia giornaliera piena, weekend = 0), lo sleeve")
    print("       a 252 (giorni di borsa): i due Sharpe NON sono confrontabili fra loro, solo")
    print("       ciascuno con se' stesso fra le due configurazioni.")



if __name__ == "__main__":
    main()