"""Harness ONESTO per overlay di OPZIONI (protezione) sulle strategie ETH. Vincolo noto (analisi ARGO/GEX 2026-06-01): lo storico per-strike dell'OI/prezzi opzioni NON e' gratuito -> non backtestabile. MA la DVOL di Deribit (indice di vol implicita ETH, annualizzato %) e' storica e gratuita in data/regime/eth_dvol.parquet (oraria 2021-03 -> 2026-06). Quindi prezziamo le opzioni SINTETICAMENTE con Black-Scholes(spot, DVOL) -> backtest del COSTO dell'overlay (premio) vs il payoff. Approssimazione dichiarata (e DELIBERATAMENTE conservativa, bias CONTRO le opzioni): - niente skew: i put reali costano piu' dell'ATM-IV -> applichiamo `skew_mult>=1` alla sigma dei put (rincara il premio). - all'uscita anticipata l'opzione vale solo l'INTRINSECO (nessun rebate di time- value) -> sotto-stima il beneficio dell'hedge. - niente costi di fill/liquidita' delle opzioni oltre lo skew_mult. Se sotto queste ipotesi pessimistiche l'overlay MIGLIORA il rischio/rendimento, l'edge e' robusto. I numeri sono INDICATIVI, non eseguibili come i perp. Convenzione coerente con explore_lab: ret per-notional, lev 3x, pos 0.15, fee 0.10% RT. L'opzione copre lo STESSO notional della posizione (lev*pos*cap), quindi il suo P&L come frazione entra in ret' = ret_base + lev*(payoff - premio)/entry. """ from __future__ import annotations import math import sys from pathlib import Path import numpy as np import pandas as pd PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) from scripts.analysis.explore_lab import get_df, atr, ema, rsi, FEE_RT, LEV, POS, OOS_FRAC # noqa HOURS_YEAR = 24 * 365.0 # --------------------------- Black-Scholes (r=0) --------------------------- def _ncdf(x: float) -> float: return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0))) def bs_put(S: float, K: float, T: float, sigma: float) -> float: """Prezzo put europea, r=0. T in anni, sigma annualizzata (frazione).""" if T <= 0 or sigma <= 0 or S <= 0 or K <= 0: return max(K - S, 0.0) sd = sigma * math.sqrt(T) d1 = (math.log(S / K) + 0.5 * sigma * sigma * T) / sd d2 = d1 - sd return K * _ncdf(-d2) - S * _ncdf(-d1) def bs_call(S: float, K: float, T: float, sigma: float) -> float: if T <= 0 or sigma <= 0 or S <= 0 or K <= 0: return max(S - K, 0.0) sd = sigma * math.sqrt(T) d1 = (math.log(S / K) + 0.5 * sigma * sigma * T) / sd d2 = d1 - sd return S * _ncdf(d1) - K * _ncdf(d2) # --------------------------- DVOL allineata (causale) --------------------------- def dvol_for(df: pd.DataFrame, asset: str = "ETH") -> np.ndarray: """DVOL (annualizzata, FRAZIONE es. 0.70) allineata causalmente ai bar di df. merge_asof backward: ogni bar riceve l'ultima DVOL con ts <= bar. NaN -> ffill, eventuale buco iniziale (pre-2021-03) -> bfill della prima disponibile.""" dv = pd.read_parquet(PROJECT_ROOT / "data" / "regime" / f"{asset.lower()}_dvol.parquet") dv = dv[["timestamp", "dvol"]].sort_values("timestamp") base = pd.DataFrame({"timestamp": df["timestamp"].values}).sort_values("timestamp") m = pd.merge_asof(base, dv, on="timestamp", direction="backward") s = (m["dvol"].values.astype(float)) / 100.0 # da % a frazione s = pd.Series(s).ffill().bfill().values return s # --------------------------- engine con overlay opzione --------------------------- def simulate_hedged(entries: list[dict], df: pd.DataFrame, dvol: np.ndarray | None = None, fee_rt: float = FEE_RT, lev: float = LEV, pos: float = POS, split: int = -1, hedge: str = "none", otm: float = 0.05, otm2: float = 0.15, skew_mult: float = 1.10, tenor_mult: float = 1.0, hedge_side: str = "both", min_dvol: float = 0.0) -> dict: """Come explore_lab.simulate ma con un overlay di opzione per-trade. hedge: "none" -> nessun overlay (identico a explore_lab.simulate) "put" -> compra protezione direzionale: put se long, call se short (floor a otm) "put_spread"-> debit spread: long protezione a otm, short a otm2 (piu' lontano) -> piu' economico, cap del payoff "collar" -> protezione a otm finanziata vendendo l'opzione opposta OTM a otm2 (riduce/azzera il premio, cappa il guadagno) otm: moneyness della protezione (frazione, es 0.05 = 5% OTM) otm2: secondo strike (per put_spread / collar) skew_mult: rincaro della sigma sull'opzione COMPRATA (conservativo) tenor_mult: scadenza = max_bars*tenor_mult ore (>=1: copre almeno l'orizzonte) hedge_side: "both" | "long" (solo i long-fade, le prese-coltello) | "short" min_dvol: copri solo se DVOL all'ingresso >= soglia (frazione) -> hedge selettivo """ h, l, c = df["high"].values, df["low"].values, df["close"].values n = len(c) ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) if dvol is None: dvol = np.full(n, 0.7) cap = peak = 1000.0 max_dd = 0.0 fee = fee_rt * lev trades = wins = 0 last_exit = -1 bars_in = 0 yearly: dict[int, float] = {} rets: list[float] = [] prem_paid = 0.0 pay_recv = 0.0 for e in entries: i, d = e["i"], e["d"] if i <= last_exit or i + 1 >= n or i < split: continue entry = c[i] tp, sl, mb = e.get("tp"), e.get("sl"), e["max_bars"] exit_p = c[min(i + mb, n - 1)] j = min(i + mb, n - 1) for k in range(1, mb + 1): j = i + k if j >= n: exit_p = c[n - 1]; break hit_sl = sl is not None and ((d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)) hit_tp = tp is not None and ((d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)) if hit_sl: exit_p = sl; break if hit_tp: exit_p = tp; break if k == mb: exit_p = c[j] base_ret = (exit_p - entry) / entry * d * lev - fee # ---- overlay opzione ---- opt_ret = 0.0 do_hedge = hedge != "none" and ( hedge_side == "both" or (hedge_side == "long" and d == 1) or (hedge_side == "short" and d == -1) ) and dvol[i] >= min_dvol if do_hedge: T = max(mb * tenor_mult, 1.0) / HOURS_YEAR sig = dvol[i] sig_buy = sig * skew_mult if d == 1: # posizione long -> rischio sotto -> PUT (floor) Kp = entry * (1.0 - otm) prem = bs_put(entry, Kp, T, sig_buy) / entry payoff = max(Kp - exit_p, 0.0) / entry if hedge == "put_spread": Kp2 = entry * (1.0 - otm2) prem -= bs_put(entry, Kp2, T, sig) / entry # vendo la coda piu' lontana payoff -= max(Kp2 - exit_p, 0.0) / entry elif hedge == "collar": Kc = entry * (1.0 + otm2) # vendo call OTM per finanziare prem -= bs_call(entry, Kc, T, sig) / entry payoff -= max(exit_p - Kc, 0.0) / entry else: # posizione short -> rischio sopra -> CALL (cap) Kc = entry * (1.0 + otm) prem = bs_call(entry, Kc, T, sig_buy) / entry payoff = max(exit_p - Kc, 0.0) / entry if hedge == "put_spread": Kc2 = entry * (1.0 + otm2) prem -= bs_call(entry, Kc2, T, sig) / entry payoff -= max(exit_p - Kc2, 0.0) / entry elif hedge == "collar": Kp = entry * (1.0 - otm2) prem -= bs_put(entry, Kp, T, sig) / entry payoff -= max(Kp - exit_p, 0.0) / entry opt_ret = lev * (payoff - prem) prem_paid += lev * prem pay_recv += lev * payoff ret = base_ret + opt_ret cb = cap cap = max(cb + cb * pos * ret, 10.0) peak = max(peak, cap); max_dd = max(max_dd, (peak - cap) / peak) trades += 1; wins += ret > 0; bars_in += (j - i) last_exit = j rets.append(ret * pos) yearly[ts.iloc[i].year] = yearly.get(ts.iloc[i].year, 0.0) + ret * 100 sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(len(rets))) if len(rets) > 1 and np.std(rets) > 0 else 0.0 return { "trades": trades, "win": wins / trades * 100 if trades else 0.0, "ret": (cap / 1000 - 1) * 100, "dd": max_dd * 100, "sharpe": sharpe, "yearly": yearly, "exposure": bars_in / n * 100 if n else 0.0, "prem_paid_pct": prem_paid * 100, "pay_recv_pct": pay_recv * 100, } def evaluate_hedged(name: str, entries: list[dict], df: pd.DataFrame, dvol: np.ndarray, fees=(0.0, 0.001, 0.002), **hcfg) -> dict: """FULL + OOS + sweep fee per una config di overlay. Stampa una riga.""" split = int(len(df) * (1 - OOS_FRAC)) full = simulate_hedged(entries, df, dvol, **hcfg) oos = simulate_hedged(entries, df, dvol, split=split, **hcfg) sweep = {f: simulate_hedged(entries, df, dvol, fee_rt=f, **hcfg)["ret"] for f in fees} sweep_oos = {f: simulate_hedged(entries, df, dvol, fee_rt=f, split=split, **hcfg)["ret"] for f in fees} yrs = full["yearly"]; pos_yrs = sum(1 for v in yrs.values() if v > 0) print(f" {name:<26s} trd={full['trades']:>5d} win={full['win']:>4.1f}% " f"FULL={full['ret']:>+8.0f}% OOS={oos['ret']:>+7.0f}% DD={full['dd']:>4.0f}% " f"oDD={oos['dd']:>4.0f}% Shrp={full['sharpe']:>4.2f} exp={full['exposure']:>4.1f}% " f"prem={full['prem_paid_pct']:>5.0f}% pay={full['pay_recv_pct']:>5.0f}% " f"anniPos={pos_yrs}/{len(yrs)} | fee0.2%: FULL={sweep[0.002]:>+7.0f} OOS={sweep_oos[0.002]:>+6.0f}") return {"full": full, "oos": oos, "sweep": sweep, "sweep_oos": sweep_oos, "pos_yrs": pos_yrs, "n_yrs": len(yrs)} # --------------------------- baseline: Donchian fade ETH --------------------------- def donchian_fade(df, n=20, sl_atr=2.0, max_bars=24, trend_max=3.0, ema_long=200, use_sl=True): """Le entries della MR02/ETH (per testarci sopra gli overlay).""" h, l, c = df["high"].values, df["low"].values, df["close"].values a = atr(df, 14) hh = pd.Series(h).rolling(n).max().shift(1).values ll = pd.Series(l).rolling(n).min().shift(1).values em = ema(c, ema_long) ents = [] for i in range(max(n, ema_long, 14) + 1, len(c)): if np.isnan(hh[i]) or np.isnan(ll[i]) or np.isnan(a[i]): continue if trend_max is not None and not np.isnan(em[i]) and a[i] > 0: if abs(c[i] - em[i]) / a[i] > trend_max: continue tp = (hh[i] + ll[i]) / 2.0 if c[i] < ll[i] and c[i - 1] >= ll[i - 1]: ents.append({"i": i, "d": 1, "tp": tp, "sl": (c[i] - sl_atr * a[i]) if use_sl else None, "max_bars": max_bars}) elif c[i] > hh[i] and c[i - 1] <= hh[i - 1]: ents.append({"i": i, "d": -1, "tp": tp, "sl": (c[i] + sl_atr * a[i]) if use_sl else None, "max_bars": max_bars}) return ents if __name__ == "__main__": df = get_df("ETH", "1h") dv = dvol_for(df, "ETH") print(f"ETH 1h {len(df)} bar | DVOL media {np.nanmean(dv)*100:.0f}% (min {np.nanmin(dv)*100:.0f} max {np.nanmax(dv)*100:.0f})") ents = donchian_fade(df) # MR02/ETH trend-filtrato, con SL ATR ents_nosl = donchian_fade(df, use_sl=False) # senza SL (per testare l'opzione COME stop) print("\n=== riferimento: MR02/ETH baseline vs overlay opzione (premio dedotto, conservativo) ===") evaluate_hedged("baseline SL-ATR (no opt)", ents, df, dv, hedge="none") evaluate_hedged("no-SL (no opt)", ents_nosl, df, dv, hedge="none") print(" -- opzione COME floor al posto dello stop (no-SL + put/call OTM) --") for otm in (0.03, 0.05, 0.08): evaluate_hedged(f"no-SL +put {int(otm*100)}%OTM", ents_nosl, df, dv, hedge="put", otm=otm, hedge_side="both") print(" -- put-spread / collar (piu' economici) sul no-SL --") evaluate_hedged("no-SL +put_spread 5/15", ents_nosl, df, dv, hedge="put_spread", otm=0.05, otm2=0.15) evaluate_hedged("no-SL +collar 5/10", ents_nosl, df, dv, hedge="collar", otm=0.05, otm2=0.10) print(" -- hedge SELETTIVO: solo long-fade (prese-coltello) in alta DVOL --") evaluate_hedged("SL +put long-only DVOL>0.8", ents, df, dv, hedge="put", otm=0.05, hedge_side="long", min_dvol=0.8)