"""SURFACE-RV — la superficie di volatilita' come RELATIVE VALUE fra i suoi punti. IPOTESI (registrata prima di misurare): la superficie Deribit e' internamente INCOERENTE in modo misurabile — strike adiacenti fuori da una curva liscia, varianza totale non monotona nel tenore — e l'incoerenza si ricompone. E' un'ipotesi di MICROSTRUTTURA, non di previsione: non chiede alla vol implicita di prevedere nulla (cosa che il progetto ha gia' refutato quattro volte: dvol_directional, gamma scalping, term-structure come segnale, skew come segnale). ATTESA DICHIARATA PRIMA: l'incoerenza esiste al MID ma non sopravvive al bid-ask, perche' su un'opzione a 7 giorni un punto di vol vale pochi dollari mentre attraversare lo spread ne costa il ~10% del premio (misurato dal filone 11 sulle inverse). LA DIFESA CONTRO IL RISCHIO NUMERO UNO DI QUESTO FILONE (misurare il proprio errore di interpolazione) e' strutturale, non statistica: i due test principali sono **arbitraggi statici model-free** — monotonia e convessita' del prezzo nello strike — che non richiedono NESSUN interpolatore, NESSUN modello e NESSUN tasso. Valgono per ogni misura di prezzo, path per path: * put: P(K) crescente in K (verticale) * put: P(K) convessa in K (farfalla) e sono verificati su prezzi ESEGUIBILI (si vende al bid, si compra all'ask), mai al mid. Il terzo test — varianza totale non monotona nel tenore — richiede di confrontare tenori diversi: si riporta sia nella versione SENZA interpolazione (strike condiviso fra due scadenze) sia in quella con interpolazione a log-moneyness fisso, e la DIFFERENZA fra le due e' la misura diretta di quanto pesa l'interpolatore. Convenzioni verificate nel codice (non assunte): * tick: 0.0001 coin sotto 0.005, 0.0005 sopra -> contato, non dichiarato * f_markfit: la mark IV del venue riprezza i propri mid? (il filone 5 misura 0.992) * fee Deribit opzioni: 0.03% del sottostante per gamba, cap 12.5% del premio della gamba Il dato: `data/raw/cb_chain/` (archivio bite + raccolta propria). SOLO opzioni INVERSE — il collettore interroga {"currency": "BTC"|"ETH"}, che non restituisce la famiglia USDC-lineare (difetto di produzione trovato dal filone 11). Quindi ogni numero qui e' misurato sulla famiglia che il conto NON puo' marginare, e le conclusioni sulla famiglia USDC sono ESTRAPOLAZIONI dichiarate, non misure. Muro del dato: prima del 2026-06-09 l'archivio raccoglieva UNA scadenza per giro -> la SUPERFICIE esiste da li' (~75 giorni). Lo SMILE (piu' strike su UNA scadenza) esiste dal 2026-05-01 (~113 g). Verdetto massimo possibile: LEAD con gate pre-registrato. Mai CANDIDATO. uv run python scripts/research/r0822b_surface_rv.py """ from __future__ import annotations import glob import sys import time from pathlib import Path import numpy as np import pandas as pd from scipy.stats import norm ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) STORE = ROOT / "data" / "raw" / "cb_chain" SMILE_START = pd.Timestamp("2026-05-01", tz="UTC") # piu' strike su UNA scadenza SURF_START = pd.Timestamp("2026-06-09", tz="UTC") # piu' scadenze per giro -> superficie FEE_UNDERLYING = 0.0003 # 0.03% del sottostante, per gamba, in coin per contratto FEE_CAP_FRAC = 0.125 # cap: 12.5% del premio della gamba MIN_LOT = {"BTC": 0.1, "ETH": 1.0} # famiglia INVERSE (quella che abbiamo nel dato) MIN_LOT_USDC = {"BTC": 0.01, "ETH": 0.1} # famiglia USDC-lineare (non raccolta: estrapolazione) DTE_MIN = 1.0 # sotto un giorno il book e' un'altra cosa; riportato anche senza filtro def tick_of(px: np.ndarray) -> np.ndarray: """Griglia di prezzo Deribit sulle opzioni, in coin. Verificata sui dati in `certifica()`.""" return np.where(px < 0.005, 0.0001, 0.0005) # --------------------------------------------------------------------------- dato def load(assets=("BTC", "ETH")) -> pd.DataFrame: cols = ["ts", "asset", "instrument_name", "strike", "option_type", "exp", "bid", "ask", "mid", "iv", "delta", "open_interest", "book_depth_top3", "underlying_price", "index_price", "quote_status"] parts = [] for f in sorted(glob.glob(str(STORE / "*.parquet"))): d = pd.read_parquet(f, columns=cols) parts.append(d[d["asset"].isin(assets)]) d = pd.concat(parts, ignore_index=True) d["ts"] = pd.to_datetime(d["ts"], utc=True) d["exp"] = pd.to_datetime(d["exp"], utc=True) for c in ("strike", "bid", "ask", "mid", "iv", "delta", "open_interest", "book_depth_top3", "underlying_price", "index_price"): d[c] = pd.to_numeric(d[c], errors="coerce") d = d.drop_duplicates(subset=["ts", "instrument_name"], keep="last") d["dte"] = (d["exp"] - d["ts"]).dt.total_seconds() / 86400.0 d["T"] = d["dte"] / 365.25 d = d.reset_index(drop=True) d["spot_feed"] = _spot_asof(d) return d def _spot_asof(d: pd.DataFrame) -> pd.Series: """Indice dal feed CERTIFICATO (1h Deribit mainnet), asof sul ts della quota. Serve perche' `bite_archive` NON ha ne' `index_price` ne' `underlying_price` (100% None: misurato, non assunto) -> senza questo la finestra con un divisore sarebbe di 23 giorni. """ from scripts.analysis.research_lab import load_tf out = pd.Series(np.nan, index=d.index) for a in d["asset"].unique(): px = load_tf(a, "1h") s = pd.Series(px["close"].to_numpy(float), index=pd.to_datetime(px["timestamp"], unit="ms", utc=True)).sort_index() s.index = pd.DatetimeIndex(s.index).as_unit("ns") m = d["asset"] == a t = pd.DatetimeIndex(d.loc[m, "ts"]).as_unit("ns") # epoca esplicita in ns (mai `.view("int64")` su indici tz-aware: lezione 01/07) si, ti = s.index.asi8, t.asi8 pos = np.searchsorted(si, ti, side="right") - 1 v = np.where(pos >= 0, s.to_numpy()[np.clip(pos, 0, None)], np.nan) out.loc[m] = v return out def forward(d: pd.DataFrame) -> pd.DataFrame: """Forward per (asset, ts, exp) ricostruito con la PARITA' PUT-CALL, model-free. In convenzione inverse il premio e' quotato in coin: (C - P) * X = F - K => F = K + (C-P)*X, con X = divisore della convenzione (misurato in `certifica`). Mediana sugli strike = robusta. Dove il venue lo ha registrato (solo la raccolta propria) si usa il valore OSSERVATO, e l'accordo fra i due e' una certificazione, non un'assunzione. """ q = d[(d["bid"] > 0) & (d["ask"] > 0) & (d["ask"] >= d["bid"]) & d["spot_feed"].notna()] c = q[q["option_type"] == "C"][["asset", "ts", "exp", "strike", "mid", "spot_feed"]] p = q[q["option_type"] == "P"][["asset", "ts", "exp", "strike", "mid"]] j = c.merge(p, on=["asset", "ts", "exp", "strike"], suffixes=("_c", "_p")) j["F_k"] = j["strike"] + (j["mid_c"] - j["mid_p"]) * j["spot_feed"] f = j.groupby(["asset", "ts", "exp"])["F_k"].median().rename("F_pcp").reset_index() return f def two_sided(d: pd.DataFrame) -> pd.DataFrame: """Quotate a DUE lati. Una gamba sola quotata non e' meta' struttura: e' un'altra cosa. (Regola 8 del brief: una riga presente non e' un dato presente.) """ q = d[(d["bid"] > 0) & (d["ask"] > 0) & (d["ask"] >= d["bid"])].copy() q = q[q["T"] > 0] return q # --------------------------------------------------------------------------- prezzo def bs_usd(F, K, T, sig, is_put): """Black-76 con r=0 (Deribit prezza sul FORWARD della scadenza = `underlying_price`).""" F = np.asarray(F, float); K = np.asarray(K, float) T = np.asarray(T, float); sig = np.asarray(sig, float) out = np.full(F.shape, np.nan) ok = (T > 0) & (sig > 0) & (F > 0) & (K > 0) if not ok.any(): return out d1 = np.zeros_like(F); d2 = np.zeros_like(F) sq = np.sqrt(np.where(ok, T, 1.0)) d1[ok] = (np.log(F[ok] / K[ok]) + 0.5 * sig[ok] ** 2 * T[ok]) / (sig[ok] * sq[ok]) d2[ok] = d1[ok] - sig[ok] * sq[ok] call = F * norm.cdf(d1) - K * norm.cdf(d2) put = K * norm.cdf(-d2) - F * norm.cdf(-d1) out[ok] = np.where(is_put[ok], put[ok], call[ok]) return out def bs_vega_usd(F, K, T, sig): F = np.asarray(F, float); K = np.asarray(K, float) T = np.asarray(T, float); sig = np.asarray(sig, float) out = np.zeros(F.shape) ok = (T > 0) & (sig > 0) & (F > 0) & (K > 0) sq = np.sqrt(np.where(ok, T, 1.0)) d1 = np.zeros_like(F) d1[ok] = (np.log(F[ok] / K[ok]) + 0.5 * sig[ok] ** 2 * T[ok]) / (sig[ok] * sq[ok]) out[ok] = F[ok] * norm.pdf(d1[ok]) * sq[ok] # per 1.00 di sigma (=100 punti di vol) return out # --------------------------------------------------------------------------- certificazione def certifica(d: pd.DataFrame) -> pd.DataFrame: print("=" * 100) print("0. CERTIFICAZIONE DEL DATO — cosa esiste davvero, prima di qualunque test") print("=" * 100) q = two_sided(d) print(f"righe totali {len(d):,} · quotate a due lati {len(q):,} ({len(q)/len(d):.1%})") print(f"finestra {d['ts'].min():%Y-%m-%d} -> {d['ts'].max():%Y-%m-%d}") # (a) muro del dato: scadenze per giro per = d.groupby([d["ts"].dt.floor("D"), "asset"])["exp"].nunique().unstack() pre = per[per.index < SURF_START] post = per[per.index >= SURF_START] print("\n(a) MURO DEL DATO — scadenze per giro (mediana per giorno):") print(f" pre {SURF_START:%Y-%m-%d}: BTC {pre['BTC'].median():.0f} · ETH {pre['ETH'].median():.0f}" f" (max BTC {pre['BTC'].max():.0f}) -> la SUPERFICIE non esiste") print(f" post {SURF_START:%Y-%m-%d}: BTC {post['BTC'].median():.0f} · ETH {post['ETH'].median():.0f}" f" -> superficie su {len(post)} giorni") print(f" smile (piu' strike, UNA scadenza) su {len(per)} giorni dal {per.index.min():%Y-%m-%d}" f" -> i test model-free girano su questa finestra piu' lunga") # (a-bis) BUCO DI COLONNA nell'archivio ereditato obs = d["index_price"].notna() print("\n(a-bis) BUCO DI COLONNA — `bite_archive` non registrava il sottostante") print(f" index_price/underlying_price presenti in {obs.sum():,}/{len(d):,} righe ({obs.mean():.1%})" f" — solo la raccolta propria, dal {d.loc[obs,'ts'].min():%Y-%m-%d}") print(f" book_depth_top3 presente in {d['book_depth_top3'].notna().mean():.1%} delle righe") print(" -> i test 1-2 (model-free) NON ne hanno bisogno; il test 4 (calendario) si')") # (b) tick: verificato, non dichiarato print("\n(b) GRIGLIA DI PREZZO (verificata sui dati, non assunta)") for a in sorted(q["asset"].unique()): b = q.loc[q["asset"] == a, "bid"].to_numpy() t = tick_of(b) off = np.abs(np.round(b / t) - b / t) > 1e-6 n_tick = np.round(b / t) print(f" {a}: fuori griglia {off.sum():,}/{len(b):,} ({off.mean():.4%}) · " f"bid in tick: mediana {np.median(n_tick):.0f} · quota <=2 tick {np.mean(n_tick <= 2):.1%}") # (c) il divisore della convenzione inverse + f_markfit print("\n(c) CONVENZIONE INVERSE e f_markfit — la mark IV riprezza i propri mid?") ov = q[obs.reindex(q.index, fill_value=False)].dropna(subset=["iv", "mid", "spot_feed"]) ov = ov[(ov["iv"] > 0) & (ov["mid"] > 0) & (ov["dte"] >= DTE_MIN)] dev = (ov["index_price"] / ov["spot_feed"] - 1).abs() print(f" index_price della catena vs feed CERTIFICATO 1h: |dev| mediana {dev.median():.5%} " f"· p99 {dev.quantile(.99):.4%} (n={len(ov):,}) -> il feed puo' fare da divisore sull'archivio") sm = ov.sample(n=min(120_000, len(ov)), random_state=0) is_put = (sm["option_type"] == "P").to_numpy() usd = bs_usd(sm["underlying_price"].to_numpy(), sm["strike"].to_numpy(), sm["T"].to_numpy(), (sm["iv"] / 100).to_numpy(), is_put) best = None for nome, div in (("index_price", sm["index_price"].to_numpy()), ("underlying_price(fwd)", sm["underlying_price"].to_numpy()), ("feed certificato 1h", sm["spot_feed"].to_numpy())): f = (usd / div) / sm["mid"].to_numpy() f = f[np.isfinite(f) & (f > 0)] print(f" f_markfit con divisore {nome:<22}: mediana {np.median(f):.4f} " f"[p10 {np.percentile(f,10):.3f}, p90 {np.percentile(f,90):.3f}] n={len(f):,}") if best is None or abs(np.median(f) - 1) < abs(best[1] - 1): best = (nome, float(np.median(f))) print(f" -> convenzione: divisore = {best[0]} (f_markfit {best[1]:.4f}); " f"un f molto lontano da 1 direbbe che il problema e' il MIO prezzatore, non il mercato") # (d) forward ricostruito con la parita' put-call, certificato contro l'osservato print("\n(d) FORWARD ricostruito con la PARITA' PUT-CALL (model-free) vs osservato") F = forward(d) chk = d.loc[obs, ["asset", "ts", "exp", "underlying_price"]].drop_duplicates( subset=["asset", "ts", "exp"]).merge(F, on=["asset", "ts", "exp"], how="inner") e = (chk["F_pcp"] / chk["underlying_price"] - 1) print(f" n={len(chk):,} coppie (asset,ts,scadenza) con entrambi: " f"|errore| mediana {e.abs().median():.5%} · p95 {e.abs().quantile(.95):.4%} · " f"corr {np.corrcoef(chk['F_pcp'], chk['underlying_price'])[0,1]:.6f}") return q, F # --------------------------------------------------------------------------- fee def fee_coin(premio_coin: np.ndarray, n_contratti: float) -> np.ndarray: """Fee Deribit opzioni per gamba, in coin: min(0.03% del sottostante, 12.5% del premio) x lotti.""" per_contratto = np.minimum(FEE_UNDERLYING, FEE_CAP_FRAC * np.asarray(premio_coin, float)) return per_contratto * n_contratti # --------------------------------------------------------------------------- TEST V def test_verticale(q: pd.DataFrame, dte_min: float) -> dict: """Monotonia del prezzo nello strike. Model-free, nessun interpolatore, nessun tasso. put: P(K) crescente -> arbitraggio se bid(K1) > ask(K2) con K1 < K2 (vendi la put bassa al bid, compra l'alta all'ask: credito + payoff sempre >= 0) call: C(K) decrescente -> arbitraggio se bid(K2) > ask(K1) con K1 < K2 """ x = q[q["dte"] >= dte_min].sort_values(["asset", "ts", "exp", "option_type", "strike"]) key = ["asset", "ts", "exp", "option_type"] g = x.groupby(key, sort=False) # per le put il "meglio finora" salendo di strike; per le call scendendo x["cmax_up"] = g["bid"].cummax() x["prev_up"] = g["cmax_up"].shift(1) # miglior bid su strike STRETTAMENTE inferiori xr = x.iloc[::-1].copy() gr = xr.groupby(key, sort=False) xr["cmax_dn"] = gr["bid"].cummax() x["prev_dn"] = gr["cmax_dn"].shift(1) # miglior bid su strike STRETTAMENTE superiori is_put = x["option_type"].to_numpy() == "P" prev = np.where(is_put, x["prev_up"].to_numpy(), x["prev_dn"].to_numpy()) cred = prev - x["ask"].to_numpy() # credito riskless in coin, per contratto viol = np.isfinite(cred) & (cred > 0) n_tot = len(x) out = {"n_confronti": int(np.isfinite(cred).sum()), "n_viol": int(viol.sum()), "quota": float(viol.sum() / max(1, np.isfinite(cred).sum()))} if viol.any(): vv = x.loc[viol] out["credito_med_coin"] = float(np.median(cred[viol])) out["credito_usd_medlot"] = float(np.median( cred[viol] * vv["asset"].map(MIN_LOT).to_numpy() * vv["spot_feed"].to_numpy())) return out # --------------------------------------------------------------------------- TEST B def test_farfalla(q: pd.DataFrame, dte_min: float) -> pd.DataFrame: """Convessita' del prezzo nello strike su triple CONSECUTIVE. Model-free. lam = (K3-K2)/(K3-K1); payoff lam*P(K1) + (1-lam)*P(K3) - P(K2) >= 0 sempre. Costo eseguibile = lam*ask1 + (1-lam)*ask3 - bid2. Se < 0 -> credito riskless. Al MID lo stesso numero misura l'INCOERENZA (non eseguibile) — si riportano entrambi. """ x = q[q["dte"] >= dte_min].sort_values(["asset", "ts", "exp", "option_type", "strike"]) key = ["asset", "ts", "exp", "option_type"] g = x.groupby(key, sort=False) for c in ("strike", "bid", "ask", "mid", "iv", "book_depth_top3"): x[c + "_p"] = g[c].shift(1) x[c + "_n"] = g[c].shift(-1) x = x.dropna(subset=["strike_p", "strike_n", "bid_p", "bid_n", "ask_p", "ask_n"]) K1, K2, K3 = x["strike_p"].to_numpy(), x["strike"].to_numpy(), x["strike_n"].to_numpy() lam = (K3 - K2) / (K3 - K1) x["lam"] = lam # eseguibile: compro le ali all'ask, vendo il corpo al bid x["cost_exec"] = lam * x["ask_p"] + (1 - lam) * x["ask_n"] - x["bid"] # incoerenza al mid (non eseguibile, ma e' la lente "la superficie e' liscia?") x["cost_mid"] = lam * x["mid_p"] + (1 - lam) * x["mid_n"] - x["mid"] # la stessa cosa in punti di vol: residuo di convessita' locale dello smile x["iv_resid"] = x["iv"] - (lam * x["iv_p"] + (1 - lam) * x["iv_n"]) return x # --------------------------------------------------------------------------- vol points def in_punti_vol(x: pd.DataFrame) -> pd.DataFrame: """Mezzo spread denaro-lettera convertito in punti di vol via vega BS (r=0, forward).""" vega_usd = bs_vega_usd(x["F"].to_numpy(), x["strike"].to_numpy(), x["T"].to_numpy(), (x["iv"] / 100).to_numpy()) vega_coin_per_pt = vega_usd / x["spot_feed"].to_numpy() / 100.0 x = x.copy() x["vega_coin_pt"] = vega_coin_per_pt x["halfspread_pt"] = (x["ask"] - x["bid"]) / 2.0 / np.clip(vega_coin_per_pt, 1e-12, None) x["relspread"] = (x["ask"] - x["bid"]) / x["mid"] # quanto vale UN TICK in punti di vol: se il tick e' piu' grosso dell'incoerenza, # l'incoerenza non e' esprimibile nella griglia di prezzo del venue. x["tick_pt"] = tick_of(x["mid"].to_numpy()) / np.clip(vega_coin_per_pt, 1e-12, None) return x # --------------------------------------------------------------------------- TEST C def test_calendario(q: pd.DataFrame, dte_min: float, ogni: int = 4) -> dict: """Varianza totale w = iv^2 * T: deve CRESCERE col tenore. Due lenti, e la loro differenza. (i) STRIKE CONDIVISO fra due scadenze adiacenti: nessuna interpolazione. (ii) LOG-MONEYNESS FISSO k = ln(K/F): richiede di interpolare lo smile lungo in k. La differenza fra (i) e (ii) e' la misura diretta del contributo dell'INTERPOLATORE. """ x = q[(q["dte"] >= dte_min) & q["iv"].notna() & (q["iv"] > 0) & q["F"].notna()].copy() x = x[x["ts"] >= SURF_START] tenuti = sorted(x["ts"].unique())[::ogni] # (i) e (ii) sugli STESSI snapshot: appaiato x = x[x["ts"].isin(tenuti)] x["w"] = (x["iv"] / 100.0) ** 2 * x["T"] x["k"] = np.log(x["strike"] / x["F"]) # --- (i) strike condiviso, adiacenti nel tempo x = x.sort_values(["asset", "ts", "option_type", "strike", "exp"]) key = ["asset", "ts", "option_type", "strike"] g = x.groupby(key, sort=False) nxt_w = g["w"].shift(-1).to_numpy() nxt_T = g["T"].shift(-1).to_numpy() nxt_iv = g["iv"].shift(-1).to_numpy() ok = np.isfinite(nxt_w) & (nxt_T > x["T"].to_numpy()) dw = nxt_w - x["w"].to_numpy() viol_i = ok & (dw < 0) # ampiezza in punti di vol: quanto dovrebbe salire l'iv lunga per ripristinare la monotonia iv_need = np.sqrt(np.maximum(x["w"].to_numpy(), 0) / np.maximum(nxt_T, 1e-9)) * 100.0 amp_pt = np.where(ok, iv_need - nxt_iv, np.nan) room = amp_pt[ok] # <0 = margine, >0 = violazione res = {"room_med_pt": float(np.nanmedian(room)), "room_p5_pt": float(np.nanpercentile(room, 95)), "i_n": int(ok.sum()), "i_viol": int(viol_i.sum()), "i_quota": float(viol_i.sum() / max(1, ok.sum())), "i_amp_med_pt": float(np.nanmedian(amp_pt[viol_i])) if viol_i.any() else np.nan, "i_amp_p95_pt": float(np.nanpercentile(amp_pt[viol_i], 95)) if viol_i.any() else np.nan} # --- (ii) log-moneyness fisso, con interpolazione y = x n_par, n_vio, amps = 0, 0, [] for (a, ts, ot), gg in y.groupby(["asset", "ts", "option_type"], sort=False): exps = sorted(gg["exp"].unique()) if len(exps) < 2: continue for e1, e2 in zip(exps[:-1], exps[1:]): s1 = gg[gg["exp"] == e1].sort_values("k") s2 = gg[gg["exp"] == e2].sort_values("k") if len(s1) < 2 or len(s2) < 2: continue k1 = s1["k"].to_numpy() inside = (k1 >= s2["k"].min()) & (k1 <= s2["k"].max()) # MAI estrapolare if not inside.any(): continue iv2 = np.interp(k1[inside], s2["k"].to_numpy(), s2["iv"].to_numpy()) T2 = float(s2["T"].iloc[0]) w2 = (iv2 / 100.0) ** 2 * T2 w1 = s1["w"].to_numpy()[inside] n_par += int(inside.sum()) bad = w2 < w1 n_vio += int(bad.sum()) if bad.any(): need = np.sqrt(w1[bad] / T2) * 100.0 amps.append(need - iv2[bad]) amps = np.concatenate(amps) if amps else np.array([]) res.update({"ii_n": n_par, "ii_viol": n_vio, "ii_quota": n_vio / max(1, n_par), "ii_amp_med_pt": float(np.median(amps)) if len(amps) else np.nan, "ii_ogni": ogni}) # valore in dollari dell'ampiezza (i) al lotto minimo, contro il costo di 2 gambe vv = x.loc[viol_i].copy() if len(vv): vv = in_punti_vol(vv) amp = pd.Series(amp_pt, index=x.index)[viol_i].to_numpy() lot = vv["asset"].map(MIN_LOT).to_numpy() val = np.abs(amp) * vv["vega_coin_pt"].to_numpy() * lot * vv["spot_feed"].to_numpy() costo = ((vv["ask"] - vv["bid"]).to_numpy() * lot * vv["spot_feed"].to_numpy() + 2 * fee_coin(vv["mid"].to_numpy(), lot) * vv["spot_feed"].to_numpy()) res.update({"i_val_usd_med": float(np.nanmedian(val)), "i_costo_usd_med": float(np.nanmedian(costo)), "i_n_val": int(np.isfinite(val).sum())}) return res # --------------------------------------------------------------------------- economia def economia(bf: pd.DataFrame, asset: str) -> dict: """Quanto vale, in dollari, l'incoerenza di una farfalla al lotto minimo — netto fee. Struttura a 3 gambe, lotti INTERI: con strike equispaziati e' 1:2:1 sul lotto minimo. Si conta la fee gamba per gamba col cap sul premio, e si usa il prezzo ESEGUIBILE. """ x = bf[bf["asset"] == asset] lot = MIN_LOT[asset] S = x["spot_feed"].to_numpy() # credito lordo della farfalla eseguibile, in coin per unita' di "corpo" cred = -x["cost_exec"].to_numpy() # >0 = credito riskless # lotti interi: corpo 2*lot, ali lot ciascuna (equivale a lam=0.5 -> solo strike equispaziati) eq = np.isclose(x["lam"].to_numpy(), 0.5, atol=0.02) cred_eq_coin = (x["bid"].to_numpy() * 2 * lot - x["ask_p"].to_numpy() * lot - x["ask_n"].to_numpy() * lot) f = (fee_coin(x["bid"].to_numpy(), 2 * lot) + fee_coin(x["ask_p"].to_numpy(), lot) + fee_coin(x["ask_n"].to_numpy(), lot)) netto_usd = (cred_eq_coin - f) * S return { "asset": asset, "n": int(len(x)), "n_equisp": int(eq.sum()), "cred_lordo_p99_coin": float(np.nanpercentile(cred, 99)), "fee_3gambe_usd_med": float(np.nanmedian(f[eq] * S[eq])) if eq.any() else np.nan, "netto_usd_max": float(np.nanmax(netto_usd[eq])) if eq.any() else np.nan, "netto_usd_p99": float(np.nanpercentile(netto_usd[eq], 99)) if eq.any() else np.nan, "n_netto_pos": int(np.nansum(netto_usd[eq] > 0)), "nozionale_min_lotto_usd": float(np.nanmedian(S) * lot), } # --------------------------------------------------------------------------- persistenza def persistenza(bf: pd.DataFrame, viol_col: str) -> dict: """Una violazione che sparisce all'ora dopo e' un guizzo di book, non un'opportunita'. Chiave = (asset, exp, tipo, tripla di strike). Si guarda se la stessa tripla e' ancora in violazione allo snapshot successivo IN CUI LA TRIPLA E' QUOTATA (non 'l'ora dopo' in astratto). """ v = bf[bf[viol_col]].copy() if v.empty: return {"n": 0} bf = bf.copy() bf["trip"] = (bf["asset"] + "|" + bf["exp"].astype(str) + "|" + bf["option_type"] + "|" + bf["strike_p"].astype(str) + "|" + bf["strike"].astype(str) + "|" + bf["strike_n"].astype(str)) bf = bf.sort_values(["trip", "ts"]) nxt = bf.groupby("trip", sort=False)[viol_col].shift(-1) m = bf[viol_col] & nxt.notna() return {"n": int(m.sum()), "resta": float(nxt[m].mean()) if m.any() else np.nan} def controfattuale_usdc(bf: pd.DataFrame, asset: str, rel_usdc: float, solo_gia_strette: bool = False) -> dict: """SE la famiglia USDC quotasse la STESSA superficie col SUO spread, sarebbe eseguibile? Controfattuale DICHIARATO, non una misura: il collettore non raccoglie le USDC (difetto di produzione del filone 11), quindi si tiene il mid osservato sulle inverse e si stringe lo spread al valore relativo misurato dagli altri filoni sulla famiglia USDC, con il lotto minimo USDC e la stessa formula di fee. Assunzione dichiarata: stessa griglia di tick (una griglia piu' grossa peggiorerebbe il risultato, quindi questo e' un LIMITE SUPERIORE). """ x = bf[(bf["asset"] == asset)].dropna(subset=["mid", "mid_p", "mid_n"]).copy() x = x[np.isclose(x["lam"], 0.5, atol=0.02)] if solo_gia_strette: # il controfattuale ingenuo prende il MID di un mercato larghissimo e finge di poterlo # quotare al 2,6%: e' proprio dove il mid e' meno informativo. Qui si tiene solo la parte # di superficie GIA' quotata stretta quanto la famiglia USDC, dove il mid vale qualcosa. rs = np.maximum.reduce([((x["ask"] - x["bid"]) / x["mid"]).to_numpy(), ((x["ask_p"] - x["bid_p"]) / x["mid_p"]).to_numpy(), ((x["ask_n"] - x["bid_n"]) / x["mid_n"]).to_numpy()]) x = x[rs <= rel_usdc] lot = MIN_LOT_USDC[asset] out = {} for nome, rel in (("inverse osservato", None), (f"{asset}_USDC controfattuale", rel_usdc)): if rel is None: b, a_p, a_n = x["bid"], x["ask_p"], x["ask_n"] l = MIN_LOT[asset] else: b = x["mid"] * (1 - rel / 2) a_p = x["mid_p"] * (1 + rel / 2) a_n = x["mid_n"] * (1 + rel / 2) l = lot cred = (b.to_numpy() * 2 * l - a_p.to_numpy() * l - a_n.to_numpy() * l) f = (fee_coin(b.to_numpy(), 2 * l) + fee_coin(a_p.to_numpy(), l) + fee_coin(a_n.to_numpy(), l)) netto = (cred - f) * x["spot_feed"].to_numpy() out[nome] = {"n": int(len(x)), "pos": int(np.nansum(netto > 0)), "pos1": int(np.nansum(netto > 1.0)), "somma": float(np.nansum(netto[netto > 0])), "max": float(np.nanmax(netto)) if len(x) else np.nan, "p999": float(np.nanpercentile(netto, 99.9)) if len(x) else np.nan, "lotto_usd": float(np.nanmedian(x["spot_feed"]) * l) if len(x) else np.nan} return out # --------------------------------------------------------------------------- controlli positivi def _finto(rows) -> pd.DataFrame: """Catena sintetica minima per i controlli positivi.""" d = pd.DataFrame(rows) d["ts"] = pd.to_datetime(d["ts"], utc=True) d["exp"] = pd.to_datetime(d["exp"], utc=True) d["dte"] = (d["exp"] - d["ts"]).dt.total_seconds() / 86400.0 d["T"] = d["dte"] / 365.25 d["mid"] = (d["bid"] + d["ask"]) / 2 for c, v in (("spot_feed", 100_000.0), ("F", 100_000.0), ("option_type", "P"), ("open_interest", 1000.0), ("book_depth_top3", 10.0), ("delta", -0.2)): if c not in d: d[c] = v return d def controlli_positivi() -> None: """Un rilevatore tarato per non segnalare e' indistinguibile da uno rotto. Regola gia' codificata nel progetto (venue_watch, 26/07). Qui i tre test vengono puntati su violazioni COSTRUITE: se non le vedono, i loro zeri non valgono niente. """ print("=" * 100) print("CONTROLLI POSITIVI — i tre rilevatori vedono una violazione COSTRUITA?") print("=" * 100) base = dict(ts="2026-08-01T00:00Z", exp="2026-08-15T08:00Z", asset="BTC", option_type="P", iv=50.0, instrument_name="x") # (1) verticale: put a strike BASSO con bid > ask della put a strike ALTO v_ok = _finto([{**base, "strike": 90000, "bid": 0.0300, "ask": 0.0305}, {**base, "strike": 95000, "bid": 0.0200, "ask": 0.0205}]) v_no = _finto([{**base, "strike": 90000, "bid": 0.0200, "ask": 0.0205}, {**base, "strike": 95000, "bid": 0.0300, "ask": 0.0305}]) r_ok, r_no = test_verticale(v_ok, 0.0), test_verticale(v_no, 0.0) print(f" TEST V: catena INVERTITA -> {r_ok['n_viol']} violazioni (attese >=1) · " f"catena sana -> {r_no['n_viol']} (attese 0)") # (2) farfalla: corpo troppo caro (concavita') b_ok = _finto([{**base, "strike": 90000, "bid": 0.0100, "ask": 0.0105}, {**base, "strike": 95000, "bid": 0.0400, "ask": 0.0405}, {**base, "strike": 100000, "bid": 0.0500, "ask": 0.0505}]) b_no = _finto([{**base, "strike": 90000, "bid": 0.0100, "ask": 0.0105}, {**base, "strike": 95000, "bid": 0.0280, "ask": 0.0285}, {**base, "strike": 100000, "bid": 0.0500, "ask": 0.0505}]) f_ok = test_farfalla(b_ok, 0.0); f_no = test_farfalla(b_no, 0.0) print(f" TEST B: corpo troppo caro -> {(f_ok['cost_exec'] < 0).sum()} violazioni eseguibili " f"(attese >=1) · smile convesso -> {(f_no['cost_exec'] < 0).sum()} (attese 0)") # (3) calendario: varianza totale che SCENDE col tenore, stesso strike c_rows = [] for e, iv in (("2026-08-08T08:00Z", 80.0), ("2026-08-15T08:00Z", 40.0)): c_rows.append({**base, "exp": e, "strike": 100000, "bid": 0.01, "ask": 0.011, "iv": iv}) c_ok = _finto(c_rows) c_ok["ts"] = pd.Timestamp("2026-07-01", tz="UTC") c_no = c_ok.copy(); c_no["iv"] = [40.0, 80.0] r1 = test_calendario(c_ok, 0.0, ogni=1); r2 = test_calendario(c_no, 0.0, ogni=1) print(f" TEST C: varianza totale DECRESCENTE -> {r1['i_viol']} violazioni (attese >=1) · " f"crescente -> {r2['i_viol']} (attese 0)") assert r_ok["n_viol"] >= 1 and r_no["n_viol"] == 0 assert (f_ok["cost_exec"] < 0).sum() >= 1 and (f_no["cost_exec"] < 0).sum() == 0 assert r1["i_viol"] >= 1 and r2["i_viol"] == 0 print(" -> 3/3 rilevatori validati in ENTRAMBI i versi: gli zeri dei test veri sono zeri.\n") # --------------------------------------------------------------------------- reversione def reversione(bf: pd.DataFrame, max_gap_h: float = 2.0) -> dict: """Il residuo di convessita' si RICOMPONE? (l'ipotesi del filone, misurata) Coppie appaiate sulla STESSA tripla di strike a snapshot consecutivi entro `max_gap_h`. beta di resid[t+1] - resid[t] su resid[t] : beta = -1 -> ricomposizione totale in un'ora, beta = 0 -> il residuo e' una proprieta' persistente dello smile (skew), non un errore. """ x = bf.dropna(subset=["iv_resid"]).copy() x["trip"] = (x["asset"] + "|" + x["exp"].astype(str) + "|" + x["option_type"] + "|" + x["strike_p"].astype(str) + "|" + x["strike"].astype(str) + "|" + x["strike_n"].astype(str)) x = x.sort_values(["trip", "ts"]) g = x.groupby("trip", sort=False) r1 = g["iv_resid"].shift(-1).to_numpy() dt = (g["ts"].shift(-1) - x["ts"]).dt.total_seconds().to_numpy() / 3600.0 r0 = x["iv_resid"].to_numpy() m = np.isfinite(r1) & np.isfinite(r0) & (dt > 0) & (dt <= max_gap_h) if m.sum() < 100: return {"n": int(m.sum())} b = np.polyfit(r0[m], r1[m] - r0[m], 1)[0] return {"n": int(m.sum()), "beta": float(b), "corr": float(np.corrcoef(r0[m], r1[m])[0, 1]), "meta_vita_h": float(np.log(0.5) / np.log(max(1e-9, 1 + b))) if -1 < b < 0 else np.nan} # --------------------------------------------------------------------------- main def main() -> None: t0 = time.time() controlli_positivi() d = load() q, F = certifica(d) q = q.merge(F, on=["asset", "ts", "exp"], how="left") q["F"] = q["underlying_price"].fillna(q["F_pcp"]) print(f"\n forward disponibile su {q['F'].notna().mean():.1%} delle quote a due lati " f"(osservato dove c'e', altrimenti parita' put-call)") for dte_min, etichetta in ((0.0, "SENZA filtro DTE"), (DTE_MIN, f"DTE >= {DTE_MIN:g}g")): print("\n" + "=" * 100) print(f"1-2. ARBITRAGGI STATICI MODEL-FREE su prezzi ESEGUIBILI — {etichetta}") print("=" * 100) v = test_verticale(q, dte_min) print(f" TEST V (monotonia nello strike): {v['n_viol']:,} violazioni su {v['n_confronti']:,} " f"confronti = {v['quota']:.4%}") if v["n_viol"]: print(f" credito mediano {v['credito_med_coin']:.5f} coin = " f"${v['credito_usd_medlot']:.2f} al lotto minimo (LORDO fee)") bf = test_farfalla(q, dte_min) bf = in_punti_vol(bf) bf["viol_exec"] = bf["cost_exec"] < 0 bf["viol_mid"] = bf["cost_mid"] < 0 print(f" TEST B (convessita' nello strike): {len(bf):,} triple consecutive") print(f" al MID (incoerenza, non eseguibile): {bf['viol_mid'].sum():,} = {bf['viol_mid'].mean():.3%}") print(f" ESEGUIBILE (compra ali all'ask, vendi corpo al bid): " f"{bf['viol_exec'].sum():,} = {bf['viol_exec'].mean():.4%}") if dte_min == DTE_MIN: bf_keep = bf bf = bf_keep print("\n" + "=" * 100) print("3. QUANTO E' GRANDE L'INCOERENZA, E QUANTO COSTA ATTRAVERSARLA") print("=" * 100) for a in sorted(bf["asset"].unique()): x = bf[bf["asset"] == a] x = x[np.isfinite(x["halfspread_pt"]) & np.isfinite(x["tick_pt"])] r = np.abs(x["iv_resid"].dropna()) hs, tk = x["halfspread_pt"], x["tick_pt"] print(f" {a} (n={len(x):,} triple con vega misurabile)") print(f" |residuo di convessita' dello smile| mediana {r.median():.3f} pt-vol · " f"p90 {r.quantile(.9):.3f} · p99 {r.quantile(.99):.3f}") print(f" mezzo spread denaro-lettera mediana {hs.median():.3f} pt-vol · " f"p25 {hs.quantile(.25):.3f} · p75 {hs.quantile(.75):.3f}") print(f" UN TICK della griglia di prezzo mediana {tk.median():.3f} pt-vol " f"-> l'incoerenza e' sotto il tick nel {np.mean(r.reindex(x.index) < tk):.1%} dei casi") both = x.dropna(subset=["iv_resid"]) rap = np.abs(both["iv_resid"]) / (2 * both["halfspread_pt"]) print(f" |residuo| / (spread PIENO delle gambe): mediana {rap.median():.3f} · " f"p99 {rap.quantile(.99):.3f} · quota > 1 = {np.mean(rap > 1):.3%}") print(f" spread relativo (ask-bid)/mid del venue: mediana {x['relspread'].median():.1%} " f"· p25 {x['relspread'].quantile(.25):.1%}") # valore in dollari dell'incoerenza al MID (il caso migliore possibile: fee 0, spread 0) eq = np.isclose(x["lam"], 0.5, atol=0.02) lot = MIN_LOT[a] val = -x.loc[eq, "cost_mid"].to_numpy() * 2 * lot * x.loc[eq, "spot_feed"].to_numpy() pos = val[val > 0] print(f" valore LORDO al mid della farfalla incoerente, lotto minimo: " f"mediana ${np.median(pos):.2f} · p99 ${np.percentile(pos,99):.2f} · max ${pos.max():.2f}" f" (n={len(pos):,})") print("\n economia della farfalla al LOTTO MINIMO (famiglia INVERSE, quella nel dato):") for a in sorted(bf["asset"].unique()): e = economia(bf, a) print(f" {a}: nozionale/lotto ${e['nozionale_min_lotto_usd']:,.0f} · " f"fee 3 gambe mediana ${e['fee_3gambe_usd_med']:.2f} · " f"triple equispaziate {e['n_equisp']:,}") print(f" credito netto: max ${e['netto_usd_max']:.2f} · p99 ${e['netto_usd_p99']:.2f} · " f"positivi {e['n_netto_pos']:,}/{e['n_equisp']:,}") print("\n ESTRAPOLAZIONE DICHIARATA alla famiglia USDC (NON nel dato: il collettore raccoglie") print(" le sole inverse — difetto di produzione del filone 11). Il lotto minimo e' 10x piu'") print(" piccolo, ma il muro NON e' il ticket: e' lo spread in punti di vol, che e' una") print(" QUOTA e non un livello -> ridurre il lotto non cambia |residuo|/spread.") for a in sorted(bf["asset"].unique()): x = bf[bf["asset"] == a] x = x[np.isfinite(x["halfspread_pt"])].dropna(subset=["iv_resid"]) rap = np.abs(x["iv_resid"]) / (2 * x["halfspread_pt"]) rel_inv = x["relspread"].median() for nome, rel_usdc in ((f"{a}_USDC", 0.082 if a == "BTC" else 0.026),): k = rel_inv / rel_usdc # di quanto lo spread sarebbe piu' stretto print(f" {a}: spread relativo inverse {rel_inv:.1%} vs {nome} {rel_usdc:.1%} " f"-> fattore {k:.1f}x; rapporto mediano {rap.median():.3f} -> {rap.median()*k:.3f}, " f"quota > 1 {np.mean(rap > 1):.2%} -> {np.mean(rap * k > 1):.2%}") print("\n" + "=" * 100) print("4. TEST C — varianza totale nel tenore (l'unico che tocca un interpolatore)") print("=" * 100) c = test_calendario(q, DTE_MIN) print(f" (i) strike CONDIVISO, nessuna interpolazione: {c['i_viol']:,}/{c['i_n']:,} = {c['i_quota']:.3%}" f" · ampiezza mediana {c['i_amp_med_pt']:.3f} pt-vol (p95 {c['i_amp_p95_pt']:.3f})") print(f" (ii) log-moneyness FISSO, con interpolazione (1 snapshot ogni {c['ii_ogni']}): " f"{c['ii_viol']:,}/{c['ii_n']:,} = {c['ii_quota']:.3%} · ampiezza mediana {c['ii_amp_med_pt']:.3f} pt-vol") print(f" -> contributo dell'INTERPOLATORE = {c['ii_quota'] - c['i_quota']:+.3%} di quota di violazione") if "i_val_usd_med" in c: print(f" valore mediano della violazione (i) al lotto minimo ${c['i_val_usd_med']:.2f} " f"contro un costo di 2 gambe (spread + fee) di ${c['i_costo_usd_med']:.2f}") print(f" margine di monotonia sull'intera superficie: la iv lunga sta {-c['room_med_pt']:.2f} pt-vol " f"SOPRA il minimo che la non-arbitraggio richiede (5% piu' stretto: {-c['room_p5_pt']:.2f}) " f"-> la superficie non e' nemmeno VICINA al bordo") print("\n" + "=" * 100) print("5. TICK E PERSISTENZA — le due difese contro 'la mia incoerenza e' un artefatto'") print("=" * 100) for col, nome in (("viol_mid", "al MID"), ("viol_exec", "ESEGUIBILE")): v = bf[bf[col]] if len(v) == 0: print(f" {nome}: nessuna violazione") continue legs = np.concatenate([v["bid"].to_numpy(), v["bid_p"].to_numpy(), v["bid_n"].to_numpy()]) legs = legs[np.isfinite(legs) & (legs > 0)] nt = legs / tick_of(legs) p = persistenza(bf, col) print(f" {nome}: n={len(v):,} · gambe a <=2 tick {np.mean(nt <= 2):.1%} " f"(mediana {np.median(nt):.0f} tick) · resta in violazione allo snapshot dopo " f"{p.get('resta', float('nan')):.1%} (n={p['n']:,})") print("\n CONTROFATTUALE (dichiarato, non misurato): la stessa superficie con lo spread USDC") for a, rel in (("BTC", 0.082), ("ETH", 0.026)): if a not in set(bf["asset"]): continue for strette, et in ((False, "tutta la superficie"), (True, "solo quote gia' strette")): cf = controfattuale_usdc(bf, a, rel, solo_gia_strette=strette) for nome, r in cf.items(): if strette and "inverse" in nome: continue print(f" {a} {nome:<26} [{et:<24}]: lotto ${r['lotto_usd']:,.0f} · " f"credito netto>0 {r['pos']:,}/{r['n']:,} · >$1 {r['pos1']:,} · " f"migliore ${r['max']:.2f} · somma ${r['somma']:,.0f}") # dove vive l'incoerenza: e' larga dove il mercato e' largo? x = bf[(bf["asset"] == a)].dropna(subset=["iv_resid", "relspread"]) x = x[np.isfinite(x["relspread"])] dec = pd.qcut(x["relspread"], 5, labels=False, duplicates="drop") t = x.assign(d=dec).groupby("d").agg(spread=("relspread", "median"), resid=("iv_resid", lambda v: v.abs().median())) print(f" |residuo| per quintile di spread relativo {a}: " + " · ".join(f"{r.spread:.1%}->{r.resid:.3f}pt" for r in t.itertuples())) print("\n QUANTO MANCA per essere eseguibile (near-miss, triple equispaziate):") for a in sorted(bf["asset"].unique()): x = bf[(bf["asset"] == a) & np.isclose(bf["lam"], 0.5, atol=0.02)] c = x["cost_exec"].to_numpy() * 2 * MIN_LOT[a] * x["spot_feed"].to_numpy() # >0 = costo print(f" {a}: costo della farfalla eseguibile — migliore ${np.nanmin(c):.2f} · " f"p0.1 ${np.nanpercentile(c, 0.1):.2f} · mediana ${np.nanmedian(c):.2f} " f"(negativo = credito riskless)") print("\n" + "=" * 100) print("6. IL RESIDUO SI RICOMPONE? (l'ipotesi del filone, misurata invece che assunta)") print("=" * 100) for a in sorted(bf["asset"].unique()): rv = reversione(bf[bf["asset"] == a]) if "beta" not in rv: print(f" {a}: coppie insufficienti (n={rv['n']})") continue print(f" {a}: n={rv['n']:,} coppie appaiate (stessa tripla, snapshot consecutivo <=2h) · " f"beta di ricomposizione {rv['beta']:+.3f} · corr(resid_t, resid_t+1) {rv['corr']:+.3f}") if np.isfinite(rv["meta_vita_h"]): print(f" -> meta-vita {rv['meta_vita_h']:.1f} h (beta=-1 sarebbe ricomposizione totale " f"in un'ora, beta=0 residuo persistente = forma dello smile, non errore)") print("\n" + "=" * 100) print("7. ROBUSTEZZA D'ANCORA — il risultato dipende dall'ora in cui guardo?") print("=" * 100) h = bf.assign(ora=bf["ts"].dt.hour).groupby("ora").agg( mid=("viol_mid", "mean"), ex=("viol_exec", "mean"), n=("viol_mid", "size")) print(f" quota di incoerenza al MID per ora UTC: mediana {h['mid'].median():.3%} · " f"min {h['mid'].min():.3%} (h{h['mid'].idxmin():02d}) · max {h['mid'].max():.3%} (h{h['mid'].idxmax():02d})") print(f" quota ESEGUIBILE per ora UTC: max {h['ex'].max():.5%} su {len(h)} ore -> " f"nessuna ora del giorno apre una finestra") print("\n" + "=" * 100) print("GRIGLIA DICHIARATA (contata al RIALZO)") print("=" * 100) print(" 3 test di arbitraggio (verticale · farfalla · calendario)") print(" x 2 asset x 2 tipi (call/put) x 2 filtri DTE (0g e 1g) x 2 convenzioni di prezzo") print(" (mid = incoerenza · bid/ask = eseguibile) = 48 celle") print(" + calendario in 2 lenti (strike condiviso / log-moneyness interpolato) = 4") print(" + controfattuale USDC x 2 asset x 2 sottocampioni = 4") print(" + reversione x 2 asset · ancora oraria 24 ore x 2 metriche = 50") print(" + 6 controlli positivi (3 rilevatori x 2 versi) = 6") print(" TOTALE dichiarato ~112 valutazioni. NESSUNA e' una selezione: non c'e' una cella") print(" scelta, c'e' un censimento. Il deflated-Sharpe non e' applicabile (nessuno stream).") print("\n" + "=" * 100) print("SINTESI") print("=" * 100) nv = int(bf["viol_exec"].sum()); nt = len(bf) print(f" 1. La superficie e' ARBITRAGE-FREE sui prezzi eseguibili: monotonia 0/{1_111_585:,},") print(f" convessita' {nv}/{nt:,} = {nv/nt:.6%} (e quell'una vale $0,00 e sparisce l'ora dopo),") print(f" calendario 6/107.177 con margine mediano di 13 punti di vol dal bordo.") print(f" 2. L'incoerenza al MID esiste ed e' grande ({bf['viol_mid'].mean():.1%} delle triple)") print(f" ma vale {np.abs(bf['iv_resid']).median():.3f} punti di vol contro un mezzo spread di 0,86-1,21,") print(f" ed e' SOTTO IL TICK del venue nel 95-97% dei casi: non e' esprimibile nella griglia.") print(f" 3. MECCANISMO: |residuo| cresce in modo monotono col LARGO del mercato") print(f" (BTC 0,080 pt a spread 2,6% -> 0,255 pt a spread 50%). L'incoerenza al mid E' la") print(f" larghezza del mercato guardata attraverso il mid, non un errore di prezzo.") print(f" 4. Il controfattuale USDC (lotto 10x piu' piccolo, spread 2,6%) sembra aprire 2.293") print(f" farfalle a credito su ETH — ma restringendolo alla parte di superficie GIA' quotata") print(f" stretta quanto le USDC diventa 0/13.287. Era un artefatto del controfattuale.") print(f" -> l'incoerenza ESISTE e vale meno del costo di attraversarla: verdetto SCARTATO.") print(f"\n[tempo {time.time()-t0:.1f}s]") if __name__ == "__main__": main()