"""CM01 — Cross-Market Momentum Filter. Squeeze su asset primario, entra SOLO se l'altro asset (BTC↔ETH) mostra momentum short-term nella STESSA direzione. Differenza da MT01: MT01 usa EMA slope su 1h (trend lento). CM01 usa rendimento grezzo degli ultimi 3-6 bar sull'asset cross (momentum veloce, stesso timeframe). Razionale: BTC e ETH sono altamente correlati ma non perfettamente. Se BTC fa squeeze breakout UP e anche ETH sta salendo (momentum 3-6 bar), la probabilità di continuazione è maggiore perché c'è consenso di mercato. Anti-overfitting: 1 parametro chiave (cross_bars 3-6), logica deterministica. Eredita antifakeout + volume da SQ02. """ 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 from src.strategies.indicators import keltner_ratio, detect_squeezes from src.data.downloader import load_data class CrossMarketMomentum(Strategy): name = "CM01_cross_momentum" description = "Squeeze + cross-asset short-term momentum filter" default_assets = ["BTC", "ETH"] default_timeframes = ["15m", "1h"] fee_rt = 0.002 leverage = 3.0 position_size = 0.15 initial_capital = 1000.0 # Map asset → cross asset _CROSS = {"BTC": "ETH", "ETH": "BTC"} def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex, **params) -> list[Signal]: """Genera segnali con cross-market momentum.""" c = df["close"].values h = df["high"].values l = df["low"].values v = df["volume"].values n = len(c) ts_ms = df["timestamp"].values asset = params.get("asset", "BTC") tf = params.get("tf", "15m") bb_w = params.get("bb_window", 14) sq_thr = params.get("sq_threshold", 0.8) retrace_limit = params.get("retrace_limit", 0.6) vol_mult = params.get("vol_multiplier", 1.3) use_vol = params.get("use_vol", True) cross_bars = params.get("cross_bars", 4) # barre momentum cross mom_min = params.get("mom_min", 0.0) # momentum minimo (0 = solo direzione) # Carica cross asset cross_asset = self._CROSS.get(asset) if cross_asset is None: return [] try: df_cross = load_data(cross_asset, tf) except Exception: return [] c_cross = df_cross["close"].values ts_cross_ms = df_cross["timestamp"].values n_cross = len(c_cross) # Momentum cross: rendimento log su cross_bars barre cross_mom = np.full(n_cross, np.nan) for i in range(cross_bars, n_cross): if c_cross[i - cross_bars] > 0: cross_mom[i] = np.log(c_cross[i] / c_cross[i - cross_bars]) kcr = keltner_ratio(c, h, l, bb_w) events = detect_squeezes(c, h, l, kcr, sq_thr) 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 direction = 1 if first_ret > 0 else -1 # Anti-fakeout br = h[i] - l[i] if br > 0: if direction == 1 and (h[i] - c[i]) / br > retrace_limit: continue elif direction == -1 and (c[i] - l[i]) / br > retrace_limit: continue # Volume confirm if use_vol: sq_start = ev["sq_start"] avg_sq_v = np.mean(v[sq_start:i]) if avg_sq_v > 0 and v[i] <= avg_sq_v * vol_mult: continue # Cross-market momentum: trova indice cross corrispondente i_cross = np.searchsorted(ts_cross_ms, ts_ms[i]) - 1 if i_cross < cross_bars or i_cross >= n_cross: continue mom = cross_mom[i_cross] if np.isnan(mom): continue # Filtra per direzione concordante if direction == 1 and mom <= mom_min: continue if direction == -1 and mom >= -mom_min: continue signals.append(Signal( idx=i, direction=direction, entry_price=c[i - 1], metadata={ "dur": ev["dur"], "cross_mom": float(mom), }, )) return signals if __name__ == "__main__": strategy = CrossMarketMomentum() configs = [ # cross_bars, mom_min, use_vol {"cross_bars": 3, "mom_min": 0.0, "use_vol": True}, {"cross_bars": 4, "mom_min": 0.0, "use_vol": True}, {"cross_bars": 6, "mom_min": 0.0, "use_vol": True}, {"cross_bars": 4, "mom_min": 0.001, "use_vol": True}, {"cross_bars": 4, "mom_min": 0.002, "use_vol": True}, {"cross_bars": 4, "mom_min": 0.0, "use_vol": False}, {"cross_bars": 3, "mom_min": 0.001, "use_vol": False}, {"cross_bars": 6, "mom_min": 0.001, "use_vol": True}, ] all_results = [] for cfg in configs: for asset in ["BTC", "ETH"]: for tf in ["15m", "1h"]: for hold in [3, 6]: r = strategy.backtest(asset, tf, hold=hold, cross_bars=cfg["cross_bars"], mom_min=cfg["mom_min"], use_vol=cfg["use_vol"]) if r and r.trades >= 20: lbl = (f"CM01 cb={cfg['cross_bars']} " f"mm={cfg['mom_min']} v={cfg['use_vol']} h={hold}") r.strategy_name = lbl all_results.append(r) all_results.sort(key=lambda r: r.accuracy, reverse=True) print(f"\n{'=' * 130}") print(" CM01 CROSS-MARKET MOMENTUM — TOP 20") print(f"{'=' * 130}") print(f" {'Nome':<50s} {'A/T':>7s} {'Trades':>6s} {'Acc':>6s} " f"{'PnL€':>10s} {'DD%':>6s} {'€/day':>7s} " f"{'Mkt%':>5s} {'Dur':>5s} {'Anni':>4s}") print(f" {'─' * 120}") 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%, €5.23/day, 9 anni") print(f" BENCHMARK MT01: 82.7% acc, 503t, DD 5.9%")