PythagorasGoal/scripts/research/r0828_gtaa_band_median_phase.py

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