PythagorasGoal/scripts/research/r0828_gtaa_band_phase.py

#!/usr/bin/env python
"""r0828_gtaa_band_phase.py — il gate (A) della banda GTAA01 fallisce di nuovo: e' la proposta, o e' ANCORA il gate?

CONTESTO. Il 27/07 fu proposta la banda al 25% della gamba. Il gate (A) — SELEZIONE IN-SAMPLE —
la valido' due volte, con due criteri diversi, e li ha rotti entrambi senza che nessuno toccasse
il codice:

  · 27/07  criterio `rank_in <= rank_oos`. Rotto il 07/08. Motivo (r0807_gtaa_gate_resolution):
           lo passa ~meta' della griglia PER COSTRUZIONE, e sulla proposta decideva su 0.00116
           di Sharpe. Una moneta. RITIRATO.
  · 07/08  criterio sostitutivo: «a cadenza di produzione, la banda scelta guardando SOLO il
           pre-2015 e' la proposta». Passava col 25% e un margine di +0.0151 sulla seconda —
           13x il criterio precedente. Rotto il 28/08: al buio esce il 60%, margine 0.0074,
           SOTTO il pavimento di 0.01 che il test stesso si era dato.

Il codice e' fermo (`git log src/portfolio/gtaa.py scripts/research/r0727_gtaa_band_gate.py`
non ha commit dopo il 07/08). Quindi si e' mosso il DATO. Questo script chiede, per la terza
volta e sullo stesso parametro, se il criterio misura cio' che dichiara.

LE DUE COSE CHE SI SONO TROVATE
-------------------------------
(0) **LA FINESTRA IN-SAMPLE NON E' FERMA.** IB ritorna una finestra ROTOLANTE di 30 anni. SPY e'
    l'unica gamba al muro (quotato 1993, storia servita dal 1996), e nel `logs/cron_daily.log`
    la sua data d'inizio avanza di un giorno di borsa OGNI NOTTE: 1996-07-02 il 24/06,
    1996-09-04 oggi. Il pre-2015 — che tutto il progetto tratta come una finestra congelata —
    perde il suo giorno piu' vecchio ogni notte. (QQQ e IWM colpiranno lo stesso muro nel
    2029-2030: non e' un difetto di SPY, e' il muro che avanza.)

(1) **L'AMPLIFICATORE E' LA FASE.** `gtaa._gated_returns` ribilancia su `i % every == 0`: l'indice
    e' la POSIZIONE NELL'ARRAY, quindi la fase e' ancorata alla prima barra del file. Se la prima
    barra scivola di un giorno, si ri-fasa ogni decisione di ribilanciamento di trent'anni di
    storia. Il progetto sa gia' che la fase conta — `r0726_loo_deluck.gtaa01_at(ph)` la tratta
    come un'ancora a 5 valori e ci de-lucka sopra — ma la assumeva FISSA alla fase canonica 0.
    Non lo e': **la fase canonica cambia da sola ogni notte**.

I DUE ESPERIMENTI, in quest'ordine
----------------------------------
  (A) **LE NOTTI.** Si taglia la testa di SPY di k=0..N-1 barre, che e' esattamente cio' che la
      finestra rotolante fa da sola in N notti, e si guarda la banda scelta al buio.
  (B) **LA FASE PURA.** A dato FISSO, si sposta la sola fase di ribilanciamento di SPY (ph=0..4).
      Serve ad attribuire: se il verdetto si muove anche qui, l'imputato e' la fase, non le barre
      perse. Col controllo su TUTTE le gambe accanto, che dice quanto sarebbe peggio quando anche
      QQQ e IWM arriveranno al muro.

CRITERIO DEL VERDETTO, dichiarato PRIMA (M13: un criterio si misura sulla sua RISOLUZIONE prima
che sul suo esito). Il gate (A) ha risoluzione se e solo se:
   (i)  la banda scelta al buio e' LA STESSA in tutte le notti e in tutte le fasi, e
   (ii) il margine sulla seconda resta sopra MARGINE_MINIMO in tutte.
Se (i) o (ii) cadono, il criterio non distingue la proposta da una cella qualunque, e va ritirato
come il suo predecessore — si congela il MOTIVO, non l'esito.

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

import r0727_gtaa_band_gate as BG                      # noqa: E402
import src.portfolio.gtaa as G                         # noqa: E402

NOTTI = 10                 # notti di finestra rotolante da simulare
MARGINE_MINIMO = 0.01      # il pavimento che il test si e' dato il 07/08, non ritoccato qui
MURO_IB_ANNI = 30.0        # finestra massima servita da IB: chi la tocca rotola
PROPOSTA_BANDA = BG.PROPOSTA[1]

# Tabella PUBBLICATA il 07/08 (docs/diary/2026-08-07-gate-gtaa-e-storia-troncata.md, sezione (2)).
# Serve a misurare di quanto si e' mossa una finestra che NON puo' acquisire dati nuovi.
SH_IN_0807 = {0.00: 0.5396, 0.05: 0.5714, 0.10: 0.5956, 0.25: 0.6398, 0.40: 0.6247, 0.60: 0.5874}


def _leg(sym: str, cap_leg: float, band_usd: float, every: int, ph: int,
         taglio: int = 0) -> pd.Series:
    """`gtaa._gated_returns` con due manopole: la FASE (`ph`) e il TAGLIO in testa alla serie.

    Copia dichiarata, non riuso: la produzione non espone ne' l'una ne' l'altra. La replica a
    (ph=0, taglio=0) deve tornare bit-exact quella di produzione — lo verifica `sanity()`.
    """
    close = G._close(sym)
    if taglio:
        close = close.iloc[taglio:]
    ex = np.nan_to_num(np.asarray(G._exposure(close).values, float))
    px = close.values.astype(float)
    ret = np.zeros(len(px))
    ret[1:] = px[1:] / px[:-1] - 1.0
    held = np.empty(len(ex))
    comm = np.zeros(len(ex))
    cur = 0.0
    for i in range(len(ex)):
        if i % every == ph:
            notional = abs(ex[i] - cur) * cap_leg
            if notional >= max(band_usd, G.IB_MIN_TRADE_USD):
                comm[i] = G.ib_commission(notional, px[i]) / cap_leg
                cur = ex[i]
        held[i] = cur
    pos = np.zeros(len(held))
    pos[1:] = held[:-1]
    net = pos * ret - comm
    net[0] = 0.0
    return pd.Series(net, index=close.index)


def cella(frac: float, ph: int = 0, taglio: int = 0, tutte: bool = False) -> pd.Series:
    """La cella (cadenza di produzione, banda `frac`). `ph`/`taglio` si applicano alla sola SPY
    salvo `tutte`: e' SPY l'unica gamba al muro dei 30 anni, e il resto della griglia e' fermo."""
    cap_leg = BG.CAP_REF / len(G.EQ_UNIVERSE)
    band = frac * BG.CAP_REF / len(G.EQ_UNIVERSE)
    cols = {a: _leg(a, cap_leg, band, BG.PROPOSTA[0],
                    ph if (tutte or a == "SPY") else 0,
                    taglio if (tutte or a == "SPY") else 0)
            for a in G.EQ_UNIVERSE}
    return pd.concat(cols, axis=1, sort=True).sort_index().mean(axis=1, skipna=True).dropna()


