diff --git a/scripts/strategies/AD01_adaptive_squeeze.py b/scripts/strategies/AD01_adaptive_squeeze.py new file mode 100644 index 0000000..b2e1200 --- /dev/null +++ b/scripts/strategies/AD01_adaptive_squeeze.py @@ -0,0 +1,205 @@ +"""AD01 — Adaptive Squeeze Threshold. + +Problema SQ02: sq_threshold fisso (0.8) non si adatta al regime di volatilità. +Soluzione: threshold adattivo basato su volatilità recente. + +Logica: + - Calcola volatilità rolling (std dei rendimenti su finestra 100 barre) + - Confronta con percentile storico (rolling 500 barre) + - Alta vol (>70° percentile) → soglia BASSA (0.65) — squeeze più "lenti" + - Bassa vol (<30° percentile) → soglia ALTA (0.90) — squeeze "stretti" + - Vol media → soglia standard (0.80) + +Razionale: in mercati calmi, il BB si stringe molto → sq_threshold alto cattura +segnali migliori. In mercati volatili, bastano squeeze minori per essere significativi. + +Anti-overfitting: solo 3 parametri (low_thr, mid_thr, high_thr), 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, TF_MINUTES +from src.strategies.indicators import keltner_ratio, ema +from src.data.downloader import load_data + + +def _adaptive_sq_threshold(close: np.ndarray, + vol_window: int = 100, + regime_window: int = 500, + low_thr: float = 0.65, + mid_thr: float = 0.80, + high_thr: float = 0.90) -> np.ndarray: + """Calcola sq_threshold adattivo per ogni barra.""" + n = len(close) + lr = np.diff(np.log(np.where(close <= 0, 1e-10, close))) + vol = np.full(n, np.nan) + for i in range(vol_window, n): + vol[i] = np.std(lr[i - vol_window:i]) + + # Percentile rolling della volatilità + thresh = np.full(n, mid_thr) + for i in range(regime_window, n): + if np.isnan(vol[i]): + continue + hist = vol[i - regime_window:i] + hist = hist[~np.isnan(hist)] + if len(hist) < 10: + continue + p30 = np.percentile(hist, 30) + p70 = np.percentile(hist, 70) + if vol[i] < p30: + thresh[i] = high_thr # vol bassa → soglia alta + elif vol[i] > p70: + thresh[i] = low_thr # vol alta → soglia bassa + else: + thresh[i] = mid_thr + return thresh + + +def _detect_adaptive_squeezes(close, high, low, kcr, adaptive_thr, + min_dur: int = 5) -> list[dict]: + """Squeeze con threshold adattivo per ogni barra.""" + events = [] + in_sq = False + sq_start = 0 + for i in range(1, len(close)): + if np.isnan(kcr[i]) or np.isnan(adaptive_thr[i]): + continue + thr = adaptive_thr[i] + is_sq = kcr[i] < thr + if is_sq and not in_sq: + in_sq = True + sq_start = i + elif not is_sq and in_sq: + in_sq = False + dur = i - sq_start + if dur < min_dur: + continue + events.append({ + "idx": i, "dur": dur, "sq_start": sq_start, + "kcr_at_release": kcr[i], + "thr_used": adaptive_thr[i], + }) + return events + + +class AdaptiveSqueeze(Strategy): + name = "AD01_adaptive_squeeze" + description = "Squeeze con threshold adattivo a regime volatilità" + default_assets = ["BTC", "ETH"] + default_timeframes = ["15m", "1h"] + fee_rt = 0.002 + leverage = 3.0 + position_size = 0.15 + initial_capital = 1000.0 + + def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex, + **params) -> list[Signal]: + 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) + low_thr = params.get("low_thr", 0.65) + mid_thr = params.get("mid_thr", 0.80) + high_thr = params.get("high_thr", 0.90) + retrace_limit = params.get("retrace_limit", 0.6) + vol_mult = params.get("vol_multiplier", 1.3) + use_vol = params.get("use_vol", True) + vol_window = params.get("vol_window", 100) + regime_window = params.get("regime_window", 500) + + kcr = keltner_ratio(c, h, l, bb_w) + adaptive_thr = _adaptive_sq_threshold( + c, vol_window, regime_window, low_thr, mid_thr, high_thr + ) + events = _detect_adaptive_squeezes(c, h, l, kcr, adaptive_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 + + signals.append(Signal( + idx=i, + direction=direction, + entry_price=c[i - 1], + metadata={ + "dur": ev["dur"], + "thr_used": ev.get("thr_used", mid_thr), + }, + )) + return signals + + +if __name__ == "__main__": + strategy = AdaptiveSqueeze() + + configs = [ + # low_thr, mid_thr, high_thr, use_vol + {"low_thr": 0.65, "mid_thr": 0.80, "high_thr": 0.90, "use_vol": True}, + {"low_thr": 0.65, "mid_thr": 0.80, "high_thr": 0.90, "use_vol": False}, + {"low_thr": 0.60, "mid_thr": 0.78, "high_thr": 0.92, "use_vol": True}, + {"low_thr": 0.70, "mid_thr": 0.82, "high_thr": 0.90, "use_vol": True}, + {"low_thr": 0.65, "mid_thr": 0.80, "high_thr": 0.95, "use_vol": True}, + {"low_thr": 0.65, "mid_thr": 0.80, "high_thr": 0.90, + "use_vol": True, "vol_multiplier": 1.2}, + ] + + 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, **cfg) + if r and r.trades >= 20: + lbl = (f"AD01 lt={cfg['low_thr']} ht={cfg['high_thr']} " + f"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(" AD01 ADAPTIVE SQUEEZE THRESHOLD — 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%") diff --git a/scripts/strategies/CM01_cross_market_momentum.py b/scripts/strategies/CM01_cross_market_momentum.py new file mode 100644 index 0000000..46af5d8 --- /dev/null +++ b/scripts/strategies/CM01_cross_market_momentum.py @@ -0,0 +1,183 @@ +"""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%") diff --git a/scripts/strategies/PD01_price_volume_divergence.py b/scripts/strategies/PD01_price_volume_divergence.py new file mode 100644 index 0000000..63b8348 --- /dev/null +++ b/scripts/strategies/PD01_price_volume_divergence.py @@ -0,0 +1,158 @@ +"""PD01 — Price-Volume Divergence Squeeze. + +Estende SQ02 con volume TREND come filtro: + - Breakout UP con volume CRESCENTE (ultimi 3 bar vs media squeeze) → ENTRA + - Breakout UP con volume CALANTE → SALTA (divergenza bearish) + - Viceversa per short + +Logica anti-fakeout: + 1. Squeeze rilascio (come SQ02) + 2. Anti-fakeout candela (come SQ02) + 3. Volume al breakout > media squeeze (come SQ02) + 4. NUOVO: volume trending UP nelle ultime 3 barre prima del breakout + +Parametri semplici, nessun overfitting. +""" +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 PriceVolumeDivergence(Strategy): + name = "PD01_price_vol_div" + description = "Squeeze + antifakeout + volume trend confirmation" + default_assets = ["BTC", "ETH"] + default_timeframes = ["15m", "1h"] + fee_rt = 0.002 + leverage = 3.0 + position_size = 0.15 + initial_capital = 1000.0 + + def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex, + **params) -> list[Signal]: + 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) + retrace_limit = params.get("retrace_limit", 0.6) + vol_mult = params.get("vol_multiplier", 1.3) + vol_trend_bars = params.get("vol_trend_bars", 3) # barre per trend volume + + 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 < vol_trend_bars + 1 or i >= n: + continue + + # Direzione breakout + 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 al breakout > media squeeze + sq_start = ev["sq_start"] + avg_sq_v = np.mean(v[sq_start:i]) + if avg_sq_v <= 0 or v[i] <= avg_sq_v * vol_mult: + continue + + # Volume TREND: slope delle ultime vol_trend_bars barre + # Usa regressione lineare semplice (rank correlation del volume) + recent_v = v[i - vol_trend_bars:i + 1] # include breakout bar + if len(recent_v) < vol_trend_bars: + continue + # slope: media seconda metà vs prima metà + mid = len(recent_v) // 2 + v_early = np.mean(recent_v[:mid]) + v_late = np.mean(recent_v[mid:]) + vol_trending_up = v_late > v_early + vol_trending_down = v_early > v_late + + # Concordanza: long richiede volume trending up, short trending down + if direction == 1 and not vol_trending_up: + continue + if direction == -1 and not vol_trending_down: + continue + + signals.append(Signal( + idx=i, + direction=direction, + entry_price=c[i - 1], + metadata={ + "dur": ev["dur"], + "vol_ratio": v[i] / avg_sq_v if avg_sq_v > 0 else 0, + "vol_trend": v_late / v_early if v_early > 0 else 1, + }, + )) + return signals + + +if __name__ == "__main__": + strategy = PriceVolumeDivergence() + + configs = [ + {"bb_window": 14, "sq_threshold": 0.8, "retrace_limit": 0.6, + "vol_multiplier": 1.3, "vol_trend_bars": 3}, + {"bb_window": 14, "sq_threshold": 0.8, "retrace_limit": 0.6, + "vol_multiplier": 1.2, "vol_trend_bars": 3}, + {"bb_window": 14, "sq_threshold": 0.8, "retrace_limit": 0.6, + "vol_multiplier": 1.3, "vol_trend_bars": 5}, + {"bb_window": 14, "sq_threshold": 0.8, "retrace_limit": 0.5, + "vol_multiplier": 1.3, "vol_trend_bars": 3}, + {"bb_window": 14, "sq_threshold": 0.75, "retrace_limit": 0.6, + "vol_multiplier": 1.3, "vol_trend_bars": 3}, + {"bb_window": 20, "sq_threshold": 0.8, "retrace_limit": 0.6, + "vol_multiplier": 1.3, "vol_trend_bars": 3}, + ] + + 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, **cfg) + if r and r.trades >= 20: + lbl = (f"PD01 vtb={cfg['vol_trend_bars']} " + f"vm={cfg['vol_multiplier']} " + f"sq={cfg['sq_threshold']} 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(" PD01 PRICE-VOLUME DIVERGENCE — 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%")