#!/usr/bin/env python """r0726_loo_deluck.py — LEAVE-ONE-OUT del book 5-sleeve DE-LUCKATO sulle ancore. DOMANDA: "quale sleeve terrei?" — cioe' quanto vale OGNI sleeve alla sua posizione attuale nel book. La risposta ovvia (togli lo sleeve, guarda quanto cala il book) e' misurata all'ANCORA CANONICA di tutti e cinque, e per questo NON e' credibile: ogni Δ misurato su griglia ancorata eredita la fortuna di quell'ancora (lezione codificata il 26/07 in `altlib.anchor_luck_delta`; corollario: *se si de-lucka una strategia va de-luckato anche il suo DEGRADO*). E qui il problema e' al quadrato — 4 sleeve su 5 sono ancorati, e per 3 di loro l'ancora canonica e' gia' stata misurata come fortunata (TP01 92° pctl su 24 ancore, SKH01 93-98° su 23 offset, XS01 fase 0 al 15° pctl di DD; VRP01 e' l'unico senza firma, la sua fase canonica e' la PEGGIORE delle 7). COSA FA (nessun file di produzione toccato, nessun peso cambiato): 0. SANITY OBBLIGATORIO — ogni replica parametrizzata sull'ancora deve riprodurre BIT-EXACT lo sleeve di produzione alla sua ancora canonica. Senza questo il resto misura un'altra strategia. 1. Precalcola le 69 serie: TP01 x24 ancore orarie, XS01 x10 fasi, VRP01 x7 fasi, SKH01 x23 offset di griglia, GTAA01 x5 fasi di cadenza settimanale. 2. ESTRAZIONI CONGIUNTE: N configurazioni d'ancora, una per sleeve, uniformi e INDIPENDENTI (il null onesto: non avevamo ragione di preferire nessuna ancora). Lo spazio pieno e' 24x10x7x23x5 = 193.200 -> si campiona. 3. Per ogni estrazione: book completo + i 5 book leave-one-out (combine_outer, pesi rinormalizzati per-riga come in produzione) -> Δ APPAIATI per sleeve. La statistica e' la MEDIANA DELLE DIFFERENZE APPAIATE, mai la differenza delle mediane (l'errore ribaltante catturato il 26/07): le due mediane cadono su ancore diverse. 4. Output: per sleeve, Δ canonico + il suo percentile nella banda, Δ mediano de-luckato, banda [p10,p90], frazione di estrazioni in cui il contributo e' positivo. Piu' il livello del book de-luckato (FULL/HOLD/DD mediani) — che aggiorna con una MISURA la stima a occhio in CLAUDE.md ("HOLD ~1.9-2.1, FULL ~2.0-2.2"). COSA NON E': non e' un gate sui pesi (quello resta `weights_tilt_null`), e non e' evidenza out-of-sample — il book e' misurato sugli stessi anni su cui XS01/SKH01 sono stati scelti e affinati. E' attribuzione, de-luckata: dice quanto di cio' che vediamo e' struttura e quanto e' l'ancora che ci e' capitata. uv run python scripts/research/r0726_loo_deluck.py [--draws N] """ from __future__ import annotations import argparse import pickle import sys import time from pathlib import Path import numpy as np import pandas as pd ROOT = Path("/opt/docker/PythagorasGoal") sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "scripts" / "research")) sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt")) from src.portfolio.portfolio import HOLDOUT, combine_outer, metrics, to_daily # noqa: E402 from src.portfolio import gtaa as G # noqa: E402 from src.portfolio.sleeves import (GTAA_DEFAULT_CAPITAL, VRP_CFG, # noqa: E402 _gtaa_daily_returns, _skyhook_returns, _tp01_returns, _vrp_combo_returns, _xsec_returns, active_sleeves) # repliche ancorate GIA' verificate bit-exact nei rispettivi audit (02/07, 03/07): # riusate, non riscritte — riscriverle significherebbe rifare gli stessi errori di convenzione. import r0702_anchor_skh01 as A_SKH # noqa: E402 skh_port(off) import r0702_anchor_xs01 as A_XS # noqa: E402 xs_phase(phase) import r0702_tp01_offset as A_TP # noqa: E402 daily_off(asset, h) import r0703_vrpimp_anchor as A_VRP # noqa: E402 combo_weekly(phase, f) + to_daily_lumped from src.strategies.trend_portfolio import CANONICAL, TrendPortfolio, simple_returns # noqa: E402 SEED = 20260726 # cache rigenerabile delle 69 serie d'ancora (~8s a ricostruirla): gitignored via `*.pkl`. # `--force` la ignora. Path stabile nel repo, non legato alla sessione che l'ha scritta. CACHE = ROOT / "data" / "_cache" / "anchor_series.pkl" WEIGHTS = {"TP01_trend_1d": 0.33, "XS01_xsec_hl": 0.15, "VRP01_shortvol": 0.12, "SKH01_skyhook": 0.20, "GTAA01_eq_trend": 0.20} SHORT = {"TP01_trend_1d": "TP01", "XS01_xsec_hl": "XS01", "VRP01_shortvol": "VRP01", "SKH01_skyhook": "SKH01", "GTAA01_eq_trend": "GTAA01"} # =========================================================================== # Generatori ancorati — uno per sleeve. Ogni serie esce GIA' normalizzata a # giornaliera con `to_daily`, come fa Sleeve.daily() in produzione. # =========================================================================== def tp01_at(h: int) -> pd.Series: """TP01 all'ancora oraria h. Replica ESATTA di sleeves._tp01_returns con il daily risampionato a offset h invece che a 00:00 UTC.""" tp = TrendPortfolio(**CANONICAL) series = {} for a in ("BTC", "ETH"): df = A_TP.daily_off(a, h) r = simple_returns(df["close"].values.astype(float)) tgt = tp.target_series(df) held = np.zeros(len(tgt)) held[1:] = tgt[:-1] net = held * r - tp.fee_side * np.abs(np.diff(held, prepend=0.0)) net[0] = 0.0 series[a] = pd.Series(np.clip(net, -0.99, None), index=pd.to_datetime(df["datetime"])) J = pd.concat(series, axis=1, join="inner").fillna(0.0) return to_daily(pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index)) def xs01_at(p: int) -> pd.Series: return to_daily(A_XS.xs_phase(p)) def vrp01_at(p: int) -> pd.Series: return to_daily(A_VRP.to_daily_lumped(A_VRP.combo_weekly(p, f=VRP_CFG["f"]))) def skh01_at(off: int) -> pd.Series: return to_daily(A_SKH.skh_port(off)) def _gtaa_leg_phase(sym: str, cap_leg: float, band_usd: float, every: int, ph: int) -> pd.Series: """Replica di gtaa._gated_returns con la cadenza in fase `ph` (i % every == ph).""" close = G._close(sym) 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 gtaa01_at(ph: int) -> pd.Series: """GTAA01 alla fase `ph` della cadenza settimanale (REBAL_EVERY=5 barre di borsa). Replica ESATTA di sleeves._gtaa_daily_returns, incluse le convenzioni di calendario.""" cap_leg = float(GTAA_DEFAULT_CAPITAL) / len(G.EQ_UNIVERSE) cols = {a: _gtaa_leg_phase(a, cap_leg, G.REBAL_BAND_USD, G.REBAL_EVERY, ph) for a in G.EQ_UNIVERSE} r = pd.concat(cols, axis=1, sort=True).sort_index().mean(axis=1, skipna=True).dropna() r = r.dropna().sort_index() if r.index.tz is None: r.index = r.index.tz_localize("UTC") days = pd.date_range(r.index.min().normalize(), r.index.max().normalize(), freq="1D", tz="UTC") r = r.reindex(days).fillna(0.0) from src.portfolio.sleeves import GTAA_BOOK_ACTIVATION return to_daily(r[r.index >= GTAA_BOOK_ACTIVATION]) # nome -> (generatore, ancore a priori, ancora canonica, sleeve di produzione) SPEC = { "TP01_trend_1d": (tp01_at, tuple(range(24)), 0, _tp01_returns), "XS01_xsec_hl": (xs01_at, tuple(range(10)), 0, _xsec_returns), "VRP01_shortvol": (vrp01_at, tuple(range(7)), 0, _vrp_combo_returns), "SKH01_skyhook": (skh01_at, tuple(range(0, 690, 30)), 0, _skyhook_returns), "GTAA01_eq_trend": (gtaa01_at, tuple(range(5)), 0, _gtaa_daily_returns), } # =========================================================================== # PART 0 — sanity: la replica ancorata == sleeve di produzione all'ancora canonica # =========================================================================== def sanity() -> None: print("[SANITY] ogni replica ancorata deve riprodurre lo sleeve di produzione " "alla sua ancora canonica") for name, (gen, _anchors, canon, prod_fn) in SPEC.items(): mine = gen(canon) ref = to_daily(prod_fn()) J = pd.concat({"m": mine, "r": ref}, axis=1, join="outer") assert len(mine) == len(ref), ( f"{name}: lunghezze diverse ({len(mine)} vs {len(ref)}) — non e' la stessa serie") dmax = float(np.nanmax(np.abs(J["m"].values - J["r"].values))) assert dmax < 1e-12, f"SANITY FAIL {name}: max|dif| = {dmax:.3e}" print(f" {SHORT[name]:<7} ancora {canon:>3} max|dif| = {dmax:.1e} " f"(n={len(mine)}) OK") # =========================================================================== # PART 1 — precalcolo delle 69 serie # =========================================================================== def build_all(force: bool = False) -> dict: if CACHE.exists() and not force: with CACHE.open("rb") as fh: cached = pickle.load(fh) if set(cached) == set(SPEC) and all(len(cached[n]) == len(SPEC[n][1]) for n in SPEC): print(f"[CACHE] serie d'ancora lette da {CACHE.name}") return cached out = {} for name, (gen, anchors, _c, _p) in SPEC.items(): t0 = time.time() out[name] = {a: gen(a) for a in anchors} print(f" {SHORT[name]:<7} {len(anchors):>3} ancore in {time.time()-t0:5.1f}s") CACHE.parent.mkdir(parents=True, exist_ok=True) with CACHE.open("wb") as fh: pickle.dump(out, fh) return out # =========================================================================== # PART 2 — book + leave-one-out a una configurazione d'ancora # =========================================================================== def book_stats(cols: dict) -> tuple[float, float, float]: """(Sharpe FULL, Sharpe HOLD-OUT, maxDD FULL) del book a pesi WEIGHTS ristretti a `cols`.""" w = {k: WEIGHTS[k] for k in cols} full = combine_outer(cols, w) mf = metrics(full) mh = metrics(full[full.index >= HOLDOUT]) return mf["sharpe"], mh["sharpe"], mf["maxdd"] def one_config(series: dict, cfg: dict) -> dict: """Un'estrazione d'ancora -> book completo + 5 leave-one-out, tutti alla STESSA ancora.""" cols = {n: series[n][cfg[n]] for n in SPEC} bf, bh, bd = book_stats(cols) res = {"book": (bf, bh, bd)} for n in SPEC: sub = {k: v for k, v in cols.items() if k != n} f, h, d = book_stats(sub) # Δ = contributo dello sleeve: >0 su Sharpe = togliendolo il book peggiora # >0 su DD = togliendolo il drawdown peggiora res[n] = (bf - f, bh - h, d - bd) return res def paired_deltas(draws: list[dict], name: str, j: int) -> np.ndarray: """Vettore dei Δ APPAIATI di uno sleeve su una metrica (j: 0=FULL, 1=HOLD, 2=maxDD). Ogni elemento e' (book con X) - (book senza X) alla STESSA configurazione d'ancora. La statistica corretta e' la mediana di QUESTO vettore — mai la differenza fra la mediana dei book-con e quella dei book-senza, perche' le due mediane cadono su ancore DIVERSE e il verdetto puo' ribaltarsi (errore catturato il 26/07 sul recupero on-book di SKH01, codificato in `altlib.anchor_luck_delta`). Isolata qui apposta per essere testabile. """ return np.array([d[name][j] for d in draws], float) def pctl_of(arr: np.ndarray, v: float) -> float: return float((arr < v).mean() * 100.0) def band(arr: np.ndarray) -> tuple[float, float, float]: return float(np.median(arr)), float(np.percentile(arr, 10)), float(np.percentile(arr, 90)) def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--draws", type=int, default=600) ap.add_argument("--force", action="store_true", help="ricalcola le serie ignorando la cache") args = ap.parse_args() t_start = time.time() print("=" * 100) print(" r0726 — LEAVE-ONE-OUT del book 5-sleeve, DE-LUCKATO sulle ancore") print("=" * 100) sanity() print(f"\n[PRECALCOLO] {sum(len(v[1]) for v in SPEC.values())} serie d'ancora " f"(spazio congiunto {np.prod([len(v[1]) for v in SPEC.values()]):,})") S = build_all(force=args.force) # --- configurazione canonica (quella di tutti i numeri del progetto) --- canon_cfg = {n: SPEC[n][2] for n in SPEC} canon = one_config(S, canon_cfg) # --- estrazioni congiunte uniformi indipendenti --- rng = np.random.default_rng(SEED) print(f"\n[ESTRAZIONI] {args.draws} configurazioni d'ancora congiunte, " f"uniformi indipendenti (seed {SEED})") t0 = time.time() draws = [] for _ in range(args.draws): cfg = {n: int(rng.choice(SPEC[n][1])) for n in SPEC} draws.append(one_config(S, cfg)) print(f" fatte in {time.time()-t0:.0f}s") # ------------------------------------------------------------------ # Livello del book # ------------------------------------------------------------------ B = np.array([d["book"] for d in draws], float) print("\n" + "=" * 100) print(" (1) LIVELLO DEL BOOK — quanto dei numeri pubblicati e' l'ancora che ci e' capitata") print("=" * 100) print(f" {'':<22}{'canonico':>10}{'pctl':>7}{'MEDIANA':>10}{'p10':>9}{'p90':>9}" f"{'fortuna':>10}") for j, lab in enumerate(["Sharpe FULL", "Sharpe HOLD-OUT", "maxDD FULL"]): med, lo, hi = band(B[:, j]) can = canon["book"][j] fmt = (lambda x: f"{x*100:8.2f}%") if j == 2 else (lambda x: f"{x:+8.3f}") # pctl a 1 decimale: un "100%" arrotondato da 99.55% e' esattamente il numero che # inganna una lettura futura (qui 9 estrazioni su 2000 battono la canonica). print(f" {lab:<22}{fmt(can)}{pctl_of(B[:, j], can):6.1f}%{fmt(med)}" f"{fmt(lo)}{fmt(hi)}{fmt(can - med)}") print("\n (fortuna = canonico - mediana. Su maxDD un valore NEGATIVO e' fortuna:") print(" il DD canonico e' piu' basso del tipico.)") # ------------------------------------------------------------------ # Leave-one-out de-luckato # ------------------------------------------------------------------ print("\n" + "=" * 100) print(" (2) LEAVE-ONE-OUT DE-LUCKATO — contributo di ogni sleeve, mediana delle") print(" DIFFERENZE APPAIATE (book con X) - (book senza X) alla STESSA ancora") print("=" * 100) order = sorted(SPEC, key=lambda n: -float(np.median(paired_deltas(draws, n, 0)))) for j, lab in enumerate(["dSharpe FULL", "dSharpe HOLD-OUT", "dMaxDD (pp, >0 = protegge)"]): print(f"\n --- {lab} ---") print(f" {'sleeve':<9}{'w':>6}{'canonico':>11}{'pctl':>7}{'MEDIANA':>11}" f"{'p10':>10}{'p90':>10}{'>0 in':>9}") for n in order: arr = paired_deltas(draws, n, j) med, lo, hi = band(arr) can = canon[n][j] k = 100.0 if j == 2 else 1.0 u = "pp" if j == 2 else "" print(f" {SHORT[n]:<9}{WEIGHTS[n]*100:5.0f}%{can*k:+10.3f}{u}" f"{pctl_of(arr, can):6.1f}%{med*k:+10.3f}{u}{lo*k:+9.3f}{u}{hi*k:+9.3f}{u}" f"{(arr > 0).mean()*100:7.1f}%") # ------------------------------------------------------------------ # Quanto sposta il de-luck ogni contributo # ------------------------------------------------------------------ print("\n" + "=" * 100) print(" (3) FORTUNA PER SLEEVE — canonico meno mediana, sul suo stesso contributo") print("=" * 100) print(f" {'sleeve':<9}{'dFULL':>22}{'dHOLD':>22}{'dDD':>22}") print(f" {'':<9}{'can':>8}{'med':>7}{'luck':>7}{'can':>8}{'med':>7}{'luck':>7}" f"{'can':>8}{'med':>7}{'luck':>7}") for n in order: row = f" {SHORT[n]:<9}" for j in range(3): arr = paired_deltas(draws, n, j) med = float(np.median(arr)) can = canon[n][j] k = 100.0 if j == 2 else 1.0 row += f"{can*k:+8.2f}{med*k:+7.2f}{(can-med)*k:+7.2f}" print(row) # ------------------------------------------------------------------ # Correlazioni: struttura o ancora? # ------------------------------------------------------------------ print("\n" + "=" * 100) print(" (4) CORRELAZIONI FRA SLEEVE — banda sulle estrazioni (finestra comune)") print("=" * 100) rng2 = np.random.default_rng(SEED + 1) pairs = [(a, b) for i, a in enumerate(SPEC) for b in list(SPEC)[i + 1:]] acc = {p: [] for p in pairs} for _ in range(min(120, args.draws)): cfg = {n: int(rng2.choice(SPEC[n][1])) for n in SPEC} D = pd.concat({n: S[n][cfg[n]] for n in SPEC}, axis=1, join="inner").fillna(0.0) C = D.corr() for a, b in pairs: acc[(a, b)].append(float(C.loc[a, b])) Dc = pd.concat({n: S[n][SPEC[n][2]] for n in SPEC}, axis=1, join="inner").fillna(0.0).corr() print(f" {'coppia':<16}{'canonica':>10}{'MEDIANA':>10}{'p10':>9}{'p90':>9}") for a, b in sorted(pairs, key=lambda p: -abs(np.median(acc[p]))): arr = np.array(acc[(a, b)], float) med, lo, hi = band(arr) print(f" {SHORT[a]+'-'+SHORT[b]:<16}{Dc.loc[a, b]:+9.3f}{med:+10.3f}" f"{lo:+9.3f}{hi:+9.3f}") print(f"\nFatto in {time.time()-t_start:.0f}s.") print("NB: attribuzione de-luckata, NON evidenza out-of-sample — il book e' misurato") print(" sugli stessi anni su cui XS01 e SKH01 sono stati scelti e affinati.") if __name__ == "__main__": main()