research(mr02eth): ricerca sostituto + integrazione dati opzioni cerbero-bite
Ricerca sostituto/miglioria a MR02/ETH (127 strategie + 18 overlay opzioni, verifica avversariale, gate PORT06). Esito: miglioria = no-SL (gia' ~catturata da EXIT-16 live); tail-hedge opzioni NO-GO empirico su prezzi reali. Infrastruttura opzioni REALE (muro ARGO/GEX caduto): - options_fetcher.py: importa lo storico per-strike + market_snapshots da cerbero-bite -> data/options/ (chain bid/ask/IV/greche/OI + pannello regime con net-GEX dealer/gamma-flip/VRP/funding/liquidation). - options_chain.py: loader + skew_curve/premium_levels (aggregati reali) + quote() causale + load_market() (pannello regime). - option_overlay_lab.py: overlay opzioni BS su DVOL (pricing sintetico). - mr02eth_port06_gate.py / eth_collar_gate.py: gate swap-sleeve e collar. - mr02eth_search/options.workflow.js: i 2 workflow. Numeri reali: skew put 10%OTM ~1.1 (liquido), premio ~1.0%/mese; niente strike 10%OTM a 24h (overlay per-trade infattibile); collar standing paga nei crash ma net-negativo a PORT06 (alza OOS DD). Diario + CLAUDE.md aggiornati. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -0,0 +1,250 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user