test(strategy2): VRP + filtri honest — 69% acc max, squeeze filter non aiuta
Regime filter migliore (+1% acc). Tutti gli anni positivi 2018-2026. Max realistico: 69.3% acc, 84% ann, 3.2% DD. 80% accuracy non raggiungibile con VRP puro. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,297 @@
|
||||
"""S2-10: VRP + filtri multipli per alzare accuracy.
|
||||
Filtri testati:
|
||||
1. NO vol sell se squeeze attivo (compressione → breakout imminente, NON vendere vol)
|
||||
2. NO vol sell se RV short-term > RV long-term (regime esplosivo)
|
||||
3. NO vol sell se move delle ultime 4h > 2% (momentum in corso)
|
||||
4. NO vol sell se volume spike > 2x media (evento in corso)
|
||||
5. COMBINAZIONI dei filtri sopra
|
||||
Test per-anno, NO compounding per PnL medio, compounding a fine report.
|
||||
"""
|
||||
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 keltner_ratio(close, high, low, window=14):
|
||||
n = len(close)
|
||||
result = np.full(n, np.nan)
|
||||
for i in range(window, n):
|
||||
wc = close[i - window : i]
|
||||
wh = high[i - window : i]
|
||||
wl = low[i - window : i]
|
||||
ma = np.mean(wc)
|
||||
bb_std = np.std(wc)
|
||||
tr = np.maximum(wh - wl, np.maximum(np.abs(wh - np.roll(wc, 1)), np.abs(wl - np.roll(wc, 1))))
|
||||
atr = np.mean(tr[1:])
|
||||
kc_r = (ma + 1.5 * atr) - (ma - 1.5 * atr)
|
||||
bb_r = (ma + 2 * bb_std) - (ma - 2 * bb_std)
|
||||
if kc_r > 0:
|
||||
result[i] = bb_r / kc_r
|
||||
return result
|
||||
|
||||
|
||||
def straddle_prem(iv, dte_h):
|
||||
if iv <= 0 or dte_h <= 0:
|
||||
return 0
|
||||
return iv * np.sqrt(dte_h / (24 * 365)) * 0.8
|
||||
|
||||
|
||||
def run_filtered(asset, dte=48):
|
||||
print(f"\n{'='*75}")
|
||||
print(f" {asset} — VRP + FILTRI (DTE={dte}h)")
|
||||
print(f"{'='*75}")
|
||||
|
||||
df = load_data(asset, "1h")
|
||||
close = df["close"].values
|
||||
high = df["high"].values
|
||||
low = df["low"].values
|
||||
volume = df["volume"].values
|
||||
n = len(close)
|
||||
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
|
||||
|
||||
rv_24 = rv_ann(close, 24)
|
||||
rv_168 = rv_ann(close, 168)
|
||||
kcr = keltner_ratio(close, high, low, 14)
|
||||
|
||||
# Pre-calcolo filtri
|
||||
vol_avg_48 = np.full(n, np.nan)
|
||||
for i in range(48, n):
|
||||
vol_avg_48[i] = np.mean(volume[i - 48 : i])
|
||||
|
||||
ret_4h = np.full(n, 0.0)
|
||||
for i in range(4, n):
|
||||
ret_4h[i] = abs(close[i] - close[i - 4]) / close[i - 4]
|
||||
|
||||
filter_configs = [
|
||||
# (name, use_squeeze, use_regime, use_momentum, use_volume, squeeze_thr, regime_thr, mom_thr, vol_thr)
|
||||
("BASELINE (no filter)", False, False, False, False, 0, 0, 0, 0),
|
||||
("squeeze_only", True, False, False, False, 0.85, 0, 0, 0),
|
||||
("regime_only", False, True, False, False, 0, 1.3, 0, 0),
|
||||
("momentum_only", False, False, True, False, 0, 0, 0.02, 0),
|
||||
("volume_only", False, False, False, True, 0, 0, 0, 2.0),
|
||||
("squeeze+regime", True, True, False, False, 0.85, 1.3, 0, 0),
|
||||
("squeeze+momentum", True, False, True, False, 0.85, 0, 0.02, 0),
|
||||
("squeeze+volume", True, False, False, True, 0.85, 0, 0, 2.0),
|
||||
("regime+momentum", False, True, True, False, 0, 1.3, 0.02, 0),
|
||||
("ALL_FILTERS", True, True, True, True, 0.85, 1.3, 0.02, 2.0),
|
||||
("squeeze+regime+mom", True, True, True, False, 0.85, 1.3, 0.02, 0),
|
||||
("squeeze_tight", True, False, False, False, 0.80, 0, 0, 0),
|
||||
("squeeze_tight+regime", True, True, False, False, 0.80, 1.3, 0, 0),
|
||||
("regime_strict", False, True, False, False, 0, 1.2, 0, 0),
|
||||
("mom_strict", False, False, True, False, 0, 0, 0.015, 0),
|
||||
("squeeze_tight+mom_strict", True, False, True, False, 0.80, 0, 0.015, 0),
|
||||
("ALL_TIGHT", True, True, True, True, 0.80, 1.2, 0.015, 1.8),
|
||||
]
|
||||
|
||||
results_table = []
|
||||
|
||||
for name, f_sq, f_reg, f_mom, f_vol, sq_thr, reg_thr, mom_thr, vol_thr in filter_configs:
|
||||
all_pnls = []
|
||||
yearly = {}
|
||||
|
||||
for i in range(170, n - dte):
|
||||
if ts.iloc[i].hour != 8:
|
||||
continue
|
||||
|
||||
rv_s = rv_24[i]
|
||||
rv_l = rv_168[i]
|
||||
if np.isnan(rv_s) or np.isnan(rv_l) or rv_s < 0.05 or rv_l < 0.05:
|
||||
continue
|
||||
|
||||
# === FILTRI ===
|
||||
skip = False
|
||||
|
||||
if f_sq and not np.isnan(kcr[i]):
|
||||
in_squeeze = kcr[i] < sq_thr
|
||||
# Controlla se squeeze nelle ultime 5 barre
|
||||
recent_squeeze = any(not np.isnan(kcr[j]) and kcr[j] < sq_thr for j in range(max(0, i - 5), i + 1))
|
||||
if recent_squeeze:
|
||||
skip = True
|
||||
|
||||
if f_reg and rv_l > 0:
|
||||
if rv_s / rv_l > reg_thr:
|
||||
skip = True
|
||||
|
||||
if f_mom:
|
||||
if ret_4h[i] > mom_thr:
|
||||
skip = True
|
||||
|
||||
if f_vol and not np.isnan(vol_avg_48[i]) and vol_avg_48[i] > 0:
|
||||
if volume[i] > vol_avg_48[i] * vol_thr:
|
||||
skip = True
|
||||
|
||||
if skip:
|
||||
continue
|
||||
|
||||
# === TRADE ===
|
||||
regime = rv_s / rv_l if rv_l > 0 else 1.0
|
||||
if regime > 2.0:
|
||||
iv_pf = 0.9
|
||||
elif regime > 1.5:
|
||||
iv_pf = 1.0
|
||||
elif regime > 1.0:
|
||||
iv_pf = 1.1
|
||||
else:
|
||||
iv_pf = 1.2
|
||||
iv = rv_l * iv_pf
|
||||
|
||||
prem = straddle_prem(iv, dte)
|
||||
spot = close[i]
|
||||
exit_idx = min(i + dte, n - 1)
|
||||
actual_move = abs(close[exit_idx] - spot) / spot
|
||||
|
||||
pos_size = INITIAL * 0.10
|
||||
if actual_move <= prem:
|
||||
raw = (prem - actual_move) * pos_size
|
||||
else:
|
||||
raw = -(actual_move - prem) * pos_size
|
||||
raw = max(raw, -pos_size * 0.05)
|
||||
|
||||
net = raw - FEE_ROUNDTRIP * pos_size
|
||||
all_pnls.append(net)
|
||||
|
||||
year = ts.iloc[i].year
|
||||
if year not in yearly:
|
||||
yearly[year] = []
|
||||
yearly[year].append(net)
|
||||
|
||||
if len(all_pnls) < 50:
|
||||
continue
|
||||
|
||||
wins = sum(1 for p in all_pnls if p > 0)
|
||||
acc = wins / len(all_pnls) * 100
|
||||
avg_pnl = np.mean(all_pnls)
|
||||
tot_pnl = np.sum(all_pnls)
|
||||
worst_trade = np.min(all_pnls)
|
||||
avg_win = np.mean([p for p in all_pnls if p > 0]) if wins > 0 else 0
|
||||
avg_loss = np.mean([p for p in all_pnls if p <= 0]) if wins < len(all_pnls) else 0
|
||||
|
||||
# Worst year
|
||||
worst_year_acc = 100
|
||||
worst_year_name = ""
|
||||
for y, ypnls in sorted(yearly.items()):
|
||||
yw = sum(1 for p in ypnls if p > 0) / len(ypnls) * 100 if ypnls else 0
|
||||
if yw < worst_year_acc:
|
||||
worst_year_acc = yw
|
||||
worst_year_name = str(y)
|
||||
|
||||
# Compounded return
|
||||
capital = float(INITIAL)
|
||||
peak = capital
|
||||
max_dd = 0
|
||||
for pnl in all_pnls:
|
||||
capital += pnl * (capital / INITIAL)
|
||||
capital = max(capital, 10)
|
||||
if capital > peak:
|
||||
peak = capital
|
||||
dd = (peak - capital) / peak
|
||||
max_dd = max(max_dd, dd)
|
||||
|
||||
n_years = len(yearly)
|
||||
ann = ((capital / INITIAL) ** (1 / n_years) - 1) * 100 if capital > 0 and n_years > 0 else -100
|
||||
|
||||
results_table.append({
|
||||
"name": name,
|
||||
"trades": len(all_pnls),
|
||||
"acc": acc,
|
||||
"avg_pnl": avg_pnl,
|
||||
"avg_win": avg_win,
|
||||
"avg_loss": avg_loss,
|
||||
"ann": ann,
|
||||
"max_dd": max_dd * 100,
|
||||
"worst_year": f"{worst_year_name}({worst_year_acc:.0f}%)",
|
||||
"capital": capital,
|
||||
})
|
||||
|
||||
# Sort by accuracy
|
||||
results_table.sort(key=lambda x: x["acc"], reverse=True)
|
||||
|
||||
print(f"\n {'Name':.<30s} {'Trades':>7s} {'Acc':>6s} {'AvgPnL':>8s} {'AvgWin':>8s} {'AvgLoss':>8s} {'Ann%':>7s} {'DD%':>5s} {'Worst_Yr':>12s} {'Capital':>10s}")
|
||||
print(f" {'-'*105}")
|
||||
for r in results_table:
|
||||
tag = "✅✅" if r["acc"] >= 75 else "✅" if r["acc"] >= 70 else ""
|
||||
print(f" {r['name']:.<30s} {r['trades']:>7d} {r['acc']:>5.1f}% {r['avg_pnl']:>+7.2f}€ {r['avg_win']:>+7.2f}€ {r['avg_loss']:>+7.2f}€ {r['ann']:>+6.1f}% {r['max_dd']:>4.1f}% {r['worst_year']:>12s} €{r['capital']:>9,.0f} {tag}")
|
||||
|
||||
# Dettaglio per anno del migliore
|
||||
best = results_table[0]
|
||||
print(f"\n MIGLIORE: {best['name']} → {best['acc']:.1f}% acc, {best['ann']:+.1f}% ann, DD {best['max_dd']:.1f}%")
|
||||
|
||||
# Rerun best per year
|
||||
best_name = best["name"]
|
||||
best_cfg = None
|
||||
for cfg in filter_configs:
|
||||
if cfg[0] == best_name:
|
||||
best_cfg = cfg
|
||||
break
|
||||
|
||||
if best_cfg:
|
||||
_, f_sq, f_reg, f_mom, f_vol, sq_thr, reg_thr, mom_thr, vol_thr = best_cfg
|
||||
yearly_detail = {}
|
||||
|
||||
for i in range(170, n - dte):
|
||||
if ts.iloc[i].hour != 8:
|
||||
continue
|
||||
rv_s = rv_24[i]
|
||||
rv_l = rv_168[i]
|
||||
if np.isnan(rv_s) or np.isnan(rv_l) or rv_s < 0.05 or rv_l < 0.05:
|
||||
continue
|
||||
|
||||
skip = False
|
||||
if f_sq:
|
||||
if any(not np.isnan(kcr[j]) and kcr[j] < sq_thr for j in range(max(0, i - 5), i + 1)):
|
||||
skip = True
|
||||
if f_reg and rv_l > 0 and rv_s / rv_l > reg_thr:
|
||||
skip = True
|
||||
if f_mom and ret_4h[i] > mom_thr:
|
||||
skip = True
|
||||
if f_vol and not np.isnan(vol_avg_48[i]) and vol_avg_48[i] > 0 and volume[i] > vol_avg_48[i] * vol_thr:
|
||||
skip = True
|
||||
if skip:
|
||||
continue
|
||||
|
||||
regime = rv_s / rv_l if rv_l > 0 else 1.0
|
||||
iv_pf = {True: 0.9, False: 1.2}[regime > 2.0] if regime > 2.0 else (1.0 if regime > 1.5 else (1.1 if regime > 1.0 else 1.2))
|
||||
iv = rv_l * iv_pf
|
||||
prem = straddle_prem(iv, dte)
|
||||
spot = close[i]
|
||||
exit_idx = min(i + dte, n - 1)
|
||||
move = abs(close[exit_idx] - spot) / spot
|
||||
pos_size = INITIAL * 0.10
|
||||
if move <= prem:
|
||||
raw = (prem - move) * pos_size
|
||||
else:
|
||||
raw = max(-(move - prem) * pos_size, -pos_size * 0.05)
|
||||
net = raw - FEE_ROUNDTRIP * pos_size
|
||||
|
||||
year = ts.iloc[i].year
|
||||
if year not in yearly_detail:
|
||||
yearly_detail[year] = []
|
||||
yearly_detail[year].append(net)
|
||||
|
||||
print(f"\n Dettaglio per anno ({best_name}):")
|
||||
for y in sorted(yearly_detail):
|
||||
pnls = yearly_detail[y]
|
||||
w = sum(1 for p in pnls if p > 0)
|
||||
a = w / len(pnls) * 100
|
||||
tag = " ← CRASH" if y in [2020, 2021, 2022] else ""
|
||||
print(f" {y}: trades={len(pnls):4d} acc={a:.1f}% avg=€{np.mean(pnls):+.2f} tot=€{np.sum(pnls):+.0f}{tag}")
|
||||
|
||||
|
||||
for asset in ["ETH", "BTC"]:
|
||||
run_filtered(asset, dte=48)
|
||||
run_filtered(asset, dte=24)
|
||||
Reference in New Issue
Block a user