""" Moteur de backtesting vectorisé. Simule l'exécution de la stratégie sur des données historiques : - Rolling window : analyse les N dernières bougies à chaque pas - Gestion du spread OANDA - Suivi des positions ouvertes (SL / TP hit) - Calcul des métriques de performance """ from dataclasses import dataclass, field from datetime import datetime from typing import Optional import numpy as np import pandas as pd from app.core.strategy.base import AbstractStrategy, TradeSignal @dataclass class BacktestTrade: direction: str entry_price: float stop_loss: float take_profit: float entry_time: datetime exit_price: Optional[float] = None exit_time: Optional[datetime] = None pnl_pips: Optional[float] = None pnl_pct: Optional[float] = None status: str = "open" # "win" | "loss" | "open" signal_type: str = "" @dataclass class BacktestMetrics: initial_balance: float final_balance: float total_pnl: float total_pnl_pct: float total_trades: int winning_trades: int losing_trades: int win_rate: float avg_win_pips: float avg_loss_pips: float expectancy: float max_drawdown: float max_drawdown_pct: float sharpe_ratio: float profit_factor: float equity_curve: list[dict] = field(default_factory=list) trades: list[BacktestTrade] = field(default_factory=list) class Backtester: def __init__( self, strategy: AbstractStrategy, initial_balance: float = 10_000.0, risk_percent: float = 1.0, spread_pips: float = 1.5, pip_value: float = 0.0001, ) -> None: self.strategy = strategy self.initial_balance = initial_balance self.risk_percent = risk_percent # % du capital risqué par trade self.spread_pips = spread_pips self.pip_value = pip_value def run( self, df: pd.DataFrame, window: int = 100, ) -> BacktestMetrics: """ Exécute le backtest sur le DataFrame complet. window : nombre de bougies analysées à chaque pas (rolling) """ df = df.copy().reset_index(drop=True) trades: list[BacktestTrade] = [] balance = self.initial_balance equity_curve: list[dict] = [{"time": str(df.iloc[0]["time"]), "balance": balance}] open_trade: Optional[BacktestTrade] = None last_signal_time: Optional[datetime] = None used_signals: set[tuple[str, int]] = set() # (direction, entry_price_rounded) for i in range(window, len(df)): candle = df.iloc[i] # 1. Gérer la position ouverte (SL/TP hit ?) if open_trade: closed, exit_price = self._check_exit(open_trade, candle) if closed: pnl_pips = self._calc_pnl_pips(open_trade, exit_price) pnl_money = self._pips_to_money(pnl_pips, balance) balance += pnl_money open_trade.exit_price = exit_price open_trade.exit_time = candle["time"] open_trade.pnl_pips = pnl_pips open_trade.pnl_pct = pnl_money / self.initial_balance * 100 open_trade.status = "win" if pnl_pips > 0 else "loss" trades.append(open_trade) open_trade = None equity_curve.append({"time": str(candle["time"]), "balance": balance}) # 2. Si pas de position ouverte, chercher un signal if not open_trade: slice_df = df.iloc[i - window:i + 1] result = self.strategy.analyze(slice_df) if result.signals: # Filtrer les signaux déjà exploités (éviter doublons) new_signals = [ s for s in result.signals if last_signal_time is None or s.time > last_signal_time ] if not new_signals: continue signal = new_signals[-1] # Signal le plus récent # Appliquer le spread entry, sl, tp = self._apply_spread(signal) # Éviter de réouvrir le même trade (même direction + même prix) sig_key = (signal.direction, round(entry / self.pip_value)) if sig_key in used_signals: continue used_signals.add(sig_key) last_signal_time = signal.time open_trade = BacktestTrade( direction=signal.direction, entry_price=entry, stop_loss=sl, take_profit=tp, entry_time=candle["time"], signal_type=signal.signal_type, ) # Fermer la position ouverte en fin de période if open_trade: last_price = df.iloc[-1]["close"] pnl_pips = self._calc_pnl_pips(open_trade, last_price) pnl_money = self._pips_to_money(pnl_pips, balance) balance += pnl_money open_trade.exit_price = last_price open_trade.exit_time = df.iloc[-1]["time"] open_trade.pnl_pips = pnl_pips open_trade.pnl_pct = pnl_money / self.initial_balance * 100 open_trade.status = "win" if pnl_pips > 0 else "loss" trades.append(open_trade) equity_curve.append({"time": str(df.iloc[-1]["time"]), "balance": balance}) return self._compute_metrics(trades, balance, equity_curve) # ─── Helpers ────────────────────────────────────────────────────────────── def _check_exit( self, trade: BacktestTrade, candle: pd.Series ) -> tuple[bool, float]: """Vérifie si le SL ou TP est touché sur la bougie. Retourne (closed, exit_price).""" if trade.direction == "buy": if candle["low"] <= trade.stop_loss: return True, trade.stop_loss if candle["high"] >= trade.take_profit: return True, trade.take_profit else: if candle["high"] >= trade.stop_loss: return True, trade.stop_loss if candle["low"] <= trade.take_profit: return True, trade.take_profit return False, 0.0 def _apply_spread(self, signal: TradeSignal) -> tuple[float, float, float]: """Applique le spread à l'entrée.""" half_spread = (self.spread_pips * self.pip_value) / 2 if signal.direction == "buy": entry = signal.entry_price + half_spread sl = signal.stop_loss tp = signal.take_profit else: entry = signal.entry_price - half_spread sl = signal.stop_loss tp = signal.take_profit return entry, sl, tp def _calc_pnl_pips(self, trade: BacktestTrade, exit_price: float) -> float: if trade.direction == "buy": return (exit_price - trade.entry_price) / self.pip_value return (trade.entry_price - exit_price) / self.pip_value def _pips_to_money(self, pips: float, balance: float) -> float: """Convertit les pips en unités monétaires selon le risque défini.""" risk_amount = balance * (self.risk_percent / 100) # Nombre de pips de risque sur ce trade return pips * (risk_amount / 10) # Approximation simple def _compute_metrics( self, trades: list[BacktestTrade], final_balance: float, equity_curve: list[dict], ) -> BacktestMetrics: if not trades: return BacktestMetrics( initial_balance=self.initial_balance, final_balance=final_balance, total_pnl=0, total_pnl_pct=0, total_trades=0, winning_trades=0, losing_trades=0, win_rate=0, avg_win_pips=0, avg_loss_pips=0, expectancy=0, max_drawdown=0, max_drawdown_pct=0, sharpe_ratio=0, profit_factor=0, equity_curve=equity_curve, trades=trades, ) winners = [t for t in trades if t.status == "win"] losers = [t for t in trades if t.status == "loss"] total_pnl = final_balance - self.initial_balance win_rate = len(winners) / len(trades) if trades else 0 avg_win = np.mean([t.pnl_pips for t in winners]) if winners else 0 avg_loss = abs(np.mean([t.pnl_pips for t in losers])) if losers else 0 expectancy = (win_rate * avg_win) - ((1 - win_rate) * avg_loss) # Max Drawdown balances = [eq["balance"] for eq in equity_curve] peak = balances[0] max_dd = 0.0 for b in balances: if b > peak: peak = b dd = (peak - b) / peak if dd > max_dd: max_dd = dd # Profit Factor gross_profit = sum(t.pnl_pips for t in winners) if winners else 0 gross_loss = abs(sum(t.pnl_pips for t in losers)) if losers else 1 pf = gross_profit / gross_loss if gross_loss > 0 else 0 # Sharpe Ratio (simplifié — daily returns) pnl_series = [t.pnl_pips or 0 for t in trades] if len(pnl_series) > 1: mean_ret = np.mean(pnl_series) std_ret = np.std(pnl_series) sharpe = (mean_ret / std_ret * np.sqrt(252)) if std_ret > 0 else 0 else: sharpe = 0 return BacktestMetrics( initial_balance=self.initial_balance, final_balance=final_balance, total_pnl=total_pnl, total_pnl_pct=total_pnl / self.initial_balance * 100, total_trades=len(trades), winning_trades=len(winners), losing_trades=len(losers), win_rate=win_rate * 100, avg_win_pips=float(avg_win), avg_loss_pips=float(avg_loss), expectancy=float(expectancy), max_drawdown=max_dd * self.initial_balance, max_drawdown_pct=max_dd * 100, sharpe_ratio=float(sharpe), profit_factor=float(pf), equity_curve=equity_curve, trades=trades, )