#!/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())