feat(strategy4): MT01 squeeze+MTF 82.7% acc — batte SQ02, 6 strategie scartate
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>
This commit is contained in:
@@ -0,0 +1,259 @@
|
||||
"""MT01 — Squeeze + Multi-Timeframe Momentum.
|
||||
|
||||
Problema SQ02: entra al breakout 15m ma non sa se il trend 1h è allineato.
|
||||
Soluzione: squeeze su 15m + conferma momentum su 1h.
|
||||
|
||||
Anti-overfitting: usa solo 2 indicatori (squeeze + EMA slope),
|
||||
nessun parametro complesso.
|
||||
|
||||
IN:
|
||||
- OHLCV 15m + 1h per lo stesso asset
|
||||
- Parametri: sq_threshold, ema_period_1h, min_slope
|
||||
|
||||
OUT:
|
||||
- Signal al breakout 15m confermato da trend 1h
|
||||
- BacktestResult
|
||||
|
||||
Logica:
|
||||
1. Squeeze release su 15m (come SQ01)
|
||||
2. Antifakeout filter (come SQ02)
|
||||
3. Check 1h: EMA slope positiva per long, negativa per short
|
||||
4. Check 1h: prezzo sopra/sotto EMA per conferma trend
|
||||
5. Entra solo se 15m e 1h concordano
|
||||
"""
|
||||
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, BacktestResult, YearlyStats, TF_MINUTES
|
||||
from src.strategies.indicators import keltner_ratio, detect_squeezes, ema
|
||||
from src.data.downloader import load_data
|
||||
|
||||
|
||||
class SqueezeMTFMomentum(Strategy):
|
||||
name = "MT01_squeeze_mtf"
|
||||
description = "Squeeze 15m + momentum trend 1h — multi-timeframe"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["15m"]
|
||||
fee_rt = 0.002
|
||||
|
||||
def generate_signals(self, df, ts, **params):
|
||||
"""Genera segnali squeeze 15m confermati da trend 1h."""
|
||||
c = df["close"].values
|
||||
h = df["high"].values
|
||||
l = df["low"].values
|
||||
v = df["volume"].values
|
||||
n = len(c)
|
||||
|
||||
asset = params.get("asset", "BTC")
|
||||
sq_thr = params.get("sq_threshold", 0.8)
|
||||
ema_period = params.get("ema_period", 50)
|
||||
min_slope_val = params.get("min_slope", 0.001)
|
||||
use_antifake = params.get("antifake", True)
|
||||
use_vol = params.get("vol_filter", False)
|
||||
|
||||
kcr = keltner_ratio(c, h, l, 14)
|
||||
events = detect_squeezes(c, h, l, kcr, sq_thr)
|
||||
|
||||
df_1h = load_data(asset, "1h")
|
||||
c1h = df_1h["close"].values
|
||||
ts1h_ms = df_1h["timestamp"].values
|
||||
n1h = len(c1h)
|
||||
ema_1h = ema(c1h, ema_period)
|
||||
ema_slope_arr = np.full(n1h, np.nan)
|
||||
for i in range(5, n1h):
|
||||
if not np.isnan(ema_1h[i]) and not np.isnan(ema_1h[i-5]) and ema_1h[i-5] > 0:
|
||||
ema_slope_arr[i] = (ema_1h[i] - ema_1h[i-5]) / ema_1h[i-5]
|
||||
|
||||
ts_ms = df["timestamp"].values
|
||||
signals = []
|
||||
|
||||
for ev in events:
|
||||
i = ev["idx"]
|
||||
if i < 1 or i >= n:
|
||||
continue
|
||||
first_ret = (c[i] - c[i-1]) / c[i-1] if c[i-1] > 0 else 0
|
||||
if abs(first_ret) < 0.001:
|
||||
continue
|
||||
if use_antifake:
|
||||
br = h[i] - l[i]
|
||||
if br > 0:
|
||||
if c[i] > c[i-1] and (h[i] - c[i]) / br > 0.6:
|
||||
continue
|
||||
elif c[i] <= c[i-1] and (c[i] - l[i]) / br > 0.6:
|
||||
continue
|
||||
if use_vol:
|
||||
avg_v = np.mean(v[ev["sq_start"]:i])
|
||||
if avg_v > 0 and v[i] <= avg_v * 1.3:
|
||||
continue
|
||||
|
||||
direction = 1 if first_ret > 0 else -1
|
||||
i1h = np.searchsorted(ts1h_ms, ts_ms[i]) - 1
|
||||
if i1h < ema_period or i1h >= n1h:
|
||||
continue
|
||||
if np.isnan(ema_1h[i1h]) or np.isnan(ema_slope_arr[i1h]):
|
||||
continue
|
||||
if direction == 1:
|
||||
if c1h[i1h] < ema_1h[i1h] or ema_slope_arr[i1h] < min_slope_val:
|
||||
continue
|
||||
else:
|
||||
if c1h[i1h] > ema_1h[i1h] or ema_slope_arr[i1h] > -min_slope_val:
|
||||
continue
|
||||
|
||||
signals.append(Signal(idx=i, direction=direction, entry_price=c[i-1]))
|
||||
|
||||
return signals
|
||||
|
||||
def backtest(self, asset, tf="15m", hold=3, **params):
|
||||
sq_thr = params.get("sq_threshold", 0.8)
|
||||
ema_period = params.get("ema_period", 50)
|
||||
min_slope = params.get("min_slope", 0.001)
|
||||
use_antifake = params.get("antifake", True)
|
||||
use_vol = params.get("vol_filter", False)
|
||||
|
||||
# Carica 15m e 1h
|
||||
df_15m = load_data(asset, "15m")
|
||||
df_1h = load_data(asset, "1h")
|
||||
|
||||
c15 = df_15m["close"].values
|
||||
h15 = df_15m["high"].values
|
||||
l15 = df_15m["low"].values
|
||||
v15 = df_15m["volume"].values
|
||||
n15 = len(c15)
|
||||
ts15 = pd.to_datetime(df_15m["timestamp"], unit="ms", utc=True)
|
||||
ts15_ms = df_15m["timestamp"].values
|
||||
|
||||
c1h = df_1h["close"].values
|
||||
ts1h_ms = df_1h["timestamp"].values
|
||||
n1h = len(c1h)
|
||||
|
||||
kcr = keltner_ratio(c15, h15, l15, 14)
|
||||
events = detect_squeezes(c15, h15, l15, kcr, sq_thr)
|
||||
|
||||
# EMA su 1h
|
||||
ema_1h = ema(c1h, ema_period)
|
||||
|
||||
# EMA slope (variazione percentuale su 5 barre)
|
||||
ema_slope = np.full(n1h, np.nan)
|
||||
for i in range(5, n1h):
|
||||
if not np.isnan(ema_1h[i]) and not np.isnan(ema_1h[i - 5]) and ema_1h[i - 5] > 0:
|
||||
ema_slope[i] = (ema_1h[i] - ema_1h[i - 5]) / ema_1h[i - 5]
|
||||
|
||||
yearly = {}
|
||||
capital = float(self.initial_capital)
|
||||
peak = capital
|
||||
max_dd = 0.0
|
||||
total_bars = 0
|
||||
|
||||
for ev in events:
|
||||
i = ev["idx"]
|
||||
if i + hold + 1 >= n15 or i < 1:
|
||||
continue
|
||||
|
||||
first_ret = (c15[i] - c15[i - 1]) / c15[i - 1] if c15[i - 1] > 0 else 0
|
||||
if abs(first_ret) < 0.001:
|
||||
continue
|
||||
|
||||
# Antifake
|
||||
if use_antifake:
|
||||
br = h15[i] - l15[i]
|
||||
if br > 0:
|
||||
if c15[i] > c15[i - 1] and (h15[i] - c15[i]) / br > 0.6:
|
||||
continue
|
||||
elif c15[i] <= c15[i - 1] and (c15[i] - l15[i]) / br > 0.6:
|
||||
continue
|
||||
|
||||
# Volume filter
|
||||
if use_vol:
|
||||
avg_v = np.mean(v15[ev["sq_start"]:i])
|
||||
if avg_v > 0 and v15[i] <= avg_v * 1.3:
|
||||
continue
|
||||
|
||||
direction = 1 if first_ret > 0 else -1
|
||||
|
||||
# Trova indice 1h corrispondente
|
||||
i1h = np.searchsorted(ts1h_ms, ts15_ms[i]) - 1
|
||||
if i1h < ema_period or i1h >= n1h or np.isnan(ema_1h[i1h]) or np.isnan(ema_slope[i1h]):
|
||||
continue
|
||||
|
||||
# Conferma trend 1h
|
||||
if direction == 1:
|
||||
if c1h[i1h] < ema_1h[i1h]:
|
||||
continue
|
||||
if ema_slope[i1h] < min_slope:
|
||||
continue
|
||||
else:
|
||||
if c1h[i1h] > ema_1h[i1h]:
|
||||
continue
|
||||
if ema_slope[i1h] > -min_slope:
|
||||
continue
|
||||
|
||||
entry = c15[i - 1]
|
||||
exit_price = c15[min(i + hold - 1, n15 - 1)]
|
||||
actual = (exit_price - entry) / entry * direction
|
||||
net = actual * self.leverage - self.fee_rt * self.leverage
|
||||
|
||||
capital += capital * self.position_size * net
|
||||
capital = max(capital, 10)
|
||||
if capital > peak: peak = capital
|
||||
dd = (peak - capital) / peak
|
||||
max_dd = max(max_dd, dd)
|
||||
total_bars += hold
|
||||
|
||||
year = ts15.iloc[i].year
|
||||
if year not in yearly:
|
||||
yearly[year] = {"w": 0, "t": 0, "pnl": 0.0}
|
||||
yearly[year]["t"] += 1
|
||||
if actual > 0: yearly[year]["w"] += 1
|
||||
yearly[year]["pnl"] += net * self.initial_capital
|
||||
|
||||
all_t = sum(d["t"] for d in yearly.values())
|
||||
all_w = sum(d["w"] for d in yearly.values())
|
||||
if all_t == 0:
|
||||
return None
|
||||
|
||||
yearly_stats = [YearlyStats(y, d["t"], d["w"], d["pnl"]) for y, d in sorted(yearly.items())]
|
||||
return BacktestResult(
|
||||
strategy_name=self.name, asset=asset, timeframe="15m", params=params,
|
||||
trades=all_t, wins=all_w, pnl=sum(d["pnl"] for d in yearly.values()),
|
||||
capital=capital, initial_capital=self.initial_capital,
|
||||
max_dd=max_dd * 100, time_in_market_pct=total_bars / n15 * 100,
|
||||
avg_trade_duration_h=hold * 15 / 60, years_active=len(yearly), yearly=yearly_stats,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = SqueezeMTFMomentum()
|
||||
|
||||
configs = [
|
||||
("ema50 sl0.1%", {"ema_period": 50, "min_slope": 0.001}),
|
||||
("ema50 sl0.05%", {"ema_period": 50, "min_slope": 0.0005}),
|
||||
("ema50 sl0.2%", {"ema_period": 50, "min_slope": 0.002}),
|
||||
("ema20 sl0.1%", {"ema_period": 20, "min_slope": 0.001}),
|
||||
("ema50 sl0.1%+vol", {"ema_period": 50, "min_slope": 0.001, "vol_filter": True}),
|
||||
("ema20 sl0.1%+vol", {"ema_period": 20, "min_slope": 0.001, "vol_filter": True}),
|
||||
("ema50 noAF", {"ema_period": 50, "min_slope": 0.001, "antifake": False}),
|
||||
("ema100 sl0.05%", {"ema_period": 100, "min_slope": 0.0005}),
|
||||
]
|
||||
|
||||
all_results = []
|
||||
for label, params in configs:
|
||||
for asset in ["BTC", "ETH"]:
|
||||
for hold in [3, 6]:
|
||||
r = strategy.backtest(asset, "15m", hold=hold, **params)
|
||||
if r and r.trades >= 30:
|
||||
r.strategy_name = f"MT01 {label} h={hold}"
|
||||
all_results.append(r)
|
||||
|
||||
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
||||
print(f"\n{'=' * 130}")
|
||||
print(f" MT01 SQUEEZE + MTF MOMENTUM — TOP 20")
|
||||
print(f"{'=' * 130}")
|
||||
for r in all_results[:20]:
|
||||
r.print_summary()
|
||||
if all_results:
|
||||
all_results[0].print_yearly()
|
||||
|
||||
print(f"\n BENCHMARK SQ02: 79.7% acc, 1250t, DD 6.5%, 9 anni, €5.23/day")
|
||||
@@ -0,0 +1,131 @@
|
||||
"""IB01 — Inside Bar Breakout.
|
||||
|
||||
Pattern di compressione a singola candela: quando una barra ha high < prev high
|
||||
E low > prev low, il prezzo si sta comprimendo. Al breakout del range della
|
||||
inside bar, segui la direzione.
|
||||
|
||||
17% delle candele 15m sono inside bars → frequenza altissima.
|
||||
|
||||
IN:
|
||||
- OHLCV DataFrame
|
||||
- Parametri: min_consecutive (N inside bars consecutivi),
|
||||
volume_filter, breakout_confirm
|
||||
|
||||
OUT:
|
||||
- Signal al breakout del range dell'inside bar
|
||||
- BacktestResult
|
||||
|
||||
Logica:
|
||||
1. Identifica N inside bars consecutivi (compressione)
|
||||
2. Quando il prezzo rompe il range → entra nella direzione del breakout
|
||||
3. Filtro: volume al breakout > media
|
||||
4. Hold fisso
|
||||
"""
|
||||
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 InsideBarBreakout(Strategy):
|
||||
name = "IB01_inside_bar"
|
||||
description = "Inside bar breakout — compressione a singola candela"
|
||||
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)
|
||||
|
||||
min_consec = params.get("min_consecutive", 2)
|
||||
use_vol = params.get("vol_filter", False)
|
||||
min_range_pct = params.get("min_range_pct", 0.002)
|
||||
|
||||
# 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 = []
|
||||
consec = 0
|
||||
mother_high = 0.0
|
||||
mother_low = 0.0
|
||||
|
||||
for i in range(1, n - 1):
|
||||
is_inside = h[i] <= h[i - 1] and l[i] >= l[i - 1]
|
||||
|
||||
if is_inside:
|
||||
if consec == 0:
|
||||
mother_high = h[i - 1]
|
||||
mother_low = l[i - 1]
|
||||
consec += 1
|
||||
else:
|
||||
if consec >= min_consec:
|
||||
range_pct = (mother_high - mother_low) / mother_low if mother_low > 0 else 0
|
||||
if range_pct < min_range_pct:
|
||||
consec = 0
|
||||
continue
|
||||
|
||||
# Breakout detection sulla barra corrente
|
||||
if c[i] > mother_high:
|
||||
direction = 1
|
||||
elif c[i] < mother_low:
|
||||
direction = -1
|
||||
else:
|
||||
consec = 0
|
||||
continue
|
||||
|
||||
# Volume filter
|
||||
if use_vol and not np.isnan(vol_ma[i]):
|
||||
if v[i] < vol_ma[i] * 1.2:
|
||||
consec = 0
|
||||
continue
|
||||
|
||||
signals.append(Signal(
|
||||
idx=i, direction=direction, entry_price=c[i],
|
||||
metadata={"consec": consec, "range_pct": round(range_pct * 100, 3)},
|
||||
))
|
||||
|
||||
consec = 0
|
||||
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = InsideBarBreakout()
|
||||
|
||||
configs = [
|
||||
("2ib", {"min_consecutive": 2}),
|
||||
("3ib", {"min_consecutive": 3}),
|
||||
("4ib", {"min_consecutive": 4}),
|
||||
("2ib+vol", {"min_consecutive": 2, "vol_filter": True}),
|
||||
("3ib+vol", {"min_consecutive": 3, "vol_filter": True}),
|
||||
("2ib r>0.3%", {"min_consecutive": 2, "min_range_pct": 0.003}),
|
||||
("3ib r>0.3%", {"min_consecutive": 3, "min_range_pct": 0.003}),
|
||||
]
|
||||
|
||||
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"IB01 {label} h={hold}"
|
||||
all_results.append(r)
|
||||
|
||||
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
||||
print(f"\n{'=' * 120}")
|
||||
print(f" IB01 INSIDE BAR BREAKOUT — TOP 20")
|
||||
print(f"{'=' * 120}")
|
||||
for r in all_results[:20]:
|
||||
r.print_summary()
|
||||
if all_results:
|
||||
all_results[0].print_yearly()
|
||||
@@ -0,0 +1,133 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,163 @@
|
||||
"""SB01 — Squeeze Breakout con Retest.
|
||||
|
||||
Il problema di SQ01/SQ02: entri al breakout, ma molti breakout sono fakeout.
|
||||
Soluzione: aspetta il RETEST. Dopo il breakout, il prezzo spesso torna a
|
||||
testare il livello di breakout prima di continuare.
|
||||
|
||||
Più selettivo di SQ02 → meno trade ma più accurati.
|
||||
Anti-overfitting: meccanismo strutturale (retest è fenomeno di mercato reale).
|
||||
|
||||
IN:
|
||||
- OHLCV DataFrame
|
||||
- Parametri: bb_window, sq_threshold, retest_window (quante barre aspettare
|
||||
il retest), retest_tolerance (quanto può tornare indietro)
|
||||
|
||||
OUT:
|
||||
- Signal al retest confermato (non al breakout iniziale)
|
||||
- BacktestResult
|
||||
|
||||
Logica:
|
||||
1. Rileva squeeze release (come SQ01)
|
||||
2. NON entrare subito — segna direzione e livello di breakout
|
||||
3. Nelle N barre successive, aspetta che il prezzo torni verso il livello
|
||||
4. Se il prezzo torna nel range di tolleranza e poi rimbalza → ENTRA
|
||||
5. Se il prezzo non torna → skip (momentum troppo forte, entry persa)
|
||||
6. Se il prezzo sfonda il livello → fakeout confermato, skip
|
||||
"""
|
||||
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
|
||||
from src.strategies.indicators import keltner_ratio, detect_squeezes
|
||||
|
||||
|
||||
class SqueezeBreakoutRetest(Strategy):
|
||||
name = "SB01_squeeze_retest"
|
||||
description = "Squeeze breakout con retest — entra solo dopo pullback confermato"
|
||||
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)
|
||||
|
||||
bb_w = params.get("bb_window", 14)
|
||||
sq_thr = params.get("sq_threshold", 0.8)
|
||||
retest_window = params.get("retest_window", 8)
|
||||
retest_tol = params.get("retest_tolerance", 0.5)
|
||||
use_vol = params.get("vol_filter", False)
|
||||
|
||||
kcr = keltner_ratio(c, h, l, bb_w)
|
||||
events = detect_squeezes(c, h, l, kcr, sq_thr)
|
||||
|
||||
vol_ma = np.full(n, np.nan)
|
||||
for i in range(20, n):
|
||||
vol_ma[i] = np.mean(v[i - 20:i])
|
||||
|
||||
signals = []
|
||||
|
||||
for ev in events:
|
||||
brk_idx = ev["idx"]
|
||||
if brk_idx + retest_window + 3 >= n or brk_idx < 1:
|
||||
continue
|
||||
|
||||
# Direzione breakout
|
||||
first_ret = (c[brk_idx] - c[brk_idx - 1]) / c[brk_idx - 1]
|
||||
if abs(first_ret) < 0.001:
|
||||
continue
|
||||
|
||||
direction = 1 if first_ret > 0 else -1
|
||||
breakout_level = c[brk_idx - 1]
|
||||
breakout_move = abs(first_ret)
|
||||
|
||||
# Aspetta retest nelle prossime N barre
|
||||
retest_found = False
|
||||
retest_idx = -1
|
||||
|
||||
for j in range(brk_idx + 1, min(brk_idx + retest_window + 1, n)):
|
||||
if direction == 1:
|
||||
# Long: il prezzo deve tornare GIÙ verso breakout_level
|
||||
pullback = (h[brk_idx] - l[j]) / (h[brk_idx] - breakout_level) if h[brk_idx] > breakout_level else 0
|
||||
if pullback >= retest_tol:
|
||||
# Tornato abbastanza — ora deve rimbalzare
|
||||
if c[j] > breakout_level:
|
||||
retest_found = True
|
||||
retest_idx = j
|
||||
break
|
||||
elif c[j] < breakout_level * 0.998:
|
||||
# Sfondato sotto → fakeout
|
||||
break
|
||||
else:
|
||||
# Short: il prezzo deve tornare SU verso breakout_level
|
||||
pullback = (h[j] - l[brk_idx]) / (breakout_level - l[brk_idx]) if breakout_level > l[brk_idx] else 0
|
||||
if pullback >= retest_tol:
|
||||
if c[j] < breakout_level:
|
||||
retest_found = True
|
||||
retest_idx = j
|
||||
break
|
||||
elif c[j] > breakout_level * 1.002:
|
||||
break
|
||||
|
||||
if not retest_found or retest_idx < 0:
|
||||
continue
|
||||
|
||||
# Volume filter al retest
|
||||
if use_vol and not np.isnan(vol_ma[retest_idx]):
|
||||
if v[retest_idx] < vol_ma[retest_idx] * 0.8:
|
||||
continue
|
||||
|
||||
signals.append(Signal(
|
||||
idx=retest_idx, direction=direction,
|
||||
entry_price=c[retest_idx],
|
||||
metadata={
|
||||
"breakout_idx": brk_idx,
|
||||
"retest_bars": retest_idx - brk_idx,
|
||||
"breakout_move": round(breakout_move * 100, 3),
|
||||
},
|
||||
))
|
||||
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = SqueezeBreakoutRetest()
|
||||
|
||||
configs = [
|
||||
("rt8 tol50%", {"retest_window": 8, "retest_tolerance": 0.5}),
|
||||
("rt6 tol50%", {"retest_window": 6, "retest_tolerance": 0.5}),
|
||||
("rt10 tol50%", {"retest_window": 10, "retest_tolerance": 0.5}),
|
||||
("rt8 tol30%", {"retest_window": 8, "retest_tolerance": 0.3}),
|
||||
("rt8 tol70%", {"retest_window": 8, "retest_tolerance": 0.7}),
|
||||
("rt8 tol50%+vol", {"retest_window": 8, "retest_tolerance": 0.5, "vol_filter": True}),
|
||||
("rt6 tol30%", {"retest_window": 6, "retest_tolerance": 0.3}),
|
||||
("rt12 tol50%", {"retest_window": 12, "retest_tolerance": 0.5}),
|
||||
]
|
||||
|
||||
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"SB01 {label} h={hold}"
|
||||
all_results.append(r)
|
||||
|
||||
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
||||
print(f"\n{'=' * 130}")
|
||||
print(f" SB01 SQUEEZE BREAKOUT RETEST — TOP 25")
|
||||
print(f"{'=' * 130}")
|
||||
for r in all_results[:25]:
|
||||
r.print_summary()
|
||||
if all_results:
|
||||
all_results[0].print_yearly()
|
||||
|
||||
# Confronto con benchmark
|
||||
print(f"\n BENCHMARK SQ02: 79.7% acc, 1250 trades, DD 6.5%, 9/9 anni")
|
||||
@@ -0,0 +1,148 @@
|
||||
"""MR01 — Mean Reversion da estremi RSI.
|
||||
|
||||
Approccio opposto allo squeeze: quando il prezzo va troppo lontano troppo veloce,
|
||||
scommetti che torni indietro. Autocorrelazione lag-1 negativa (-0.21 BTC, -0.35 ETH)
|
||||
conferma che il mercato a 15m è mean-reverting.
|
||||
|
||||
IN:
|
||||
- OHLCV DataFrame
|
||||
- Parametri: rsi_period, rsi_oversold, rsi_overbought, hold_bars,
|
||||
volume_filter (volume > N× media), atr_filter (move > N×ATR)
|
||||
|
||||
OUT:
|
||||
- Signal: long quando RSI < oversold, short quando RSI > overbought
|
||||
- BacktestResult con metriche
|
||||
|
||||
Logica:
|
||||
1. RSI scende sotto soglia oversold → LONG (prezzo tornerà su)
|
||||
2. RSI sale sopra soglia overbought → SHORT (prezzo tornerà giù)
|
||||
3. Filtro opzionale: volume spike conferma l'eccesso
|
||||
4. Filtro opzionale: move recente > N×ATR (eccesso di prezzo)
|
||||
5. Hold fisso, poi chiudi
|
||||
"""
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
result[i + 1] = 100 if al == 0 else 100 - 100 / (1 + ag / al)
|
||||
return result
|
||||
|
||||
|
||||
class MeanReversionRSI(Strategy):
|
||||
name = "MR01_mean_reversion_rsi"
|
||||
description = "Mean reversion da estremi RSI — fade eccessi direzionali"
|
||||
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)
|
||||
|
||||
rsi_period = params.get("rsi_period", 14)
|
||||
oversold = params.get("rsi_oversold", 25)
|
||||
overbought = params.get("rsi_overbought", 75)
|
||||
use_vol_filter = params.get("vol_filter", False)
|
||||
use_atr_filter = params.get("atr_filter", False)
|
||||
cooldown = params.get("cooldown", 4)
|
||||
|
||||
rsi_vals = rsi(c, rsi_period)
|
||||
|
||||
# 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])
|
||||
|
||||
# ATR
|
||||
tr = np.maximum(h[1:] - l[1:],
|
||||
np.maximum(np.abs(h[1:] - c[:-1]), np.abs(l[1:] - c[:-1])))
|
||||
atr_vals = np.full(n, np.nan)
|
||||
for i in range(15, len(tr)):
|
||||
atr_vals[i + 1] = np.mean(tr[i - 14:i])
|
||||
|
||||
signals = []
|
||||
last_signal_idx = -cooldown
|
||||
|
||||
for i in range(20, n):
|
||||
if i - last_signal_idx < cooldown:
|
||||
continue
|
||||
|
||||
direction = 0
|
||||
if rsi_vals[i] < oversold:
|
||||
direction = 1 # oversold → long
|
||||
elif rsi_vals[i] > overbought:
|
||||
direction = -1 # overbought → short
|
||||
|
||||
if direction == 0:
|
||||
continue
|
||||
|
||||
# Volume filter
|
||||
if use_vol_filter and not np.isnan(vol_ma[i]):
|
||||
if v[i] < vol_ma[i] * 1.5:
|
||||
continue
|
||||
|
||||
# ATR filter: il move recente deve essere > 1.5× ATR
|
||||
if use_atr_filter and not np.isnan(atr_vals[i]):
|
||||
recent_move = abs(c[i] - c[max(0, i - 3)]) / c[max(0, i - 3)]
|
||||
if recent_move < atr_vals[i] / c[i] * 1.5:
|
||||
continue
|
||||
|
||||
signals.append(Signal(
|
||||
idx=i, direction=direction, entry_price=c[i],
|
||||
metadata={"rsi": float(rsi_vals[i])},
|
||||
))
|
||||
last_signal_idx = i
|
||||
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = MeanReversionRSI()
|
||||
|
||||
configs = [
|
||||
("RSI25/75", {}),
|
||||
("RSI20/80", {"rsi_oversold": 20, "rsi_overbought": 80}),
|
||||
("RSI25/75+vol", {"vol_filter": True}),
|
||||
("RSI20/80+vol", {"rsi_oversold": 20, "rsi_overbought": 80, "vol_filter": True}),
|
||||
("RSI25/75+atr", {"atr_filter": True}),
|
||||
("RSI20/80+vol+atr", {"rsi_oversold": 20, "rsi_overbought": 80, "vol_filter": True, "atr_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"MR01 {label} h={hold}"
|
||||
all_results.append(r)
|
||||
|
||||
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
||||
print(f"\n{'=' * 120}")
|
||||
print(f" MR01 MEAN REVERSION RSI — TOP 20")
|
||||
print(f"{'=' * 120}")
|
||||
for r in all_results[:20]:
|
||||
r.print_summary()
|
||||
if all_results:
|
||||
all_results[0].print_yearly()
|
||||
@@ -0,0 +1,133 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,169 @@
|
||||
"""HY01 — Squeeze + Mean Reversion Ibrida.
|
||||
|
||||
Insight: durante lo squeeze (bassa volatilità), il prezzo mean-reverte
|
||||
DENTRO il range compresso. Autocorrelazione negativa a 15m conferma.
|
||||
Invece di aspettare il BREAKOUT, tradi la MEAN REVERSION dentro lo squeeze.
|
||||
|
||||
Completamente diverso da SQ01-SQ04 che aspettano il RILASCIO.
|
||||
|
||||
IN:
|
||||
- OHLCV DataFrame
|
||||
- Parametri: bb_window, sq_threshold, rsi_period, rsi_levels,
|
||||
vol_filter, bb_touch (prezzo tocca banda Bollinger)
|
||||
|
||||
OUT:
|
||||
- Signal: long quando RSI oversold DURANTE squeeze, short quando overbought
|
||||
- BacktestResult
|
||||
|
||||
Logica:
|
||||
1. Verifica che siamo IN squeeze (BB dentro KC)
|
||||
2. Prezzo tocca banda inferiore BB → LONG (tornerà alla media)
|
||||
3. Prezzo tocca banda superiore BB → SHORT (tornerà alla media)
|
||||
4. Conferma RSI: deve essere estremo nella direzione
|
||||
5. Hold corto (2-3 barre) — target: ritorno alla media
|
||||
6. Stop: se prezzo rompe lo squeeze → chiudi subito
|
||||
"""
|
||||
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
|
||||
from src.strategies.indicators import keltner_ratio
|
||||
|
||||
|
||||
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
|
||||
result[i + 1] = 100 if al == 0 else 100 - 100 / (1 + ag / al)
|
||||
return result
|
||||
|
||||
|
||||
def bollinger(close, window=14):
|
||||
n = len(close)
|
||||
upper = np.full(n, np.nan)
|
||||
lower = np.full(n, np.nan)
|
||||
mid = np.full(n, np.nan)
|
||||
for i in range(window, n):
|
||||
wc = close[i - window:i]
|
||||
m = np.mean(wc)
|
||||
s = np.std(wc)
|
||||
mid[i] = m
|
||||
upper[i] = m + 2 * s
|
||||
lower[i] = m - 2 * s
|
||||
return upper, mid, lower
|
||||
|
||||
|
||||
class SqueezeMeanReversion(Strategy):
|
||||
name = "HY01_squeeze_mr"
|
||||
description = "Mean reversion DENTRO lo squeeze — fade estremi in range compresso"
|
||||
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)
|
||||
|
||||
bb_w = params.get("bb_window", 14)
|
||||
sq_thr = params.get("sq_threshold", 0.8)
|
||||
rsi_period = params.get("rsi_period", 14)
|
||||
rsi_low = params.get("rsi_oversold", 30)
|
||||
rsi_high = params.get("rsi_overbought", 70)
|
||||
use_bb_touch = params.get("bb_touch", True)
|
||||
cooldown = params.get("cooldown", 3)
|
||||
|
||||
kcr = keltner_ratio(c, h, l, bb_w)
|
||||
rsi_vals = rsi(c, rsi_period)
|
||||
bb_upper, bb_mid, bb_lower = bollinger(c, bb_w)
|
||||
|
||||
signals = []
|
||||
last_idx = -cooldown
|
||||
|
||||
for i in range(bb_w + 1, n):
|
||||
if i - last_idx < cooldown:
|
||||
continue
|
||||
if np.isnan(kcr[i]) or np.isnan(bb_lower[i]):
|
||||
continue
|
||||
|
||||
# Must be IN squeeze
|
||||
if kcr[i] >= sq_thr:
|
||||
continue
|
||||
|
||||
direction = 0
|
||||
|
||||
if use_bb_touch:
|
||||
# Prezzo tocca/rompe BB lower → long (mean reversion up)
|
||||
if c[i] <= bb_lower[i] and rsi_vals[i] < rsi_low:
|
||||
direction = 1
|
||||
# Prezzo tocca/rompe BB upper → short (mean reversion down)
|
||||
elif c[i] >= bb_upper[i] and rsi_vals[i] > rsi_high:
|
||||
direction = -1
|
||||
else:
|
||||
# Solo RSI
|
||||
if rsi_vals[i] < rsi_low:
|
||||
direction = 1
|
||||
elif rsi_vals[i] > rsi_high:
|
||||
direction = -1
|
||||
|
||||
if direction == 0:
|
||||
continue
|
||||
|
||||
signals.append(Signal(
|
||||
idx=i, direction=direction, entry_price=c[i],
|
||||
metadata={
|
||||
"rsi": float(rsi_vals[i]),
|
||||
"kcr": float(kcr[i]),
|
||||
"bb_pos": "lower" if direction == 1 else "upper",
|
||||
},
|
||||
))
|
||||
last_idx = i
|
||||
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = SqueezeMeanReversion()
|
||||
|
||||
configs = [
|
||||
("bb+rsi30/70", {"bb_touch": True, "rsi_oversold": 30, "rsi_overbought": 70}),
|
||||
("bb+rsi25/75", {"bb_touch": True, "rsi_oversold": 25, "rsi_overbought": 75}),
|
||||
("bb+rsi35/65", {"bb_touch": True, "rsi_oversold": 35, "rsi_overbought": 65}),
|
||||
("rsi30/70 only", {"bb_touch": False, "rsi_oversold": 30, "rsi_overbought": 70}),
|
||||
("rsi25/75 only", {"bb_touch": False, "rsi_oversold": 25, "rsi_overbought": 75}),
|
||||
("sq<0.7 bb+rsi30", {"bb_touch": True, "sq_threshold": 0.7, "rsi_oversold": 30, "rsi_overbought": 70}),
|
||||
("sq<0.9 bb+rsi30", {"bb_touch": True, "sq_threshold": 0.9, "rsi_oversold": 30, "rsi_overbought": 70}),
|
||||
("sq<0.9 rsi35/65", {"bb_touch": False, "sq_threshold": 0.9, "rsi_oversold": 35, "rsi_overbought": 65}),
|
||||
]
|
||||
|
||||
all_results = []
|
||||
for label, params in configs:
|
||||
for asset in ["BTC", "ETH"]:
|
||||
for tf in ["15m", "1h"]:
|
||||
for hold in [2, 3, 4]:
|
||||
r = strategy.backtest(asset, tf, hold=hold, **params)
|
||||
if r and r.trades >= 30:
|
||||
r.strategy_name = f"HY01 {label} h={hold}"
|
||||
all_results.append(r)
|
||||
|
||||
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
||||
print(f"\n{'=' * 130}")
|
||||
print(f" HY01 SQUEEZE MEAN REVERSION — TOP 25")
|
||||
print(f"{'=' * 130}")
|
||||
for r in all_results[:25]:
|
||||
r.print_summary()
|
||||
if all_results:
|
||||
all_results[0].print_yearly()
|
||||
Reference in New Issue
Block a user