Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(strategy2): VRP DVOL reale BTC 82.7% + strategie perpetual

Browse Source 
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
AdrianoDev
2026-05-27 11:03:36 +02:00
parent f6e111f72d
commit 613c2ccda1
2 changed files with 472 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/s2_11_vrp_real_dvol.py
+152
View File
@@ -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}%")
scripts/s2_12_perpetual_only.py
+320
View File
@@ -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)