#!/usr/bin/env python3 """ r0822_basis_calendar.py — filone BASIS-CALENDAR (ondata 2026-08-22) DOMANDA Il progetto ha chiuso il FUNDING su tre lati (CC01 carry spot-perp, time-series, cross-sectional) ma non ha mai guardato la STRUTTURA A TERMINE dei futures datati di Deribit — l'unico strumento delta-neutro eseguibile sul NOSTRO venue, a 2 gambe. Q1 Il basis (datato vs indice/perp) e la sua pendenza sono PREVEDIBILI, cioe' esiste un roll-down incassabile, o e' un martingala rumoroso? Q2 Un CALENDAR SPREAD (front vs back) su segnale di pendenza produce un edge netto di fee e spread bid-ask REALI? Q3 E' ortogonale al libro ed eseguibile a $600-3.000? IL DATO — costruito qui, non esisteva `data/raw/fut_*` sono futures INDICE AZIONARIO da IB (ES/NQ/DAX...), niente a che vedere. Su Deribit non c'era nulla. Ma: * /public/get_instruments?expired=true ritorna UN SOLO strumento (Deribit purga subito le liste), quindi la via ovvia non funziona; * i nomi dei trimestrali sono DETERMINISTICI (ultimo venerdi' di MAR/GIU/SET/DIC, 08:00 UTC) e `get_tradingview_chart_data` serve la storia COMPLETA di un contratto SCADUTO se il nome lo si costruisce a mano. Verificato: 30/30 trimestrali 2019Q1..2026Q2 rispondono `status=ok`. => **7+ anni di storia**, che includono il deleveraging 2022. E' esattamente la coda che a CC01 mancava per costruzione (funding HL dal 2023-05): qui il 2022 c'e'. La chiamata funding porta in regalo l'INDICE Deribit orario (`index_price`), quindi spot e funding arrivano insieme e il basis si misura sull'indice vero, non su una proxy. DISCIPLINA DI RETE (la VPS ha un IP solo e ci gira il libro) <=2 richieste/s, e il fetcher DORME nella finestra :25-:29 di ogni ora, che e' `cron_chain` (il collettore della catena opzioni). Il progetto ha gia' pagato un'ora di guasto per un IP saturato. Tutto e' in cache su scratchpad: un ri-run non tocca la rete. CONVENZIONE DI ROLL — dichiarata, e mai una serie incollata I prezzi NON vengono mai splicciati. La posizione e' su una COPPIA DI CONTRATTI specifica; al roll si chiude e si riapre pagando il round-trip pieno su entrambe le gambe. L'incollaggio avviene solo nella serie dei RENDIMENTI, che e' l'unico posto in cui e' lecito: e' l'errore che il progetto ha gia' commesso con le serie equity ri-aggiustate, e qui e' strutturalmente impossibile. USO nice -n 19 timeout 900 uv run python scripts/research/r0822_basis_calendar.py """ from __future__ import annotations import calendar import datetime as dt import json import math import os import sys import time import urllib.parse import urllib.request 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" / "alt")) import altlib as A # noqa: E402 SCRATCH = Path( os.environ.get( "R0822_SCRATCH", "/tmp/claude-1001/-opt-docker-PythagorasGoal/" "b6cc75e7-14f8-4c32-bd07-ab8a0d2aaee6/scratchpad", ) ) CACHE = SCRATCH / "basis_cache" CACHE.mkdir(parents=True, exist_ok=True) API = "https://www.deribit.com/api/v2" UTC = dt.timezone.utc MONTHS = ["JAN", "FEB", "MAR", "APR", "MAY", "JUN", "JUL", "AUG", "SEP", "OCT", "NOV", "DEC"] ASSETS = ("BTC", "ETH") FEE_SIDE = 0.00035 # 3,5 bps/lato per GAMBA (taker Deribit, letto dal venue) HOLDOUT = pd.Timestamp("2025-01-01", tz="UTC") HOURS_Y = 365.25 * 24 _last_call = [0.0] # ========================================================================== # 0. FETCHER — rate-limited, cache-first, consapevole dei cron della VPS # ========================================================================== def _polite_sleep() -> None: """<=2 req/s E fuori dalla finestra :25-:29 (cron_chain, stesso IP).""" now = dt.datetime.now(UTC) if 25 <= now.minute <= 29: wait = (30 - now.minute) * 60 - now.second print(f" [rete] finestra cron_chain (:25-:29) -> dormo {wait}s") time.sleep(max(1, wait)) gap = time.time() - _last_call[0] if gap < 0.55: time.sleep(0.55 - gap) _last_call[0] = time.time() def _api(path: str, **kw) -> dict: url = API + path + "?" + urllib.parse.urlencode(kw) for attempt in range(4): _polite_sleep() try: with urllib.request.urlopen(url, timeout=30) as r: return json.load(r) except Exception as exc: # noqa: BLE001 if attempt == 3: raise print(f" [rete] {type(exc).__name__} -> retry {attempt+1}/3") time.sleep(2.0 * (attempt + 1)) raise RuntimeError("unreachable") def last_friday(year: int, month: int) -> dt.date: d = dt.date(year, month, calendar.monthrange(year, month)[1]) while d.weekday() != 4: d -= dt.timedelta(days=1) return d def quarterly_universe(asset: str, y0: int = 2019, y1: int = 2027) -> list[tuple[str, pd.Timestamp]]: """Nomi DETERMINISTICI dei trimestrali Deribit: ultimo venerdi' di MAR/GIU/SET/DIC 08:00 UTC.""" out = [] for y in range(y0, y1 + 1): for m in (3, 6, 9, 12): lf = last_friday(y, m) exp = pd.Timestamp(lf.year, lf.month, lf.day, 8, tz="UTC") out.append((f"{asset}-{lf.day}{MONTHS[m-1]}{str(y)[2:]}", exp)) return out def fetch_ohlc(instrument: str, t0: pd.Timestamp, t1: pd.Timestamp) -> pd.DataFrame: """OHLCV orario di UNO strumento (anche SCADUTO), a chunk di 90 giorni. Cache su disco.""" cf = CACHE / f"ohlc_{instrument.replace('/', '_')}.parquet" if cf.exists(): return pd.read_parquet(cf) frames = [] cur = t0 while cur < t1: end = min(cur + pd.Timedelta(days=90), t1) try: res = _api("/public/get_tradingview_chart_data", instrument_name=instrument, resolution="60", start_timestamp=int(cur.timestamp() * 1000), end_timestamp=int(end.timestamp() * 1000)).get("result", {}) except Exception as exc: # noqa: BLE001 print(f" {instrument}: chunk {cur.date()} FALLITO ({type(exc).__name__})") cur = end continue ticks = res.get("ticks") or [] if res.get("status") == "ok" and ticks: frames.append(pd.DataFrame({ "ts": pd.to_datetime(ticks, unit="ms", utc=True), "open": res["open"], "high": res["high"], "low": res["low"], "close": res["close"], "volume": res.get("volume", [np.nan] * len(ticks)), })) cur = end df = (pd.concat(frames).drop_duplicates("ts").sort_values("ts").reset_index(drop=True) if frames else pd.DataFrame(columns=["ts", "open", "high", "low", "close", "volume"])) df.to_parquet(cf, index=False) return df def fetch_funding_and_index(asset: str, t0: pd.Timestamp, t1: pd.Timestamp) -> pd.DataFrame: """Funding orario del perp E indice Deribit (la chiamata li porta insieme).""" cf = CACHE / f"fund_{asset}.parquet" if cf.exists(): return pd.read_parquet(cf) frames, cur = [], t0 while cur < t1: end = min(cur + pd.Timedelta(days=30), t1) try: res = _api("/public/get_funding_rate_history", instrument_name=f"{asset}-PERPETUAL", start_timestamp=int(cur.timestamp() * 1000), end_timestamp=int(end.timestamp() * 1000)).get("result", []) except Exception: # noqa: BLE001 cur = end continue if res: frames.append(pd.DataFrame({ "ts": pd.to_datetime([r["timestamp"] for r in res], unit="ms", utc=True), "index": [r["index_price"] for r in res], "f1h": [r["interest_1h"] for r in res], })) cur = end df = (pd.concat(frames).drop_duplicates("ts").sort_values("ts").reset_index(drop=True) if frames else pd.DataFrame(columns=["ts", "index", "f1h"])) df.to_parquet(cf, index=False) return df def fetch_live_books(instruments: list[str]) -> pd.DataFrame: """Calibrazione DATATA dello spread bid-ask: il book VIVO di oggi. Non e' storia — e' un punto di ancoraggio per una stima, e va detto.""" cf = CACHE / "live_books.parquet" if cf.exists(): return pd.read_parquet(cf) rows = [] for nm in instruments: try: r = _api("/public/get_order_book", instrument_name=nm, depth=5)["result"] bid, ask = r.get("best_bid_price"), r.get("best_ask_price") if bid and ask and bid > 0: rows.append(dict(instrument=nm, bid=bid, ask=ask, mid=(bid + ask) / 2, spread_bps=1e4 * (ask - bid) / ((ask + bid) / 2), bid_sz=r.get("best_bid_amount"), ask_sz=r.get("best_ask_amount"))) except Exception: # noqa: BLE001 continue df = pd.DataFrame(rows) df.to_parquet(cf, index=False) return df # ========================================================================== # 1. PANEL + CERTIFICAZIONE (prima di ogni strategia — regola di prim'ordine) # ========================================================================== def build_panel(asset: str) -> tuple[pd.DataFrame, pd.DataFrame]: """Ritorna (wide, meta). `wide` = indice orario x contratto -> close. `meta` = una riga per contratto con scadenza, barre, copertura, quota barre FLAT.""" uni = quarterly_universe(asset) today = pd.Timestamp.now(tz="UTC") closes, vols, meta = {}, {}, [] for nm, exp in uni: if exp > today + pd.Timedelta(days=400): continue t0 = exp - pd.Timedelta(days=400) t1 = min(exp + pd.Timedelta(hours=2), today) if t1 <= t0: continue df = fetch_ohlc(nm, t0, t1) if df.empty or len(df) < 200: meta.append(dict(contract=nm, exp=exp, bars=len(df), status="VUOTO/CORTO")) continue s = df.set_index("ts")["close"].astype(float) s = s[s > 0] closes[nm] = s vols[nm] = df.set_index("ts")["volume"].astype(float) flat = float((s.diff() == 0).mean()) meta.append(dict(contract=nm, exp=exp, bars=len(s), first=s.index.min(), last=s.index.max(), flat_frac=flat, status="ok")) wide = pd.DataFrame(closes).sort_index() vwide = pd.DataFrame(vols).sort_index() return wide, vwide, pd.DataFrame(meta) def certify(asset: str, wide: pd.DataFrame, vwide: pd.DataFrame, meta: pd.DataFrame, fund: pd.DataFrame, perp: pd.Series) -> dict: """Quattro difetti sorvegliati. Un edge su un book fermo NON e' un edge (regola 4).""" print(f"\n --- CERTIFICAZIONE {asset} ---") ok = meta[meta.status == "ok"] print(f" contratti con dato: {len(ok)}/{len(meta)} " f"finestra {wide.index.min().date()} .. {wide.index.max().date()}") # (a) barre FLAT per bucket di giorni-a-scadenza = liquidita' vera del datato rows = [] for nm in wide.columns: exp = ok.loc[ok.contract == nm, "exp"] if exp.empty: continue s = wide[nm].dropna() dte = (exp.iloc[0] - s.index).total_seconds() / 86400.0 flat = (s.diff() == 0).astype(float) for lo, hi, lab in [(0, 30, "0-30g"), (30, 90, "30-90g"), (90, 180, "90-180g"), (180, 400, ">180g")]: m = (dte >= lo) & (dte < hi) if m.sum() > 50: rows.append(dict(bucket=lab, flat=float(flat[m].mean()), n=int(m.sum()))) fb = pd.DataFrame(rows).groupby("bucket").agg(flat=("flat", "mean"), n=("n", "sum")) order = ["0-30g", "30-90g", "90-180g", ">180g"] print(" barre FLAT (proxy di illiquidita') per giorni-a-scadenza:") for b in order: if b in fb.index: print(f" {b:9s} flat={fb.loc[b,'flat']*100:5.1f}% barre={int(fb.loc[b,'n']):,}") # (b) volume orario mediano per bucket (il flat da solo non basta) vrows = [] for nm in vwide.columns: exp = ok.loc[ok.contract == nm, "exp"] if exp.empty: continue s = vwide[nm].dropna() dte = (exp.iloc[0] - s.index).total_seconds() / 86400.0 for lo, hi, lab in [(0, 30, "0-30g"), (30, 90, "30-90g"), (90, 180, "90-180g"), (180, 400, ">180g")]: m = (dte >= lo) & (dte < hi) if m.sum() > 50: vrows.append(dict(bucket=lab, med=float(s[m].median()), zero=float((s[m] == 0).mean()))) vb = pd.DataFrame(vrows).groupby("bucket").agg(med=("med", "median"), zero=("zero", "mean")) print(" volume orario mediano (USD di nozionale) e quota ore a volume ZERO:") for b in order: if b in vb.index: print(f" {b:9s} mediano=${vb.loc[b,'med']:>12,.0f} ore a vol 0 = {vb.loc[b,'zero']*100:5.1f}%") # (c) accordo con l'INDICE alla scadenza: il datato DEVE convergere conv = [] for nm in wide.columns: e = ok.loc[ok.contract == nm, "exp"] if e.empty: continue exp = e.iloc[0] s = wide[nm].dropna() tail = s[s.index >= exp - pd.Timedelta(hours=6)] idx = fund.set_index("ts")["index"].reindex(tail.index, method="nearest", tolerance=pd.Timedelta("2h")) if len(tail) and idx.notna().any(): conv.append(float(np.abs(np.log(tail / idx)).dropna().iloc[-1]) * 1e4) if conv: print(f" convergenza a scadenza |ln(F/indice)| ultima ora: " f"mediana {np.median(conv):.1f} bps, max {np.max(conv):.1f} bps (n={len(conv)})") # (d) CROSS-CHECK INDIPENDENTE: il forward implicito nella catena opzioni xchk = cross_check_option_forward(asset, wide, ok) return dict(flat=fb, vol=vb, conv=conv, xchk=xchk) def cross_check_option_forward(asset: str, wide: pd.DataFrame, meta: pd.DataFrame) -> dict: """Il `underlying_price` di un'opzione Deribit E' il forward della SUA scadenza. La catena raccolta dal 2026-07-30 e' quindi una MISURA INDIPENDENTE del basis: se il mio F(T) e il forward delle opzioni sulla stessa scadenza non coincidono, uno dei due e' sbagliato. (L'archivio ereditato da bite ha underlying_price NULL.)""" import glob files = sorted(glob.glob(str(PROJECT_ROOT / "data/raw/cb_chain/2026-*.parquet"))) if not files: return dict(status="catena assente") keep = [] for f in files[-30:]: try: d = pd.read_parquet(f, columns=["asset", "exp", "ts", "underlying_price", "index_price"]) except Exception: # noqa: BLE001 continue d = d[(d.asset == asset) & d.underlying_price.notna()] if len(d): keep.append(d) if not keep: return dict(status="nessun underlying_price") ch = pd.concat(keep) ch["ts_h"] = ch["ts"].dt.floor("1h") # forward mediano per (ora, scadenza) — tutti gli strike condividono lo stesso forward fw = ch.groupby(["ts_h", "exp"]).agg(fwd=("underlying_price", "median")).reset_index() rows = [] for nm in wide.columns: e = meta.loc[meta.contract == nm, "exp"] if e.empty: continue exp = e.iloc[0] sub = fw[fw.exp == exp] if sub.empty: continue mine = wide[nm].dropna() j = sub.set_index("ts_h")["fwd"].reindex(mine.index).dropna() if len(j) < 5: continue dev = (np.log(mine.reindex(j.index) / j) * 1e4).dropna() rows.append(dict(contract=nm, n=len(dev), med_bps=float(dev.median()), p95_bps=float(dev.abs().quantile(0.95)))) return dict(status="ok", table=pd.DataFrame(rows)) # ========================================================================== # 2. CURVA — front/back trimestrali, basis annualizzato, slope forward # ========================================================================== def curve_frame(asset: str, wide: pd.DataFrame, meta: pd.DataFrame, fund: pd.DataFrame, roll_dte: int) -> pd.DataFrame: """Per ogni ora: front = trimestrale piu' vicino con dte >= roll_dte, back = il successivo. Ritorna prezzi, tau, basis annualizzati, slope forward e funding.""" ok = meta[meta.status == "ok"].sort_values("exp") exps = {r.contract: r.exp for r in ok.itertuples()} idx = wide.index F = fund.set_index("ts").reindex(idx) n = len(idx) front = np.full(n, "", dtype=object) back = np.full(n, "", dtype=object) contracts = list(wide.columns) tau = {c: (exps[c] - idx).total_seconds() / 3600.0 for c in contracts if c in exps} avail = {c: wide[c].notna().values for c in contracts} for c in contracts: if c not in exps: continue alive = avail[c] & (tau[c] >= roll_dte * 24) # front = il primo (per scadenza) ancora vivo; i contratti sono ordinati per exp order = [c for c in ok.contract if c in wide.columns] for i in range(n): picked = [] for c in order: if avail[c][i] and tau[c][i] >= roll_dte * 24: picked.append(c) if len(picked) == 2: break if len(picked) == 2: front[i], back[i] = picked out = pd.DataFrame(index=idx) out["front"], out["back"] = front, back m = (front != "") & (back != "") Ff = np.full(n, np.nan); Fb = np.full(n, np.nan) tf = np.full(n, np.nan); tb = np.full(n, np.nan) W = wide.values colpos = {c: k for k, c in enumerate(wide.columns)} for i in np.where(m)[0]: Ff[i] = W[i, colpos[front[i]]]; Fb[i] = W[i, colpos[back[i]]] tf[i] = tau[front[i]][i]; tb[i] = tau[back[i]][i] out["F_front"], out["F_back"] = Ff, Fb out["tau_front_h"], out["tau_back_h"] = tf, tb out["index"] = F["index"].values out["f1h"] = F["f1h"].values # basis annualizzato di ciascuna gamba vs INDICE out["c_front"] = np.log(Ff / out["index"]) * HOURS_Y / tf out["c_back"] = np.log(Fb / out["index"]) * HOURS_Y / tb # slope = tasso forward annualizzato FRA le due scadenze (stazionario, interpretabile) out["slope"] = np.log(Fb / Ff) * HOURS_Y / (tb - tf) out["s_raw"] = np.log(Fb / Ff) return out def describe_curve(asset: str, cv: pd.DataFrame) -> None: """Q1 — descrittiva. Il basis e' positivo? persiste? il suo CAMBIO e' prevedibile?""" c = cv.dropna(subset=["c_front", "slope"]) print(f"\n --- Q1 CURVA {asset} (n={len(c):,} ore) ---") print(f" basis annualizzato del FRONT: mediana {c.c_front.median()*100:+6.2f}%/anno " f"media {c.c_front.mean()*100:+6.2f}% sd {c.c_front.std()*100:5.2f} " f"contango {float((c.c_front>0).mean())*100:.1f}% del tempo") print(f" slope forward front->back : mediana {c.slope.median()*100:+6.2f}%/anno " f"media {c.slope.mean()*100:+6.2f}% sd {c.slope.std()*100:5.2f} " f"positiva {float((c.slope>0).mean())*100:.1f}% del tempo") fnd = cv["f1h"].dropna() * HOURS_Y print(f" funding perp annualizzato : mediana {fnd.median()*100:+6.2f}%/anno " f"media {fnd.mean()*100:+6.2f}%") print(" per anno (mediane annualizzate):") g = c.groupby(c.index.year).agg(basis_front=("c_front", "median"), slope=("slope", "median")) fy = (cv["f1h"] * HOURS_Y).groupby(cv.index.year).median() for y, r in g.iterrows(): print(f" {y} basis_front {r.basis_front*100:+7.2f}% slope {r.slope*100:+7.2f}% " f"funding {fy.get(y, np.nan)*100:+7.2f}%") # prevedibilita': mean-reversion dello slope a 7 giorni (AR sul livello) for h, lab in [(24, "1g"), (24 * 7, "7g"), (24 * 30, "30g")]: x = c["slope"] dx = x.shift(-h) - x j = pd.concat([x, dx], axis=1).dropna() if len(j) > 500: b = np.polyfit(j.iloc[:, 0], j.iloc[:, 1], 1)[0] rho = float(j.iloc[:, 0].corr(j.iloc[:, 1])) print(f" mean-reversion slope a {lab:>3s}: beta {b:+.3f} corr {rho:+.3f} " f"(beta -1 = reversione totale, 0 = martingala)") # ========================================================================== # 3. STRATEGIE — posizioni in spazio CONTRATTO (il roll si paga da solo) # ========================================================================== def _z(x: pd.Series, win: int) -> pd.Series: m = x.rolling(win, min_periods=win // 3).mean() s = x.rolling(win, min_periods=win // 3).std() return ((x - m) / s.replace(0, np.nan)) def run_strategy(cv: pd.DataFrame, wide: pd.DataFrame, family: str, sig: str, win_h: int, thr: float, slip_bps: float, fee_side: float = FEE_SIDE) -> pd.Series: """Ritorno ORARIO netto per $1 di nozionale LORDO PER GAMBA. Le posizioni vivono in spazio CONTRATTO: quando la coppia cambia (roll) la variazione per contratto e' automaticamente piena => il roll paga il round-trip completo su entrambe le gambe. Nessuna serie di prezzi viene MAI incollata. Causalita': w deciso con dati <= t, TENUTO durante t -> t+1 (shift esplicito).""" idx = cv.index if family == "CAL": base = cv["slope"] else: # FvP: basis annualizzato del front base = cv["c_front"] if sig == "static_long": w = pd.Series(1.0, index=idx) elif sig == "static_short": w = pd.Series(-1.0, index=idx) elif sig == "carry": # segno del carry corrente (causale) w = -np.sign(base) # contango -> vendi il datato elif sig == "zrev": z = _z(base, win_h) w = -np.tanh(z) w = w.where(z.abs() >= thr, 0.0) elif sig == "zmom": z = _z(base, win_h) w = np.tanh(z) w = w.where(z.abs() >= thr, 0.0) else: raise ValueError(sig) w = pd.Series(np.asarray(w, float), index=idx).fillna(0.0).clip(-1, 1) # --- posizioni per CONTRATTO --- cols = list(wide.columns) cpos = {c: k for k, c in enumerate(cols)} P = np.zeros((len(idx), len(cols))) fr = cv["front"].values; bk = cv["back"].values wv = w.values for i in range(len(idx)): if fr[i] == "" or bk[i] == "": continue if family == "CAL": # long spread = long back / short front P[i, cpos[bk[i]]] = wv[i] P[i, cpos[fr[i]]] = -wv[i] else: # FvP: gamba datata; il perp e' a parte P[i, cpos[fr[i]]] = wv[i] Pd = pd.DataFrame(P, index=idx, columns=cols) held = Pd.shift(1).fillna(0.0) # decisa a t, TENUTA in t->t+1 R = wide.reindex(idx).pct_change().replace([np.inf, -np.inf], np.nan).fillna(0.0) gross = (held * R).sum(axis=1) if family == "FvP": # gamba perp: nozionale opposto alla gamba datata + funding pagato/incassato perp_r = pd.Series(cv["index"].pct_change().fillna(0.0).values, index=idx) wheld = held.sum(axis=1) # = w shiftato (una sola gamba datata) gross = gross - wheld * perp_r - wheld * (-1.0) * cv["f1h"].fillna(0.0) # long datato (w>0) => short perp => INCASSA funding se f>0 turn = (Pd.diff().abs().sum(axis=1) * 2.0) # datato + perp else: turn = Pd.diff().abs().sum(axis=1) cost = turn * (fee_side + slip_bps * 1e-4) return (gross - cost).fillna(0.0) # ========================================================================== # 4. METRICHE / GATE # ========================================================================== def to_daily(h: pd.Series) -> pd.Series: return ((1.0 + h.fillna(0.0)).resample("1D").prod() - 1.0).dropna() def stats(daily: pd.Series) -> dict: r = daily.dropna() if len(r) < 30 or r.std() == 0: return dict(sharpe=0.0, holdout=0.0, maxdd=0.0, cagr=0.0, n=len(r), vol=0.0) sh = float(r.mean() / r.std() * math.sqrt(365.25)) eq = (1 + r).cumprod() dd = float(((eq.cummax() - eq) / eq.cummax()).max()) yrs = (r.index[-1] - r.index[0]).days / 365.25 cagr = float(eq.iloc[-1] ** (1 / yrs) - 1) if yrs > 0.5 and eq.iloc[-1] > 0 else float("nan") ho = r[r.index >= HOLDOUT] sho = float(ho.mean() / ho.std() * math.sqrt(365.25)) if len(ho) > 30 and ho.std() > 0 else float("nan") return dict(sharpe=sh, holdout=sho, maxdd=dd, cagr=cagr, n=len(r), vol=float(r.std() * math.sqrt(365.25))) def delevering_null(cand: pd.Series, base: pd.Series) -> dict: """NULL DEL DE-LEVERING (5 occorrenze in questo progetto). Esiste k<1 che, applicato al BASELINE, da' lo STESSO maxDD del candidato con Sharpe MIGLIORE? Se si', il candidato non protegge: e' solo meno leva.""" cs = stats(cand) best = None for k in np.arange(0.05, 1.01, 0.05): s = stats(base * k) if s["maxdd"] <= cs["maxdd"] + 1e-9: best = (float(k), s) break if best is None: return dict(refuted=False, note="nessun k<=1 raggiunge quel DD sul baseline") k, s = best return dict(refuted=bool(s["sharpe"] > cs["sharpe"]), k=k, base_sharpe_at_k=s["sharpe"], cand_sharpe=cs["sharpe"], base_dd_at_k=s["maxdd"], cand_dd=cs["maxdd"]) # ========================================================================== # 5. MAIN # ========================================================================== def main() -> None: t_start = time.time() print("=" * 78) print("r0822 BASIS-CALENDAR — struttura a termine dei futures datati Deribit") print("=" * 78) today = pd.Timestamp.now(tz="UTC") panels, funds, metas, vwides = {}, {}, {}, {} # ---------------- STEP 1: dato ---------------- print("\n[1] DATO — ricostruzione da Deribit pubblico (cache su scratchpad)") for a in ASSETS: print(f" {a}: trimestrali (ultimo venerdi' MAR/GIU/SET/DIC 08:00 UTC)...") wide, vwide, meta = build_panel(a) fund = fetch_funding_and_index(a, pd.Timestamp("2019-01-01", tz="UTC"), today) panels[a], vwides[a], metas[a], funds[a] = wide, vwide, meta, fund ok = meta[meta.status == "ok"] print(f" {len(ok)} contratti, {int(meta.bars.sum()):,} barre orarie, " f"funding+indice {len(fund):,} ore") # ---------------- STEP 2: certificazione ---------------- print("\n[2] CERTIFICAZIONE (regola 4: un edge su un book fermo non e' un edge)") certs = {} for a in ASSETS: perp = A.get(a, "1h").set_index(pd.DatetimeIndex( pd.to_datetime(A.get(a, "1h")["datetime"], utc=True)))["close"] certs[a] = certify(a, panels[a], vwides[a], metas[a], funds[a], perp) x = certs[a]["xchk"] if x.get("status") == "ok" and len(x["table"]): t = x["table"] print(f" CROSS-CHECK INDIPENDENTE vs forward della catena opzioni " f"({int(t.n.sum())} ore, {len(t)} scadenze):") print(f" deviazione mediana {t.med_bps.median():+.1f} bps, " f"p95 |dev| {t.p95_bps.max():.1f} bps") else: print(f" CROSS-CHECK catena: {x.get('status')}") # ---------------- STEP 3: Q1 curva ---------------- print("\n[3] Q1 — LA CURVA E' PREVEDIBILE?") curves = {} for a in ASSETS: cv = curve_frame(a, panels[a], metas[a], funds[a], roll_dte=7) curves[a] = cv describe_curve(a, cv) # ---------------- STEP 3b: spread bid-ask (calibrazione DATATA) ---------------- print("\n[3b] SPREAD BID-ASK — calibrazione sul book VIVO di oggi (non e' storia)") live = [] for a in ASSETS: for nm, exp in quarterly_universe(a): if today < exp <= today + pd.Timedelta(days=400): live.append(nm) lb = fetch_live_books(live + [f"{a}-PERPETUAL" for a in ASSETS]) slip_est = 5.0 if len(lb): for r in lb.itertuples(): print(f" {r.instrument:16s} bid {r.bid:>10,.1f} ask {r.ask:>10,.1f} " f"spread {r.spread_bps:6.1f} bps size {r.bid_sz:>8,.0f}/{r.ask_sz:>8,.0f}") dated = lb[~lb.instrument.str.contains("PERPETUAL")] if len(dated): slip_est = float(dated.spread_bps.median()) / 2.0 print(f" => mezzo-spread MEDIANO del datato = {slip_est:.1f} bps/lato " f"(STIMA datata oggi, NON storia: il book storico dei datati non e' disponibile)") # ---------------- STEP 4: griglia ---------------- print("\n[4] GRIGLIA — dichiarata PRIMA, contata AL RIALZO") FAM = {"CAL": ["static_long", "static_short", "zrev", "zmom"], "FvP": ["carry", "static_long", "zrev", "zmom"]} WINS = [720, 2160, 4320] # 30 / 90 / 180 giorni in ore ROLLS = [7, 21] THRS = [0.0, 1.0] cells = [(f, s, w, rl, th) for f, sigs in FAM.items() for s in sigs for w in WINS for rl in ROLLS for th in THRS] print(f" famiglie {list(FAM)} x segnali x finestre {WINS} x roll {ROLLS} x soglie {THRS}") print(f" = {len(cells)} celle valutate (le varianti statiche non usano W/soglia:" f" le conto lo stesso, al rialzo)") cvcache = {(a, rl): curve_frame(a, panels[a], metas[a], funds[a], rl) for a in ASSETS for rl in ROLLS} rows = [] for (fam, sig, win, rl, th) in cells: hs = [] for a in ASSETS: h = run_strategy(cvcache[(a, rl)], panels[a], fam, sig, win, th, slip_est) hs.append(h) J = pd.concat(hs, axis=1).fillna(0.0) comb = J.mean(axis=1) # 50/50 BTC+ETH, come il resto del progetto d = to_daily(comb) st = stats(d) rows.append(dict(family=fam, sig=sig, win=win, roll=rl, thr=th, daily=d, **st)) G = pd.DataFrame([{k: v for k, v in r.items() if k != "daily"} for r in rows]) dailies = {i: r["daily"] for i, r in enumerate(rows)} print("\n migliori 8 celle per Sharpe FULL (NON e' la selezione — solo panorama):") for i in G.sharpe.nlargest(8).index: r = G.loc[i] print(f" {r.family:4s} {r.sig:12s} W={int(r.win):5d}h roll={int(r.roll):2d}g " f"thr={r.thr:.1f} | Sh {r.sharpe:+6.2f} hold {r.holdout:+6.2f} " f"DD {r.maxdd*100:5.2f}% CAGR {r.cagr*100:+6.2f}% vol {r.vol*100:5.2f}%") # ---------------- STEP 5: selezione IN-SAMPLE ONLY ---------------- print("\n[5] SELEZIONE IN-SAMPLE-ONLY (mai la cella col miglior hold-out)") ins = {} for i, d in dailies.items(): pre = d[d.index < HOLDOUT] ins[i] = (float(pre.mean() / pre.std() * math.sqrt(365.25)) if len(pre) > 60 and pre.std() > 0 else -9.9) best_i = max(ins, key=ins.get) b = G.loc[best_i] bd = dailies[best_i] bs = stats(bd) print(f" cella scelta AL BUIO: {b.family} {b.sig} W={int(b.win)}h roll={int(b.roll)}g thr={b.thr}") print(f" Sharpe in-sample {ins[best_i]:+.2f} -> FULL {bs['sharpe']:+.2f} " f"HOLD-OUT {bs['holdout']:+.2f} maxDD {bs['maxdd']*100:.2f}% " f"CAGR {bs['cagr']*100:+.2f}% vol {bs['vol']*100:.2f}%") best_full = G.sharpe.idxmax() print(f" (per confronto, la cella col miglior FULL sarebbe stata " f"{G.loc[best_full,'family']} {G.loc[best_full,'sig']} " f"Sh {G.loc[best_full,'sharpe']:+.2f} — NON e' quella scelta)") # ---------------- STEP 6: GATE ---------------- print("\n[6] GATE") dsr, sr0 = A.deflated_sharpe(bs["sharpe"], list(G.sharpe.values), bd) print(f" deflated-Sharpe : DSR {dsr:.3f} (null max atteso {sr0:+.2f} su {len(G)} trial) " f"-> {'PASS' if dsr >= 0.95 else 'FAIL'}") imp = A.implausible_sharpe(bd) print(f" implausible_sharpe: implausible={imp['implausible']} " f"Sh {imp.get('sharpe', float('nan')):+.2f} maxDD {imp.get('maxdd', float('nan'))*100:.2f}% " f"quota barre in perdita {imp.get('loss_frac', float('nan'))*100:.1f}% " f"Calmar {imp.get('calmar', float('nan')):.1f}") if imp["reasons"]: for rr in imp["reasons"]: print(f" - {rr}") tp = A.tp01_baseline_daily() try: mg = A.marginal_vs_tp01(bd) print(f" marginal_vs_tp01 : {mg.get('verdict')} corr {mg.get('corr', float('nan')):+.3f} " f"robust_oos={mg.get('robust_oos')} insample_edge={mg.get('has_insample_edge')} " f"is_hedge={mg.get('is_hedge')} beats_noise={mg.get('beats_noise_null')}") except Exception as exc: # noqa: BLE001 mg = {} print(f" marginal_vs_tp01 : NON GIRATO ({type(exc).__name__}: {exc})") dn = delevering_null(bd, tp) print(f" null de-levering : refuted={dn.get('refuted')} {dn}") # anchor: l'ora del giorno in cui si valuta il segnale print(" anchor_luck_band (ora del giorno di decisione, 24 ancore):") def by_off(off): hs = [] for a in ASSETS: cv = cvcache[(a, int(b.roll))] h = run_strategy(cv, panels[a], b.family, b.sig, int(b.win), float(b.thr), slip_est) hh = h.copy() mask = (hh.index.hour != off) hh[mask] = 0.0 # decide/agisce solo a quell'ora hs.append(hh) return to_daily(pd.concat(hs, axis=1).fillna(0.0).mean(axis=1)) try: ab = A.anchor_luck_band(by_off, list(range(24)), canonical=0) print(f" canonica {ab.get('canonical'):+.3f} MEDIANA(onesta) {ab.get('median'):+.3f} " f"banda [{ab.get('p10'):+.3f},{ab.get('p90'):+.3f}] " f"frazione>0 {ab.get('frac_positive'):.2f} gate_pass={ab.get('gate_pass')}") except Exception as exc: # noqa: BLE001 print(f" NON GIRATO ({type(exc).__name__}: {exc})") try: print(f" causality_ok : la posizione e' shiftata in run_strategy " f"(held = pos.shift(1)); verifica strutturale, non A.causality_ok " f"(che vuole un target_fn su BTC/ETH direzionale)") except Exception: # noqa: BLE001 pass # ---------------- STEP 7: quanto spread uccide l'edge ---------------- print("\n[7] A CHE SPREAD MUORE L'EDGE (la stima di slippage e' il rischio #1)") for sl in [0.0, 1.0, 2.5, 5.0, 7.5, 10.0, 20.0]: hs = [] for a in ASSETS: hs.append(run_strategy(cvcache[(a, int(b.roll))], panels[a], b.family, b.sig, int(b.win), float(b.thr), sl)) s2 = stats(to_daily(pd.concat(hs, axis=1).fillna(0.0).mean(axis=1))) mark = " <= stima di oggi" if abs(sl - slip_est) < 0.6 else "" print(f" mezzo-spread {sl:5.1f} bps/lato -> Sharpe {s2['sharpe']:+6.2f} " f"CAGR {s2['cagr']*100:+6.2f}%{mark}") # ---------------- STEP 8: eseguibilita' ---------------- print("\n[8] Q3 — ESEGUIBILITA' a $600-3.000") for a in ASSETS: try: sp = _api("/public/get_instrument", instrument_name=f"{a}-25DEC26")["result"] print(f" {a}-25DEC26: min_trade_amount ${sp['min_trade_amount']:,.0f} di nozionale, " f"contract_size ${sp['contract_size']:,.0f}, tick {sp['tick_size']}, " f"taker {sp['taker_commission']*1e4:.1f} bps") except Exception as exc: # noqa: BLE001 print(f" {a}: specs non lette ({type(exc).__name__})") print(f" vol ANNUA del candidato per $1 di nozionale lordo/gamba: {bs['vol']*100:.2f}%") if bs["vol"] > 0: lev = 0.20 / bs["vol"] print(f" per portarlo al vol-target 20% del progetto servirebbe leva " f"{lev:.1f}x per gamba = ${lev*2:,.0f} di nozionale LORDO per $1 di conto") for cap in (600, 3000): print(f" a ${cap}: nozionale lordo ${lev*2*cap:,.0f} " f"(margine ~2-4%/gamba => ${lev*2*cap*0.03:,.0f} di IM richiesto)") print(f"\n[fine] {time.time()-t_start:.0f}s") if __name__ == "__main__": main()