#!/usr/bin/env python """r0702_anchor_xs01.py — TIMING-LUCK della FASE del ciclo di ribilanciamento di XS01. XS01 (`src/portfolio/sleeves._xsec_returns`, XS_CFG H=10) ribilancia ogni 10 giorni quando `i % H == 0` con `i` = indice di riga della matrice prezzi inner-joined (start 2024-01-01). La FASE del ciclo (quale dei 10 giorni possibili) e' quindi un artefatto della prima riga dei dati — esattamente come l'ancora oraria di TP01 (vedi r0702_tp01_offset.py + r0702_skeptic_offset.py, diario 2026-07-02-timing-crt-wave.md). Spazio di luck: 10 fasi. Questo script (metodologia identica all'audit TP01, ZERO tuning per-fase): 1. REPLICA `_xsec_returns` con parametro `phase` (i % H == phase). SANITY OBBLIGATORIO: phase=0 deve riprodurre ESATTAMENTE la serie di `_xsec_returns()` (max|dif| ~ 0). Il gate di dispersione (percentile ESPANDENTE causale) e' ricalcolato PER FASE: disp_hist accumula solo nei giorni di ribilanciamento di QUELLA fase (replica fedele). 2. Tabella delle 10 fasi: Sharpe/CAGR/maxDD FULL (serie dal 2024) e HOLD-OUT 2025-26; min/mediana/max + percentile della fase canonica (phase=0). 3. ENSEMBLE delle 10 fasi (media dei ritorni = 1/10 del capitale per fase) vs canonica. 4. IMPATTO SUL BOOK 5-sleeve (TP 33 / XS 15 / VRP 12 / SKH 20 / GTAA 20, combine_outer con pesi rinormalizzati, attivazione era crypto): XS canonica vs fase mediana vs fase peggiore vs ensemble vs senza-XS -> HOLD e FULL del book. 5. GATE DI AMMISSIONE: plateau (lookback singoli/blend, disp_pct 15-35) ricalcolato alla fase MEDIANA e alla canonica: i numeri di ammissione (FULL 1.50/HOLD 1.71/DD 11%) reggono? 6. BOOTSTRAP alla maniera dello scettico: block-bootstrap congiunto delle 10 fasi sull'hold-out -> P(una fase qualsiasi mostri uno spike >= quello della canonica) + CI95 (storia ~2.5y -> quantifica l'ampiezza). Tecnica: mai .view su tz-aware (epoca via pd.to_datetime unit='ms', come l'originale); posizioni shiftate come nell'originale (gross[1:] = W[:-1]*dret[1:]); vol-target rolling ricalcolato per fase sulla PROPRIA serie netta. Nessun file toccato fuori da questo script. Runtime ~2-4 min (il grosso e' SKH01 dal 5m per il book). """ from __future__ import annotations import sys from pathlib import Path import numpy as np import pandas as pd ROOT = Path("/opt/docker/PythagorasGoal") sys.path.insert(0, str(ROOT)) from src.portfolio.portfolio import HOLDOUT, combine_outer, to_daily # noqa: E402 from src.portfolio.sleeves import (_HL_DIR, XS_CFG, XS_UNIVERSE, # noqa: E402 _xsec_returns) RNG_SEED = 42 B_BOOT = 4000 H = XS_CFG["H"] # 10 -> 10 fasi possibili FULL_START = pd.Timestamp("2024-01-01", tz="UTC") # --------------------------------------------------------------------------- # Metriche (convenzioni identiche ad altlib._sh/_dd_ret e a r0702_tp01_offset.dmetrics) # --------------------------------------------------------------------------- def _sh(s) -> float: r = np.asarray(pd.Series(s).dropna().values, float) return float(np.mean(r) / np.std(r) * np.sqrt(365.25)) if len(r) > 2 and np.std(r) > 0 else 0.0 def _dd(s) -> float: eq = np.cumprod(1.0 + np.asarray(pd.Series(s).dropna().values, float)) pk = np.maximum.accumulate(eq) return float(np.max((pk - eq) / pk)) if len(eq) else 0.0 def _cagr(s) -> float: r = np.asarray(pd.Series(s).dropna().values, float) eq = float(np.prod(1.0 + r)) yrs = len(r) / 365.25 return eq ** (1 / yrs) - 1 if eq > 0 and yrs > 0 else -1.0 def dmetrics(s: pd.Series) -> dict: s = s.dropna() ho = s[s.index >= HOLDOUT] return dict(sh_full=_sh(s), cagr_full=_cagr(s), dd_full=_dd(s), sh_hold=_sh(ho), cagr_hold=_cagr(ho), dd_hold=_dd(ho), n=len(s)) def pctl_of(v: np.ndarray, x: float) -> float: return float((v < x).mean() + 0.5 * (v == x).mean()) * 100.0 # --------------------------------------------------------------------------- # Prezzi: caricamento IDENTICO a _xsec_returns (stesso ordine di operazioni) # --------------------------------------------------------------------------- _PX_CACHE: pd.DataFrame | None = None def load_prices() -> pd.DataFrame: global _PX_CACHE if _PX_CACHE is not None: return _PX_CACHE cols = {} for sym in XS_UNIVERSE: p = _HL_DIR / f"hl_{sym.lower()}_1d.parquet" if not p.exists(): continue d = pd.read_parquet(p) cols[sym] = pd.Series(d["close"].values.astype(float), index=pd.to_datetime(d["timestamp"], unit="ms", utc=True)) if len(cols) < 10: raise FileNotFoundError("universo Hyperliquid XS01 incompleto") _PX_CACHE = pd.concat(cols, axis=1, join="inner").sort_index().dropna() return _PX_CACHE # --------------------------------------------------------------------------- # REPLICA di _xsec_returns con parametro `phase` (+ override cfg per il plateau). # Codice copiato 1:1 da src/portfolio/sleeves._xsec_returns; UNICA differenza: # `i % H == 0` -> `i % H == phase`. Il disp_hist (percentile espandente causale) # accumula SOLO nei giorni di ribilanciamento della fase -> ricalcolato per fase. # --------------------------------------------------------------------------- def xs_phase(phase: int, lookbacks=None, disp_pct=None) -> pd.Series: cfg = dict(XS_CFG) if lookbacks is not None: cfg["lookbacks"] = lookbacks if disp_pct is not None: cfg["disp_pct"] = disp_pct C = load_prices() px = C.values n, A = px.shape lookbacks_, Hc, k, mode, tv = (cfg["lookbacks"], cfg["H"], cfg["k"], cfg["mode"], cfg["target_vol"]) disp_pct_ = cfg.get("disp_pct", 0) minhist = cfg.get("disp_minhist", 20) mlb = max(lookbacks_) dret = np.vstack([np.zeros(A), px[1:] / px[:-1] - 1.0]) W = np.zeros((n, A)) w = np.zeros(A) disp_hist = [] for i in range(n): if i >= mlb and i % Hc == phase: rLs = [px[i] / px[i - L] - 1.0 for L in lookbacks_] disp_i = float(np.mean([r.std() for r in rLs])) thr = (np.percentile(disp_hist, disp_pct_) if (disp_pct_ > 0 and len(disp_hist) >= minhist) else -np.inf) if disp_i >= thr: score = np.zeros(A) cnt = 0 for rL in rLs: sd = rL.std() if sd > 0: score += (rL - rL.mean()) / sd cnt += 1 if cnt: score /= cnt order = np.argsort(score) w = np.zeros(A) lo, hi = order[:k], order[-k:] if mode == "mom": w[hi] = 0.5 / k w[lo] = -0.5 / k else: w[lo] = 0.5 / k w[hi] = -0.5 / k else: w = np.zeros(A) disp_hist.append(disp_i) W[i] = w gross = np.zeros(n) gross[1:] = np.sum(W[:-1] * dret[1:], axis=1) # posizioni SHIFTATE (held t+1) turn = np.zeros(n) turn[0] = np.abs(W[0]).sum() turn[1:] = np.abs(np.diff(W, axis=0)).sum(axis=1) net = gross - turn * (0.001 / 2.0) s = pd.Series(net, index=C.index) rv = s.rolling(30, min_periods=15).std().shift(1) * np.sqrt(365.25) scale = np.clip(np.nan_to_num(tv / rv.replace(0, np.nan).values, nan=0.0), 0, 3.0) return pd.Series(s.values * scale, index=C.index) # --------------------------------------------------------------------------- # 1. SANITY: replica a fase canonica == _xsec_returns() (bit-exact) # --------------------------------------------------------------------------- def sanity() -> None: ref = _xsec_returns() mine = xs_phase(0) assert len(mine) == len(ref) and mine.index.equals(ref.index), "SANITY index mismatch" dmax = float(np.max(np.abs(mine.values - ref.values))) assert dmax == 0.0, f"SANITY FAIL: replica phase=0 != _xsec_returns (max|dif|={dmax:.3e})" m = dmetrics(ref) print(f"[SANITY] replica phase=0 == _xsec_returns(): OK (max|dif| = {dmax:.1e}, " f"n={len(ref)}, start {ref.index.min().date()})") print(f"[SANITY] canonica oggi: FULL Sh {m['sh_full']:.3f} / HOLD {m['sh_hold']:.3f} / " f"DD {m['dd_full']:.1%} (dichiarati all'ammissione: 1.50 / 1.71 / 11% — dati mossi da allora)") # --------------------------------------------------------------------------- # 6. BOOTSTRAP alla maniera dello scettico (r0702_skeptic_offset.block_boot_stats) # --------------------------------------------------------------------------- def _sh_mat(R: np.ndarray) -> np.ndarray: mu = R.mean(axis=1) sd = R.std(axis=1) return np.where(sd > 0, mu / sd, 0.0) * np.sqrt(365.25) def block_boot(M: np.ndarray, B: int, block: int, rng) -> dict: n, K = M.shape nblocks = int(np.ceil(n / block)) g0s, gmaxs, meds, sh0s = [], [], [], [] done = 0 while done < B: b = min(500, B - done) starts = rng.integers(0, n, size=(b, nblocks)) idx = (starts[:, :, None] + np.arange(block)[None, None, :]) % n idx = idx.reshape(b, -1)[:, :n] R = M[idx] # (b, n, K) Sh = np.stack([_sh_mat(R[:, :, kk]) for kk in range(K)], axis=1) gg = np.empty_like(Sh) for h in range(K): gg[:, h] = Sh[:, h] - np.median(np.delete(Sh, h, axis=1), axis=1) g0s.append(gg[:, 0]) gmaxs.append(gg.max(axis=1)) meds.append(np.median(Sh, axis=1)) sh0s.append(Sh[:, 0]) done += b return dict(g0=np.concatenate(g0s), gmax=np.concatenate(gmaxs), med=np.concatenate(meds), sh0=np.concatenate(sh0s)) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main() -> None: print("=" * 100) print("r0702 — XS01 rebalance PHASE timing-luck: 10 fasi del ciclo H=10 " "(metodologia = audit ancore TP01)") print(f"XS_CFG={XS_CFG} universo={len(load_prices().columns)} major HL " f"HOLD-OUT >= {HOLDOUT.date()}") print("=" * 100) sanity() # ---- (2) tabella delle 10 fasi ---------------------------------------- phases = {p: xs_phase(p) for p in range(H)} T = pd.DataFrame({p: dmetrics(s) for p, s in phases.items()}).T print("\n--- (2) PER-FASE (parametri IDENTICI, zero tuning; fase = i % 10 del ciclo) ---") print(f"{'fase':>4} {'ShFULL':>7} {'CAGRf':>7} {'DDfull':>7} {'ShHOLD':>7} " f"{'CAGRh':>7} {'DDhold':>7}") for p, r in T.iterrows(): tag = " <- canonica" if p == 0 else "" print(f"{p:>4} {r.sh_full:>7.3f} {r.cagr_full:>6.1%} {r.dd_full:>6.1%} " f"{r.sh_hold:>7.3f} {r.cagr_hold:>6.1%} {r.dd_hold:>6.1%}{tag}") print("\nDistribuzione fra le 10 fasi (min / mediana / max / std) [percentile della canonica]:") for col, lbl in (("sh_full", "Sharpe FULL"), ("sh_hold", "Sharpe HOLD"), ("cagr_full", "CAGR FULL"), ("dd_full", "maxDD FULL"), ("dd_hold", "maxDD HOLD")): v = T[col].values.astype(float) print(f" {lbl:<14} {v.min():>7.3f} / {np.median(v):>7.3f} / {v.max():>7.3f} " f"/ std {v.std():.3f} canonica al {pctl_of(v, v[0]):.0f}° pctl (val {v[0]:.3f})") hold_v = T["sh_hold"].values.astype(float) full_v = T["sh_full"].values.astype(float) med_hold = float(np.median(hold_v)) p_med = int(min(range(H), key=lambda p: (abs(hold_v[p] - med_hold), p))) p_worst = int(np.argmin(hold_v)) p_best = int(np.argmax(hold_v)) print(f"\n fase MEDIANA (per ShHOLD) = {p_med} | PEGGIORE = {p_worst} | MIGLIORE = {p_best}") # correlazione fra fasi (contestualizza il bootstrap) Jf = pd.concat(phases, axis=1, join="inner").dropna() Mh = Jf[Jf.index >= HOLDOUT].values cor = np.corrcoef(Mh.T) iu = np.triu_indices(H, 1) print(f" correlazione daily fra fasi (hold-out): mediana {np.median(cor[iu]):.3f}, " f"min {cor[iu].min():.3f}") # ---- (3) ensemble delle 10 fasi ---------------------------------------- ens = pd.Series(Jf.values.mean(axis=1), index=Jf.index) me, m0 = dmetrics(ens), dmetrics(phases[0]) print("\n--- (3) ENSEMBLE (media dei ritorni delle 10 fasi = 1/10 capitale per fase) ---") print(f"{'config':<18} {'ShFULL':>7} {'CAGRf':>7} {'DDfull':>7} {'ShHOLD':>7} {'DDhold':>7}") for name, m in (("canonica (f0)", m0), ("ensemble 10 fasi", me)): print(f"{name:<18} {m['sh_full']:>7.3f} {m['cagr_full']:>6.1%} {m['dd_full']:>6.1%} " f"{m['sh_hold']:>7.3f} {m['dd_hold']:>6.1%}") # ---- (4) impatto sul book 5-sleeve ------------------------------------- print("\n--- (4) BOOK 5-sleeve (TP 33 / XS 15 / VRP 12 / SKH 20 / GTAA 20, combine_outer, " "attivazione era crypto) ---") from src.portfolio.sleeves import (_gtaa_daily_returns, _skyhook_returns, _tp01_returns, _vrp_combo_returns) TPd = to_daily(_tp01_returns()) fixed = dict(TP=TPd, VRP=to_daily(_vrp_combo_returns()), SKH=to_daily(_skyhook_returns()), GTAA=to_daily(_gtaa_daily_returns())) Wt = dict(TP=0.33, XS=0.15, VRP=0.12, SKH=0.20, GTAA=0.20) lo = TPd.index.min() def book(xs: pd.Series | None) -> pd.Series: cols = dict(fixed) if xs is not None: cols["XS"] = to_daily(xs) wt = {k: v for k, v in Wt.items() if k in cols} return combine_outer(cols, wt, lo=lo) rows = [("XS canonica (f0)", phases[0]), (f"XS fase mediana ({p_med})", phases[p_med]), (f"XS fase peggiore ({p_worst})", phases[p_worst]), (f"XS fase migliore ({p_best})", phases[p_best]), ("XS ensemble 10f", ens), ("senza XS01 (rinorm.)", None)] print(f"{'book con':<24} {'ShHOLD':>7} {'ShFULL':>7} {'DDfull':>7} {'DDhold':>7}") book_stats = {} for name, xs in rows: b = book(xs) bh = b[b.index >= HOLDOUT] book_stats[name] = (_sh(bh), _sh(b)) print(f"{name:<24} {_sh(bh):>7.3f} {_sh(b):>7.3f} {_dd(b):>6.1%} {_dd(bh):>6.1%}") bh0 = book_stats["XS canonica (f0)"][0] print(f"\n fortuna di fase ereditata dal book HOLD: canonica {bh0:.3f} vs mediana " f"{book_stats[f'XS fase mediana ({p_med})'][0]:.3f} " f"(delta {bh0 - book_stats[f'XS fase mediana ({p_med})'][0]:+.3f}) | vs peggiore " f"{book_stats[f'XS fase peggiore ({p_worst})'][0]:.3f} " f"(delta {bh0 - book_stats[f'XS fase peggiore ({p_worst})'][0]:+.3f}) | vs senza-XS " f"{book_stats['senza XS01 (rinorm.)'][0]:.3f}") # ---- (5) gate di ammissione alla fase mediana: plateau ------------------ print("\n--- (5) GATE DI AMMISSIONE — plateau ricalcolato alla fase MEDIANA " f"({p_med}) vs canonica (0) ---") print("(ammissione XS01: FULL 1.50 / HOLD 1.71 / DD 11%, plateau lookback + disp_pct 15-35)") print(f"{'variante':<28} {'f0 FULL':>8} {'f0 HOLD':>8} {'f0 DD':>7} " f"{'fMED FULL':>9} {'fMED HOLD':>9} {'fMED DD':>8}") grid = ([(f"disp_pct={dp} (lb 30,90)", dict(disp_pct=dp)) for dp in (15, 20, 25, 30, 35)] + [(f"lookbacks={lb} (p30)", dict(lookbacks=lb)) for lb in ((30,), (60,), (90,), (30, 60, 90))]) for name, kw in grid: a = dmetrics(xs_phase(0, **kw)) b = dmetrics(xs_phase(p_med, **kw)) star = " *" if kw == dict(disp_pct=30) or kw.get("lookbacks") == (30, 90) else "" print(f"{name:<28} {a['sh_full']:>8.3f} {a['sh_hold']:>8.3f} {a['dd_full']:>6.1%} " f"{b['sh_full']:>9.3f} {b['sh_hold']:>9.3f} {b['dd_full']:>7.1%}{star}") mmed = dmetrics(phases[p_med]) print(f"\n cella canonica alla fase mediana: FULL {mmed['sh_full']:.3f} " f"(ammesso con 1.50) / HOLD {mmed['sh_hold']:.3f} (ammesso con 1.71) / " f"DD {mmed['dd_full']:.1%} (ammesso con 11%)") n_full_pos = int((full_v >= 1.0).sum()) n_hold_pos = int((hold_v >= 1.0).sum()) print(f" fasi con FULL>=1.0: {n_full_pos}/10 | fasi con HOLD>=1.0: {n_hold_pos}/10 | " f"fasi con HOLD<=0: {int((hold_v <= 0).sum())}/10") # ---- (6) bootstrap: la canonica e' speciale? ---------------------------- print("\n--- (6) BOOTSTRAP (block, congiunto sulle 10 fasi, hold-out) ---") sh_hold_obs = _sh_mat(Mh.T) g0_obs = float(sh_hold_obs[0] - np.median(sh_hold_obs[1:])) print(f"hold-out: {Mh.shape[0]} giorni, 10 fasi; Sh canonica {sh_hold_obs[0]:.3f}, " f"mediana altre {np.median(sh_hold_obs[1:]):.3f}, spike osservato g0 = {g0_obs:+.3f}") for blk in (10, 20, 40): bs = block_boot(Mh, B_BOOT, blk, np.random.default_rng(RNG_SEED + blk)) p_any = float(np.mean(bs["gmax"] >= g0_obs)) ci_g0 = np.percentile(bs["g0"], [2.5, 97.5]) ci_med = np.percentile(bs["med"], [2.5, 97.5]) ci_sh0 = np.percentile(bs["sh0"], [2.5, 97.5]) print(f" block={blk:>2}: P(spike di UNA QUALSIASI fase >= {g0_obs:+.2f}) = {p_any:.3f} | " f"CI95 g0 [{ci_g0[0]:+.2f},{ci_g0[1]:+.2f}] | CI95 Sh mediana-fasi " f"[{ci_med[0]:+.2f},{ci_med[1]:+.2f}] | CI95 Sh canonica [{ci_sh0[0]:+.2f},{ci_sh0[1]:+.2f}]") # CI sulla FULL della canonica (storia corta -> quanto sono larghi?) Mf = Jf.values bsf = block_boot(Mf, B_BOOT, 20, np.random.default_rng(RNG_SEED)) ci_f0 = np.percentile(bsf["sh0"], [2.5, 97.5]) ci_fmed = np.percentile(bsf["med"], [2.5, 97.5]) print(f" FULL (block=20): CI95 Sh canonica [{ci_f0[0]:+.2f},{ci_f0[1]:+.2f}] | " f"CI95 Sh mediana-fasi [{ci_fmed[0]:+.2f},{ci_fmed[1]:+.2f}] " f"(~2.5 anni di storia: intervalli larghi)") # ---- (7) caveat --------------------------------------------------------- print("\n--- (7) CAVEAT DICHIARATI ---") print(" (a) storia corta ~2.5y (914g, hold-out ~548g): i CI95 qui sopra quantificano " "l'incertezza — nessuna stima puntuale di Sharpe e' affidabile a +/-1.") print(" (b) dimensione ORA-DEL-GIORNO non auditata: le barre HL in data/raw sono solo 1d " "(ancora daily fissa del feed) -> luck residua non testabile con i dati certificati.") print(" (c) XS01 e' STAT-MODE (19 gambe, non eseguito live): l'impatto e' sulla STIMA del " "book (reporting/ammissione), non sull'operativita' del book live Deribit.") print("\nFatto. Nessuna selezione per-fase: parametri identici, giudizio su distribuzione " "delle fasi + ensemble; l'hold-out e' solo riportato.") if __name__ == "__main__": main()