"""SNAPSHOT catena opzioni DAILY (0-1DTE) Deribit BTC/ETH — cattura quote REALI per il dossier 0DTE. CONTESTO (2026-07-24). VRP01 (put credit spread settimanale) e' deploy-bloccato dalla regola "niente short-vol da modello": serve una serie di premi REALI, non BS-su-DVOL. Le scadenze DAILY accumulano 365 expiry/anno (7x le weekly): in ~6 mesi di cattura si ottiene una serie di ~180 premi reali a tenor giornaliero + ~26 weekly, abbastanza per stimare il VRP daily NETTO di bid/ask e fee (misura live 2026-07-24: haircut mid->netto ~30-50% a tenor daily, fee cap 12.5% quasi sempre binding perche' i premi daily sono < 0.0024 base ccy). COSA FA (sola lettura, API pubblica, nessun ordine, nessun token): - per BTC e ETH: public/get_instruments (kind=option) -> tutte le scadenze entro --max-dte-h (default 50h = la daily 0DTE + la daily 1DTE appena listata); - per ogni strumento nel ladder di moneyness (default 75-125%): public/ticker -> bid/ask/depth, mark, mark_iv, bid_iv/ask_iv, greche, OI, volume, underlying; - 1 riga JSON per strumento ("rec":"chain") + 1 riga meta per valuta ("rec":"meta", con index price e DVOL corrente) APPESE a data/options_daily/snapshots.jsonl. CADENZA RACCOMANDATA (quando/se si decide di cablarla — NON e' in cron adesso): - 08:05 UTC: subito dopo il listing della nuova daily (~24h DTE) = il premio "vendibile"; - 07:55 UTC: subito prima del settle (08:00 UTC) = chiude il ciclo (payoff realizzato). Con 2 run/giorno: ~160 strumenti/run, ~350 byte/riga -> ~120 KB/giorno, ~40 MB/anno. Banale. Ogni run extra (es. 12:00/20:00) aggiunge la dimensione intraday dello spread: opzionale. USO: uv run python scripts/research/r0724_daily_opt_snapshot.py uv run python scripts/research/r0724_daily_opt_snapshot.py --currencies BTC --max-dte-h 30 NB ONESTO: questo script MISURA, non decide. La serie che produce serve a rispondere fra ~6 mesi a: (1) IV daily vs RV daily netto haircut, (2) quanto spesso il gate IV-rank aprirebbe a tenor daily, (3) f di stress reale quando capita un crash dentro la finestra di cattura. """ from __future__ import annotations import argparse import json import sys import time from pathlib import Path import requests PROJECT_ROOT = Path(__file__).resolve().parents[2] OUT_DIR = PROJECT_ROOT / "data" / "options_daily" OUT_FILE = OUT_DIR / "snapshots.jsonl" API = "https://www.deribit.com/api/v2/public" SESSION = requests.Session() def api(endpoint: str, **params): """GET pubblico con retry breve. Ritorna result o solleva.""" last = None for _ in range(4): try: r = SESSION.get(f"{API}/{endpoint}", params=params, timeout=20) j = r.json() if "result" in j: return j["result"] last = j.get("error") except Exception as e: # rete/JSON: ritenta last = str(e) time.sleep(0.7) raise RuntimeError(f"Deribit API fail {endpoint} {params}: {last}") def snapshot_currency(cur: str, max_dte_h: float, mny_lo: float, mny_hi: float) -> list[dict]: now_ms = int(time.time() * 1000) snap_ts = now_ms rows: list[dict] = [] instruments = api("get_instruments", currency=cur, kind="option", expired="false") spot = api("get_index_price", index_name=f"{cur.lower()}_usd")["index_price"] try: dvol_data = api("get_volatility_index_data", currency=cur, start_timestamp=now_ms - 3_600_000, end_timestamp=now_ms, resolution=3600).get("data", []) dvol = float(dvol_data[-1][4]) if dvol_data else None except Exception: dvol = None chain = [i for i in instruments if (i["expiration_timestamp"] - now_ms) / 3.6e6 <= max_dte_h and mny_lo * spot <= i["strike"] <= mny_hi * spot] chain.sort(key=lambda i: (i["expiration_timestamp"], i["strike"], i["option_type"])) expiries = sorted({i["expiration_timestamp"] for i in chain}) rows.append({ "rec": "meta", "snap_ts": snap_ts, "currency": cur, "index_price": spot, "dvol": dvol, "n_instruments": len(chain), "expiries": expiries, "max_dte_h": max_dte_h, "moneyness": [mny_lo, mny_hi], }) for inst in chain: name = inst["instrument_name"] try: t = api("ticker", instrument_name=name) except RuntimeError as e: print(f" WARN ticker {name}: {e}", file=sys.stderr) continue g = t.get("greeks") or {} st = t.get("stats") or {} rows.append({ "rec": "chain", "snap_ts": snap_ts, "currency": cur, "instrument": name, "expiry_ts": inst["expiration_timestamp"], "dte_h": round((inst["expiration_timestamp"] - snap_ts) / 3.6e6, 3), "strike": inst["strike"], "type": inst["option_type"], "settlement_period": inst.get("settlement_period"), "min_trade_amount": inst.get("min_trade_amount"), "taker_comm": inst.get("taker_commission"), "bid": t.get("best_bid_price"), "ask": t.get("best_ask_price"), "bid_amount": t.get("best_bid_amount"), "ask_amount": t.get("best_ask_amount"), "mark": t.get("mark_price"), "mark_iv": t.get("mark_iv"), "bid_iv": t.get("bid_iv"), "ask_iv": t.get("ask_iv"), "delta": g.get("delta"), "gamma": g.get("gamma"), "vega": g.get("vega"), "theta": g.get("theta"), "oi": t.get("open_interest"), "volume_24h": st.get("volume"), "underlying": t.get("underlying_price"), "index_price": t.get("index_price"), }) time.sleep(0.05) # rate-limit gentile (pubblico: 20 req/s, stiamo larghi) return rows def main(): ap = argparse.ArgumentParser(description="Snapshot catena daily-expiry Deribit -> JSONL") ap.add_argument("--currencies", nargs="+", default=["BTC", "ETH"]) ap.add_argument("--max-dte-h", type=float, default=50.0, help="cattura tutte le scadenze entro N ore (default 50 = 0DTE+1DTE)") ap.add_argument("--moneyness", nargs=2, type=float, default=[0.75, 1.25], metavar=("LO", "HI"), help="ladder strike in frazione dello spot") args = ap.parse_args() OUT_DIR.mkdir(parents=True, exist_ok=True) total = 0 with OUT_FILE.open("a") as f: for cur in args.currencies: try: rows = snapshot_currency(cur, args.max_dte_h, *args.moneyness) except Exception as e: print(f"ERRORE {cur}: {e}", file=sys.stderr) continue for r in rows: f.write(json.dumps(r, separators=(",", ":")) + "\n") total += len(rows) n_chain = sum(1 for r in rows if r["rec"] == "chain") print(f"{cur}: {n_chain} strumenti (+1 meta) appesi") print(f"OK: {total} righe -> {OUT_FILE}") if __name__ == "__main__": main()