""" Backtester pour stratégies de trading degen Simule des trades avec leverage extrême et génère des métriques """ import pandas as pd import numpy as np from typing import List, Dict, Optional from dataclasses import dataclass, asdict import logging from datetime import datetime import json import sys sys.path.append('..') import config from signals.short_term_signals import ShortTermSignalGenerator, Signal @dataclass class Trade: """Trade exécuté""" entry_time: pd.Timestamp exit_time: pd.Timestamp symbol: str side: str strategy: str entry_price: float exit_price: float stop_loss: float take_profit: float quantity: float leverage: int pnl_pct: float pnl_usd: float exit_reason: str # 'TP', 'SL', 'TIME' max_favorable_excursion: float # MFE - meilleur moment max_adverse_excursion: float # MAE - pire moment duration_minutes: int @dataclass class BacktestResult: """Résultat d'un backtest""" strategy_name: str symbol: str timeframe: str start_date: str end_date: str initial_capital: float final_capital: float total_return_pct: float total_trades: int winning_trades: int losing_trades: int win_rate: float avg_win_pct: float avg_loss_pct: float best_trade_pct: float worst_trade_pct: float profit_factor: float max_drawdown_pct: float sharpe_ratio: float trades: List[Trade] class DegenBacktester: """Backtester avec simulation de leverage extrême""" def __init__(self, initial_capital: float = 1000.0, leverage: int = 250): self.logger = logging.getLogger(self.__class__.__name__) self.initial_capital = initial_capital self.leverage = leverage # Paramètres de simulation self.trading_fee = config.TRADING_FEE_TAKER # 0.05% self.slippage = config.SLIPPAGE_PCT # 0.05% # Signal generator self.signal_generator = ShortTermSignalGenerator() self.logger.info(f"✅ Backtester initialisé: ${initial_capital} capital, {leverage}x leverage") def simulate_trade(self, signal: Signal, df: pd.DataFrame, start_idx: int) -> Optional[Trade]: """ Simule un trade depuis un signal Args: signal: Signal d'entrée df: DataFrame avec données OHLCV start_idx: Index de départ dans df Returns: Trade si exécuté, None sinon """ entry_time = df.iloc[start_idx]['timestamp'] entry_price = signal.entry_price stop_loss = signal.stop_loss take_profit = signal.take_profit # Appliquer slippage sur entrée if signal.side == 'LONG': entry_price *= (1 + self.slippage) else: entry_price *= (1 - self.slippage) # Calculer la position size # Pour leverage 250x: on risque 0.3% du capital par trade risk_pct = config.STOP_LOSS_PCT[self.leverage] position_size_usd = self.initial_capital * risk_pct * self.leverage quantity = position_size_usd / entry_price # Frais d'entrée entry_fee = position_size_usd * self.trading_fee # Parcourir les chandeliers suivants jusqu'à sortie max_favorable_excursion = 0.0 max_adverse_excursion = 0.0 for i in range(start_idx + 1, len(df)): current = df.iloc[i] high = current['high'] low = current['low'] close = current['close'] # Calculer excursions if signal.side == 'LONG': favorable_move = (high - entry_price) / entry_price adverse_move = (entry_price - low) / entry_price max_favorable_excursion = max(max_favorable_excursion, favorable_move) max_adverse_excursion = max(max_adverse_excursion, adverse_move) # Check stop-loss (liquidation) if low <= stop_loss: exit_price = stop_loss * (1 - self.slippage) # Slippage défavorable exit_reason = 'SL' exit_time = current['timestamp'] break # Check take-profit if high >= take_profit: exit_price = take_profit * (1 - self.slippage) exit_reason = 'TP' exit_time = current['timestamp'] break else: # SHORT favorable_move = (entry_price - low) / entry_price adverse_move = (high - entry_price) / entry_price max_favorable_excursion = max(max_favorable_excursion, favorable_move) max_adverse_excursion = max(max_adverse_excursion, adverse_move) # Check stop-loss if high >= stop_loss: exit_price = stop_loss * (1 + self.slippage) exit_reason = 'SL' exit_time = current['timestamp'] break # Check take-profit if low <= take_profit: exit_price = take_profit * (1 + self.slippage) exit_reason = 'TP' exit_time = current['timestamp'] break # Timeout après 100 chandeliers (100min sur 1m timeframe) if i - start_idx >= 100: exit_price = close exit_reason = 'TIME' exit_time = current['timestamp'] break else: # Fin des données - sortie au dernier prix exit_price = df.iloc[-1]['close'] exit_reason = 'EOD' exit_time = df.iloc[-1]['timestamp'] # Calculer PnL if signal.side == 'LONG': pnl_pct = (exit_price - entry_price) / entry_price else: pnl_pct = (entry_price - exit_price) / entry_price # PnL en USD avec leverage pnl_usd = position_size_usd * pnl_pct # Frais de sortie exit_fee = position_size_usd * self.trading_fee # PnL net pnl_usd -= (entry_fee + exit_fee) # Duration duration_minutes = int((exit_time - entry_time).total_seconds() / 60) return Trade( entry_time=entry_time, exit_time=exit_time, symbol=signal.symbol, side=signal.side, strategy=signal.strategy, entry_price=entry_price, exit_price=exit_price, stop_loss=stop_loss, take_profit=take_profit, quantity=quantity, leverage=self.leverage, pnl_pct=pnl_pct * 100, # En % pnl_usd=pnl_usd, exit_reason=exit_reason, max_favorable_excursion=max_favorable_excursion * 100, max_adverse_excursion=max_adverse_excursion * 100, duration_minutes=duration_minutes ) def backtest_strategy(self, df: pd.DataFrame, symbol: str, strategy_name: Optional[str] = None) -> BacktestResult: """ Backtest une stratégie sur des données historiques Args: df: DataFrame OHLCV symbol: Symbole testé strategy_name: Si spécifié, teste seulement cette stratégie """ self.logger.info(f"\n{'='*60}") self.logger.info(f"BACKTEST: {symbol} | {len(df)} chandeliers") self.logger.info(f"{'='*60}") capital = self.initial_capital trades = [] equity_curve = [capital] # Parcourir les données i = 50 # Commencer après 50 périodes (warmup pour indicateurs) while i < len(df) - 100: # Garder 100 périodes pour simuler le trade # Extraire fenêtre pour analyse window = df.iloc[:i+1].copy() # Générer signaux signals = self.signal_generator.generate_signals(window, symbol) # Filtrer par stratégie si demandé if strategy_name: signals = [s for s in signals if s.strategy == strategy_name] if not signals: i += 1 continue # Prendre le meilleur signal best_signal = max(signals, key=lambda x: x.score) # Simuler le trade trade = self.simulate_trade(best_signal, df, i) if trade: trades.append(trade) capital += trade.pnl_usd equity_curve.append(capital) self.logger.debug( f"Trade #{len(trades)}: {trade.side} {trade.exit_reason} " f"PnL: {trade.pnl_pct:+.2f}% (${trade.pnl_usd:+.2f}) | " f"Capital: ${capital:.2f}" ) # Avancer après la sortie du trade exit_idx = df[df['timestamp'] == trade.exit_time].index[0] i = exit_idx + 1 else: i += 1 # Calculer métriques if not trades: self.logger.warning("⚠️ Aucun trade généré") return None winning_trades = [t for t in trades if t.pnl_usd > 0] losing_trades = [t for t in trades if t.pnl_usd <= 0] total_wins = sum(t.pnl_usd for t in winning_trades) total_losses = abs(sum(t.pnl_usd for t in losing_trades)) win_rate = len(winning_trades) / len(trades) if trades else 0 avg_win_pct = np.mean([t.pnl_pct for t in winning_trades]) if winning_trades else 0 avg_loss_pct = np.mean([t.pnl_pct for t in losing_trades]) if losing_trades else 0 best_trade = max(trades, key=lambda t: t.pnl_pct) worst_trade = min(trades, key=lambda t: t.pnl_pct) profit_factor = total_wins / total_losses if total_losses > 0 else float('inf') # Max drawdown equity_series = pd.Series(equity_curve) running_max = equity_series.expanding().max() drawdown = (equity_series - running_max) / running_max * 100 max_drawdown = drawdown.min() # Sharpe ratio (simplifié - annualisé) returns = equity_series.pct_change().dropna() if len(returns) > 1 and returns.std() != 0: sharpe = (returns.mean() / returns.std()) * np.sqrt(365 * 24 * 60) # Annualisé pour 1min else: sharpe = 0.0 total_return = (capital - self.initial_capital) / self.initial_capital * 100 result = BacktestResult( strategy_name=strategy_name or 'ALL', symbol=symbol, timeframe=config.TIMEFRAME_PRIMARY, start_date=str(df.iloc[0]['timestamp']), end_date=str(df.iloc[-1]['timestamp']), initial_capital=self.initial_capital, final_capital=capital, total_return_pct=total_return, total_trades=len(trades), winning_trades=len(winning_trades), losing_trades=len(losing_trades), win_rate=win_rate, avg_win_pct=avg_win_pct, avg_loss_pct=avg_loss_pct, best_trade_pct=best_trade.pnl_pct, worst_trade_pct=worst_trade.pnl_pct, profit_factor=profit_factor, max_drawdown_pct=max_drawdown, sharpe_ratio=sharpe, trades=trades ) self.print_result(result) return result def compare_strategies(self, df: pd.DataFrame, symbol: str) -> Dict[str, BacktestResult]: """Compare toutes les stratégies""" strategies = [s.name for s in self.signal_generator.strategies] results = {} for strategy_name in strategies: self.logger.info(f"\n🧪 Test de stratégie: {strategy_name}") result = self.backtest_strategy(df, symbol, strategy_name=strategy_name) if result: results[strategy_name] = result # Afficher comparaison self.print_comparison(results) return results def print_result(self, result: BacktestResult): """Affiche les résultats d'un backtest""" self.logger.info(f"\n{'='*60}") self.logger.info(f"📊 RÉSULTATS: {result.strategy_name} - {result.symbol}") self.logger.info(f"{'='*60}") self.logger.info(f"Capital initial: ${result.initial_capital:,.2f}") self.logger.info(f"Capital final: ${result.final_capital:,.2f}") self.logger.info(f"Retour total: {result.total_return_pct:+.2f}%") self.logger.info(f"\n📈 TRADES:") self.logger.info(f"Total: {result.total_trades}") self.logger.info(f"Gagnants: {result.winning_trades} ({result.win_rate*100:.1f}%)") self.logger.info(f"Perdants: {result.losing_trades}") self.logger.info(f"\n💰 PERFORMANCE:") self.logger.info(f"Gain moyen: {result.avg_win_pct:+.2f}%") self.logger.info(f"Perte moyenne: {result.avg_loss_pct:+.2f}%") self.logger.info(f"Meilleur trade: {result.best_trade_pct:+.2f}%") self.logger.info(f"Pire trade: {result.worst_trade_pct:+.2f}%") self.logger.info(f"Profit factor: {result.profit_factor:.2f}") self.logger.info(f"\n📉 RISQUE:") self.logger.info(f"Max drawdown: {result.max_drawdown_pct:.2f}%") self.logger.info(f"Sharpe ratio: {result.sharpe_ratio:.2f}") self.logger.info(f"{'='*60}\n") def print_comparison(self, results: Dict[str, BacktestResult]): """Affiche la comparaison des stratégies""" self.logger.info(f"\n{'='*80}") self.logger.info("🏆 COMPARAISON DES STRATÉGIES") self.logger.info(f"{'='*80}") # Tableau comparatif headers = ['Stratégie', 'Return %', 'Trades', 'Win Rate', 'PF', 'Sharpe', 'Max DD'] self.logger.info(f"{headers[0]:<30} {headers[1]:>10} {headers[2]:>8} {headers[3]:>10} {headers[4]:>6} {headers[5]:>8} {headers[6]:>10}") self.logger.info("-" * 80) sorted_results = sorted(results.items(), key=lambda x: x[1].total_return_pct, reverse=True) for strategy_name, result in sorted_results: self.logger.info( f"{strategy_name:<30} " f"{result.total_return_pct:>9.2f}% " f"{result.total_trades:>8} " f"{result.win_rate*100:>9.1f}% " f"{result.profit_factor:>6.2f} " f"{result.sharpe_ratio:>8.2f} " f"{result.max_drawdown_pct:>9.2f}%" ) # Identifier la meilleure best_strategy = sorted_results[0][0] best_return = sorted_results[0][1].total_return_pct self.logger.info(f"\n🥇 MEILLEURE STRATÉGIE: {best_strategy} ({best_return:+.2f}%)") self.logger.info(f"{'='*80}\n") def save_results(self, results: Dict[str, BacktestResult], filename: str = 'backtest_results.json'): """Sauvegarde les résultats en JSON""" output = {} for strategy_name, result in results.items(): output[strategy_name] = { **asdict(result), 'trades': [asdict(t) for t in result.trades] } # Convertir timestamps en strings def convert_timestamps(obj): if isinstance(obj, dict): return {k: convert_timestamps(v) for k, v in obj.items()} elif isinstance(obj, list): return [convert_timestamps(item) for item in obj] elif isinstance(obj, pd.Timestamp): return str(obj) return obj output = convert_timestamps(output) with open(filename, 'w') as f: json.dump(output, f, indent=2) self.logger.info(f"✅ Résultats sauvegardés: {filename}") if __name__ == '__main__': # Test basique avec données simulées logging.basicConfig(level=logging.INFO) # Créer données de test (simulation d'un mini-pump) dates = pd.date_range('2025-01-01', periods=500, freq='1min') base_price = 100 trend = np.linspace(0, 10, 500) # Trend haussier noise = np.random.randn(500) * 0.5 closes = base_price + trend + noise df = pd.DataFrame({ 'timestamp': dates, 'open': closes + np.random.randn(500) * 0.1, 'high': closes + abs(np.random.randn(500)) * 0.3, 'low': closes - abs(np.random.randn(500)) * 0.3, 'close': closes, 'volume': np.random.randint(1000, 10000, 500) }) print("📊 Test du backtester avec données simulées...\n") backtester = DegenBacktester(initial_capital=1000, leverage=250) results = backtester.compare_strategies(df, 'TESTUSDT') backtester.save_results(results, 'test_backtest_results.json')