""" Test parallèle de toutes les optimisations de la stratégie INVERSE Compare les 6 variantes + baseline """ import logging import pandas as pd import numpy as np from apollo_client import ApolloDegenClient from strategies.inverse_strategy import AdaptiveInverseBacktester as BaselineBacktester from strategies.inverse_strategy_optimized import ( InverseStrategyV1_EarlyExit, InverseStrategyV2_MultiTimeframe, InverseStrategyV3_DynamicSizing, InverseStrategyV4_TrailingStop, InverseStrategyV5_StrictFilter, InverseStrategyV6_Combined ) import config logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Désactiver logs verbeux logging.getLogger('InverseStrategy').setLevel(logging.WARNING) logging.getLogger('ShortTermSignalGenerator').setLevel(logging.WARNING) logging.getLogger('ApolloDegenClient').setLevel(logging.WARNING) logging.getLogger('MarketRegimeDetectorAdvanced').setLevel(logging.WARNING) logging.getLogger('InverseStrategyV1_EarlyExit').setLevel(logging.WARNING) logging.getLogger('InverseStrategyV2_MultiTimeframe').setLevel(logging.WARNING) logging.getLogger('InverseStrategyV3_DynamicSizing').setLevel(logging.WARNING) logging.getLogger('InverseStrategyV4_TrailingStop').setLevel(logging.WARNING) logging.getLogger('InverseStrategyV5_StrictFilter').setLevel(logging.WARNING) logging.getLogger('InverseStrategyV6_Combined').setLevel(logging.WARNING) class OptimizedBacktester: """Backtester générique pour toutes les variantes""" def __init__(self, strategy, leverage: int = 200, initial_capital: float = 1000): self.strategy = strategy self.leverage = leverage self.initial_capital = initial_capital self.trading_fee = config.TRADING_FEE_TAKER def backtest(self, df: pd.DataFrame, symbol: str) -> dict: capital = self.initial_capital trades = [] i = 50 while i < len(df) - 100: window = df.iloc[:i+1].copy() signal = self.strategy.generate_inverse_signal(window, symbol) if not signal: i += 1 continue # Position sizing if hasattr(self.strategy, 'get_position_size'): position_size_usd = self.strategy.get_position_size( self.leverage, signal.score, self.initial_capital ) else: risk_pct = 0.02 if self.leverage >= 200 else 0.015 position_size_usd = self.initial_capital * risk_pct * self.leverage entry_time = df.iloc[i]['timestamp'] entry_price = signal.entry_price sl = signal.stop_loss tp = signal.take_profit entry_regime = 'RANGE' # Tracking pour trailing stop best_price = entry_price current_sl = sl # Chercher sortie exit_found = False for j in range(i + 1, min(i + 100, len(df))): current = df.iloc[j] high = current['high'] low = current['low'] close = current['close'] # Update best price pour trailing if signal.side == 'LONG': best_price = max(best_price, high) else: best_price = min(best_price, low) # Trailing stop if hasattr(self.strategy, 'update_trailing_stop'): current_sl = self.strategy.update_trailing_stop( signal.side, entry_price, best_price, current_sl ) # Early exit should_exit = False exit_reason = None if hasattr(self.strategy, 'should_exit_early'): sub_window = df.iloc[:j+1].copy() should_exit, exit_reason = self.strategy.should_exit_early( sub_window, symbol, entry_regime, signal.side ) if should_exit: exit_price = close exit_reason = f'EARLY_{exit_reason}' else: # SL/TP normal exit_price = None if signal.side == 'LONG': if low <= current_sl: exit_price = current_sl exit_reason = 'SL' elif high >= tp: exit_price = tp exit_reason = 'TP' else: if high >= current_sl: exit_price = current_sl exit_reason = 'SL' elif low <= tp: exit_price = tp exit_reason = 'TP' if exit_price: # Calculer PnL if signal.side == 'LONG': pnl_pct = (exit_price - entry_price) / entry_price else: pnl_pct = (entry_price - exit_price) / entry_price pnl_usd = position_size_usd * pnl_pct pnl_usd -= position_size_usd * self.trading_fee * 2 capital += pnl_usd trades.append({ 'entry_time': entry_time, 'exit_time': current['timestamp'], 'side': signal.side, 'pnl_pct': pnl_pct * 100, 'pnl_usd': pnl_usd, 'exit_reason': exit_reason }) i = j + 1 exit_found = True break if not exit_found: i += 1 if not trades: return None winning = [t for t in trades if t['pnl_usd'] > 0] losing = [t for t in trades if t['pnl_usd'] <= 0] total_return = (capital - self.initial_capital) / self.initial_capital * 100 win_rate = len(winning) / len(trades) if trades else 0 # Profit factor total_wins = sum(t['pnl_usd'] for t in winning) if winning else 0 total_losses = abs(sum(t['pnl_usd'] for t in losing)) if losing else 1 profit_factor = total_wins / total_losses if total_losses > 0 else 999 return { 'total_return': total_return, 'win_rate': win_rate, 'profit_factor': profit_factor, 'trades': trades, 'num_trades': len(trades), 'wins': len(winning), 'losses': len(losing), 'avg_win': np.mean([t['pnl_pct'] for t in winning]) if winning else 0, 'avg_loss': np.mean([t['pnl_pct'] for t in losing]) if losing else 0, 'final_capital': capital } if __name__ == '__main__': logger.info("🚀 TEST PARALLÈLE - 7 VARIANTES DE LA STRATÉGIE INVERSE\n") logger.info("="*80) # Données client = ApolloDegenClient() df = client.get_klines('BTCUSDT', '1m', limit=1500) # Pré-charger données 5m pour V2 et V6 logger.info("Chargement données 5m...") df_5m = client.get_klines('BTCUSDT', '5m', limit=300) logger.info(f"Données: {len(df)} chandeliers 1m, {len(df_5m)} chandeliers 5m\n") leverage = 200 results = {} # BASELINE (stratégie actuelle optimisée) logger.info("Testing BASELINE (current optimized)...") try: baseline_bt = BaselineBacktester(initial_capital=1000, leverage=leverage) baseline_result = baseline_bt.backtest(df, 'BTCUSDT') if baseline_result: # Ajouter profit_factor au résultat winning = [t for t in baseline_result['trades'] if t['pnl_usd'] > 0] losing = [t for t in baseline_result['trades'] if t['pnl_usd'] <= 0] total_wins = sum(t['pnl_usd'] for t in winning) if winning else 0 total_losses = abs(sum(t['pnl_usd'] for t in losing)) if losing else 1 baseline_result['profit_factor'] = total_wins / total_losses if total_losses > 0 else 999 baseline_result['num_trades'] = len(baseline_result['trades']) baseline_result['wins'] = len(winning) baseline_result['losses'] = len(losing) baseline_result['avg_win'] = np.mean([t['pnl_pct'] for t in winning]) if winning else 0 baseline_result['avg_loss'] = np.mean([t['pnl_pct'] for t in losing]) if losing else 0 results['BASELINE'] = baseline_result except Exception as e: logger.error(f"BASELINE failed: {e}") results['BASELINE'] = None # V1: Early Exit logger.info("Testing V1: Early Exit...") try: v1_strategy = InverseStrategyV1_EarlyExit() v1_bt = OptimizedBacktester(v1_strategy, leverage=leverage) results['V1_EARLY_EXIT'] = v1_bt.backtest(df, 'BTCUSDT') except Exception as e: logger.error(f"V1 failed: {e}") results['V1_EARLY_EXIT'] = None # V3: Dynamic Sizing (Skip V2 for now - too slow) logger.info("Testing V3: Dynamic Sizing...") try: v3_strategy = InverseStrategyV3_DynamicSizing() v3_bt = OptimizedBacktester(v3_strategy, leverage=leverage) results['V3_DYNAMIC_SIZE'] = v3_bt.backtest(df, 'BTCUSDT') except Exception as e: logger.error(f"V3 failed: {e}") results['V3_DYNAMIC_SIZE'] = None # V4: Trailing Stop logger.info("Testing V4: Trailing Stop...") try: v4_strategy = InverseStrategyV4_TrailingStop() v4_bt = OptimizedBacktester(v4_strategy, leverage=leverage) results['V4_TRAILING'] = v4_bt.backtest(df, 'BTCUSDT') except Exception as e: logger.error(f"V4 failed: {e}") results['V4_TRAILING'] = None # V5: Strict Filter logger.info("Testing V5: Strict Filter...") try: v5_strategy = InverseStrategyV5_StrictFilter() v5_bt = OptimizedBacktester(v5_strategy, leverage=leverage) results['V5_STRICT'] = v5_bt.backtest(df, 'BTCUSDT') except Exception as e: logger.error(f"V5 failed: {e}") results['V5_STRICT'] = None # RÉSULTATS logger.info("\n" + "="*80) logger.info("📊 RÉSULTATS COMPARATIFS") logger.info("="*80 + "\n") # Tableau comparatif logger.info(f"{'Version':<20} {'Return%':<10} {'WR%':<8} {'PF':<8} {'Trades':<8} {'Avg Win%':<10} {'Avg Loss%':<10}") logger.info("-"*90) sorted_results = sorted( [(k, v) for k, v in results.items() if v is not None], key=lambda x: x[1]['total_return'], reverse=True ) for version, res in sorted_results: logger.info( f"{version:<20} " f"{res['total_return']:<10.2f} " f"{res['win_rate']*100:<8.0f} " f"{res['profit_factor']:<8.2f} " f"{res['num_trades']:<8} " f"{res['avg_win']:<10.2f} " f"{res['avg_loss']:<10.2f}" ) # Gagnant if sorted_results: winner_name, winner = sorted_results[0] baseline_return = results['BASELINE']['total_return'] if results['BASELINE'] else 0 logger.info("\n" + "="*80) logger.info("🏆 MEILLEURE STRATÉGIE") logger.info("="*80) logger.info(f"Version: {winner_name}") logger.info(f"Return: {winner['total_return']:+.2f}%") logger.info(f"Win Rate: {winner['win_rate']*100:.1f}%") logger.info(f"Profit Factor: {winner['profit_factor']:.2f}") logger.info(f"Trades: {winner['num_trades']}") logger.info(f"Wins/Losses: {winner['wins']}/{winner['losses']}") logger.info(f"") logger.info(f"Amélioration vs BASELINE: {winner['total_return'] - baseline_return:+.2f} points") logger.info("="*80) # Analyse des exits logger.info("\n📈 ANALYSE DES EXITS:") exit_reasons = {} for t in winner['trades']: reason = t['exit_reason'] exit_reasons[reason] = exit_reasons.get(reason, 0) + 1 for reason, count in sorted(exit_reasons.items(), key=lambda x: x[1], reverse=True): logger.info(f" {reason}: {count} trades") logger.info("\n")