feat(strategie): 3 nuove fade mean-reversion validate OOS fee-aware (MR02/MR03/MR07)

Trovate e promosse 3 strategie con edge netto distinto da MR01, stessa metodologia (ingresso close[i], netto fee 0.10% RT + leva 3x, OOS ultimo 30%, robustezza su griglia + sweep fee 0.00-0.20%): - MR02 Donchian Fade: fade rottura canale H/L, TP al centro. BTC +172% OOS. - MR03 Keltner Fade: canale ATR su EMA (indipendente da Bollinger). BTC +112%. - MR07 Return Reversal: fade movimento di barra estremo (z dei rendimenti). BTC +105%. Tutte positive netto OOS su entrambi gli asset e su tutto lo sweep fee, anche 0.20% RT pessimista (validate anche via oos_validation live-path). Scartate MR04 (= MR01 riparametrizzato), MR05 (ADX non robusto), MR06 (RSI2 ETH neg). Base condivisa fade_base.FadeStrategy (backtest intrabar TP/SL/max_bars). Aggiunte a strategy_loader e strategies.yml (BTC+ETH 1h). Ricerca in strategy_research_v2.py. Diario e CLAUDE.md aggiornati. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-28 23:26:21 +02:00
parent 48435f6858
commit 21d3ba609d
10 changed files with 788 additions and 8 deletions
@@ -0,0 +1,101 @@
+"""Base condivisa per strategie mean-reversion con exit TP/SL/max_bars.
+
+Tutte le strategie fade (MR02/MR03/MR07) generano Signal con metadata
+{tp, sl, max_bars} e usano lo stesso backtest fedele: ingresso a close[i]
+(eseguibile dal vivo), uscita su take-profit / stop-loss intrabar (high/low)
+o time-limit, una posizione per volta (non-overlap), capitale composto,
+fee+leva nette. Identico all'engine di scripts/analysis/strategy_research.py.
+
+Le sottoclassi implementano solo generate_signals().
+"""
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+from src.strategies.base import Strategy, BacktestResult, YearlyStats, TF_MINUTES
+from src.data.downloader import load_data
+
+
+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 FadeStrategy(Strategy):
+    """Strategy con backtest intrabar TP/SL/max_bars (exit guidati dai metadata)."""
+
+    fee_rt = 0.001          # Deribit perp realistico (taker 0.05%/lato)
+    leverage = 3.0
+    position_size = 0.15
+    initial_capital = 1000.0
+
+    def backtest(self, asset: str, tf: str = "1h", hold: int = 3,
+                 **params) -> BacktestResult | None:
+        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)
+        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_sl = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
+                hit_tp = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
+                if hit_sl:                          # conservativo: SL prima del TP nello stesso bar
+                    exit_p = sl; break
+                if hit_tp:
+                    exit_p = tp; 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,
+        )