"""VRP-f WATCH — misura f sulle strutture VRP dalle quote REALI, e avvisa quando basta. PERCHE' ESISTE. Il gate del tenore (30/07, `r0730_vrp_tenor_gate`) ha bocciato il candidato **10g / delta -0.28 / -0.05** sul deflated-Sharpe (0.948 < 0.95), e fra le ragioni di cautela avevo scritto che quel candidato *"sta dove il modello sbaglia di piu'"* — compra l'ala piu' lontana, che il 30/07 aveva misurato sottoprezzata ~2.3x. ⚠️ **Quel sospetto e' stato MISURATO e la conclusione era sbagliata.** L'ala lontana e' molto piu' mispriced in RELATIVO (f_long 5.85 contro 2.23) ma pesa il **3.2%** del premio corto invece del **18.4%**, quindi sul credito NETTO l'effetto e' minore: f_net **0.852** contro **0.718**. Meccanismo giusto, segno della conclusione sbagliato. E' pero' una misura su **16 coppie**: la differenza appaiata e' **+0.109 con IC95 [-0.047, +0.193]**, cioe' **compatibile con zero**. Serve piu' campione, e "quando ce ne sara' abbastanza" e' un criterio, non un promemoria — un lead senza sorvegliante e senza data e' un lead perso (lezione DVOLSPREAD, 26/07). Questo script fa la misura a ogni giro e **avvisa da solo** quando il campione basta. CRITERIO DI SUFFICIENZA — PRE-REGISTRATO IL 2026-07-30, PRIMA DI AVERE IL CAMPIONE: (a) >= 40 coppie appaiate (asset x scadenza), contro le 16 di oggi (b) ampiezza dell'IC95 della differenza appaiata <= 0.12 (oggi 0.240) Con ~2 coppie/settimana (scadenze settimanali x 2 asset) (a) cade verso **fine ottobre 2026**. Quando entrambe sono soddisfatte lo script manda una notifica UNA volta e si zittisce. ⚠️ **COSA QUESTA MISURA NON FA.** Non promuove niente. Il candidato e' bocciato sul deflated-Sharpe, che non dipende da f; e la regola "niente short-vol da modello in deploy" resta. Un f favorevole toglie UN argomento di cautela, non aggiunge un edge. Il valore operativo vero e' sull'altra riga: il f del **canonico** e' cio' che rende onesti i numeri di uno sleeve che sta nel book. uv run python scripts/live/vrp_f_watch.py uv run python scripts/live/vrp_f_watch.py --quiet # per il cron """ from __future__ import annotations import argparse import json import sys from pathlib import Path import numpy as np import pandas as pd PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) sys.path.insert(0, str(PROJECT_ROOT / "scripts" / "research")) from cblib import ( # noqa: E402 dvol_series, f_factors, load_chain, pick_legs, puts, spot_series, ) from src.live.notifier import notify # noqa: E402 STATE = PROJECT_ROOT / "data" / "vrp_f_watch" / "state.json" # --- config CONGELATA il 2026-07-30 --------------------------------------- STRUCTS = { # nome (dte_lo, dte_hi, dte_target, short_delta, long_delta) "canonico": (4.0, 10.0, 7.0, -0.28, -0.10), # VRP01 in produzione "candidato": (8.0, 13.0, 10.0, -0.28, -0.05), # cella scelta al buio dal gate 30/07 } MIN_PAIRS = 40 # criterio (a), pre-registrato MAX_CI_WIDTH = 0.12 # criterio (b), pre-registrato BOOT = 20000 SEED = 7 ASSETS = ("BTC", "ETH") def _boot_median_ci(x: np.ndarray, seed: int = SEED) -> tuple[float, float]: if len(x) < 3: return float("nan"), float("nan") r = np.random.default_rng(seed) b = np.median(r.choice(x, (BOOT, len(x)), replace=True), axis=1) return float(np.percentile(b, 2.5)), float(np.percentile(b, 97.5)) def measure(chain: pd.DataFrame | None = None) -> pd.DataFrame: """Un'osservazione per (asset, struttura, scadenza): lo snapshot col DTE piu' vicino al bersaglio in cui ENTRAMBE le gambe sono quotate.""" ch = load_chain() if chain is None else chain rows = [] for asset in ASSETS: p = puts(asset, ch) S, V = spot_series(asset), dvol_series(asset) for name, (lo, hi, tgt, sd, ld) in STRUCTS.items(): w = p[(p["dte"] >= lo) & (p["dte"] <= hi)] for exp, g in w.groupby("exp"): order = sorted(g["ts"].unique(), key=lambda t: abs((exp - pd.Timestamp(t)).total_seconds() / 86400.0 - tgt)) for ts in order: legs = pick_legs(g[g["ts"] == ts], sd, ld) if legs is None: continue t = pd.Timestamp(ts).as_unit("ns") dte = (exp - pd.Timestamp(ts)).total_seconds() / 86400.0 ff = f_factors(legs, float(S.asof(t)), float(V.asof(t)) / 100.0, dte) rows.append(dict(asset=asset, struct=name, exp=exp, ts=t, dte=dte, f_short=ff["f_short"], f_long=ff["f_long"], f_net=ff["f_net"], f_net_mid=ff["f_net_mid"], quota_lunga=ff["mod_long"] / ff["mod_short"] if ff["mod_short"] > 0 else np.nan)) break return pd.DataFrame(rows) def assess(R: pd.DataFrame) -> dict: """Statistica appaiata per (asset, scadenza): stessa scadenza, due strutture.""" if R.empty: return dict(pairs=0, ready=False, reason="nessuna osservazione") piv = R.pivot_table(index=["asset", "exp"], columns="struct", values="f_net").dropna() if len(piv) < 3: return dict(pairs=int(len(piv)), ready=False, reason="troppo poche coppie") d = (piv["candidato"] - piv["canonico"]).to_numpy() lo, hi = _boot_median_ci(d) c_lo, c_hi = _boot_median_ci(piv["canonico"].to_numpy()) k_lo, k_hi = _boot_median_ci(piv["candidato"].to_numpy()) width = hi - lo out = dict( pairs=int(len(piv)), f_canonico=float(np.median(piv["canonico"])), f_canonico_ci=[c_lo, c_hi], f_candidato=float(np.median(piv["candidato"])), f_candidato_ci=[k_lo, k_hi], diff=float(np.median(d)), diff_ci=[lo, hi], ci_width=float(width), n_positive=int((d > 0).sum()), esclude_zero=bool(lo > 0 or hi < 0), ) out["ready"] = bool(out["pairs"] >= MIN_PAIRS and width <= MAX_CI_WIDTH) return out def _load_state() -> dict: try: return json.loads(STATE.read_text()) except Exception: return {} def _save_state(d: dict) -> None: STATE.parent.mkdir(parents=True, exist_ok=True) STATE.write_text(json.dumps(d, indent=2, default=str)) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--quiet", action="store_true", help="stampa solo se c'e' qualcosa da dire") args = ap.parse_args() R = measure() a = assess(R) prev = _load_state() if not args.quiet: print("=" * 88) print(" VRP-f WATCH — f dalle quote reali, canonico vs candidato del gate 30/07") print("=" * 88) if R.empty: print("\n nessuna osservazione nella catena.") else: print(f"\n {'struttura':<12}{'oss.':>6}{'f_short':>10}{'f_long':>10}" f"{'f_net':>10}{'quota ala/corta':>18}") for name, g in R.groupby("struct"): print(f" {name:<12}{len(g):>6}{g['f_short'].median():>10.3f}" f"{g['f_long'].median():>10.3f}{g['f_net'].median():>10.3f}" f"{g['quota_lunga'].median()*100:>17.1f}%") print(f"\n coppie appaiate (stesso asset, stessa scadenza): {a['pairs']}") print(f" f canonico {a['f_canonico']:.3f} IC95 [{a['f_canonico_ci'][0]:.3f}, {a['f_canonico_ci'][1]:.3f}]") print(f" f candidato {a['f_candidato']:.3f} IC95 [{a['f_candidato_ci'][0]:.3f}, {a['f_candidato_ci'][1]:.3f}]") print(f" DIFFERENZA {a['diff']:+.3f} IC95 [{a['diff_ci'][0]:+.3f}, {a['diff_ci'][1]:+.3f}]" f" ampiezza {a['ci_width']:.3f} ({a['n_positive']}/{a['pairs']} positive)") print(f"\n criterio pre-registrato 2026-07-30: >= {MIN_PAIRS} coppie E ampiezza IC95 <= {MAX_CI_WIDTH}") print(f" coppie {a['pairs']:>3} / {MIN_PAIRS} {'OK' if a['pairs'] >= MIN_PAIRS else 'non ancora'}") print(f" ampiezza {a['ci_width']:.3f} / {MAX_CI_WIDTH} " f"{'OK' if a['ci_width'] <= MAX_CI_WIDTH else 'non ancora'}") print(f"\n -> {'CAMPIONE SUFFICIENTE' if a['ready'] else 'campione ancora insufficiente: si ASPETTA, non si decide'}") print("\n ⚠ Anche a campione pieno questa misura NON promuove il candidato: e' bocciato") print(" sul deflated-Sharpe, che non dipende da f. Toglie un argomento di cautela.") print() if a.get("ready") and not prev.get("notified"): notify("📐 VRP-f WATCH — campione sufficiente", f"Le coppie appaiate sono {a['pairs']} (soglia {MIN_PAIRS}) e l'IC95 della " f"differenza è ampio {a['ci_width']:.3f} (soglia {MAX_CI_WIDTH}).\n\n" f"f canonico {a['f_canonico']:.3f} · f candidato {a['f_candidato']:.3f} · " f"differenza {a['diff']:+.3f} IC95 [{a['diff_ci'][0]:+.3f}, {a['diff_ci'][1]:+.3f}]" f"{' — ESCLUDE lo zero' if a['esclude_zero'] else ' — compatibile con zero'}.\n\n" f"Non promuove nulla: il candidato resta bocciato sul deflated-Sharpe.") a["notified"] = True else: a["notified"] = bool(prev.get("notified", False)) _save_state(a) return 0 if __name__ == "__main__": raise SystemExit(main())