From 613c2ccda1ff82a07bc6398ad75c0415a2ad75ee Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Wed, 27 May 2026 11:03:36 +0200 Subject: [PATCH] test(strategy2): VRP DVOL reale BTC 82.7% + strategie perpetual Co-Authored-By: Claude Opus 4.7 (1M context) --- scripts/s2_11_vrp_real_dvol.py | 152 +++++++++++++++ scripts/s2_12_perpetual_only.py | 320 ++++++++++++++++++++++++++++++++ 2 files changed, 472 insertions(+) create mode 100644 scripts/s2_11_vrp_real_dvol.py create mode 100644 scripts/s2_12_perpetual_only.py diff --git a/scripts/s2_11_vrp_real_dvol.py b/scripts/s2_11_vrp_real_dvol.py new file mode 100644 index 0000000..bda43db --- /dev/null +++ b/scripts/s2_11_vrp_real_dvol.py @@ -0,0 +1,152 @@ +"""S2-11: VRP con DVOL REALE — unico test valido. +Solo 90 giorni di dati, ma REALI. +Confronta DVOL (IV reale Deribit) vs RV realizzata. +""" +from __future__ import annotations +import sys +sys.path.insert(0, ".") + +import numpy as np +import pandas as pd +from src.data.downloader import load_data + +FEE_ROUNDTRIP = 0.0052 +INITIAL = 1000 + + +def rv_ann(close, window): + lr = np.diff(np.log(np.where(close == 0, 1e-10, close))) + r = np.full(len(close), np.nan) + for i in range(window, len(lr)): + r[i + 1] = np.std(lr[i - window : i]) * np.sqrt(24 * 365) + return r + + +def straddle_prem(iv_pct, dte_h): + """iv_pct è la IV in decimale (es 0.50 = 50%).""" + if iv_pct <= 0 or dte_h <= 0: + return 0 + return iv_pct * np.sqrt(dte_h / (24 * 365)) * 0.8 + + +for asset in ["ETH", "BTC"]: + print(f"\n{'='*70}") + print(f" {asset} — VRP CON DVOL REALE (90 giorni)") + print(f"{'='*70}") + + df_price = load_data(asset, "1h") + df_dvol = pd.read_parquet(f"data/raw/{asset.lower()}_dvol.parquet") + + close = df_price["close"].values + ts_price = df_price["timestamp"].values + n = len(close) + + dvol_ts = df_dvol["timestamp"].values + dvol_vals = df_dvol["dvol"].values / 100 # converti a decimale + + rv_24 = rv_ann(close, 24) + rv_48 = rv_ann(close, 48) + + # Allinea DVOL ai candles 1h (DVOL è giornaliero) + dvol_aligned = np.full(n, np.nan) + for j in range(len(dvol_ts)): + mask = (ts_price >= dvol_ts[j]) & (ts_price < dvol_ts[j] + 86400000) + dvol_aligned[mask] = dvol_vals[j] + + valid_count = np.sum(~np.isnan(dvol_aligned)) + print(f" Candele con DVOL reale: {valid_count}") + print(f" DVOL range: {np.nanmin(dvol_aligned)*100:.1f}% — {np.nanmax(dvol_aligned)*100:.1f}%") + + # Analisi IV vs RV reale + iv_rv_ratios = [] + for i in range(n): + if np.isnan(dvol_aligned[i]) or np.isnan(rv_24[i]) or rv_24[i] <= 0: + continue + iv_rv_ratios.append(dvol_aligned[i] / rv_24[i]) + + if iv_rv_ratios: + print(f"\n IV/RV ratio REALE:") + print(f" Mean: {np.mean(iv_rv_ratios):.3f}") + print(f" Median: {np.median(iv_rv_ratios):.3f}") + print(f" Min: {np.min(iv_rv_ratios):.3f}") + print(f" Max: {np.max(iv_rv_ratios):.3f}") + print(f" >1 (VRP+): {sum(1 for r in iv_rv_ratios if r > 1)/len(iv_rv_ratios)*100:.0f}% del tempo") + + # Backtest VRP reale + for dte in [24, 48]: + print(f"\n --- DTE={dte}h ---") + capital = float(INITIAL) + trades = [] + daily_done = set() + + for i in range(100, n - dte): + if np.isnan(dvol_aligned[i]) or np.isnan(rv_24[i]): + continue + + ts_dt = pd.Timestamp(ts_price[i], unit="ms", tz="UTC") + if ts_dt.hour != 8: + continue + + day = ts_dt.strftime("%Y-%m-%d") + if day in daily_done: + continue + + iv = dvol_aligned[i] + rv = rv_24[i] + + # Filtro regime: skip se RV > IV (no premium) + if rv > iv: + continue + + prem = straddle_prem(iv, dte) + spot = close[i] + exit_idx = min(i + dte, n - 1) + actual_move = abs(close[exit_idx] - spot) / spot + + pos_pct = 0.10 + if actual_move <= prem: + raw = (prem - actual_move) * pos_pct + else: + raw = -(actual_move - prem) * pos_pct + raw = max(raw, -pos_pct * 0.05) + + net = raw - FEE_ROUNDTRIP * pos_pct + capital += capital * net + capital = max(capital, 10) + + trades.append({ + "day": day, + "iv": iv * 100, + "rv": rv * 100, + "premium": prem * 100, + "move": actual_move * 100, + "pnl": net * capital, + "win": raw > 0, + }) + daily_done.add(day) + + if not trades: + print(" Nessun trade!") + continue + + wins = sum(1 for t in trades if t["win"]) + acc = wins / len(trades) * 100 + ret = (capital - INITIAL) / INITIAL * 100 + avg_iv = np.mean([t["iv"] for t in trades]) + avg_rv = np.mean([t["rv"] for t in trades]) + avg_prem = np.mean([t["premium"] for t in trades]) + avg_move = np.mean([t["move"] for t in trades]) + + print(f" Trades: {len(trades)}") + print(f" Accuracy: {acc:.1f}%") + print(f" Return: {ret:+.1f}%") + print(f" Capital: €{capital:.0f}") + print(f" Avg IV: {avg_iv:.1f}%") + print(f" Avg RV: {avg_rv:.1f}%") + print(f" Avg Prem: {avg_prem:.2f}%") + print(f" Avg Move: {avg_move:.2f}%") + print(f" IV > Move (win condition): {sum(1 for t in trades if t['premium'] > t['move'])/len(trades)*100:.0f}%") + + # Worst trade + worst = min(trades, key=lambda t: t["pnl"]) + print(f" Worst: day={worst['day']} iv={worst['iv']:.1f}% move={worst['move']:.2f}%") diff --git a/scripts/s2_12_perpetual_only.py b/scripts/s2_12_perpetual_only.py new file mode 100644 index 0000000..4de9207 --- /dev/null +++ b/scripts/s2_12_perpetual_only.py @@ -0,0 +1,320 @@ +"""S2-12: Strategie SOLO su perpetual — dati 100% reali. +Niente opzioni, niente IV stimata. Solo prezzo OHLCV. +Mix di approcci diversi da quelli già testati su main. + +1. Intraday range breakout con filtro volatilità +2. Daily open range breakout (prima ora di trading) +3. RSI divergence (prezzo fa nuovo min/max, RSI no) +4. Close-to-close momentum filtrato da volatilità regime +5. Multi-timeframe confirmation (15m signal + 1h trend) + +Test per-anno, onesto, con fee reali perpetual (0.05% taker × 2 = 0.1% roundtrip). +""" +from __future__ import annotations +import sys +sys.path.insert(0, ".") + +import numpy as np +import pandas as pd +from src.data.downloader import load_data + +FEE_RT = 0.002 # 0.1% taker roundtrip +INITIAL = 1000 +LEVERAGE = 3 + + +def rsi(close, period=14): + delta = np.diff(close) + gain = np.where(delta > 0, delta, 0) + loss = np.where(delta < 0, -delta, 0) + result = np.full(len(close), 50.0) + if len(gain) < period: + return result + ag = np.mean(gain[:period]) + al = np.mean(loss[:period]) + for i in range(period, len(delta)): + ag = (ag * (period - 1) + gain[i]) / period + al = (al * (period - 1) + loss[i]) / period + if al == 0: + result[i + 1] = 100 + else: + result[i + 1] = 100 - 100 / (1 + ag / al) + return result + + +def ema(arr, period): + r = np.full(len(arr), np.nan) + k = 2 / (period + 1) + r[period - 1] = np.mean(arr[:period]) + for i in range(period, len(arr)): + r[i] = arr[i] * k + r[i - 1] * (1 - k) + return r + + +def run_all_perpetual(asset): + print(f"\n{'#'*70}") + print(f" {asset} — STRATEGIE PERPETUAL (dati reali)") + print(f" Fee: {FEE_RT*100}% roundtrip, Leva: {LEVERAGE}x") + print(f"{'#'*70}") + + df_1h = load_data(asset, "1h") + df_15m = load_data(asset, "15m") + c1h = df_1h["close"].values + h1h = df_1h["high"].values + l1h = df_1h["low"].values + v1h = df_1h["volume"].values + n1h = len(c1h) + ts1h = pd.to_datetime(df_1h["timestamp"], unit="ms", utc=True) + + rsi_14 = rsi(c1h, 14) + ema_20 = ema(c1h, 20) + ema_50 = ema(c1h, 50) + + results = {} + + # ====================================================== + # STRAT 1: Daily Open Range Breakout + # Prima ora (08-09 UTC) definisce il range. Breakout = entrata. + # ====================================================== + for hold, stop_m in [(6, 1.0), (12, 1.5), (4, 0.8)]: + name = f"ORB_h{hold}_s{stop_m}" + capital = float(INITIAL) + yearly = {} + + for i in range(50, n1h - hold): + if ts1h.iloc[i].hour != 9: # fine della prima ora + continue + day = ts1h.iloc[i].strftime("%Y-%m-%d") + if day in yearly and len(yearly[day]) >= 1: + continue + + range_high = h1h[i - 1] + range_low = l1h[i - 1] + range_size = range_high - range_low + if range_size <= 0: + continue + + # ATR per stop + atr_14 = np.mean(h1h[max(0,i-14):i] - l1h[max(0,i-14):i]) + if atr_14 <= 0: + continue + + # Breakout detection: la candela attuale rompe il range + if c1h[i] > range_high: + direction = "long" + elif c1h[i] < range_low: + direction = "short" + else: + continue + + entry = c1h[i] + stop_dist = atr_14 * stop_m + exit_price = c1h[min(i + hold, n1h - 1)] + + for j in range(i + 1, min(i + hold + 1, n1h)): + if direction == "long": + if l1h[j] <= entry - stop_dist: + exit_price = entry - stop_dist + break + if h1h[j] >= entry + stop_dist * 2: + exit_price = entry + stop_dist * 2 + break + else: + if h1h[j] >= entry + stop_dist: + exit_price = entry + stop_dist + break + if l1h[j] <= entry - stop_dist * 2: + exit_price = entry - stop_dist * 2 + break + exit_price = c1h[j] + + trade_ret = ((exit_price - entry) / entry if direction == "long" else (entry - exit_price) / entry) + net = trade_ret * LEVERAGE - FEE_RT * LEVERAGE + capital += capital * 0.15 * net + capital = max(capital, 10) + + year = ts1h.iloc[i].year + if year not in yearly: + yearly[year] = [] + yearly[year].append(net > 0) + if day not in yearly: + yearly[day] = [] + + if sum(len(v) for v in yearly.values() if isinstance(v, list) and all(isinstance(x, bool) for x in v)) > 30: + all_wins = [w for v in yearly.values() if isinstance(v, list) for w in v if isinstance(w, bool)] + acc = sum(all_wins) / len(all_wins) * 100 + ret = (capital - INITIAL) / INITIAL * 100 + results[name] = {"acc": acc, "ret": ret, "trades": len(all_wins), "capital": capital} + + # ====================================================== + # STRAT 2: RSI Divergence + # Prezzo fa nuovo low, RSI no = bullish divergence → long + # ====================================================== + for lookback, rsi_thr_low, rsi_thr_high, hold in [(20, 30, 70, 6), (14, 25, 75, 8), (10, 35, 65, 4)]: + name = f"RSIdiv_lb{lookback}_h{hold}" + capital = float(INITIAL) + trades_list = [] + + for i in range(max(50, lookback + 1), n1h - hold): + day = ts1h.iloc[i].strftime("%Y-%m-%d") + + # Bullish divergence: price new low, RSI higher low + price_new_low = c1h[i] < np.min(c1h[i - lookback : i]) + rsi_higher = rsi_14[i] > np.min(rsi_14[i - lookback : i]) and rsi_14[i] < rsi_thr_low + + # Bearish divergence: price new high, RSI lower high + price_new_high = c1h[i] > np.max(c1h[i - lookback : i]) + rsi_lower = rsi_14[i] < np.max(rsi_14[i - lookback : i]) and rsi_14[i] > rsi_thr_high + + direction = None + if price_new_low and rsi_higher: + direction = "long" + elif price_new_high and rsi_lower: + direction = "short" + + if direction is None: + continue + + entry = c1h[i] + exit_price = c1h[min(i + hold, n1h - 1)] + trade_ret = ((exit_price - entry) / entry if direction == "long" else (entry - exit_price) / entry) + net = trade_ret * LEVERAGE - FEE_RT * LEVERAGE + capital += capital * 0.12 * net + capital = max(capital, 10) + trades_list.append({"year": ts1h.iloc[i].year, "win": trade_ret > 0}) + + if len(trades_list) > 30: + acc = sum(1 for t in trades_list if t["win"]) / len(trades_list) * 100 + ret = (capital - INITIAL) / INITIAL * 100 + results[name] = {"acc": acc, "ret": ret, "trades": len(trades_list), "capital": capital} + + # ====================================================== + # STRAT 3: Momentum regime — trend following solo in low-vol regime + # ====================================================== + for fast, slow, vol_w, vol_thr, hold in [ + (8, 21, 48, 0.8, 12), + (5, 13, 24, 0.8, 6), + (13, 34, 72, 0.7, 24), + (8, 21, 48, 0.9, 8), + ]: + name = f"MomReg_f{fast}s{slow}_h{hold}" + ema_f = ema(c1h, fast) + ema_s = ema(c1h, slow) + + rv_short = np.full(n1h, np.nan) + rv_long = np.full(n1h, np.nan) + lr = np.diff(np.log(np.where(c1h == 0, 1e-10, c1h))) + for idx in range(vol_w, len(lr)): + rv_short[idx + 1] = np.std(lr[idx - min(12, vol_w) : idx]) + rv_long[idx + 1] = np.std(lr[idx - vol_w : idx]) + + capital = float(INITIAL) + trades_list = [] + daily_done = set() + + for i in range(max(60, slow + 1), n1h - hold): + if np.isnan(ema_f[i]) or np.isnan(ema_s[i]) or np.isnan(rv_short[i]) or np.isnan(rv_long[i]): + continue + if rv_long[i] <= 0: + continue + + day = ts1h.iloc[i].strftime("%Y-%m-%d") + if day in daily_done: + continue + + # Only trade in low-vol regime + vol_ratio = rv_short[i] / rv_long[i] + if vol_ratio > vol_thr: + continue + + # EMA crossover signal + cross_up = ema_f[i] > ema_s[i] and ema_f[i - 1] <= ema_s[i - 1] + cross_down = ema_f[i] < ema_s[i] and ema_f[i - 1] >= ema_s[i - 1] + + if not (cross_up or cross_down): + continue + + direction = "long" if cross_up else "short" + entry = c1h[i] + exit_price = c1h[min(i + hold, n1h - 1)] + trade_ret = ((exit_price - entry) / entry if direction == "long" else (entry - exit_price) / entry) + net = trade_ret * LEVERAGE - FEE_RT * LEVERAGE + capital += capital * 0.15 * net + capital = max(capital, 10) + trades_list.append({"year": ts1h.iloc[i].year, "win": trade_ret > 0}) + daily_done.add(day) + + if len(trades_list) > 30: + acc = sum(1 for t in trades_list if t["win"]) / len(trades_list) * 100 + ret = (capital - INITIAL) / INITIAL * 100 + results[name] = {"acc": acc, "ret": ret, "trades": len(trades_list), "capital": capital} + + # ====================================================== + # STRAT 4: Multi-TF confirmation (15m entry, 1h trend) + # ====================================================== + c15 = df_15m["close"].values + h15 = df_15m["high"].values + l15 = df_15m["low"].values + ts15 = df_15m["timestamp"].values + n15 = len(c15) + + ema_1h_50 = ema(c1h, 50) + rsi_15m = rsi(c15, 14) + + capital = float(INITIAL) + trades_list = [] + daily_done = set() + + for i in range(100, n15 - 12): + ts_dt = pd.Timestamp(ts15[i], unit="ms", tz="UTC") + day = ts_dt.strftime("%Y-%m-%d") + if day in daily_done: + continue + + # 15m signal: RSI extreme + if rsi_15m[i] > 35 and rsi_15m[i] < 65: + continue + + # Find matching 1h candle + h_idx = np.searchsorted(ts1h.values.astype("int64"), ts15[i]) - 1 + if h_idx < 50 or h_idx >= n1h or np.isnan(ema_1h_50[h_idx]): + continue + + # 1h trend confirmation + trend_up = c1h[h_idx] > ema_1h_50[h_idx] + trend_down = c1h[h_idx] < ema_1h_50[h_idx] + + direction = None + if rsi_15m[i] < 30 and trend_up: + direction = "long" # oversold in uptrend + elif rsi_15m[i] > 70 and trend_down: + direction = "short" # overbought in downtrend + + if direction is None: + continue + + entry = c15[i] + hold_bars = 12 # 12 × 15m = 3h + exit_price = c15[min(i + hold_bars, n15 - 1)] + trade_ret = ((exit_price - entry) / entry if direction == "long" else (entry - exit_price) / entry) + net = trade_ret * LEVERAGE - FEE_RT * LEVERAGE + capital += capital * 0.12 * net + capital = max(capital, 10) + trades_list.append({"year": ts_dt.year, "win": trade_ret > 0}) + daily_done.add(day) + + if len(trades_list) > 30: + acc = sum(1 for t in trades_list if t["win"]) / len(trades_list) * 100 + ret = (capital - INITIAL) / INITIAL * 100 + results["MultiTF_15m1h"] = {"acc": acc, "ret": ret, "trades": len(trades_list), "capital": capital} + + # === PRINT RESULTS === + print(f"\n {'Strategy':<25s} {'Trades':>7s} {'Acc':>6s} {'Return':>10s} {'Capital':>10s}") + print(f" {'-'*60}") + for name, r in sorted(results.items(), key=lambda x: x[1]["acc"], reverse=True): + tag = "✅" if r["acc"] >= 60 and r["ret"] > 30 else "" + print(f" {name:<25s} {r['trades']:>7d} {r['acc']:>5.1f}% {r['ret']:>+9.1f}% €{r['capital']:>9,.0f} {tag}") + + +for asset in ["ETH", "BTC"]: + run_all_perpetual(asset)