Files
Adriano Dal Pastro 491411ac77 research(wave-0701): 6 filoni multi-agente — 0 nuovi sleeve, pesi confermati, gate weights_tilt_null
Ondata onesta su angoli non coperti: funding-TS (chiude il filone funding su 3
lati), breadth alt (non-ridondante ma DSR 0.43, rivisitabile con storia),
XS-residmom (REDUNDANT), pesi+guardia-DD (EW-STR refutato dallo scettico come
selezione-sull'hold-out di 2° ordine, firma best-of-15), VRP-refine (filone
esaurito), stagionalità-XS (morta allo step statistico).

Lezione codificata: weights_tilt_null + combine_outer in src/portfolio
(ogni cambio-pesi vs null di tilt casuali cap-respecting + delta in-sample>=0);
5 test nuovi, suite 165/165.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-01 23:21:59 +00:00

370 lines
18 KiB
Python

"""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()