"""r0701_vrp_refine — AFFINAMENTO VRP01 (gate/sizing) dentro i limiti del modello (2026-07-01). Baseline = VRP01 combo (sleeves._vrp_combo_returns): put credit spread settimanale -0.28/-0.10, f=1.0, tenor 7d, gate VRP>0 (DVOL>RV30 causale) AND IV-rank>0.30 AND crash-skip IV-rank>0.90, fee 12.5% del premio. FULL Sh ~1.10 / HOLD ~0.60 / DD ~12%. Celle NUOVE (mai provate — verificato nei diari; l'active management intra-trade e' gia' SCARTATO in 2026-06-20-vrp-active-management.md e NON si ripete): 1. SIZING sul gap IV-RV (il carry atteso): size lineare clip(vrp/scale,0,1) o percentile espandente causale del VRP, invece del (o in aggiunta al) gate binario IV-rank. NB: il gate composito "IV-rank>0.30 AND IV-RV>0" e' GIA' il baseline (gate_vrp=True). 2. Filtro DVOL-MOMENTUM: non vendere vol mentre DVOL sta salendo (dv[i]-dv[i-k] > thr). (Diverso da dvol_directional 2026-06-29: la' il DVOL-mom era segnale DIREZIONALE sul perp.) 3. Gate di REGIME da TP01: de-risk (skip o half-size) quando TP01 e' flat su BTC e ETH (risk-off). Rischio ridondanza col trend -> riporto la frequenza d'intervento REALE. 4. Croce completa delle manopole (griglia contenuta, 105 celle, TUTTE contate nel DSR). Metodo: stessa pipeline di options_vrp_v2 (pricing BS su DVOL reale, payoff sul path certificato, stesse fee) — cambiano SOLO gate/sizing. Selezione cella IN-SAMPLE (pre-2025), hold-out 2025-26, multi-cut (5 tagli), deflated-Sharpe su tutti i trial, effetto a livello portafoglio 4-sleeve (TP01 41.25 / XS01 18.75 / VRP 15 / SKH01 25). ONESTA': il premio resta MODELLATO su DVOL ATM (no skew), book 1d, f di stress non catturato. Il verdetto massimo possibile e' "sleeve modellato migliorato", MAI deploy pieno. uv run python scripts/research/r0701_vrp_refine.py [--skip-portfolio] """ from __future__ import annotations import sys from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) sys.path.insert(0, str(PROJECT_ROOT / "scripts" / "research" / "alt")) from collections import Counter from functools import lru_cache import numpy as np import pandas as pd from scripts.research.options_vrp_lab import bs_put, strike_from_delta, load_series, m_weekly, per_year from altlib import deflated_sharpe HOLDOUT = pd.Timestamp("2025-01-01", tz="UTC") WK_PER_YEAR = 365.25 / 7.0 CUTS = [pd.Timestamp(c, tz="UTC") for c in ("2023-01-01", "2023-07-01", "2024-01-01", "2024-07-01", "2025-01-01")] MIN_IS_ACTIVE = 0.20 # attivita' minima in-sample per candidarsi (baseline ~41%) # --- parametri FISSI del baseline VRP01 (NON toccati: cambia solo gate/sizing) --- SHORT_DELTA, LONG_DELTA, F, TENOR_D = -0.28, -0.10, 1.0, 7 CRASH_SKIP, FEE_FRAC = 0.90, 0.125 # ----------------------------- pre-compute per asset (causale) ----------------------------- @lru_cache(maxsize=None) def prep(asset: str): """px/dvol allineati + VRP causale (DVOL - RV30) e IV-rank espandente per OGNI giorno. vrp[i] usa i 30 log-ret che finiscono a close[i]; ivr[i] = percentile di dv[i] in dv[:i].""" J = load_series(asset) px = J["px"].values.astype(float) dv = J["dvol"].values.astype(float) / 100.0 idx = J.index n = len(px) lr = np.diff(np.log(px)) # lr[k] = log(px[k+1]/px[k]) vrp = np.full(n, np.nan) for i in range(31, n): vrp[i] = dv[i] - float(np.std(lr[i - 30:i]) * np.sqrt(365.25)) # come baseline (ddof=0) ivr = np.full(n, np.nan) for i in range(60, n): ivr[i] = float((dv[:i] < dv[i]).mean()) return px, dv, idx, vrp, ivr @lru_cache(maxsize=None) def tp01_avg_target(): """Serie giornaliera del target medio TP01 (BTC+ETH)/2. target[i] usa solo dati <= close[i] -> noto alla sell-date del VRP (stessa close). Long-flat: 0.0 = risk-off pieno.""" from src.data.downloader import load_data from src.strategies.trend_portfolio import TrendPortfolio, CANONICAL, resample_1d tp = TrendPortfolio(**CANONICAL) cols = {} for a in ("BTC", "ETH"): df = resample_1d(load_data(a, "1h")) t = pd.Series(np.nan_to_num(tp.target_series(df), nan=0.0), index=pd.to_datetime(df["datetime"])) if t.index.tz is None: t.index = t.index.tz_localize("UTC") cols[a] = t J = pd.concat(cols, axis=1, join="inner") return J.mean(axis=1) # ----------------------------- motore settimanale (unica differenza: gate/sizing) ----------------------------- def vrp_weekly(asset: str, sizing="bin", prop_scale=0.10, ivr_gate=0.30, mom_k=0, mom_thr=0.0, tp_mode="off") -> tuple[pd.Series, Counter]: """Put credit spread settimanale come VRP01, con gate/sizing parametrici. CAUSALE: strike/premio/gate/size usano solo dati <= sell-date; payoff a scadenza sul path certificato. Ordine gate: prima i gate BASELINE (vrp/crash/ivr), poi i NUOVI (mom, tp) -> i counter dei nuovi gate contano l'intervento MARGINALE (settimane altrimenti tradabili).""" px, dv, idx, vrp_a, ivr_a = prep(asset) n = len(px); T = TENOR_D / 365.25 tpv = None if tp_mode != "off": tpv = tp01_avg_target().reindex(idx, method="ffill").values rets = {}; st = Counter() i = 60 while i + TENOR_D < n: st["weeks"] += 1 S0 = px[i]; sig = dv[i]; vrp = vrp_a[i]; ivr = ivr_a[i] blocked = None # --- gate BASELINE (identici a VRP01) --- if np.isnan(vrp) or vrp <= 0: blocked = "vrp" elif not np.isnan(ivr) and ivr > CRASH_SKIP: blocked = "crash" elif ivr_gate > 0 and not np.isnan(ivr) and ivr < ivr_gate: blocked = "ivr" # --- gate NUOVI (contati sul residuo tradabile) --- if blocked is None and mom_k > 0 and i >= mom_k: if (dv[i] - dv[i - mom_k]) > mom_thr: blocked = "mom" size = 1.0 if blocked is None and tp_mode != "off" and tpv is not None and tpv[i] <= 1e-12: if tp_mode == "skip": blocked = "tp" else: # half-size in risk-off size *= 0.5; st["tp_half"] += 1 if blocked is None and sizing != "bin": if sizing == "lin": # size ∝ gap IV-RV (carry atteso) size *= float(np.clip(vrp / prop_scale, 0.0, 1.0)) elif sizing == "rank": # percentile espandente causale del VRP hist = vrp_a[31:i]; hist = hist[~np.isnan(hist)] size *= float((hist < vrp).mean()) if len(hist) >= 30 else 0.5 if blocked is not None: st[f"blk_{blocked}"] += 1 rets[idx[i + TENOR_D]] = 0.0 i += TENOR_D continue st["traded"] += 1; st["size_sum"] += size Ks = strike_from_delta(S0, T, sig, SHORT_DELTA) Kl = strike_from_delta(S0, T, sig, LONG_DELTA) net_prem = (bs_put(S0, Ks, T, sig) - bs_put(S0, Kl, T, sig)) * F S1 = px[i + TENOR_D] payoff = max(0.0, Ks - S1) - max(0.0, Kl - S1) pnl = net_prem - payoff - FEE_FRAC * abs(net_prem) rets[idx[i + TENOR_D]] = size * pnl / Ks # cash-secured su strike corto i += TENOR_D return pd.Series(rets), st def book(**kw) -> tuple[pd.Series, Counter]: rB, sB = vrp_weekly("BTC", **kw) rE, sE = vrp_weekly("ETH", **kw) b = pd.concat({"B": rB, "E": rE}, axis=1, join="inner").mean(axis=1).sort_index() return b, sB + sE # ----------------------------- metriche ----------------------------- def sh_wk(r: pd.Series) -> float: r = r.dropna() if len(r) < 8 or r.std() == 0: return float("nan") return float(r.mean() / r.std() * np.sqrt(WK_PER_YEAR)) def cell_metrics(b: pd.Series) -> dict: is_ = b[b.index < HOLDOUT]; ho = b[b.index >= HOLDOUT] full = m_weekly(b) return dict(full_sh=full["sh"], full_dd=full["dd"], full_cagr=full["cagr"], is_sh=sh_wk(is_), hold_sh=sh_wk(ho), worst=float(b.min()), active=float((b != 0).mean()), is_active=float((is_ != 0).mean())) def multicut(cand: pd.Series, base: pd.Series) -> list[tuple[str, float, float, float]]: out = [] for c in CUTS: sc, sb = sh_wk(cand[cand.index >= c]), sh_wk(base[base.index >= c]) out.append((str(c.date()), sc, sb, sc - sb)) return out # ----------------------------- griglia ----------------------------- def grid_cells(): sizings = [("bin", 0.0, 0.30), ("lin", 0.08, 0.30), ("lin", 0.08, 0.0), ("lin", 0.12, 0.30), ("lin", 0.12, 0.0), ("rank", 0.0, 0.30), ("rank", 0.0, 0.0)] moms = [(0, 0.0), (5, 0.0), (5, 0.05), (10, 0.0), (10, 0.05)] tps = ["off", "skip", "half"] cells = [] for sz, scale, ivr in sizings: for mk, mth in moms: for tp in tps: name = (f"{sz}{f'{scale:g}' if sz == 'lin' else ''}" f"|ivr{ivr:g}|mom{mk}k{mth:g}|tp-{tp}") cells.append(dict(name=name, sizing=sz, prop_scale=scale, ivr_gate=ivr, mom_k=mk, mom_thr=mth, tp_mode=tp)) return cells BASELINE_NAME = "bin|ivr0.3|mom0k0|tp-off" # ----------------------------- portafoglio 4-sleeve ----------------------------- def weekly_to_daily_lump(wk: pd.Series) -> pd.Series: """Come sleeves._vrp_combo_returns: rendimento settimanale sul giorno di scadenza, 0 altrove.""" days = pd.date_range(wk.index.min().normalize(), wk.index.max().normalize(), freq="1D", tz="UTC") daily = pd.Series(0.0, index=days) daily.loc[wk.index.normalize()] = wk.values return daily def portfolio_compare(base_wk: pd.Series, cand_wk: pd.Series, cand_name: str): """4-sleeve con VRP baseline vs VRP variante (stessi TP01/XS01/SKH01, cache condivisa).""" from src.portfolio.sleeves import tp01_sleeve, xsec_sleeve, skyhook_sleeve from src.portfolio.portfolio import Sleeve, StrategyPortfolio, metrics tp, xs, sk = tp01_sleeve(weight=0.4125), xsec_sleeve(weight=0.1875), skyhook_sleeve(weight=0.25) rows = [] for tag, wk in (("VRP01 baseline", base_wk), (f"VRP variante [{cand_name}]", cand_wk)): daily = weekly_to_daily_lump(wk) vrp = Sleeve("VRP01_shortvol", 0.15, lambda d=daily: d) port = StrategyPortfolio([tp, xs, vrp, sk]) full = metrics(port.combined_daily()) hold = metrics(port.combined_daily(lo=HOLDOUT)) rows.append((tag, full, hold)) print(f" {tag:<38} FULL Sh {full['sharpe']:>5.2f} DD {full['maxdd']*100:>4.1f}% " f"CAGR {full['cagr']*100:>+5.1f}% | HOLD Sh {hold['sharpe']:>5.2f} DD {hold['maxdd']*100:>4.1f}%") return rows # ----------------------------- main ----------------------------- def main(): skip_port = "--skip-portfolio" in sys.argv print("=" * 110) print(" r0701 VRP REFINE — sizing IV-RV / filtro DVOL-momentum / gate TP01 (griglia onesta, sel. in-sample)") print("=" * 110) cells = grid_cells() print(f" griglia: {len(cells)} celle (TUTTE contate nel deflated-Sharpe). " f"IS = pre-2025, HOLD = 2025-01-01+.\n") results = {} for c in cells: b, st = book(sizing=c["sizing"], prop_scale=c["prop_scale"], ivr_gate=c["ivr_gate"], mom_k=c["mom_k"], mom_thr=c["mom_thr"], tp_mode=c["tp_mode"]) results[c["name"]] = dict(cfg=c, b=b, st=st, **cell_metrics(b)) base = results[BASELINE_NAME] print(f" (0) BASELINE riprodotto [{BASELINE_NAME}]:") print(f" FULL Sh {base['full_sh']:.2f} DD {base['full_dd']*100:.0f}% CAGR {base['full_cagr']*100:+.0f}% " f"worst {base['worst']*100:+.1f}% IS Sh {base['is_sh']:.2f} HOLD Sh {base['hold_sh']:.2f} " f"attivo {base['active']*100:.0f}% (atteso ~ FULL 1.10 / HOLD 0.60 / DD 12%)") # ---- frequenza d'intervento dei gate NUOVI (sul baseline + singola manopola) ---- print("\n (1) FREQUENZA D'INTERVENTO dei gate nuovi (settimane altrimenti tradabili, book BTC+ETH):") probes = [("mom k=5 thr=0", dict(mom_k=5, mom_thr=0.0)), ("mom k=5 thr=5pt", dict(mom_k=5, mom_thr=0.05)), ("mom k=10 thr=0", dict(mom_k=10, mom_thr=0.0)), ("mom k=10 thr=5pt", dict(mom_k=10, mom_thr=0.05)), ("tp01-skip", dict(tp_mode="skip")), ("tp01-half", dict(tp_mode="half"))] base_traded = base["st"]["traded"] for label, kw in probes: _, st = book(**kw) blk = st.get("blk_mom", 0) + st.get("blk_tp", 0) half = st.get("tp_half", 0) extra = f" (+{half} sett. a mezza size)" if half else "" print(f" {label:<18} blocca {blk:>3} / {base_traded} settimane-trade del baseline " f"({100*blk/max(base_traded,1):>4.1f}%){extra}") tgt = tp01_avg_target() pxB, _, idxB, _, _ = prep("BTC") tp_on_grid = tgt.reindex(idxB, method="ffill") print(f" [contesto] TP01 flat (BTC+ETH entrambi 0): {100*float((tp_on_grid <= 1e-12).mean()):.0f}% dei giorni della finestra DVOL") # ---- classifica IN-SAMPLE (selezione onesta: nessuno sguardo all'hold-out) ---- ranked = sorted((r for r in results.values() if r["is_active"] >= MIN_IS_ACTIVE), key=lambda r: r["is_sh"], reverse=True) print(f"\n (2) TOP-10 per Sharpe IN-SAMPLE (pre-2025; filtro attivita' IS >= {MIN_IS_ACTIVE:.0%}):") print(f" {'cella':<34}{'IS Sh':>7}{'FULL':>7}{'HOLD':>7}{'DD':>6}{'worst':>8}{'att.':>6}") for r in ranked[:10]: print(f" {r['cfg']['name']:<34}{r['is_sh']:>7.2f}{r['full_sh']:>7.2f}{r['hold_sh']:>7.2f}" f"{r['full_dd']*100:>5.0f}%{r['worst']*100:>+7.1f}%{r['active']*100:>5.0f}%") # ---- varianti a SINGOLA manopola vs baseline (tabella diario) ---- print("\n (2b) VARIANTI A SINGOLA MANOPOLA vs baseline (stessa tabella, nessuna selezione):") singles = ["lin0.08|ivr0.3|mom0k0|tp-off", "lin0.12|ivr0.3|mom0k0|tp-off", "rank|ivr0.3|mom0k0|tp-off", "lin0.08|ivr0|mom0k0|tp-off", "rank|ivr0|mom0k0|tp-off", "bin|ivr0.3|mom5k0.05|tp-off", "bin|ivr0.3|mom10k0.05|tp-off", "bin|ivr0.3|mom5k0|tp-off", "bin|ivr0.3|mom10k0|tp-off", "bin|ivr0.3|mom0k0|tp-skip", "bin|ivr0.3|mom0k0|tp-half"] print(f" {'cella':<34}{'IS Sh':>7}{'FULL':>7}{'HOLD':>7}{'DD':>6}{'worst':>8}{'att.':>6}") r = base print(f" {'BASELINE ' + BASELINE_NAME:<34}{r['is_sh']:>7.2f}{r['full_sh']:>7.2f}{r['hold_sh']:>7.2f}" f"{r['full_dd']*100:>5.0f}%{r['worst']*100:>+7.1f}%{r['active']*100:>5.0f}%") for nm in singles: r = results[nm] print(f" {nm:<34}{r['is_sh']:>7.2f}{r['full_sh']:>7.2f}{r['hold_sh']:>7.2f}" f"{r['full_dd']*100:>5.0f}%{r['worst']*100:>+7.1f}%{r['active']*100:>5.0f}%") n_beat_hold = sum(1 for r in results.values() if r["hold_sh"] > base["hold_sh"]) print(f" [onesta'] celle che battono l'HOLD-OUT del baseline: {n_beat_hold}/{len(results)} — " f"NON selezionabili (sarebbe selection-on-holdout, gate 2026-06-29).") cand = ranked[0] is_baseline_best = cand["cfg"]["name"] == BASELINE_NAME print(f"\n -> cella scelta IN-SAMPLE: [{cand['cfg']['name']}] IS Sh {cand['is_sh']:.2f} " f"(baseline IS {base['is_sh']:.2f}, Δ {cand['is_sh']-base['is_sh']:+.2f})") # ---- hold-out multi-cut vs baseline ---- print("\n (3) MULTI-CUT hold-out (Sharpe da ogni taglio a fine storia; uplift = cand - baseline):") mc = multicut(cand["b"], base["b"]) pos = sum(1 for _, _, _, u in mc if u > 0) for cut, sc, sb, u in mc: print(f" cut {cut}: cand {sc:>5.2f} base {sb:>5.2f} uplift {u:>+5.2f}") print(f" uplift positivo in {pos}/{len(mc)} tagli (richiesti >= 4/5)") # ---- deflated Sharpe (tutti i trial della griglia) ---- all_sh = [r["full_sh"] for r in results.values()] dsr_c, null_max = deflated_sharpe(cand["full_sh"], all_sh, cand["b"].values, dpy=WK_PER_YEAR) dsr_b, _ = deflated_sharpe(base["full_sh"], all_sh, base["b"].values, dpy=WK_PER_YEAR) print(f"\n (4) DEFLATED SHARPE (N={len(all_sh)} trial di questa griglia; PASS >= 0.95):") print(f" cand DSR {dsr_c:.3f} (null-max Sh {null_max:.2f}) | baseline DSR {dsr_b:.3f}") print(" NB: le celle della griglia sono fortemente correlate fra loro (stesso trade sottostante)") print(" -> il DSR qui e' anti-conservativo sul multiple-testing; in piu' VRP01 stesso viene da") print(" ~20 config precedenti (options_vrp_lab/_v2). Leggere il DSR come limite SUPERIORE.") # ---- per-anno cand vs base ---- print("\n (5) PER-ANNO (ritorno composto):") pyc, pyb = per_year(cand["b"]), per_year(base["b"]) print(" base: " + " ".join(f"{y}:{v*100:+.0f}%" for y, v in pyb.items())) print(" cand: " + " ".join(f"{y}:{v*100:+.0f}%" for y, v in pyc.items())) # ---- portafoglio 4-sleeve ---- if not skip_port: print("\n (6) PORTAFOGLIO 4-SLEEVE (TP01 41.25 / XS01 18.75 / VRP 15 / SKH01 25), VRP base vs variante:") try: portfolio_compare(base["b"], cand["b"], cand["cfg"]["name"]) except Exception as e: # dati HL/5m mancanti in qualche ambiente print(f" [saltato: {type(e).__name__}: {e}]") else: print("\n (6) portafoglio: saltato (--skip-portfolio)") # ---- verdetto ---- print("\n" + "=" * 110) improves = (not is_baseline_best and cand["is_sh"] > base["is_sh"] and pos >= 4 and (cand["hold_sh"] > base["hold_sh"]) and dsr_c >= 0.95) if is_baseline_best: print(" VERDETTO: NON MIGLIORA — il baseline VRP01 vince gia' la selezione in-sample.") elif improves: print(f" VERDETTO: MIGLIORA (variante {cand['cfg']['name']}) — batte il baseline in-sample,") print(f" su hold-out multi-cut ({pos}/{len(mc)}) e DSR {dsr_c:.2f}>=0.95. Resta SLEEVE MODELLATO") print(" (premio DVOL ATM, book 1d, f di stress non catturato): NON deploy pieno.") else: why = [] if cand["is_sh"] <= base["is_sh"]: why.append("non batte il baseline in-sample") if pos < 4: why.append(f"multi-cut {pos}/{len(mc)} (<4)") if cand["hold_sh"] <= base["hold_sh"]: why.append("hold-out non migliore") if dsr_c < 0.95: why.append(f"DSR {dsr_c:.2f}<0.95") print(f" VERDETTO: NON MIGLIORA — cella IS-best [{cand['cfg']['name']}] bocciata: " + "; ".join(why) + ".") print("=" * 110) if __name__ == "__main__": main()