Adriano
f42fec9fac
Nuova strategia MT01: squeeze 15m + momentum EMA 1h BTC 15m: 82.7% acc, 503 trades, DD 5.9%, 9/9 anni, worst 72% ETH 15m: 81.2% acc, 404 trades, DD 2.9%, 9/9 anni, worst 73% Strategie testate e scartate (waste W23-W28): IB01 inside bar (58.7%, no edge) DC01 donchian (48%, sotto random) SB01 retest (52%, no edge) MR01 mean reversion RSI (62.9%, DD 29%) VO01 volume spike (64.2%, DD 34%) HY01 squeeze+MR (64.6%, DD 14.5%) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
134 lines
4.3 KiB
Python
134 lines
4.3 KiB
Python
"""VO01 — Volume Spike Reversal.
|
||
|
||
Quando il volume esplode (>3× media) con un forte move direzionale,
|
||
il mercato è in eccesso → fade il move (mean reversion).
|
||
|
||
Diverso dallo squeeze: non cerca compressione, cerca ECCESSO.
|
||
Il volume spike indica panico/euforia → reversal probabile.
|
||
|
||
IN:
|
||
- OHLCV DataFrame
|
||
- Parametri: vol_mult (3), move_threshold (0.005), hold
|
||
|
||
OUT:
|
||
- Signal: fade la direzione del volume spike
|
||
- BacktestResult
|
||
|
||
Logica:
|
||
1. Volume > vol_mult × media 20 periodi
|
||
2. Move nella candela > move_threshold (0.5%)
|
||
3. Direzione: opposta al move (mean reversion)
|
||
4. Filtro: non entrare se già in trend forte (EMA slope)
|
||
"""
|
||
from __future__ import annotations
|
||
import sys
|
||
sys.path.insert(0, ".")
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
from src.strategies.base import Strategy, Signal
|
||
|
||
|
||
class VolumeSpikeReversal(Strategy):
|
||
name = "VO01_vol_spike_reversal"
|
||
description = "Volume spike reversal — fade eccessi di volume/prezzo"
|
||
default_assets = ["BTC", "ETH"]
|
||
default_timeframes = ["15m", "1h"]
|
||
fee_rt = 0.002
|
||
|
||
def generate_signals(self, df, ts, **params):
|
||
c = df["close"].values
|
||
o = df["open"].values
|
||
h = df["high"].values
|
||
l = df["low"].values
|
||
v = df["volume"].values
|
||
n = len(c)
|
||
|
||
vol_mult = params.get("vol_mult", 3.0)
|
||
move_thr = params.get("move_threshold", 0.005)
|
||
use_trend_filter = params.get("trend_filter", False)
|
||
cooldown = params.get("cooldown", 4)
|
||
|
||
# Volume media rolling
|
||
vol_ma = np.full(n, np.nan)
|
||
for i in range(20, n):
|
||
vol_ma[i] = np.mean(v[i - 20:i])
|
||
|
||
# EMA per trend filter
|
||
ema_20 = np.full(n, np.nan)
|
||
k = 2 / 21
|
||
ema_20[19] = np.mean(c[:20])
|
||
for i in range(20, n):
|
||
ema_20[i] = c[i] * k + ema_20[i - 1] * (1 - k)
|
||
|
||
signals = []
|
||
last_idx = -cooldown
|
||
|
||
for i in range(21, n):
|
||
if i - last_idx < cooldown:
|
||
continue
|
||
if np.isnan(vol_ma[i]):
|
||
continue
|
||
|
||
# Volume spike
|
||
if v[i] < vol_ma[i] * vol_mult:
|
||
continue
|
||
|
||
# Price move
|
||
move = (c[i] - o[i]) / o[i] if o[i] > 0 else 0
|
||
if abs(move) < move_thr:
|
||
continue
|
||
|
||
# Fade: opposto al move
|
||
direction = -1 if move > 0 else 1
|
||
|
||
# Trend filter: non fare mean reversion contro trend forte
|
||
if use_trend_filter and not np.isnan(ema_20[i]):
|
||
ema_slope = (ema_20[i] - ema_20[max(0, i - 5)]) / ema_20[max(0, i - 5)]
|
||
if direction == -1 and ema_slope > 0.005:
|
||
continue
|
||
if direction == 1 and ema_slope < -0.005:
|
||
continue
|
||
|
||
signals.append(Signal(
|
||
idx=i, direction=direction, entry_price=c[i],
|
||
metadata={"vol_ratio": float(v[i] / vol_ma[i]), "move_pct": round(move * 100, 3)},
|
||
))
|
||
last_idx = i
|
||
|
||
return signals
|
||
|
||
|
||
if __name__ == "__main__":
|
||
strategy = VolumeSpikeReversal()
|
||
|
||
configs = [
|
||
("v3x m0.5%", {"vol_mult": 3.0, "move_threshold": 0.005}),
|
||
("v3x m1%", {"vol_mult": 3.0, "move_threshold": 0.01}),
|
||
("v4x m0.5%", {"vol_mult": 4.0, "move_threshold": 0.005}),
|
||
("v4x m1%", {"vol_mult": 4.0, "move_threshold": 0.01}),
|
||
("v3x m0.5%+tf", {"vol_mult": 3.0, "move_threshold": 0.005, "trend_filter": True}),
|
||
("v3x m1%+tf", {"vol_mult": 3.0, "move_threshold": 0.01, "trend_filter": True}),
|
||
("v5x m1%", {"vol_mult": 5.0, "move_threshold": 0.01}),
|
||
("v5x m1%+tf", {"vol_mult": 5.0, "move_threshold": 0.01, "trend_filter": True}),
|
||
]
|
||
|
||
all_results = []
|
||
for label, params in configs:
|
||
for asset in ["BTC", "ETH"]:
|
||
for tf in ["15m", "1h"]:
|
||
for hold in [3, 6]:
|
||
r = strategy.backtest(asset, tf, hold=hold, **params)
|
||
if r and r.trades >= 30:
|
||
r.strategy_name = f"VO01 {label} h={hold}"
|
||
all_results.append(r)
|
||
|
||
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
||
print(f"\n{'=' * 120}")
|
||
print(f" VO01 VOLUME SPIKE REVERSAL — TOP 20")
|
||
print(f"{'=' * 120}")
|
||
for r in all_results[:20]:
|
||
r.print_summary()
|
||
if all_results:
|
||
all_results[0].print_yearly()
|