"""MR01 — Bollinger Fade (mean-reversion). L'UNICA famiglia con edge netto reale dopo l'analisi out-of-sample fee-aware (vedi scripts/analysis/strategy_research.py). Contrario della tesi squeeze: i breakout RIENTRANO, quindi si fada l'estremo verso la media. Logica: 1. Bollinger Band (window n, k deviazioni) sul close 2. ENTRY: close esce sotto la banda inferiore -> long (o sopra la superiore -> short) 3. EXIT: take-profit alla media mobile (il rientro atteso), stop-loss a sl_atr*ATR oltre l'estremo, oppure time-limit max_bars 4. ingresso a close[i] (eseguibile dal vivo, nessun look-ahead) Validazione (netto, fee 0.10% RT reale Deribit, leva 3x, OOS = ultimo 30%): BTC 1h n=50 k=2.5: +201% OOS, DD 15%, ~tutti gli anni positivi ETH 1h n=50 k=2.0: +1238% OOS, DD 23% Robusto su TUTTA la griglia n in {14,20,30,50} x k in {2.0,2.5,3.0} e su tutte le fee 0.00-0.20% RT (margine di sicurezza ampio). NOTA LIVE: usa TP alla media + SL ad ATR + max_bars. Lo StrategyWorker attuale esce solo a hold_bars/stop -2% fisso: per tradarla come validata il worker deve supportare gli exit TP/SL passati in metadata (vedi metadata di ogni Signal). """ 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.data.downloader import load_data from src.strategies.fade_base import hurst_skip_mask def _atr(df: pd.DataFrame, n: int = 14) -> np.ndarray: h, l, c = df["high"].values, df["low"].values, df["close"].values pc = np.roll(c, 1); pc[0] = c[0] tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc))) return pd.Series(tr).rolling(n).mean().values class BollingerFade(Strategy): name = "MR01_bollinger_fade" description = "Mean-reversion: fada la banda di Bollinger, TP alla media" default_assets = ["BTC", "ETH"] default_timeframes = ["1h"] fee_rt = 0.001 # Deribit perp realistico (taker 0.05%/lato) 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 n_len = len(c) bb_w = params.get("bb_window", 50) k = params.get("k", 2.5) sl_atr = params.get("sl_atr", 2.0) max_bars = params.get("max_bars", 24) trend_max = params.get("trend_max") # None = filtro disattivo ema_long = params.get("ema_long", 200) # Edge minimo: salta i segnali il cui TP (la media) è più vicino dell'entry del # costo round-trip -> perdenti garantiti anche colpendo il TP. 0 = off. min_tp_frac = params.get("min_tp_frac", 0.0) # Loss-guard Hurst: salta in regime persistente/trending (hurst >= soglia). None = off. hurst_max = params.get("hurst_max") ma = pd.Series(c).rolling(bb_w).mean().values sd = pd.Series(c).rolling(bb_w).std().values a = _atr(df, 14) up, lo = ma + k * sd, ma - k * sd el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values if trend_max is not None else None skip = hurst_skip_mask(df, hurst_max, params.get("hurst_win", 100)) signals: list[Signal] = [] for i in range(bb_w + 14, n_len): if np.isnan(up[i]) or np.isnan(a[i]): continue if skip[i]: continue # loss-guard: regime persistente if el is not None and (a[i] == 0 or np.isnan(el[i]) or abs(c[i] - el[i]) / a[i] > trend_max): continue if c[i] < lo[i] and c[i - 1] >= lo[i - 1]: d, sl = 1, c[i] - sl_atr * a[i] elif c[i] > up[i] and c[i - 1] <= up[i - 1]: d, sl = -1, c[i] + sl_atr * a[i] else: continue if min_tp_frac > 0 and abs(ma[i] - c[i]) / c[i] <= min_tp_frac: continue # TP entro le fee -> non eseguibile in utile signals.append(Signal( idx=i, direction=d, entry_price=c[i], metadata={"tp": float(ma[i]), "sl": float(sl), "max_bars": max_bars}, )) return signals def backtest(self, asset: str, tf: str = "1h", hold: int = 3, **params) -> BacktestResult | None: """Backtest fedele: TP alla media / SL ad ATR / time-limit, fee+leva nette.""" df = load_data(asset, tf) ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) signals = self.generate_signals(df, ts, **params) if not signals: return None h, l, c = df["high"].values, df["low"].values, df["close"].values n = len(c) # EXIT-16 close-confirm SL (2026-06-04): come in fade_base.FadeStrategy.backtest. # None = comportamento storico. Vedi docs/diary/2026-06-04-exit-lab.md. sl_confirm = params.get("sl_confirm_atr") a14 = _atr(df, 14) if sl_confirm else None fee = self.fee_rt * self.leverage capital = peak = float(self.initial_capital) max_dd = 0.0 total_bars = 0 last_exit = -1 yearly: dict[int, dict] = {} for sig in signals: i, d = sig.idx, sig.direction if i <= last_exit or i + 1 >= n: continue entry = c[i] tp, sl, mb = sig.metadata["tp"], sig.metadata["sl"], sig.metadata["max_bars"] exit_p = c[min(i + mb, n - 1)] j = min(i + mb, n - 1) for step in range(1, mb + 1): j = i + step if j >= n: j = n - 1; exit_p = c[j]; break hit_tp = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp) if sl_confirm is None: hit_sl = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl) if hit_sl: exit_p = sl; break if hit_tp: exit_p = tp; break else: # close-confirm: TP intrabar al livello; SL valutato sul CLOSE if hit_tp: exit_p = tp; break buf = sl_confirm * a14[j] if (d == 1 and c[j] < sl - buf) or (d == -1 and c[j] > sl + buf): exit_p = c[j]; break if step == mb: exit_p = c[j] ret = (exit_p - entry) / entry * d * self.leverage - fee capital = max(capital + capital * self.position_size * ret, 10.0) if capital > peak: peak = capital max_dd = max(max_dd, (peak - capital) / peak) total_bars += (j - i) last_exit = j year = ts.iloc[i].year yr = yearly.setdefault(year, {"w": 0, "t": 0, "pnl": 0.0}) yr["t"] += 1 if ret > 0: yr["w"] += 1 yr["pnl"] += ret * self.initial_capital all_t = sum(v["t"] for v in yearly.values()) all_w = sum(v["w"] for v in yearly.values()) if all_t == 0: return None yearly_stats = [YearlyStats(y, v["t"], v["w"], v["pnl"]) for y, v in sorted(yearly.items())] return BacktestResult( strategy_name=self.name, asset=asset, timeframe=tf, params=params, trades=all_t, wins=all_w, pnl=sum(v["pnl"] for v in yearly.values()), capital=capital, initial_capital=self.initial_capital, max_dd=max_dd * 100, time_in_market_pct=total_bars / n * 100, avg_trade_duration_h=total_bars / all_t * TF_MINUTES.get(tf, 60) / 60, years_active=len(yearly), yearly=yearly_stats, ) if __name__ == "__main__": strat = BollingerFade() print(f"{'=' * 110}") print(f" MR01 BOLLINGER FADE — netto fee {strat.fee_rt*100:.2f}% RT, leva {strat.leverage:.0f}x") print(f"{'=' * 110}") results = [] for asset in ["BTC", "ETH"]: for k in [2.0, 2.5]: r = strat.backtest(asset, "1h", bb_window=50, k=k, sl_atr=2.0, max_bars=24) if r: r.strategy_name = f"MR01 {asset} 1h n50 k{k}" results.append(r) for r in results: r.print_summary() if results: results[0].print_yearly()