Adriano Dal Pastro
5ac4e16af8
Ondata di ricerca onesta a largo spettro su BTC/ETH+DVOL certificati: 104 ipotesi distinte (11 famiglie), un agente-finder per ipotesi, verifica avversariale a 3 scettici sui promettenti, sintesi (153 agenti totali). Esito: NIENTE di nuovo regge -> conferma del soffitto strutturale ~1.3 BTC/ETH-direzionale; lo stack TP01+XS01+VRP01 resta imbattuto. - altlib.py: harness condiviso vettoriale leak-free (eval_weights/study_weights, fee-sweep, both-asset + hold-out 2025+). Riproduce i numeri canonici di TP01. - MARGINAL SCORER (study_marginal/marginal_vs_tp01): Sharpe INCREMENTALE vs baseline TP01 (corr, blend uplift OOS, alpha residua) + jackknife OOS (clean-year + drop-best-month). earns_slot = abs!=FAIL & ADDS & robust_oos. Smaschera gli overlay su TSMOM con PASS assoluti fasulli (CMB04, VOL11, ...) e il falso positivo KAMA (ADDS ma muore al jackknife). - runs/*.py (104) script riproducibili per ipotesi; wf_altstrat.js workflow. - Verdetto: 0 candidati deployabili; 2 LEAD fragili (VOL08, STA05_LS) da forward-monitor. - test_marginal_scorer.py blocca baseline + invarianti. Suite: 32 verde. Diario: docs/diary/2026-06-20-alt-strategies-100agent-sweep.md Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
378 lines
15 KiB
Python
378 lines
15 KiB
Python
"""OPT04 — Iron Condor Weekly (DVOL-gated).
|
|
|
|
IDEA: Sell OTM call+put spreads weekly, collect premium from both sides. Iron condor =
|
|
- Sell OTM put (delta ~-0.20), Buy further OTM put (delta ~-0.08) <- put credit spread
|
|
- Sell OTM call (delta ~+0.20), Buy further OTM call (delta ~+0.08) <- call credit spread
|
|
|
|
Premium collected from BOTH sides. Profit if underlying stays within the wings (range-bound week).
|
|
Max loss = wing width - net premium (total of both spreads).
|
|
|
|
MODELED APPROACH:
|
|
- DVOL used as ATM vol proxy (symmetric BS, no skew).
|
|
- Gate: IV-rank > 0.30 (sell only when IV is rich relative to own history).
|
|
- Optional crash-skip: IV-rank > 0.90 -> vol already exploded, skip.
|
|
- Capital = put wing width + call wing width (total defined risk).
|
|
- Fee: 12.5% of net premium (Deribit options cap, per side; 4 legs total = 2 round-trips).
|
|
|
|
GRID (4 configs on 1d TF):
|
|
A) delta ±0.20/±0.08, ivr_gate=0.30, no crash-skip
|
|
B) delta ±0.25/±0.10, ivr_gate=0.30, no crash-skip
|
|
C) delta ±0.20/±0.08, ivr_gate=0.30, crash-skip at 0.90
|
|
D) delta ±0.25/±0.10, ivr_gate=0.30, crash-skip at 0.90
|
|
|
|
CAVEAT: premiums MODELED on DVOL ATM (no skew, no real chain). Lead quantification only.
|
|
DVOL history starts 2021-03 -> effective backtest from ~2021-Q3.
|
|
"""
|
|
import sys
|
|
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
|
|
import altlib as al
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
from scipy.stats import norm
|
|
|
|
# ─── Black-Scholes helpers ────────────────────────────────────────────────────
|
|
|
|
def bs_put(S: float, K: float, T: float, sig: float) -> float:
|
|
"""Black-Scholes put price, r=0."""
|
|
if T <= 0 or sig <= 0:
|
|
return max(K - S, 0.0)
|
|
d1 = (np.log(S / K) + 0.5 * sig**2 * T) / (sig * np.sqrt(T))
|
|
d2 = d1 - sig * np.sqrt(T)
|
|
return K * norm.cdf(-d2) - S * norm.cdf(-d1)
|
|
|
|
|
|
def bs_call(S: float, K: float, T: float, sig: float) -> float:
|
|
"""Black-Scholes call price, r=0."""
|
|
if T <= 0 or sig <= 0:
|
|
return max(S - K, 0.0)
|
|
d1 = (np.log(S / K) + 0.5 * sig**2 * T) / (sig * np.sqrt(T))
|
|
d2 = d1 - sig * np.sqrt(T)
|
|
return S * norm.cdf(d1) - K * norm.cdf(d2)
|
|
|
|
|
|
def strike_put_from_delta(S: float, T: float, sig: float, delta: float) -> float:
|
|
"""Strike for a put with given delta (delta < 0, e.g. -0.20).
|
|
put_delta = -N(-d1) = delta -> N(-d1) = -delta -> -d1 = N^{-1}(-delta)
|
|
d1 = -N^{-1}(-delta)
|
|
K = S * exp(0.5*sig^2*T - d1*sig*sqrt(T))."""
|
|
d1 = -norm.ppf(-delta)
|
|
return S * np.exp(0.5 * sig**2 * T - d1 * sig * np.sqrt(T))
|
|
|
|
|
|
def strike_call_from_delta(S: float, T: float, sig: float, delta: float) -> float:
|
|
"""Strike for a call with given delta (delta > 0, e.g. +0.20).
|
|
call_delta = N(d1) = delta -> d1 = N^{-1}(delta)
|
|
K = S * exp(-d1*sig*sqrt(T) + 0.5*sig^2*T)."""
|
|
d1 = norm.ppf(delta)
|
|
return S * np.exp(-d1 * sig * np.sqrt(T) + 0.5 * sig**2 * T)
|
|
|
|
|
|
# ─── IV-rank (causal, expanding window) ──────────────────────────────────────
|
|
|
|
def iv_rank_series(dv_pts: np.ndarray, min_history: int = 60) -> np.ndarray:
|
|
"""Causal expanding-window IV rank: fraction of past DVOL values below current.
|
|
NaN until min_history valid bars are available."""
|
|
n = len(dv_pts)
|
|
ivr = np.full(n, np.nan)
|
|
valid = np.where(np.isfinite(dv_pts))[0]
|
|
if len(valid) < min_history:
|
|
return ivr
|
|
start = valid[0]
|
|
for i in valid:
|
|
hist_len = i - start
|
|
if hist_len >= min_history:
|
|
hist = dv_pts[start:i]
|
|
hist = hist[np.isfinite(hist)]
|
|
if len(hist) >= min_history:
|
|
ivr[i] = float((hist < dv_pts[i]).mean())
|
|
return ivr
|
|
|
|
|
|
# ─── Standalone iron condor backtest ─────────────────────────────────────────
|
|
|
|
def backtest_ic(
|
|
df: pd.DataFrame,
|
|
asset: str,
|
|
short_delta_put: float = -0.20,
|
|
long_delta_put: float = -0.08,
|
|
short_delta_call: float = 0.20,
|
|
long_delta_call: float = 0.08,
|
|
ivr_gate: float = 0.30,
|
|
crash_skip: float = 1.01, # >1 disables crash-skip
|
|
tenor_d: int = 7,
|
|
fee_side: float = al.FEE_SIDE,
|
|
) -> dict:
|
|
"""Honest backtest of weekly iron condor on daily bars.
|
|
|
|
P&L mechanics:
|
|
- Every tenor_d bars (weekly) decide at bar i (close[i] known), settle at i+tenor_d.
|
|
- Net premium = put_net + call_net (both modeled with BS on DVOL, no skew).
|
|
- Payoff realized on close[i+tenor_d].
|
|
- Capital basis = put_wing + call_wing (total defined risk).
|
|
- Return_week = (net_premium - payoffs - fee) / capital.
|
|
- Booked at settlement bar; 0 elsewhere.
|
|
|
|
Returns al.eval_weights-compatible dict.
|
|
"""
|
|
close = df["close"].values.astype(float)
|
|
dts = pd.to_datetime(df["datetime"], utc=True)
|
|
n = len(close)
|
|
T_yr = tenor_d / 365.25
|
|
|
|
dv_pts = al.dvol(df, asset)
|
|
dv = dv_pts / 100.0
|
|
ivr = iv_rank_series(dv_pts, min_history=60)
|
|
|
|
daily_pnl = np.zeros(n)
|
|
in_trade = np.zeros(n, dtype=bool)
|
|
|
|
# Start from first bar where we have at least 60 bars of DVOL history
|
|
valid_dvol = np.where(np.isfinite(dv_pts))[0]
|
|
if len(valid_dvol) < 60:
|
|
return _empty_result(df, dts)
|
|
|
|
i_start = valid_dvol[60] # first bar with 60 history points
|
|
i = i_start
|
|
|
|
trades = 0
|
|
while i + tenor_d < n:
|
|
S0 = close[i]
|
|
sig = dv[i]
|
|
|
|
# DVOL must be available
|
|
if not np.isfinite(sig) or sig <= 0.0:
|
|
i += tenor_d
|
|
continue
|
|
|
|
# IV-rank must be available
|
|
if not np.isfinite(ivr[i]):
|
|
i += tenor_d
|
|
continue
|
|
|
|
# Gate: sell only when IV rank above threshold
|
|
if ivr_gate > 0.0 and ivr[i] < ivr_gate:
|
|
i += tenor_d
|
|
continue
|
|
|
|
# Crash-skip: do not sell when vol already exploded
|
|
if crash_skip < 1.0 and ivr[i] > crash_skip:
|
|
i += tenor_d
|
|
continue
|
|
|
|
# ── PUT credit spread ──────────────────────────────────────────────
|
|
Ks_put = strike_put_from_delta(S0, T_yr, sig, short_delta_put) # short (less OTM)
|
|
Kl_put = strike_put_from_delta(S0, T_yr, sig, long_delta_put) # long (more OTM)
|
|
prem_s_put = bs_put(S0, Ks_put, T_yr, sig)
|
|
prem_l_put = bs_put(S0, Kl_put, T_yr, sig)
|
|
net_put = prem_s_put - prem_l_put
|
|
wing_put = Ks_put - Kl_put # put short strike > long strike -> positive
|
|
|
|
# ── CALL credit spread ─────────────────────────────────────────────
|
|
Ks_call = strike_call_from_delta(S0, T_yr, sig, short_delta_call) # short (less OTM)
|
|
Kl_call = strike_call_from_delta(S0, T_yr, sig, long_delta_call) # long (more OTM)
|
|
prem_s_call = bs_call(S0, Ks_call, T_yr, sig)
|
|
prem_l_call = bs_call(S0, Kl_call, T_yr, sig)
|
|
net_call = prem_s_call - prem_l_call
|
|
wing_call = Kl_call - Ks_call # call long strike > short strike -> positive
|
|
|
|
# Sanity: net premiums must be positive (should always be true by construction)
|
|
if net_put <= 0.0 or net_call <= 0.0 or wing_put <= 0.0 or wing_call <= 0.0:
|
|
i += tenor_d
|
|
continue
|
|
|
|
S1 = close[i + tenor_d]
|
|
|
|
# ── PUT spread payoff ──────────────────────────────────────────────
|
|
payoff_put = max(0.0, Ks_put - S1) - max(0.0, Kl_put - S1)
|
|
|
|
# ── CALL spread payoff ─────────────────────────────────────────────
|
|
payoff_call = max(0.0, S1 - Ks_call) - max(0.0, S1 - Kl_call)
|
|
|
|
# ── Net P&L ────────────────────────────────────────────────────────
|
|
gross_pnl = (net_put - payoff_put) + (net_call - payoff_call)
|
|
|
|
# Capital basis: total defined risk (both wings)
|
|
cap = wing_put + wing_call
|
|
|
|
# Deribit options fee: 0.03% of notional per leg, cap 12.5% of premium.
|
|
# 4 legs total for an iron condor. Conservative: cap 12.5% of gross net premium.
|
|
FEE_FRAC = 0.125
|
|
fee_cost = FEE_FRAC * (net_put + net_call)
|
|
|
|
ret_week = (gross_pnl - fee_cost) / cap
|
|
|
|
# Book at settlement bar
|
|
settle = i + tenor_d
|
|
daily_pnl[settle] += ret_week
|
|
in_trade[i:settle] = True
|
|
trades += 1
|
|
|
|
i += tenor_d
|
|
|
|
idx = pd.DatetimeIndex(dts)
|
|
net = daily_pnl
|
|
full = al._metrics_from_net(net, idx)
|
|
hmask = idx >= al.HOLDOUT
|
|
hold = al._metrics_from_net(net[hmask], idx[hmask]) if hmask.sum() > 3 else dict(sharpe=0.0, n=0)
|
|
bpy_d = al.bars_per_day(df) * 365.25
|
|
|
|
return dict(
|
|
full=full, holdout=hold, yearly=al._yearly(net, idx),
|
|
time_in_market=round(float(np.mean(in_trade)), 3),
|
|
turnover_per_year=round(float(trades * 2) / max(1, len(net) / bpy_d), 1),
|
|
net=net, idx=idx,
|
|
)
|
|
|
|
|
|
def _empty_result(df, dts):
|
|
idx = pd.DatetimeIndex(pd.to_datetime(dts, utc=True))
|
|
net = np.zeros(len(df))
|
|
return dict(
|
|
full=al._metrics_from_net(net, idx), holdout=dict(sharpe=0.0, n=0),
|
|
yearly=al._yearly(net, idx), time_in_market=0.0, turnover_per_year=0.0,
|
|
net=net, idx=idx,
|
|
)
|
|
|
|
|
|
# ─── Config grid ──────────────────────────────────────────────────────────────
|
|
|
|
CONFIGS = [
|
|
# (label, sdp, ldp, ivr_gate, crash_skip)
|
|
("w20-08-ivr30", -0.20, -0.08, 0.30, 1.01), # wider wing, gate only
|
|
("w25-10-ivr30", -0.25, -0.10, 0.30, 1.01), # narrower wing, gate only
|
|
("w20-08-ivr30-cs90", -0.20, -0.08, 0.30, 0.90), # wider + crash-skip
|
|
("w25-10-ivr30-cs90", -0.25, -0.10, 0.30, 0.90), # narrower + crash-skip
|
|
]
|
|
|
|
|
|
def run_config(label, sdp, ldp, ivr_gate, cs, tf="1d") -> dict:
|
|
name = f"OPT04-IC-{label}"
|
|
per_asset = {}
|
|
fee_ok_all = True
|
|
|
|
for asset in ("BTC", "ETH"):
|
|
df = al.get(asset, tf)
|
|
base = backtest_ic(df, asset,
|
|
short_delta_put=sdp, long_delta_put=ldp,
|
|
short_delta_call=-sdp, long_delta_call=-ldp,
|
|
ivr_gate=ivr_gate, crash_skip=cs)
|
|
|
|
# Fee sweep: re-run with different fee fracs via fee_side proxy
|
|
# (fee_side not directly used in our custom backtest; we scale FEE_FRAC)
|
|
sweep = {}
|
|
for f_side in al.FEE_SWEEP:
|
|
# Map taker fee to options fee frac: baseline is 0.125 at f_side=FEE_SIDE=0.0005
|
|
# Scale proportionally
|
|
scale = f_side / al.FEE_SIDE if al.FEE_SIDE > 0 else 1.0
|
|
fee_frac_scaled = 0.125 * scale
|
|
|
|
# Recompute with scaled fee
|
|
net_scaled = _recompute_net_scaled(df, asset, sdp, ldp, ivr_gate, cs, fee_frac_scaled)
|
|
net_arr = net_scaled["net"]
|
|
idx_arr = net_scaled["idx"]
|
|
m = al._metrics_from_net(net_arr, idx_arr)
|
|
sweep[f"{2*f_side*100:.2f}%RT"] = m["sharpe"]
|
|
|
|
fee_ok = sweep.get("0.20%RT", -9) > 0
|
|
fee_ok_all = fee_ok_all and fee_ok
|
|
|
|
per_asset[asset] = dict(
|
|
full=base["full"], holdout=base["holdout"],
|
|
tim=base["time_in_market"], turnover=base["turnover_per_year"],
|
|
fee_sweep=sweep, yearly=base["yearly"],
|
|
)
|
|
|
|
min_full = min(per_asset[a]["full"]["sharpe"] for a in ("BTC", "ETH"))
|
|
min_hold = min(per_asset[a]["holdout"].get("sharpe", 0.0) for a in ("BTC", "ETH"))
|
|
cells = [dict(
|
|
tf=tf, per_asset=per_asset,
|
|
min_asset_full_sharpe=round(min_full, 3),
|
|
min_asset_holdout_sharpe=round(min_hold, 3),
|
|
full_sharpe=round(np.mean([per_asset[a]["full"]["sharpe"] for a in ("BTC", "ETH")]), 3),
|
|
fee_survives=fee_ok_all,
|
|
)]
|
|
return dict(name=name, kind="weights", cells=cells, verdict=al._verdict(cells))
|
|
|
|
|
|
def _recompute_net_scaled(df, asset, sdp, ldp, ivr_gate, cs, fee_frac):
|
|
"""Recompute iron condor returns with a different fee fraction."""
|
|
close = df["close"].values.astype(float)
|
|
dts = pd.to_datetime(df["datetime"], utc=True)
|
|
n = len(close)
|
|
T_yr = 7 / 365.25
|
|
|
|
dv_pts = al.dvol(df, asset)
|
|
dv = dv_pts / 100.0
|
|
ivr = iv_rank_series(dv_pts, min_history=60)
|
|
|
|
daily_pnl = np.zeros(n)
|
|
|
|
valid_dvol = np.where(np.isfinite(dv_pts))[0]
|
|
if len(valid_dvol) < 60:
|
|
return dict(net=daily_pnl, idx=pd.DatetimeIndex(pd.to_datetime(dts, utc=True)))
|
|
|
|
i = valid_dvol[60]
|
|
while i + 7 < n:
|
|
S0 = close[i]; sig = dv[i]
|
|
if not np.isfinite(sig) or sig <= 0:
|
|
i += 7; continue
|
|
if not np.isfinite(ivr[i]):
|
|
i += 7; continue
|
|
if ivr_gate > 0 and ivr[i] < ivr_gate:
|
|
i += 7; continue
|
|
if cs < 1.0 and ivr[i] > cs:
|
|
i += 7; continue
|
|
|
|
Ks_put = strike_put_from_delta(S0, T_yr, sig, sdp)
|
|
Kl_put = strike_put_from_delta(S0, T_yr, sig, ldp)
|
|
net_put = bs_put(S0, Ks_put, T_yr, sig) - bs_put(S0, Kl_put, T_yr, sig)
|
|
wing_put = Ks_put - Kl_put
|
|
|
|
Ks_call = strike_call_from_delta(S0, T_yr, sig, -sdp)
|
|
Kl_call = strike_call_from_delta(S0, T_yr, sig, -ldp)
|
|
net_call = bs_call(S0, Ks_call, T_yr, sig) - bs_call(S0, Kl_call, T_yr, sig)
|
|
wing_call = Kl_call - Ks_call
|
|
|
|
if net_put <= 0 or net_call <= 0 or wing_put <= 0 or wing_call <= 0:
|
|
i += 7; continue
|
|
|
|
S1 = close[i + 7]
|
|
payoff_put = max(0.0, Ks_put - S1) - max(0.0, Kl_put - S1)
|
|
payoff_call = max(0.0, S1 - Ks_call) - max(0.0, S1 - Kl_call)
|
|
|
|
gross = (net_put - payoff_put) + (net_call - payoff_call)
|
|
fee = fee_frac * (net_put + net_call)
|
|
cap = wing_put + wing_call
|
|
|
|
daily_pnl[i + 7] += (gross - fee) / cap
|
|
i += 7
|
|
|
|
return dict(net=daily_pnl, idx=pd.DatetimeIndex(pd.to_datetime(dts, utc=True)))
|
|
|
|
|
|
# ─── Main ─────────────────────────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
print("OPT04 — Iron Condor Weekly (DVOL-gated)")
|
|
print("CAVEAT: premiums MODELED on DVOL ATM (no skew). Lead quantification only.")
|
|
print("DVOL history starts 2021-03 -> effective backtest from ~2021-Q3.")
|
|
print()
|
|
|
|
results = []
|
|
for label, sdp, ldp, ivr_gate, cs in CONFIGS:
|
|
print(f"Running: {label}")
|
|
rep = run_config(label, sdp, ldp, ivr_gate, cs, tf="1d")
|
|
results.append(rep)
|
|
print(al.fmt(rep))
|
|
print()
|
|
|
|
best = max(results, key=lambda r: max(
|
|
(c["min_asset_holdout_sharpe"] for c in r["cells"]), default=-9.0))
|
|
|
|
print("=" * 70)
|
|
print("BEST CONFIG:", best["name"])
|
|
print(al.fmt(best))
|
|
print()
|
|
print("JSON:", al.as_json(best))
|