#!/usr/bin/env python3 """ Backtest de la stratégie Degen Scalping Teste la stratégie sur données historiques pour évaluer les performances """ import pandas as pd import numpy as np from datetime import datetime, timedelta from degen_scalping_strategy import DegenScalpingStrategy from price_feed import BinancePriceFeed import logging logging.basicConfig( level=logging.INFO, format='%(message)s' ) logger = logging.getLogger(__name__) class DegenScalpingBacktest: """Backtester pour la stratégie degen scalping""" def __init__(self, strategy: DegenScalpingStrategy, initial_capital: float = 100.0): """ Args: strategy: Stratégie à tester initial_capital: Capital initial en USDT """ self.strategy = strategy self.initial_capital = initial_capital self.capital = initial_capital # Résultats self.trades = [] self.equity_curve = [] def run(self, df: pd.DataFrame, symbol: str = "BTCUSD_DEGEN"): """ Lance le backtest Args: df: DataFrame avec OHLCV data symbol: Symbole du token Returns: Dict avec les résultats """ logger.info("="*80) logger.info("🧪 BACKTEST DEGEN SCALPING STRATEGY") logger.info("="*80) logger.info(f"Période: {df['timestamp'].iloc[0]} → {df['timestamp'].iloc[-1]}") logger.info(f"Bougies: {len(df)}") logger.info(f"Capital initial: ${self.initial_capital:.2f}") logger.info("="*80) current_position = None signals_checked = 0 signals_generated = 0 # Boucle principale for i in range(100, len(df)): # Start at 100 for indicators # Get data window window = df.iloc[max(0, i-100):i+1].copy() current_price = float(window['close'].iloc[-1]) current_time = window['timestamp'].iloc[-1] signals_checked += 1 # Si position ouverte, vérifier sortie if current_position: should_exit, exit_reason = self._check_exit( current_position, current_price ) if should_exit: self._close_trade( current_position, current_price, current_time, exit_reason ) current_position = None continue # Ne cherche pas nouveau signal si en position # Sinon, chercher signal signal = self.strategy.analyze(symbol, window) if signal: signals_generated += 1 # Ouvrir position position_size = self.capital * 0.95 # 95% du capital (garde 5% marge) margin = position_size # Pour ApolloX, margin = position size current_position = { 'signal': signal, 'entry_price': current_price, 'entry_time': current_time, 'margin': margin, 'leverage': signal.leverage, 'side': signal.side } # Fermer position finale si ouverte if current_position: final_price = float(df['close'].iloc[-1]) final_time = df['timestamp'].iloc[-1] self._close_trade(current_position, final_price, final_time, "END_OF_BACKTEST") # Calculer résultats results = self._calculate_results(signals_checked, signals_generated) return results def _check_exit(self, position, current_price: float) -> tuple: """Vérifie si doit sortir de la position""" signal = position['signal'] # Vérifier TP/SL if signal.side == 'LONG': if current_price >= signal.take_profit: return True, "TP_HIT" elif current_price <= signal.stop_loss: return True, "SL_HIT" else: # SHORT if current_price <= signal.take_profit: return True, "TP_HIT" elif current_price >= signal.stop_loss: return True, "SL_HIT" return False, None def _close_trade(self, position, exit_price: float, exit_time, reason: str): """Ferme un trade et enregistre les résultats""" signal = position['signal'] entry_price = position['entry_price'] margin = position['margin'] leverage = position['leverage'] # Calculer PnL if signal.side == 'LONG': price_change_pct = (exit_price - entry_price) / entry_price else: # SHORT price_change_pct = (entry_price - exit_price) / entry_price # PnL en % du margin avec leverage pnl_pct = price_change_pct * leverage pnl_usdt = margin * pnl_pct # Fees (environ 0.3% total - 0.15% open + 0.15% close) fees = margin * 0.003 # PnL net pnl_net = pnl_usdt - fees # Update capital self.capital += pnl_net # Enregistrer trade trade = { 'entry_time': position['entry_time'], 'exit_time': exit_time, 'side': signal.side, 'entry_price': entry_price, 'exit_price': exit_price, 'leverage': leverage, 'margin': margin, 'pnl_pct': pnl_pct * 100, # En % 'pnl_usdt': pnl_net, 'reason': reason, 'capital_after': self.capital } self.trades.append(trade) self.equity_curve.append({ 'time': exit_time, 'capital': self.capital }) def _calculate_results(self, signals_checked: int, signals_generated: int) -> dict: """Calcule les métriques de performance""" if len(self.trades) == 0: logger.warning("⚠️ Aucun trade effectué pendant le backtest") return {} # Convertir en DataFrame df_trades = pd.DataFrame(self.trades) # Métriques de base total_trades = len(df_trades) wins = len(df_trades[df_trades['pnl_usdt'] > 0]) losses = len(df_trades[df_trades['pnl_usdt'] <= 0]) win_rate = wins / total_trades * 100 if total_trades > 0 else 0 # PnL total_pnl = df_trades['pnl_usdt'].sum() total_pnl_pct = (self.capital - self.initial_capital) / self.initial_capital * 100 avg_win = df_trades[df_trades['pnl_usdt'] > 0]['pnl_usdt'].mean() if wins > 0 else 0 avg_loss = df_trades[df_trades['pnl_usdt'] <= 0]['pnl_usdt'].mean() if losses > 0 else 0 # Profit factor gross_profit = df_trades[df_trades['pnl_usdt'] > 0]['pnl_usdt'].sum() if wins > 0 else 0 gross_loss = abs(df_trades[df_trades['pnl_usdt'] <= 0]['pnl_usdt'].sum()) if losses > 0 else 1 profit_factor = gross_profit / gross_loss if gross_loss > 0 else 0 # Drawdown df_equity = pd.DataFrame(self.equity_curve) df_equity['peak'] = df_equity['capital'].cummax() df_equity['drawdown'] = (df_equity['capital'] - df_equity['peak']) / df_equity['peak'] * 100 max_drawdown = df_equity['drawdown'].min() # Distribution PnL pnl_std = df_trades['pnl_pct'].std() pnl_mean = df_trades['pnl_pct'].mean() results = { 'signals_checked': signals_checked, 'signals_generated': signals_generated, 'total_trades': total_trades, 'wins': wins, 'losses': losses, 'win_rate': win_rate, 'total_pnl_usdt': total_pnl, 'total_pnl_pct': total_pnl_pct, 'avg_win': avg_win, 'avg_loss': avg_loss, 'profit_factor': profit_factor, 'max_drawdown': max_drawdown, 'pnl_mean': pnl_mean, 'pnl_std': pnl_std, 'final_capital': self.capital, 'trades': df_trades } return results def print_results(self, results: dict): """Affiche les résultats de manière lisible""" if not results: return logger.info("\n" + "="*80) logger.info("📊 RÉSULTATS DU BACKTEST") logger.info("="*80) logger.info(f"\n🔍 SIGNAUX:") logger.info(f" Checks: {results['signals_checked']}") logger.info(f" Signaux générés: {results['signals_generated']}") logger.info(f" Taux de signal: {results['signals_generated']/results['signals_checked']*100:.2f}%") logger.info(f"\n📈 TRADES:") logger.info(f" Total: {results['total_trades']}") logger.info(f" Wins: {results['wins']} ({results['win_rate']:.1f}%)") logger.info(f" Losses: {results['losses']}") logger.info(f"\n💰 PERFORMANCE:") logger.info(f" PnL total: ${results['total_pnl_usdt']:+.2f} ({results['total_pnl_pct']:+.1f}%)") logger.info(f" Capital final: ${results['final_capital']:.2f}") logger.info(f" Win moyen: ${results['avg_win']:.2f}") logger.info(f" Loss moyen: ${results['avg_loss']:.2f}") logger.info(f" Profit Factor: {results['profit_factor']:.2f}") logger.info(f"\n📉 RISQUE:") logger.info(f" Max Drawdown: {results['max_drawdown']:.1f}%") logger.info(f" PnL moyen: {results['pnl_mean']:+.1f}%") logger.info(f" PnL volatilité: {results['pnl_std']:.1f}%") # Distribution des trades logger.info(f"\n📊 DISTRIBUTION DES TRADES:") df = results['trades'] tp_hits = len(df[df['reason'] == 'TP_HIT']) sl_hits = len(df[df['reason'] == 'SL_HIT']) logger.info(f" TP hits: {tp_hits} ({tp_hits/len(df)*100:.1f}%)") logger.info(f" SL hits: {sl_hits} ({sl_hits/len(df)*100:.1f}%)") # Top 5 meilleurs trades logger.info(f"\n🏆 TOP 5 MEILLEURS TRADES:") top_trades = df.nlargest(5, 'pnl_usdt')[['entry_time', 'side', 'pnl_usdt', 'pnl_pct', 'reason']] for idx, trade in top_trades.iterrows(): logger.info(f" {trade['entry_time']} - {trade['side']}: ${trade['pnl_usdt']:+.2f} ({trade['pnl_pct']:+.1f}%) - {trade['reason']}") # Top 5 pires trades logger.info(f"\n💀 TOP 5 PIRES TRADES:") worst_trades = df.nsmallest(5, 'pnl_usdt')[['entry_time', 'side', 'pnl_usdt', 'pnl_pct', 'reason']] for idx, trade in worst_trades.iterrows(): logger.info(f" {trade['entry_time']} - {trade['side']}: ${trade['pnl_usdt']:+.2f} ({trade['pnl_pct']:+.1f}%) - {trade['reason']}") logger.info("\n" + "="*80) if __name__ == '__main__': import sys # Params hours = int(sys.argv[1]) if len(sys.argv) > 1 else 24 interval = sys.argv[2] if len(sys.argv) > 2 else "1m" logger.info(f"\n⏱️ Récupération données historiques: {hours}h ({interval} timeframe)...") # Get data feed = BinancePriceFeed() if interval == "1m": limit = min(hours * 60, 1000) # Max 1000 bougies elif interval == "5m": limit = min(hours * 12, 1000) else: limit = 100 df = feed.get_klines("BTCUSDT", interval, limit) if df is None: logger.error("❌ Impossible de récupérer les données") exit(1) logger.info(f"✅ {len(df)} bougies récupérées") logger.info(f" De: {df['timestamp'].iloc[0]}") logger.info(f" À: {df['timestamp'].iloc[-1]}") logger.info(f" Prix: ${df['close'].iloc[0]:,.2f} → ${df['close'].iloc[-1]:,.2f}") # Run backtest strategy = DegenScalpingStrategy() backtest = DegenScalpingBacktest(strategy, initial_capital=100.0) results = backtest.run(df, "BTCUSD_DEGEN") # Print results backtest.print_results(results) logger.info("\n✅ Backtest terminé!")