"""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")