def riga(ph: int = 0, taglio: int = 0, tutte: bool = False) -> dict:
    sh = {f: BG.met(cella(f, ph, taglio, tutte).loc[: BG.HOLDOUT])["sharpe"] for f in BG.FRAC_GRID}
    best = max(sh, key=sh.get)
    secondo = max(v for f, v in sh.items() if f != best)
    return dict(scelta=best, margine=sh[best] - secondo, sh=sh)


def sanity() -> bool:
    """La replica a (fase 0, taglio 0) DEVE essere quella di produzione. Senza questo, ogni
    numero sotto misura la mia copia invece dello sleeve che gira."""
    mia = cella(PROPOSTA_BANDA, 0, 0)
    vera = BG.run_cell(BG.PROPOSTA[0], PROPOSTA_BANDA, BG.CAP_REF)
    j = pd.concat({"m": mia, "v": vera}, axis=1, join="outer")
    delta = float((j["m"] - j["v"]).abs().max())
    ok = bool(np.isfinite(delta) and delta < 1e-12 and len(j) == len(vera))
    print(f"[SANITY] replica (fase 0, taglio 0) vs produzione: max|Δ| = {delta:.2e}  "
          f"barre {len(j)} vs {len(vera)}  -> {'OK' if ok else 'DIVERGE'}")
    return ok


def storia() -> pd.DataFrame:
    """(0) Quali gambe sono al muro dei 30 anni — cioe' quali hanno una finestra ROTOLANTE."""
    rows = []
    for a in G.EQ_UNIVERSE:
        s = G._close(a)
        anni = (s.index[-1] - s.index[0]).days / 365.25
        rows.append(dict(gamba=a, dal=str(s.index[0].date()), barre=len(s), anni=round(anni, 1),
                         al_muro="SI — rotola" if anni >= MURO_IB_ANNI - 0.15 else "no",
                         anni_al_muro=round(MURO_IB_ANNI - anni, 1)))
    return pd.DataFrame(rows)


