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.1 KiB
Python
134 lines
4.1 KiB
Python
"""DC01 — Donchian Channel Breakout con filtri.
|
|
|
|
Trend-following classico: quando il prezzo rompe il massimo/minimo degli
|
|
ultimi N periodi, entra nella direzione del breakout.
|
|
|
|
Completamente diverso dallo squeeze (che usa Bollinger/Keltner).
|
|
Donchian cattura breakout di RANGE, non di VOLATILITÀ.
|
|
|
|
IN:
|
|
- OHLCV DataFrame
|
|
- Parametri: channel_period, volume_filter, atr_stop, trend_filter
|
|
|
|
OUT:
|
|
- Signal al breakout del canale Donchian
|
|
- BacktestResult
|
|
|
|
Logica:
|
|
1. Donchian upper = max(high, N periodi), lower = min(low, N periodi)
|
|
2. Close > upper → LONG (breakout rialzista)
|
|
3. Close < lower → SHORT (breakout ribassista)
|
|
4. Filtri: volume, trend EMA, ATR minimo
|
|
"""
|
|
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 DonchianBreakout(Strategy):
|
|
name = "DC01_donchian"
|
|
description = "Donchian Channel breakout — trend-following su range"
|
|
default_assets = ["BTC", "ETH"]
|
|
default_timeframes = ["15m", "1h"]
|
|
fee_rt = 0.002
|
|
|
|
def generate_signals(self, df, ts, **params):
|
|
c = df["close"].values
|
|
h = df["high"].values
|
|
l = df["low"].values
|
|
v = df["volume"].values
|
|
n = len(c)
|
|
|
|
period = params.get("channel_period", 48)
|
|
use_vol = params.get("vol_filter", False)
|
|
use_trend = params.get("trend_filter", False)
|
|
cooldown = params.get("cooldown", 6)
|
|
|
|
# EMA per trend filter
|
|
ema_50 = np.full(n, np.nan)
|
|
k = 2 / 51
|
|
ema_50[49] = np.mean(c[:50])
|
|
for i in range(50, n):
|
|
ema_50[i] = c[i] * k + ema_50[i - 1] * (1 - k)
|
|
|
|
# Volume media
|
|
vol_ma = np.full(n, np.nan)
|
|
for i in range(20, n):
|
|
vol_ma[i] = np.mean(v[i - 20:i])
|
|
|
|
signals = []
|
|
last_signal_idx = -cooldown
|
|
|
|
for i in range(period + 1, n):
|
|
if i - last_signal_idx < cooldown:
|
|
continue
|
|
|
|
upper = np.max(h[i - period:i])
|
|
lower = np.min(l[i - period:i])
|
|
|
|
direction = 0
|
|
if c[i] > upper:
|
|
direction = 1
|
|
elif c[i] < lower:
|
|
direction = -1
|
|
|
|
if direction == 0:
|
|
continue
|
|
|
|
# Trend filter: breakout must align with EMA trend
|
|
if use_trend and not np.isnan(ema_50[i]):
|
|
if direction == 1 and c[i] < ema_50[i]:
|
|
continue
|
|
if direction == -1 and c[i] > ema_50[i]:
|
|
continue
|
|
|
|
# Volume filter
|
|
if use_vol and not np.isnan(vol_ma[i]):
|
|
if v[i] < vol_ma[i] * 1.3:
|
|
continue
|
|
|
|
signals.append(Signal(
|
|
idx=i, direction=direction, entry_price=c[i],
|
|
metadata={"upper": float(upper), "lower": float(lower)},
|
|
))
|
|
last_signal_idx = i
|
|
|
|
return signals
|
|
|
|
|
|
if __name__ == "__main__":
|
|
strategy = DonchianBreakout()
|
|
|
|
configs = [
|
|
("p=24", {"channel_period": 24}),
|
|
("p=48", {"channel_period": 48}),
|
|
("p=96", {"channel_period": 96}),
|
|
("p=48+trend", {"channel_period": 48, "trend_filter": True}),
|
|
("p=48+vol", {"channel_period": 48, "vol_filter": True}),
|
|
("p=48+t+v", {"channel_period": 48, "trend_filter": True, "vol_filter": True}),
|
|
("p=96+t+v", {"channel_period": 96, "trend_filter": True, "vol_filter": True}),
|
|
]
|
|
|
|
all_results = []
|
|
for label, params in configs:
|
|
for asset in ["BTC", "ETH"]:
|
|
for tf in ["15m", "1h"]:
|
|
for hold in [3, 6, 12]:
|
|
r = strategy.backtest(asset, tf, hold=hold, **params)
|
|
if r and r.trades >= 30:
|
|
r.strategy_name = f"DC01 {label} h={hold}"
|
|
all_results.append(r)
|
|
|
|
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
|
print(f"\n{'=' * 120}")
|
|
print(f" DC01 DONCHIAN BREAKOUT — TOP 20")
|
|
print(f"{'=' * 120}")
|
|
for r in all_results[:20]:
|
|
r.print_summary()
|
|
if all_results:
|
|
all_results[0].print_yearly()
|