Adriano Dal Pastro
14522262e6
Reset del progetto su fondamenta verificate dopo la scoperta che l'intera libreria "validata OOS" era artefatto di feed contaminato (print fantasma del feed Cerbero TESTNET + storico Binance/USDT). - Storico ricostruito da Deribit MAINNET (ccxt pubblico, tokenless) e CERTIFICATO (certify_feed.py): BTC/ETH puliti su TUTTA la storia (mediana 2-6 bps vs Coinbase USD), integrita' OHLC + coerenza resample (maxΔ 0.00) + cross-venue OK. Alt esclusi (illiquidi/divergenti: LTC/DOGE 50-82% barre flat; XRP/BNB non certificabili). - Verdetto sul feed pulito: FADE / PAIRS / XS01 / TSM01 morti (ogni portafoglio Sharpe -2.3..-3.0, DD ~40%); solo SH01 e frammenti HONEST con segnale residuo, da ri-validare in isolamento. - Cleanup "restart pulito": strategie, stack live (src/live, src/portfolio, runner/executor, yml, docker), ~100 script ricerca/gate, waste/games/ portfolios, dati non certificati + cache e 60+ diari -> archiviati in Old/ (preservati, non cancellati). Diario consolidato in un unico documento. - Skeleton ricerca tenuto: Strategy ABC + indicatori + src/fractal + src/backtest/engine + load_data; tool dati certificati (rebuild_history, certify_feed, audit_feed, multi_source_check). - Universo dati ATTIVO: solo BTC/ETH (5m/15m/1h); guardrail fisico (load_data su alt -> FileNotFoundError). Esecuzione DISABILITATA, conto flat. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
251 lines
12 KiB
Python
251 lines
12 KiB
Python
"""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)
|