def main() -> int:
    print(__doc__.split("\n\n")[0])
    print()
    if not sanity():
        print("!! la replica non riproduce la produzione: i numeri sotto non sono confrontabili")
        return 2

    print("\n" + "=" * 96)
    print("(0) LA FINESTRA E' ROTOLANTE — quali gambe sono al muro dei 30 anni di IB")
    print("=" * 96)
    H = storia()
    print(H.to_string(index=False))
    rotolanti = list(H[H.al_muro.str.startswith("SI")].gamba)
    print(f"\n  gambe con finestra rotolante: {rotolanti or 'nessuna'} — per queste, il pre-2015 "
          "perde il giorno piu' vecchio ogni notte.")

    print("\n" + "=" * 96)
    print("  quanto si e' mossa una finestra che NON puo' acquisire dati nuovi (07/08 -> oggi)")
    print("=" * 96)
    oggi = riga(0, 0)["sh"]
    D = pd.DataFrame([dict(banda=f"{f:.0%}", sh_in_0807=SH_IN_0807[f], sh_in_oggi=round(oggi[f], 4),
                           delta=round(oggi[f] - SH_IN_0807[f], 4)) for f in BG.FRAC_GRID])
    print(D.to_string(index=False))
    print(f"\n  spostamento massimo in valore assoluto: {D.delta.abs().max():.4f} di Sharpe, "
          f"su una finestra CHIUSA (pre-2015) e a codice fermo.")

    print("\n" + "=" * 96)
    print(f"(A) LE NOTTI — banda scelta al buio in {NOTTI} notti consecutive di finestra rotolante")
    print("=" * 96)
    spy = G._close("SPY")
    A = []
    for k in range(NOTTI):
        r = riga(0, k)
        A.append(dict(notte=k, inizio_SPY=str(spy.index[k].date()), scelta=f"{r['scelta']:.0%}",
                      margine=round(r["margine"], 4),
                      **{f"{f:.0%}": round(r["sh"][f], 4) for f in BG.FRAC_GRID}))
    DA = pd.DataFrame(A)
    print(DA.to_string(index=False))
    scelte_A = set(DA.scelta)
    quota_proposta = float((DA.scelta == f"{PROPOSTA_BANDA:.0%}").mean())
    print(f"\n  celle distinte scelte: {sorted(scelte_A)}   "
          f"| la proposta ({PROPOSTA_BANDA:.0%}) vince {quota_proposta:.0%} delle notti")
    print(f"  margine: min {DA.margine.min():.4f}  mediana {DA.margine.median():.4f}  "
          f"max {DA.margine.max():.4f}   (pavimento {MARGINE_MINIMO})")
    print(f"  notti sotto il pavimento: {int((DA.margine < MARGINE_MINIMO).sum())}/{NOTTI}")

    print("\n" + "=" * 96)
    print("(B) LA FASE PURA — dato FISSO, si muove solo la fase di ribilanciamento")
    print("=" * 96)
    for tutte, eti in ((False, f"fase mossa solo su {rotolanti or ['SPY']} (cio' che succede DAVVERO oggi)"),
                       (True, "fase mossa su TUTTE le gambe (controllo: quando anche QQQ e IWM saranno al muro)")):
        B = []
        for ph in range(BG.PROPOSTA[0]):
            r = riga(ph, 0, tutte)
            B.append(dict(fase=ph, scelta=f"{r['scelta']:.0%}", margine=round(r["margine"], 4),
                          **{f"{f:.0%}": round(r["sh"][f], 4) for f in BG.FRAC_GRID}))
        DB = pd.DataFrame(B)
        print(f"\n  {eti}")
        print(DB.to_string(index=False))
        print(f"  celle distinte: {sorted(set(DB.scelta))} | margine min {DB.margine.min():.4f}")
        if not tutte:
            scelte_B, marg_B = set(DB.scelta), DB.margine

    print("\n" + "=" * 96)
    print("VERDETTO — coi criteri dichiarati in testa allo script, calcolati adesso")
    print("=" * 96)
    stabile = len(scelte_A) == 1 and len(scelte_B) == 1
    sopra = bool(DA.margine.min() >= MARGINE_MINIMO and marg_B.min() >= MARGINE_MINIMO)
    print(f"  (i)  scelta stabile su notti e fasi : {'SI' if stabile else 'NO'}  "
          f"— notti {sorted(scelte_A)}, fasi {sorted(scelte_B)}")
    print(f"  (ii) margine sempre >= {MARGINE_MINIMO}      : {'SI' if sopra else 'NO'}  "
          f"— minimo osservato {min(DA.margine.min(), marg_B.min()):.4f}")
    if stabile and sopra:
        print("\n  -> il gate (A) HA risoluzione: il fallimento del test e' un fatto sulla PROPOSTA.")
        return 0
    print("\n  -> il gate (A) NON ha risoluzione. Il verdetto non e' una proprieta' della banda:")
    print("     e' una proprieta' della notte in cui lo si legge. Come il criterio a ranghi")
    print("     ritirato il 07/08, va ritirato — congelando il MOTIVO, non l'esito.")
    print(f"     La banda della proposta ({PROPOSTA_BANDA:.0%}) resta la scelta al buio in "
          f"{quota_proposta:.0%} delle notti: ne' confermata ne' smentita, NON DECIDIBILE cosi'.")
    return 1


if __name__ == "__main__":
    raise SystemExit(main())