#!/usr/bin/env python3 """ Backtest de la stratégie adaptative sur données massives """ import pandas as pd import numpy as np from datetime import datetime import logging from adaptive_strategy import AdaptiveStrategy from dynamic_params_adapter import DynamicParameterAdapter logging.basicConfig(level=logging.WARNING) logger = logging.getLogger(__name__) class AdaptiveBacktest: """Backtest pour stratégie adaptative""" def __init__(self, strategy: AdaptiveStrategy, initial_capital: float = 500.0, use_trailing_stop: bool = False, trailing_pct: float = 0.15, early_stop_on_losses: bool = True, min_capital_pct: float = 0.50, use_dynamic_adaptation: bool = False): self.strategy = strategy self.initial_capital = initial_capital self.capital = initial_capital self.current_position = None self.trades = [] # Trailing stop parameters self.use_trailing_stop = use_trailing_stop self.trailing_pct = trailing_pct # % de retrait du PnL max avant exit # Early stopping self.early_stop_on_losses = early_stop_on_losses self.min_capital_pct = min_capital_pct # % minimum du capital initial (ex: 0.50 = 50%) # Dynamic adaptation (adapt params based on volatility) self.use_dynamic_adaptation = use_dynamic_adaptation self.dynamic_adapter = None if use_dynamic_adaptation: # Récupérer les paramètres de base de la stratégie base_params = { 'trailing_pct': trailing_pct, 'rsi_oversold': getattr(strategy, 'params', {}).get('rsi_oversold', 30), 'rsi_overbought': getattr(strategy, 'params', {}).get('rsi_overbought', 70), 'sl_atr_mult': getattr(strategy, 'params', {}).get('sl_atr_multiplier', 1.0) } self.dynamic_adapter = DynamicParameterAdapter(base_params) def run(self, df: pd.DataFrame, symbol: str) -> dict: """Execute backtest""" logger.warning(f"\n{'='*80}") logger.warning(f"🧪 BACKTEST STRATÉGIE ADAPTATIVE") logger.warning(f"{'='*80}") logger.warning(f"Période: {df['timestamp'].iloc[0]} → {df['timestamp'].iloc[-1]}") logger.warning(f"Bougies: {len(df):,}") logger.warning(f"Capital initial: ${self.initial_capital:.2f}") logger.warning(f"{'='*80}\n") # Parcourir les données for i in range(100, len(df)): df_slice = df.iloc[:i+1].copy() current_price = df_slice['close'].iloc[-1] timestamp = df_slice['timestamp'].iloc[-1] # EARLY STOPPING: Si balance tombe sous le minimum (ex: 50%) min_capital = self.initial_capital * self.min_capital_pct if self.capital < min_capital: logger.warning(f"⚠️ EARLY STOP: Capital ${self.capital:.2f} < ${min_capital:.2f} ({self.min_capital_pct*100:.0f}% initial)") return None # Rejeter cette config # EARLY STOPPING: Si 3 premières positions perdantes, abort if self.early_stop_on_losses and len(self.trades) >= 3: first_3_trades = self.trades[:3] if all(t['pnl_usd'] < 0 for t in first_3_trades): logger.warning(f"⚠️ EARLY STOP: 3 premières positions perdantes") return None # Rejeter cette config # Si en position, check exit if self.current_position: self._check_exit(current_price, timestamp) continue # Sinon, check entry if self.capital > 0: # ADAPTATION DYNAMIQUE: Adapter les params selon l'ATR actuel if self.use_dynamic_adaptation and self.dynamic_adapter: adapted_params = self.dynamic_adapter.adapt_params(df_slice) # Appliquer les params adaptés à la stratégie if hasattr(self.strategy, 'params'): self.strategy.params['rsi_oversold'] = adapted_params['rsi_oversold'] self.strategy.params['rsi_overbought'] = adapted_params['rsi_overbought'] self.strategy.params['sl_atr_multiplier'] = adapted_params['sl_atr_mult'] # Adapter aussi le trailing stop self.trailing_pct = adapted_params['trailing_pct'] logger.debug(f"Dynamic adapt: ATR {adapted_params['atr_pct']:.2f}% | " f"Trail {adapted_params['trailing_pct']:.2%} | " f"RSI {adapted_params['rsi_oversold']:.0f}/{adapted_params['rsi_overbought']:.0f}") signal = self.strategy.analyze(symbol, df_slice) if signal and signal.confidence >= 65: # Min confidence self._open_position(signal, timestamp) # Close dernière position si ouverte if self.current_position: self._close_position(df['close'].iloc[-1], df['timestamp'].iloc[-1], 'END') # Calculer stats return self._calculate_stats() def _open_position(self, signal, timestamp): """Ouvre une position""" # Position size: 20% du capital (ISOLATED - gestion du risque!) position_size = self.capital * 0.20 self.current_position = { 'signal': signal, 'entry_price': signal.entry_price, 'entry_time': timestamp, 'position_size': position_size, 'entry_capital': self.capital, # Trailing stop tracking 'max_pnl_pct': 0.0, 'trailing_stop_price': None, 'max_price_reached': signal.entry_price } logger.info(f"\n📈 OPEN {signal.side} @ ${signal.entry_price:,.2f}") logger.info(f" Stratégie: {signal.strategy_used.upper()}") logger.info(f" Confiance: {signal.confidence:.1f}%") logger.info(f" Capital: ${self.capital:.2f}") def _check_exit(self, current_price: float, timestamp): """Check si doit sortir""" signal = self.current_position['signal'] entry_price = self.current_position['entry_price'] if self.use_trailing_stop: # Mode TRAILING STOP self._check_trailing_stop_exit(current_price, timestamp) else: # Mode TP/SL FIXE (original) if signal.side == 'LONG': if current_price >= signal.take_profit: self._close_position(current_price, timestamp, 'TP') elif current_price <= signal.stop_loss: self._close_position(current_price, timestamp, 'SL') else: # SHORT if current_price <= signal.take_profit: self._close_position(current_price, timestamp, 'TP') elif current_price >= signal.stop_loss: self._close_position(current_price, timestamp, 'SL') def _check_trailing_stop_exit(self, current_price: float, timestamp): """Check exit avec trailing stop dynamique""" signal = self.current_position['signal'] entry_price = self.current_position['entry_price'] # Calculer PnL actuel if signal.side == 'LONG': pnl_pct = (current_price - entry_price) / entry_price else: pnl_pct = (entry_price - current_price) / entry_price # Check SL initial (protection pertes) if signal.side == 'LONG' and current_price <= signal.stop_loss: self._close_position(current_price, timestamp, 'INITIAL_SL') return elif signal.side == 'SHORT' and current_price >= signal.stop_loss: self._close_position(current_price, timestamp, 'INITIAL_SL') return # Update max PnL if pnl_pct > self.current_position['max_pnl_pct']: self.current_position['max_pnl_pct'] = pnl_pct self.current_position['max_price_reached'] = current_price # Activer trailing stop seulement si en profit if pnl_pct > 0: # Trailing stop = prix qui donne (max_pnl * (1 - trailing_pct)) target_pnl = pnl_pct * (1 - self.trailing_pct) if signal.side == 'LONG': self.current_position['trailing_stop_price'] = entry_price * (1 + target_pnl) else: self.current_position['trailing_stop_price'] = entry_price * (1 - target_pnl) # Check trailing stop trailing_stop = self.current_position['trailing_stop_price'] if trailing_stop is not None: if signal.side == 'LONG' and current_price <= trailing_stop: self._close_position(current_price, timestamp, 'TRAILING_STOP') elif signal.side == 'SHORT' and current_price >= trailing_stop: self._close_position(current_price, timestamp, 'TRAILING_STOP') def _close_position(self, exit_price: float, timestamp, reason: str): """Ferme la position""" signal = self.current_position['signal'] entry_price = self.current_position['entry_price'] position_size = self.current_position['position_size'] # Calculer PnL if signal.side == 'LONG': pnl_pct = (exit_price - entry_price) / entry_price else: pnl_pct = (entry_price - exit_price) / entry_price pnl_pct_leveraged = pnl_pct * signal.leverage pnl_usd = position_size * pnl_pct_leveraged # Update capital self.capital += pnl_usd # Record trade trade = { 'entry_time': self.current_position['entry_time'], 'exit_time': timestamp, 'side': signal.side, 'strategy': signal.strategy_used, 'entry_price': entry_price, 'exit_price': exit_price, 'pnl_pct': pnl_pct_leveraged * 100, 'pnl_usd': pnl_usd, 'reason': reason, 'confidence': signal.confidence } self.trades.append(trade) logger.info(f"\n🏁 CLOSE {signal.side} @ ${exit_price:,.2f} - {reason}") logger.info(f" PnL: {pnl_pct_leveraged*100:+.1f}% (${pnl_usd:+.2f})") logger.info(f" Capital: ${self.capital:.2f}") self.current_position = None def _calculate_stats(self) -> dict: """Calcule les statistiques""" if not self.trades: return {} df_trades = pd.DataFrame(self.trades) wins = df_trades[df_trades['pnl_usd'] > 0] losses = df_trades[df_trades['pnl_usd'] < 0] total_pnl = df_trades['pnl_usd'].sum() total_pnl_pct = (self.capital - self.initial_capital) / self.initial_capital * 100 # Stats par stratégie strategy_stats = {} for strategy in df_trades['strategy'].unique(): strat_trades = df_trades[df_trades['strategy'] == strategy] strat_wins = strat_trades[strat_trades['pnl_usd'] > 0] strategy_stats[strategy] = { 'trades': len(strat_trades), 'wins': len(strat_wins), 'win_rate': len(strat_wins) / len(strat_trades) * 100 if len(strat_trades) > 0 else 0, 'total_pnl': strat_trades['pnl_usd'].sum() } return { 'total_trades': len(self.trades), 'wins': len(wins), 'losses': len(losses), 'win_rate': len(wins) / len(self.trades) * 100, 'total_pnl_usd': total_pnl, 'total_pnl_pct': total_pnl_pct, 'final_capital': self.capital, 'avg_win': wins['pnl_usd'].mean() if len(wins) > 0 else 0, 'avg_loss': losses['pnl_usd'].mean() if len(losses) > 0 else 0, 'profit_factor': wins['pnl_usd'].sum() / abs(losses['pnl_usd'].sum()) if len(losses) > 0 else float('inf'), 'strategy_stats': strategy_stats, 'trades': df_trades } def print_results(results: dict): """Affiche les résultats""" logger.warning(f"\n{'='*80}") logger.warning(f"📊 RÉSULTATS DU BACKTEST") logger.warning(f"{'='*80}\n") logger.warning(f"📈 TRADES:") logger.warning(f" Total: {results['total_trades']}") logger.warning(f" Wins: {results['wins']} ({results['win_rate']:.1f}%)") logger.warning(f" Losses: {results['losses']}\n") logger.warning(f"💰 PERFORMANCE:") logger.warning(f" PnL total: ${results['total_pnl_usd']:+.2f} ({results['total_pnl_pct']:+.1f}%)") logger.warning(f" Capital final: ${results['final_capital']:.2f}") logger.warning(f" Win moyen: ${results['avg_win']:.2f}") logger.warning(f" Loss moyen: ${results['avg_loss']:.2f}") logger.warning(f" Profit Factor: {results['profit_factor']:.2f}\n") logger.warning(f"📊 PAR STRATÉGIE:") for strategy, stats in results['strategy_stats'].items(): logger.warning(f" {strategy.upper()}:") logger.warning(f" Trades: {stats['trades']} | WR: {stats['win_rate']:.1f}% | PnL: ${stats['total_pnl']:+.2f}") logger.warning(f"\n{'='*80}\n") if __name__ == '__main__': import sys # Paramètres timeframe = sys.argv[1] if len(sys.argv) > 1 else '5m' leverage = int(sys.argv[2]) if len(sys.argv) > 2 else 225 logger.warning(f"\n⏱️ Chargement données {timeframe}...") # Charger données if timeframe == '1m': df = pd.read_csv('historical_data/BTCUSDT_1m_44000candles.csv') elif timeframe == '5m': df = pd.read_csv('historical_data/BTCUSDT_5m_9000candles.csv') else: # 15m df = pd.read_csv('historical_data/BTCUSDT_15m_3000candles.csv') df['timestamp'] = pd.to_datetime(df['timestamp']) logger.warning(f"✅ {len(df):,} bougies chargées") logger.warning(f" Période: {df['timestamp'].iloc[0]} → {df['timestamp'].iloc[-1]}") # Run backtest strategy = AdaptiveStrategy(leverage=leverage) backtest = AdaptiveBacktest(strategy, initial_capital=100.0) results = backtest.run(df, "BTCUSD_DEGEN") # Print results if results: print_results(results) # Save trades to CSV results['trades'].to_csv(f'backtest_results_{timeframe}_{leverage}x.csv', index=False) logger.warning(f"💾 Trades sauvegardés: backtest_results_{timeframe}_{leverage}x.csv") else: logger.warning("❌ Aucun trade généré!")