#!/usr/bin/env python """r0828_gtaa_band_median_phase.py — criterio (C): la banda giudicata sulla MEDIANA delle 5 fasi. ⚠️ DA LEGGERE PRIMA DI CITARE QUALUNQUE NUMERO DI QUI. Questo criterio e' stato scelto **DOPO** aver visto la tabella delle fasi di `r0828_gtaa_band_phase.py`, che mostra il criterio a fase singola cambiare verdetto da una notte all'altra. Sceglierne uno nuovo guardando l'esito del vecchio e' selezione, esattamente come scegliere una cella guardando l'hold-out. Quindi: · questo script NON valida la proposta del 27/07 e NON puo' essere citato come sua conferma; · l'unica domanda che puo' porre onestamente e' quella sullo STRUMENTO — «un criterio costruito cosi' avrebbe risoluzione e potenza?» — che si risponde senza guardare se la cella che vince e' quella che ci piace; · se la risposta e' si', il criterio va poi DICHIARATO e la validazione RIFATTA da capo su di esso, con la sua data. Il verdetto di oggi sulla banda resta *non decidibile*. L'IDEA. Il difetto misurato in `r0828_gtaa_band_phase.py` e' che `gtaa._gated_returns` fasa il ribilanciamento sulla POSIZIONE nell'array (`i % every == 0`), e la prima barra di SPY scivola di un giorno ogni notte perche' IB serve una finestra rotolante di 30 anni. La fase canonica quindi cambia da sola. La mediana sulle 5 fasi toglie di mezzo la componente di FASE — e' la stessa mossa che il progetto fa gia' in `r0726_loo_deluck` (M7: su ancore appaiate la statistica e' la mediana). Resta la componente di DATO: le barre perse. La domanda e' se basti. I CRITERI, DICHIARATI QUI PRIMA DI GUARDARE I NUMERI. Il pavimento resta **0.01**, quello che il test si diede il 07/08: non lo si ritocca dopo aver visto l'esito, o si sta tarando il metro sul risultato. Il criterio (C) ha risoluzione e potenza se e solo se, sulla statistica mediana: (i) la banda scelta al buio e' LA STESSA in tutte le NOTTI notti simulate; (ii) il margine sulla seconda resta >= 0.01 in tutte; (iii) POTENZA — la banda scelta sull'hold-out e' DIVERSA da quella scelta al buio. Se coincidono, il gate lo passerebbe anche una proposta selezionata sull'hold-out, e non distingue niente (controllo positivo, come il 07/08). E una lettura di contorno che vale piu' del verdetto: la mediana di 5 fasi non e' una mediana di 5 osservazioni. Si stampa la correlazione media fra le serie di fase e l'N_eff che ne segue (§2 del CLAUDE.md: «positivo in N/N ancore NON e' N osservazioni»). uv run python scripts/research/r0828_gtaa_band_median_phase.py """ from __future__ import annotations import sys from functools import lru_cache 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 MARGINE_MINIMO = 0.01 # invariato dal 07/08: il metro non si ritocca sul risultato EVERY = BG.PROPOSTA[0] FASI = tuple(range(EVERY)) PROPOSTA_BANDA = BG.PROPOSTA[1] ROTOLANTE = ("SPY",) # l'unica gamba al muro dei 30 anni (misurato in r0828_gtaa_band_phase) @lru_cache(maxsize=None) def _serie(sym: str, taglio: int): """(prezzi, esposizione) di una gamba, tagliata in testa di `taglio` barre. In cache: e' il pezzo caro (l'esposizione e' un doppio ciclo su 30 anni) e non dipende da banda ne' fase.""" close = G._close(sym) if taglio: close = close.iloc[taglio:] ex = np.nan_to_num(np.asarray(G._exposure(close).values, float)) return close, ex def _leg(sym: str, cap_leg: float, band_usd: float, ph: int, taglio: int) -> pd.Series: close, ex = _serie(sym, taglio) 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, taglio: int) -> pd.Series: 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, ph if a in ROTOLANTE else 0, taglio if a in ROTOLANTE 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 sh(frac: float, ph: int, taglio: int, finestra: str) -> float: s = cella(frac, ph, taglio) s = s.loc[: BG.HOLDOUT] if finestra == "in" else s.loc[BG.HOLDOUT:] return BG.met(s)["sharpe"] def mediana_fasi(frac: float, taglio: int, finestra: str = "in") -> float: return float(np.median([sh(frac, ph, taglio, finestra) for ph in FASI])) def scelta(valori: dict) -> tuple[float, float]: best = max(valori, key=valori.get) return best, valori[best] - max(v for f, v in valori.items() if f != best) def sanity() -> bool: mia = cella(PROPOSTA_BANDA, 0, 0) vera = BG.run_cell(EVERY, PROPOSTA_BANDA, BG.CAP_REF) j = pd.concat({"m": mia, "v": vera}, axis=1, join="outer") d = float((j["m"] - j["v"]).abs().max()) ok = bool(np.isfinite(d) and d < 1e-12 and len(j) == len(vera)) print(f"[SANITY] replica (fase 0, taglio 0) vs produzione: max|Δ| = {d:.2e} -> {'OK' if ok else 'DIVERGE'}") return ok def main() -> int: print(__doc__.split("\n\n")[0]) print("\n⚠️ criterio scelto DOPO aver visto l'esito del precedente: misura lo STRUMENTO, " "non valida la banda.\n") if not sanity(): return 2 # ---------------------------------------------------------------- contorno: quante osservazioni print("=" * 100) print("CONTORNO — la mediana di 5 fasi, quante osservazioni sono davvero?") print("=" * 100) serie = {ph: cella(PROPOSTA_BANDA, ph, 0).loc[: BG.HOLDOUT] for ph in FASI} M = pd.concat(serie, axis=1, join="inner").dropna() C = M.corr().values rho = float((C.sum() - len(FASI)) / (len(FASI) * (len(FASI) - 1))) n_eff = len(FASI) / (1.0 + (len(FASI) - 1) * rho) print(f" correlazione media fra le 5 serie di fase (banda {PROPOSTA_BANDA:.0%}, in-sample): {rho:.3f}") print(f" N_eff = {len(FASI)} / (1 + {len(FASI)-1}·ρ) = {n_eff:.2f}") print(" -> la mediana media su ~2 osservazioni indipendenti, non su 5: toglie la fase,") print(" non compra precisione. Va citata come robustezza alla fase, mai come campione.") # ---------------------------------------------------------------- (i) e (ii) print("\n" + "=" * 100) print(f"(C) LA MEDIANA DELLE 5 FASI — banda scelta al buio, su {NOTTI} notti di finestra rotolante") print("=" * 100) spy = G._close("SPY") righe, righe_singola = [], [] for k in range(NOTTI): med = {f: mediana_fasi(f, k) for f in BG.FRAC_GRID} b, m = scelta(med) righe.append(dict(notte=k, inizio_SPY=str(spy.index[k].date()), scelta=f"{b:.0%}", margine=round(m, 4), **{f"{f:.0%}": round(med[f], 4) for f in BG.FRAC_GRID})) sing = {f: sh(f, 0, k, "in") for f in BG.FRAC_GRID} bs, ms = scelta(sing) righe_singola.append(dict(notte=k, scelta=f"{bs:.0%}", margine=round(ms, 4))) D = pd.DataFrame(righe) pd.set_option("display.width", 250) print(D.to_string(index=False)) scelte = set(D.scelta) quota = float((D.scelta == f"{PROPOSTA_BANDA:.0%}").mean()) print(f"\n celle distinte scelte: {sorted(scelte)} | la proposta ({PROPOSTA_BANDA:.0%}) vince {quota:.0%} delle notti") print(f" margine: min {D.margine.min():.4f} mediana {D.margine.median():.4f} max {D.margine.max():.4f}") print(f" notti sotto il pavimento {MARGINE_MINIMO}: {int((D.margine < MARGINE_MINIMO).sum())}/{NOTTI}") S = pd.DataFrame(righe_singola) print("\n confronto diretto col criterio a FASE SINGOLA (quello che fallisce oggi):") print(f" fase singola : celle {sorted(set(S.scelta))} " f"| sotto pavimento {int((S.margine < MARGINE_MINIMO).sum())}/{NOTTI} " f"| margine min {S.margine.min():.4f}") print(f" mediana fasi : celle {sorted(scelte)} " f"| sotto pavimento {int((D.margine < MARGINE_MINIMO).sum())}/{NOTTI} " f"| margine min {D.margine.min():.4f}") esc_s = {f: 0.0 for f in BG.FRAC_GRID} for f in BG.FRAC_GRID: vs = [sh(f, 0, k, "in") for k in range(NOTTI)] vm = [D[f"{f:.0%}"].iloc[k] for k in range(NOTTI)] esc_s[f] = (max(vs) - min(vs), max(vm) - min(vm)) print("\n escursione dello Sharpe in-sample su 10 notti, per banda (fase singola -> mediana):") for f in BG.FRAC_GRID: a, b = esc_s[f] print(f" banda {f:>4.0%}: {a:.4f} -> {b:.4f} ({'−' if b < a else '+'}{abs(b-a)/a*100:4.0f}%)") # ---------------------------------------------------------------- (iii) potenza print("\n" + "=" * 100) print("(iii) POTENZA — chi guardasse l'HOLD-OUT sceglierebbe la stessa banda?") print("=" * 100) med_in = {f: mediana_fasi(f, 0, "in") for f in BG.FRAC_GRID} med_oos = {f: mediana_fasi(f, 0, "oos") for f in BG.FRAC_GRID} b_in, m_in = scelta(med_in) b_oos, m_oos = scelta(med_oos) T = pd.DataFrame([dict(banda=f"{f:.0%}", med_in_sample=round(med_in[f], 4), med_hold_out=round(med_oos[f], 4)) for f in BG.FRAC_GRID]) print(T.to_string(index=False)) print(f"\n scelta al buio (pre-2015): {b_in:.0%} (margine {m_in:.4f})") print(f" scelta sull'hold-out : {b_oos:.0%} (margine {m_oos:.4f})") potenza = b_in != b_oos print(f" -> {'DIVERSE: il gate ha potenza' if potenza else 'UGUALI: il gate NON ha potenza'}") # ---------------------------------------------------------------- verdetto print("\n" + "=" * 100) print("VERDETTO — coi criteri dichiarati in testa, calcolati adesso") print("=" * 100) stabile = len(scelte) == 1 sopra = bool(D.margine.min() >= MARGINE_MINIMO) print(f" (i) scelta stabile su {NOTTI} notti : {'SI' if stabile else 'NO'} — {sorted(scelte)}") print(f" (ii) margine sempre >= {MARGINE_MINIMO} : {'SI' if sopra else 'NO'} — minimo {D.margine.min():.4f}") print(f" (iii) potenza (in ≠ hold-out) : {'SI' if potenza else 'NO'}") if stabile and sopra and potenza: print("\n -> (C) e' uno strumento utilizzabile. NON e' una validazione della banda:") print(" va DICHIARATO come criterio, con la sua data, e la validazione va RIFATTA su di esso.") return 0 manca = [n for n, v in (("i", stabile), ("ii", sopra), ("iii", potenza)) if not v] print(f"\n -> (C) NON basta: cade su ({'), ('.join(manca)}).") print(" La mediana toglie la FASE ma non le barre perse: se il verdetto si muove ancora,") print(" l'instabilita' e' nel DATO, e nessun criterio la aggira. Si ripara la fonte.") return 1 if __name__ == "__main__": raise SystemExit(main())