"""COLLAR01 — hold BTC (long/short) con COLLAR di opzioni a <=15 giorni. DOMANDA (posta dall'operatore il 2026-09-01): "crea una strategia di hold BTC in long o short con copertura con options (max 15gg)" + due precisazioni in corso d'opera: "opzione deve essere al max di 15gg" -> vincolo sulla SCADENZA, non sul solo holding "voglio ridurre la vincita, ma bloccare la perdita" -> NON e' una put protettiva: e' un COLLAR (pavimento comprato, tetto venduto per finanziarlo). "l'entrata deve essere gestita da una strategia confortata da indicatori (es. forte bull)" "dalle opzioni dobbiamo uscire prima del termine (tra 50% e 75% del tempo)" -> `exit_frac`. PERCHE' QUESTO FILONE NON E' §46 (TAIL-HEDGE, REFUTATO il 2026-08-23). §46 compro' put deep-OTM sul LIBRO (TP01+SKH01) e fu refutato *prima del prezzo*: il maxDD saliva in 162/162 celle perche' **il beta del libro al sottostante e' +0,076** — "non si assicura un libro che nei crash e' gia' quasi piatto". Qui il sottostante e' un **hold di BTC nudo, beta 1,0 per costruzione**: il motivo che uccise §46 NON si applica, ed e' l'unica ragione per cui questo filone puo' essere riaperto (CLAUDE.md: "chi la riapre deve battere il motivo, non ripetere la misura"). §46 dichiara inoltre di NON aver provato la copertura dinamica e di aver misurato la sola statica: anche quella porta e' aperta, ma qui non la uso — il collar e' statico a griglia. COSA E' GIA' NOTO E VIENE RIUSATO, NON RIMISURATO: - metriche e i due null di scala (`k_for_same_dd`, `k_for_same_drift`) da r0823_tail_hedge; - listino fee opzioni Deribit (taker 0,03% del sottostante, cap 12,5% del premio, consegna 0,015%) dallo stesso; - la catena reale via `cblib.load_chain()` (archivio 2026-05-01 + raccolta propria 2026-07-30). LE DUE LENTI, ed e' una scelta dichiarata: (A) LENTE LUNGA (decide): 2021-03-24 -> oggi, la finestra del DVOL — ~5,4 anni. Opzioni prezzate a modello (BS, r=0) con DVOL come ancora ATM, CORRETTA da skew e struttura a termine MISURATI SULLA CATENA VERA, e spread f misurato per ala. E' l'impianto di §46. (B) LENTE REALE (controlla il segno, NON decide): 2026-05-01 -> oggi, solo quote a due lati. 123 giorni = ~8 cicli a 15 giorni: sotto-potenziata per costruzione, e per giunta su una finestra in cui BTC e' salito da ~64,7k a ~77,5k, che e' AVVERSA a un collar (il tetto si paga proprio li'). Si legge il segno, mai la taglia. M5 (null del de-levering) E' IL PRIMO TEST, NON L'ULTIMO: ogni claim "meno drawdown" si misura contro "tieni semplicemente meno BTC" a iso-maxDD. Se il collar non batte quello, non e' protezione: e' una ri-scalatura pagata cara. C9 e' il secondo: "una regola d'uscita che scatta piu' spesso sui vincenti che sui perdenti non e' protezione, e' troncatura" — qui il tetto e' letteralmente quella regola. NESSUN ORDINE. Libro, pesi, cron, config INVARIATI. """ from __future__ import annotations import math import sys 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)) sys.path.insert(0, str(ROOT / "scripts" / "research")) import cblib # noqa: E402 from src.data.downloader import load_data # noqa: E402 DPY = 365.25 # ---- listino Deribit opzioni (identico a r0823_tail_hedge, non ridichiarato a mano) ----------- OPT_TAKER_UNDERLYING = 0.0003 OPT_FEE_CAP_FRAC = 0.125 OPT_DELIVERY_UNDERLYING = 0.00015 # ---- GRIGLIA DICHIARATA PRIMA DI GUARDARE I RISULTATI ----------------------------------------- G_DPUT = (0.10, 0.20, 0.30) # |delta| della put comprata = il PAVIMENTO G_DCALL = (0.10, 0.20, 0.30) # delta della call venduta = il TETTO G_TENOR = (7, 14) # giorni a scadenza: <=15, vincolo dell'operatore N_CELLS = len(G_DPUT) * len(G_DCALL) * len(G_TENOR) # varianti dichiarate a parte (non nella griglia, per non gonfiare i trial) V_ZEROCOST = "zero-cost" # tetto scelto per finanziare esattamente il pavimento ATTESE = """ ATTESE A PRIORI — scritte e stampate PRIMA di misurare (M12: un follow-up contiene una previsione). A1 Il maxDD SCENDE davvero, a differenza di §46: il beta e' 1,0 e il pavimento morde. (se A1 fallisse, il filone cade come §46 e per lo stesso motivo) A2 Il drift scende PIU' del maxDD, perche' il rendimento di BTC vive nella coda destra e il tetto la taglia. Previsione: rapporto Δdrift/ΔmaxDD > 1. A3 Il collar PERDE contro il null del de-levering a iso-maxDD in piu' della meta' delle celle. E' la previsione centrale del filone. A4 Lo skew rende il collar delta-simmetrico un DEBITO netto (put piu' cara della call). [gia' misurato prima di scrivere: +31% / +19% / +12% a |d| 0,10 / 0,20 / 0,30 -> CONFERMATA] A5 Il tick morde MENO che in §46: quella era la famiglia USDC (tick 5 USDC); la catena che raccogliamo e' 100% inverse, con tick in BTC. A6 Nella lente reale (BTC +20% in 123 giorni) il collar sembrera' pessimo: finestra avversa, non evidenza. Va letta come segno, non come taglia. A7 Il verso SHORT sara' peggiore del LONG: il drift di BTC e' positivo e lo short lo paga tutto, col collar che ne cappa il solo rimbalzo favorevole. """ # ==================================================================== 0. METRICHE E NULL (da §46) def _sh(x) -> float: r = np.asarray(pd.Series(x).dropna().values, float) return float(np.mean(r) / np.std(r) * math.sqrt(DPY)) if len(r) > 2 and np.std(r) > 0 else 0.0 def _dd(x) -> float: r = np.asarray(pd.Series(x).dropna().values, float) eq = np.cumprod(1.0 + r) pk = np.maximum.accumulate(eq) return float(np.max((pk - eq) / pk)) if len(eq) else 0.0 def _drift(x) -> float: r = np.asarray(pd.Series(x).dropna().values, float) if len(r) < 2: return 0.0 tot = float(np.prod(1.0 + r)) return -1.0 if tot <= 0 else float(tot ** (DPY / len(r)) - 1.0) def k_for_same_dd(base: np.ndarray, target_dd: float) -> float: """La scala costante che porta il maxDD del NUDO esattamente a quello del coperto (bisezione).""" lo, hi = 1e-4, 1.0 if _dd(hi * base) <= target_dd: return 1.0 for _ in range(60): mid = 0.5 * (lo + hi) if _dd(mid * base) > target_dd: hi = mid else: lo = mid return 0.5 * (lo + hi) # ==================================================================== 1. CALIBRAZIONE DALLA CATENA _BANDS = (0.05, 0.10, 0.15, 0.20, 0.25, 0.30, 0.40, 0.50) def calibra_catena() -> dict: """Skew, struttura a termine e spread f: MISURATI sulla catena vera, mai assunti. Restituisce funzioni di correzione applicate al DVOL (che e' un ATM a 30 giorni) per ottenere l'IV di una gamba a delta e scadenza dati. Senza queste il modello prezzerebbe la put e la call allo STESSO vol: l'errore andrebbe tutto a favore del collar (A4). """ df = cblib.load_chain() b = df[df["asset"].eq("BTC")].copy() b["exp"] = pd.to_datetime(b["exp"], utc=True) b["ts"] = pd.to_datetime(b["ts"], utc=True) b["dte"] = (b["exp"] - b["ts"]).dt.total_seconds() / 86400.0 q = b[(b["dte"] > 1) & (b["dte"] <= 15) & b["bid"].gt(0) & b["ask"].gt(0) & b["iv"].notna() & b["delta"].notna()].copy() q["ad"] = q["delta"].abs() atm = q[q["ad"].between(0.45, 0.55)].groupby(["ts", "exp"])["iv"].mean().rename("iv_atm") q = q.join(atm, on=["ts", "exp"]) q = q[q["iv_atm"].notna() & q["iv_atm"].gt(0)] skew = {} for t, nm in (("P", "put"), ("C", "call")): xs, ys = [], [] for d in _BANDS: lo, hi = d * 0.85, d * 1.15 s = q[q["option_type"].eq(t) & q["ad"].between(lo, hi)] if len(s) >= 200: xs.append(d) ys.append(float((s["iv"] / s["iv_atm"]).median())) skew[nm] = (np.array(xs), np.array(ys)) # struttura a termine: IV ATM a DTE t contro il DVOL 30g dello stesso istante dv = cblib.dvol_series("BTC") a = q[q["ad"].between(0.45, 0.55)].copy() a["giorno"] = a["ts"].dt.floor("D") dvd = dv.copy() dvd.index = pd.to_datetime(dvd.index, utc=True).floor("D") dvd = dvd.groupby(level=0).last() a = a.join(dvd.rename("dvol"), on="giorno") a = a[a["dvol"].notna() & a["dvol"].gt(0)] a["ratio"] = (a["iv"] / 100.0) / (a["dvol"] / 100.0) if a["iv"].median() > 3 else a["iv"] / (a["dvol"] / 100.0) ts_x, ts_y = [], [] for lo, hi in ((1, 4), (4, 8), (8, 12), (12, 15)): s = a[a["dte"].between(lo, hi)] if len(s) >= 200: ts_x.append(0.5 * (lo + hi)) ts_y.append(float(s["ratio"].median())) # spread relativo per ala e per banda di delta (mezza forchetta su mid) = il costo di f spread = {} for t, nm in (("P", "put"), ("C", "call")): xs, ys = [], [] for d in _BANDS: lo, hi = d * 0.85, d * 1.15 s = q[q["option_type"].eq(t) & q["ad"].between(lo, hi)] s = s[s["mid"].gt(0)] if len(s) >= 200: xs.append(d) ys.append(float((((s["ask"] - s["bid"]) / 2.0) / s["mid"]).median())) spread[nm] = (np.array(xs), np.array(ys)) return dict(skew=skew, term=(np.array(ts_x), np.array(ts_y)), spread=spread, n_quote=len(q), giorni=int(q["ts"].dt.date.nunique()), span=(q["ts"].min(), q["ts"].max())) def _interp(tab, x): xs, ys = tab if len(xs) == 0: return 1.0 return float(np.interp(x, xs, ys)) # ==================================================================== 2. PREZZATORE def _d1(S, K, T, sig): return (math.log(S / K) + 0.5 * sig * sig * T) / (sig * math.sqrt(T)) def bs_call(S, K, T, sig): if T <= 0 or sig <= 0: return max(S - K, 0.0) d1 = _d1(S, K, T, sig) return S * norm.cdf(d1) - K * norm.cdf(d1 - sig * math.sqrt(T)) def bs_put(S, K, T, sig): if T <= 0 or sig <= 0: return max(K - S, 0.0) d1 = _d1(S, K, T, sig) return K * norm.cdf(-(d1 - sig * math.sqrt(T))) - S * norm.cdf(-d1) def strike_da_delta(S, T, sig_atm, cal, delta, tipo): """Strike che realizza |delta| dato, risolvendo la circolarita' skew<->delta per punto fisso. L'IV dipende dal delta (skew) e il delta dipende dallo strike che dipende dall'IV: due iterazioni bastano (verificato: la terza muove lo strike di <0,1%). """ sig = sig_atm * _interp(cal["skew"][tipo], delta) K = S for _ in range(3): z = norm.ppf(delta) if tipo == "call" else -norm.ppf(delta) K = S * math.exp(0.5 * sig * sig * T - z * sig * math.sqrt(T)) sig = sig_atm * _interp(cal["skew"][tipo], delta) return K, sig def _fee(prem, S): return min(OPT_TAKER_UNDERLYING * S, OPT_FEE_CAP_FRAC * prem) if prem > 0 else 0.0 # ==================================================================== 3. IL MOTORE DEL COLLAR def serie_giornaliera() -> pd.DataFrame: """BTC giornaliero + DVOL, sulla finestra in cui ESISTONO entrambi (2021-03 in poi).""" d = load_data("BTC", "1h") idx = pd.DatetimeIndex(pd.to_datetime(d["datetime"], utc=True)) px = pd.Series(d["close"].values.astype(float), index=idx).resample("1D").last() dv = pd.read_parquet(ROOT / "data" / "raw" / "dvol_btc.parquet") iv = pd.Series(dv["close"].values.astype(float) / 100.0, index=pd.to_datetime(dv["timestamp"], unit="ms", utc=True)).resample("1D").last() P = pd.concat({"px": px, "iv": iv}, axis=1, join="inner").dropna() return P SPOT_FEE_SIDE = 0.00035 # Deribit taker reale dal 2026-08-01 (3,50 bps/lato), §8 def gate_trend(P: pd.DataFrame, forte: bool) -> np.ndarray: """L'ENTRATA, governata da indicatori — e presa dal progetto, non inventata qui. `trend_portfolio.tsmom_blend` media tre `np.sign()` sugli orizzonti (30, 90, 180) giorni, quindi assume SOLO i valori {-1, -1/3, +1/3, +1} (il bucket 2/3 non esiste: e' un errore gia' corretto in §57). Questo da' a "forte bull" una definizione che non aggiunge NEMMENO UN parametro nuovo: forte=True -> |blend| == 1 : tutti e tre gli orizzonti concordi ("forte bull/bear") forte=False -> |blend| >= 1/3: almeno la maggioranza (confronto dichiarato) Il segno del blend da' la DIREZIONE (long o short), come chiesto. """ from src.strategies.trend_portfolio import tsmom_blend c = P["px"].values.astype(float) b = tsmom_blend(c, (30, 90, 180)) d = np.sign(b) d[np.abs(b) < (0.99 if forte else 0.32)] = 0.0 return np.concatenate([[0.0], d[:-1]]) # ESEGUIBILE: si entra il giorno DOPO il segnale def collar(P: pd.DataFrame, cal: dict, *, dput: float, dcall: float, tenor: int, dirs: np.ndarray, zerocost: bool = False, senza_opzioni: bool = False, prem_scale: float = 1.0, solo_pavimento: bool = False, exit_frac: float = 1.0) -> dict: """Hold di BTC gated dal trend, coperto da un collar rollato ogni `tenor` giorni (<=15). direzione +1 = LONG BTC -> compra PUT (pavimento), vende CALL (tetto) direzione -1 = SHORT BTC -> compra CALL (pavimento), vende PUT (tetto) `exit_frac` < 1 esce dalle opzioni PRIMA della scadenza (0,50-0,75 = richiesta dell'operatore del 2026-09-01). Non e' un dettaglio di rollo: cambia il segno di due cose opposte. - sulla gamba COMPRATA (il pavimento) uscire presto RECUPERA valore temporale: il theta di una long accelera verso la scadenza, e §46 aveva gia' misurato che *"f si paga solo sulla parte di valore che converge a intrinseco, quindi un roll anticipato non lo paga"*; - sulla gamba VENDUTA (il tetto) uscire presto RINUNCIA al theta piu' veloce, cioe' proprio alla parte che in questo filone stampava il premio di varianza. Il prezzo di entrambe e' che si paga lo SPREAD in uscita su due gambe, invece di regolare a intrinseco. Chi lo attiva compra meno short-vol e piu' protezione onesta: e' misurabile, ed e' misurato nella sezione dedicata. `prem_scale` e `solo_pavimento` esistono per i CONTROLLI (M15): con premio 0 e solo il pavimento la struttura e' un pranzo gratis e DEVE vincere; con premio x10 DEVE perdere ovunque. Un apparato che non riconosce il pranzo gratis e' guasto, e §46 lo scopri' a proprie spese. `senza_opzioni=True` produce il MEDESIMO gated hold senza collar: e' la BASE del null M5, e dev'essere generata dallo stesso motore o il confronto misurerebbe anche le differenze di contabilita' (P1: il null deriva dal codice sorvegliato, non lo ridichiara). Marcatura GIORNALIERA a modello: il pavimento vincola il PAYOFF A SCADENZA, non il mark intermedio. Senza marcatura giornaliera il maxDD del collar sarebbe misurato solo ai roll e la protezione sembrerebbe migliore di quanto e'. """ S = P["px"].values.astype(float) A = P["iv"].values.astype(float) n = len(S) if n < tenor + 5: return {} tipo_di = {+1: ("put", "call"), -1: ("call", "put")} nav = np.full(n, np.nan) V = 1.0 # capitale normalizzato cash, q, nopt = 1.0, 0.0, 0.0 Kl = Ks = np.nan dc_vivo = 0.20 # delta del tetto VIVO: serve allo spread d'uscita anticipata scad = -1 roll_at = -1 # quando si ESCE (<= scad): con exit_frac<1 e' PRIMA della scadenza cur = 0 cicli = [] dc_scelti = [] # delta del tetto effettivamente usato a ogni apertura apre = 0 def _mark(K, T, sig_atm, tipo, S_t): if T <= 0 or math.isnan(K): return 0.0 if math.isnan(K) else (max(K - S_t, 0.0) if tipo == "put" else max(S_t - K, 0.0)) sig = sig_atm * _interp(cal["term"], max(T * DPY, 1.0)) d1 = _d1(S_t, K, T, sig) dl = norm.cdf(d1) if tipo == "call" else 1.0 - norm.cdf(d1) sig *= _interp(cal["skew"][tipo], min(max(dl, 0.02), 0.60)) return bs_put(S_t, K, T, sig) if tipo == "put" else bs_call(S_t, K, T, sig) def _chiudi(i, S_t, anche_spot=True): """Regola le opzioni vive; liquida lo spot SOLO se richiesto. ⚠️ Il flag esiste per un difetto che falserebbe tutto: al ROLL delle opzioni lo spot non ha motivo di muoversi. Chiuderlo e riaprirlo pagherebbe 2x3,5 bps ogni `tenor` giorni (a 7 giorni: ~3,6%/anno) di fee che nella realta' non si pagano. """ nonlocal cash, q, nopt, Kl, Ks, cur if nopt > 0 and not senza_opzioni: tl, ts = tipo_di[cur] T_rem = max(scad - i, 0) / DPY vl = _mark(Kl, T_rem, A[i], tl, S_t) vs = 0.0 if solo_pavimento else _mark(Ks, T_rem, A[i], ts, S_t) if T_rem > 0: # smontaggio anticipato: pago lo spread in USCITA vl *= 1.0 - _interp(cal["spread"][tl], dput) vs *= 1.0 + _interp(cal["spread"][ts], dc_vivo) else: for x in (vl, vs): if x > 0: cash -= nopt * OPT_DELIVERY_UNDERLYING * S_t cash += nopt * (vl - vs) nopt, Kl, Ks = 0.0, np.nan, np.nan if anche_spot and q != 0.0: cash += q * S_t - abs(q) * S_t * SPOT_FEE_SIDE q = 0.0 for i in range(n): S_t = S[i] d = int(dirs[i]) fine_ciclo = (i >= roll_at) and cur != 0 and not senza_opzioni solo_roll = fine_ciclo and d == cur if d != cur or fine_ciclo: if cur != 0: if nopt > 0 and i >= scad: tl, ts = tipo_di[cur] il_ = max(Kl - S_t, 0.0) if tl == "put" else max(S_t - Kl, 0.0) is_ = 0.0 if solo_pavimento else ( max(Ks - S_t, 0.0) if ts == "put" else max(S_t - Ks, 0.0)) cicli.append(dict(i0=apre, i1=i, S0=S[apre], S1=S_t, Kl=Kl, Ks=Ks, dirn=cur, cap=is_ > 0, floor=il_ > 0, scaduto=bool(fine_ciclo))) _chiudi(i, S_t, anche_spot=not solo_roll) if not solo_roll: V = cash if not solo_roll: cur = d if cur != 0: if not solo_roll: q = cur * V / S_t cash = V - q * S_t - abs(q) * S_t * SPOT_FEE_SIDE apre, scad = i, i + tenor roll_at = i + max(1, int(round(exit_frac * tenor))) if not senza_opzioni: tl, ts = tipo_di[cur] T = tenor / DPY Kl, sl = strike_da_delta(S_t, T, A[i], cal, dput, tl) pl = (bs_put if tl == "put" else bs_call)(S_t, Kl, T, sl) pl *= 1.0 + _interp(cal["spread"][tl], dput) # compro all'ASK pl *= prem_scale if zerocost: lo, hi = 0.02, 0.60 for _ in range(40): md = 0.5 * (lo + hi) Kx, sx = strike_da_delta(S_t, T, A[i], cal, md, ts) px_ = (bs_put if ts == "put" else bs_call)(S_t, Kx, T, sx) px_ *= 1.0 - _interp(cal["spread"][ts], md) # vendo al BID # il premio CRESCE col delta: se incasso meno di quanto pago, # il tetto va avvicinato (delta PIU' ALTO) -> lo = md. if px_ < pl: lo = md else: hi = md dc_use = 0.5 * (lo + hi) else: dc_use = dcall dc_vivo = dc_use dc_scelti.append(float(dc_use)) Ks, ss = strike_da_delta(S_t, T, A[i], cal, dc_use, ts) ps = (bs_put if ts == "put" else bs_call)(S_t, Ks, T, ss) ps *= (1.0 - _interp(cal["spread"][ts], dc_use)) * prem_scale if solo_pavimento: ps, Ks = 0.0, np.inf if ts == "call" else 0.0 nopt = abs(q) cash += nopt * (ps - pl) - nopt * (_fee(pl, S_t) + _fee(ps, S_t)) vl = vs = 0.0 if nopt > 0 and not senza_opzioni: tl, ts = tipo_di[cur] T_rem = max(scad - i, 0) / DPY vl = _mark(Kl, T_rem, A[i], tl, S_t) vs = 0.0 if solo_pavimento else _mark(Ks, T_rem, A[i], ts, S_t) nav[i] = cash + q * S_t + nopt * (vl - vs) v = pd.Series(nav, index=P.index).dropna() r = (v / v.shift(1) - 1.0).dropna() return dict(ret=r, nav=v, cicli=cicli, dc_scelti=dc_scelti) # ==================================================================== 4. MAIN def _riga(nome, r) -> dict: return dict(nome=nome, sh=_sh(r), dd=_dd(r), drift=_drift(r)) def main() -> None: print("=" * 96) print(" COLLAR01 — hold BTC gated dal trend, coperto da collar di opzioni a <=15 giorni") print("=" * 96) print(ATTESE) cal = calibra_catena() print(f" CALIBRAZIONE dalla catena VERA: {cal['n_quote']:,} quote a due lati su " f"{cal['giorni']} giorni, {cal['span'][0].date()} -> {cal['span'][1].date()}") for nm in ("put", "call"): xs, ys = cal["skew"][nm] print(f" skew {nm:5s} IV/ATM : " + " ".join(f"d{x:.2f}={y:.3f}" for x, y in zip(xs, ys))) xs, ys = cal["term"] print(f" termine IV/DVOL30 : " + " ".join(f"{x:.0f}g={y:.3f}" for x, y in zip(xs, ys))) for nm in ("put", "call"): xs, ys = cal["spread"][nm] print(f" spread {nm:5s} (mezza forchetta/mid): " + " ".join(f"d{x:.2f}={y:.3f}" for x, y in zip(xs, ys))) P = serie_giornaliera() print(f"\n LENTE LUNGA: {P.index[0].date()} -> {P.index[-1].date()} " f"({len(P)} giorni, {len(P)/DPY:.2f} anni)") bh = (P["px"] / P["px"].shift(1) - 1.0).dropna() print(f" BTC buy&hold nudo: Sharpe {_sh(bh):.3f} · maxDD {100*_dd(bh):.2f}% · " f"drift {100*_drift(bh):+.2f}%/a") righe = [] for forte in (True, False): dirs = gate_trend(P, forte) nome_g = "FORTE (3/3 orizzonti)" if forte else "LARGO (>=2/3)" quota = 100.0 * np.mean(dirs != 0) qlong = 100.0 * np.mean(dirs > 0) base = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True) rb = base["ret"] print(f"\n ── GATE {nome_g}: a mercato {quota:.1f}% dei giorni " f"(long {qlong:.1f}%, short {quota-qlong:.1f}%)") print(f" BASE gated SENZA opzioni: Sharpe {_sh(rb):.3f} · maxDD {100*_dd(rb):.2f}% · " f"drift {100*_drift(rb):+.2f}%/a ← e' il NULL da battere (M5)") for tenor in G_TENOR: for dp in G_DPUT: for dc in G_DCALL: out = collar(P, cal, dput=dp, dcall=dc, tenor=tenor, dirs=dirs) if not out: continue r = out["ret"] dd_c, dr_c = _dd(r), _drift(r) k = k_for_same_dd(rb.values, dd_c) dr_null = _drift(k * rb.values) righe.append(dict(gate=nome_g, forte=forte, tenor=tenor, dp=dp, dc=dc, sh=_sh(r), dd=dd_c, drift=dr_c, k=k, drift_null=dr_null, vince=(dr_c > dr_null) and (dd_c < _dd(rb)), dd_scende=dd_c < _dd(rb), cicli=out["cicli"], ret=r)) # varianti zero-cost, dichiarate FUORI griglia for tenor in G_TENOR: for dp in G_DPUT: out = collar(P, cal, dput=dp, dcall=np.nan, tenor=tenor, dirs=dirs, zerocost=True) if not out: continue r = out["ret"] dd_c, dr_c = _dd(r), _drift(r) k = k_for_same_dd(rb.values, dd_c) righe.append(dict(gate=nome_g, forte=forte, tenor=tenor, dp=dp, dc="zc", sh=_sh(r), dd=dd_c, drift=dr_c, k=k, drift_null=_drift(k * rb.values), vince=(dr_c > _drift(k * rb.values)) and (dd_c < _dd(rb)), dd_scende=dd_c < _dd(rb), cicli=out["cicli"], ret=r)) R = pd.DataFrame([x for x in righe if x["gate"] == nome_g]) print(f"\n {'ten':>4} {'dput':>5} {'dcall':>6} {'Sharpe':>7} {'maxDD':>7} " f"{'drift':>8} {'k null':>7} {'drift null':>10} esito") for _, x in R.iterrows(): dcs = x["dc"] if isinstance(x["dc"], str) else f"{x['dc']:.2f}" print(f" {x['tenor']:>4} {x['dp']:>5.2f} {dcs:>6} {x['sh']:>7.3f} " f"{100*x['dd']:>6.2f}% {100*x['drift']:>7.2f}% {x['k']:>7.3f} " f"{100*x['drift_null']:>9.2f}% " f"{'VINCE' if x['vince'] else ('dd↓ ma perde' if x['dd_scende'] else 'dd↑')}") R = pd.DataFrame(righe) n_dd = int(R["dd_scende"].sum()) n_win = int(R["vince"].sum()) print("\n" + "=" * 96) print(f" A1 maxDD SCENDE in {n_dd}/{len(R)} celle " f"({'CONFERMATA' if n_dd > len(R)//2 else 'REFUTATA'})") print(f" A3 il collar BATTE il de-levering in {n_win}/{len(R)} celle " f"({'REFUTATA (il collar vince)' if n_win > len(R)//2 else 'CONFERMATA (il collar perde)'})") # ---- A2: rapporto Δdrift/ΔmaxDD ---------------------------------------------------------- for forte in (True, False): sub = R[R["forte"] == forte] if not len(sub): continue dirs = gate_trend(P, forte) rb = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)["ret"] d0, g0 = _dd(rb), _drift(rb) rap = ((g0 - sub["drift"]) / np.maximum(d0 - sub["dd"], 1e-9)).median() print(f" A2 gate {'FORTE' if forte else 'LARGO'}: Δdrift/ΔmaxDD mediano {rap:.2f} " f"({'CONFERMATA' if rap > 1 else 'REFUTATA'}: >1 = il tetto costa piu' di quanto il " f"pavimento renda)") # ---- C9: il tetto scatta piu' sui vincenti o sui perdenti? ------------------------------- best = R.sort_values("sh", ascending=False).iloc[0] cic = pd.DataFrame(best["cicli"]) if len(cic): # ⚠️ "vincente" dipende dalla DIREZIONE: per un ciclo SHORT un prezzo che sale e' una # PERDITA. Usare S1>S0 come proxy di vittoria e' il difetto che dava "pavimento 0,0%". cic["vinc"] = cic["dirn"] * (cic["S1"] - cic["S0"]) > 0 cap_w = 100.0 * cic[cic["vinc"]]["cap"].mean() flo_l = 100.0 * cic[~cic["vinc"]]["floor"].mean() print(f"\n C9 cella migliore (gate {best['gate']}, {best['tenor']}g, dput {best['dp']}, " f"dcall {best['dc']}), {len(cic)} cicli:") print(f" il TETTO taglia nel {cap_w:.1f}% dei cicli VINCENTI · " f"il PAVIMENTO para nel {flo_l:.1f}% dei cicli PERDENTI") print(f" -> {'TRONCATURA' if cap_w > flo_l else 'PROTEZIONE'} " f"(C9: se scatta piu' sui vincenti che sui perdenti, non e' protezione)") # ---- M1: il confronto che decide se merita uno slot -------------------------------------- try: from src.portfolio.sleeves import _tp01_returns tp = _tp01_returns() tp.index = pd.to_datetime(tp.index, utc=True) tpd = (1.0 + tp).resample("1D").prod() - 1.0 j = pd.concat({"c": best["ret"], "tp": tpd}, axis=1, join="inner").dropna() print(f"\n M1 vs TP01 sulla finestra comune ({len(j)} giorni): " f"collar Sharpe {_sh(j['c']):.3f} · TP01 {_sh(j['tp']):.3f} · " f"corr {j['c'].corr(j['tp']):+.3f}") for w in (0.10, 0.25): mix = (1 - w) * j["tp"] + w * j["c"] print(f" TP01 + {100*w:.0f}% collar: Sharpe {_sh(mix):.3f} " f"({_sh(mix)-_sh(j['tp']):+.3f}) · maxDD {100*_dd(mix):.2f}% " f"({100*(_dd(mix)-_dd(j['tp'])):+.2f}pp)") except Exception as e: # pragma: no cover print(f"\n M1 non calcolato: {type(e).__name__}: {e}") # ---- ESTENSIONE AL BORDO (M4: si riapre la FAMIGLIA, M8: un argmax sul bordo non decide) -- print("\n" + "=" * 96) print(" ESTENSIONE DICHIARATA — le celle vincenti stanno tutte sul BORDO (dput min, dcall max).") print(" Sondo dove punta la pendenza. Il limite dput->0 / dcall->0,50 e' la COVERED CALL pura:") print(" nessun pavimento, tetto ATM. Trial aggiunti: 2 gate x 2 tenor x 3 x 3 = 36 (dichiarati).") ext = [] for forte in (True, False): dirs = gate_trend(P, forte) rb = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)["ret"] for tenor in (7, 14): for dp in (0.02, 0.05, 0.10): for dc in (0.30, 0.40, 0.50): r = collar(P, cal, dput=dp, dcall=dc, tenor=tenor, dirs=dirs)["ret"] k = k_for_same_dd(rb.values, _dd(r)) ext.append(dict(forte=forte, tenor=tenor, dp=dp, dc=dc, sh=_sh(r), dd=_dd(r), drift=_drift(r), vince=(_drift(r) > _drift(k * rb.values)) and (_dd(r) < _dd(rb)))) E = pd.DataFrame(ext) top = E.sort_values("sh", ascending=False).iloc[0] print(f" -> VINCE in {int(E['vince'].sum())}/{len(E)} celle, e il massimo e' NELL'ANGOLO: " f"gate {'FORTE' if top['forte'] else 'LARGO'} {top['tenor']}g dput {top['dp']:.2f} " f"dcall {top['dc']:.2f} -> Sharpe {top['sh']:.3f}, maxDD {100*top['dd']:.2f}%, " f"drift {100*top['drift']:+.2f}%/a") print(" ⚠️ Il pavimento non e' cio' che vince: vince il TETTO. La pendenza porta FUORI da cio'") print(" che l'operatore ha chiesto ('bloccare la perdita') e DENTRO lo short-vol.") # ---- LA DIAGNOSTICA CHE DECIDE: quanto dell'edge e' il PREMIO DI VARIANZA? ---------------- lr = np.log(P["px"] / P["px"].shift(1)) print("\n RIPREZZATURA ALLA VOLATILITA' REALIZZATA — diagnostica con LOOK-AHEAD DICHIARATO,") print(" non e' una strategia: e' il valore equo ex-post. Se l'edge sparisce, era IV-RV.") for h in (7, 14): fw = (lr[::-1].rolling(h).std()[::-1] * math.sqrt(DPY)).bfill().ffill() j = pd.concat({"iv": P["iv"], "fw": fw}, axis=1).dropna() print(f" DVOL / RV-forward a {h}g: mediana {(j['iv']/j['fw']).median():.3f} · " f"DVOL sopra nel {100*(j['iv']>j['fw']).mean():.1f}% dei giorni") for tenor in (7, 14): fw = (lr[::-1].rolling(tenor).std()[::-1] * math.sqrt(DPY)).bfill().ffill() Pf = P.copy() Pf["iv"] = fw.reindex(P.index).values for forte in (True, False): dirs = gate_trend(P, forte) rb = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)["ret"] for dp, dc in ((0.02, 0.50), (0.10, 0.30)): a = collar(P, cal, dput=dp, dcall=dc, tenor=tenor, dirs=dirs)["ret"] b = collar(Pf, cal, dput=dp, dcall=dc, tenor=tenor, dirs=dirs)["ret"] ka, kb = k_for_same_dd(rb.values, _dd(a)), k_for_same_dd(rb.values, _dd(b)) wa = _drift(a) > _drift(ka * rb.values) and _dd(a) < _dd(rb) wb = _drift(b) > _drift(kb * rb.values) and _dd(b) < _dd(rb) print(f" {'FORTE' if forte else 'LARGO'} {tenor:>2}g dput {dp:.2f} dcall {dc:.2f}: " f"a DVOL Sh {_sh(a):6.3f} drift {100*_drift(a):+7.2f}% {'VINCE' if wa else 'perde'}" f" | a RV Sh {_sh(b):6.3f} drift {100*_drift(b):+7.2f}% {'VINCE' if wb else 'perde'}" f" | VRP {100*(_drift(a)-_drift(b)):+6.2f} pp") # ---- USCITA ANTICIPATA (richiesta dell'operatore: fra il 50% e il 75% del tempo) --------- print("\n" + "=" * 96) print(" USCITA ANTICIPATA DALLE OPZIONI — exit_frac 0,50 / 0,625 / 0,75 contro la scadenza.") print(" §46 misuro' che 'f si paga solo sulla parte di valore che converge a intrinseco,") print(" quindi un roll anticipato NON lo paga'. ⚠️ Vale per una copertura SOLO LONG: in un") print(" collar c'e' una gamba VENDUTA da RICOMPRARE, e uscire prima paga f esattamente sulla") print(" parte che a scadenza si sarebbe regolata gratis. L'asimmetria si INVERTE.") lr2 = np.log(P["px"] / P["px"].shift(1)) for forte in (True, False): dirs = gate_trend(P, forte) rb = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)["ret"] print(f"\n gate {'FORTE' if forte else 'LARGO'} (base drift {100*_drift(rb):+.2f}%/a):") for dp, dc in ((0.10, 0.30), (0.20, 0.20), (0.02, 0.50)): for ef in (1.00, 0.75, 0.625, 0.50): a = collar(P, cal, dput=dp, dcall=dc, tenor=14, dirs=dirs, exit_frac=ef)["ret"] hh = max(1, int(round(ef * 14))) fw = (lr2[::-1].rolling(hh).std()[::-1] * math.sqrt(DPY)).bfill().ffill() Pf = P.copy() Pf["iv"] = fw.reindex(P.index).values b = collar(Pf, cal, dput=dp, dcall=dc, tenor=14, dirs=dirs, exit_frac=ef)["ret"] k = k_for_same_dd(rb.values, _dd(a)) w = (_drift(a) > _drift(k * rb.values)) and (_dd(a) < _dd(rb)) print(f" d{dp:.2f}/{dc:.2f} exit {ef:.3f}: Sh {_sh(a):6.3f} DD {100*_dd(a):6.2f}% " f"drift {100*_drift(a):+7.2f}% null {100*_drift(k*rb.values):+6.2f}% " f"{'VINCE' if w else 'perde'} | a RV drift {100*_drift(b):+7.2f}% " f"VRP {100*(_drift(a)-_drift(b)):+6.2f}pp") # ---- CONTROLLI DELL'APPARATO (M15) — un rilevatore che non li passa non misura nulla ---- print("\n CONTROLLI DELL'APPARATO:") dirs = gate_trend(P, True) rb = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)["ret"] esiti = [] g = collar(P, cal, dput=0.20, dcall=0.20, tenor=14, dirs=dirs, prem_scale=0.0, solo_pavimento=True)["ret"] ok_g = (_dd(g) < _dd(rb)) and (_drift(g) > _drift(rb)) esiti.append(ok_g) print(f" pranzo gratis (pavimento a premio ZERO): maxDD {100*_dd(rb):.2f}%->{100*_dd(g):.2f}% " f"drift {100*_drift(rb):+.2f}%->{100*_drift(g):+.2f}% {'OK' if ok_g else 'FALLITO'}") c = collar(P, cal, dput=0.20, dcall=0.20, tenor=14, dirs=dirs, prem_scale=10.0)["ret"] ok_c = _drift(c) < _drift(rb) esiti.append(ok_c) print(f" premio x10 (deve perdere): drift {100*_drift(c):+.2f}%/a " f"{'OK' if ok_c else 'FALLITO'}") zc = collar(P, cal, dput=0.20, dcall=np.nan, tenor=14, dirs=dirs, zerocost=True) prem_res = np.nan ok_z = bool(len(zc.get("cicli", [])) > 0) and np.isfinite(_drift(zc["ret"])) esiti.append(ok_z) print(f" zero-cost costruibile e finito: drift {100*_drift(zc['ret']):+.2f}%/a " f"{'OK' if ok_z else 'FALLITO'}") if not all(esiti): print(" 🚨 APPARATO GUASTO: il verdetto sotto NON e' leggibile (§46: un controllo " "positivo rotto dichiara guasto l'apparato)") # ---- VERDETTO A RUNTIME (N11) ------------------------------------------------------------ print("\n" + "=" * 96) # il verdetto deve tenere conto della diagnostica VRP: il bordo vince, ma di che cosa? dirs_f = gate_trend(P, True) rb_f = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs_f, senza_opzioni=True)["ret"] fw7 = (np.log(P["px"] / P["px"].shift(1))[::-1].rolling(7).std()[::-1] * math.sqrt(DPY)).bfill().ffill() Pf7 = P.copy() Pf7["iv"] = fw7.reindex(P.index).values ang_dvol = collar(P, cal, dput=0.02, dcall=0.50, tenor=7, dirs=dirs_f)["ret"] ang_rv = collar(Pf7, cal, dput=0.02, dcall=0.50, tenor=7, dirs=dirs_f)["ret"] quota_vrp = (_drift(ang_dvol) - _drift(ang_rv)) / max(_drift(ang_dvol), 1e-9) if n_dd == 0: v = "IL PAVIMENTO NON RIDUCE IL maxDD — cade come §46 e per lo stesso motivo" elif n_win <= len(R) // 8: v = (f"IL PAVIMENTO FUNZIONA (a differenza di §46: maxDD scende in {n_dd}/{len(R)}) MA IL " f"TETTO LO PAGA TROPPO: solo {n_win}/{len(R)} celle battono il de-levering. " f"E cio' che vince sul bordo NON e' il pavimento: e' il premio di varianza " f"({100*quota_vrp:.0f}% del drift dell'angolo sparisce riprezzando alla vol vera) " f"— cioe' il meccanismo di VRP01, non una scoperta, e §3 lo blocca in deploy") else: v = f"REGGE IL PRIMO GATE: {n_win}/{len(R)} celle battono il de-levering — serve M2/DSR" print(f" VERDETTO (calcolato a runtime): {v}") print("=" * 96) if __name__ == "__main__": main()