""" Optimiseur automatique pour Apollo Degen Bot Teste des centaines de combinaisons de paramètres pour trouver la config optimale """ import logging import sys import json from datetime import datetime from itertools import product import pandas as pd from apollo_client import ApolloDegenClient from backtesting.backtester import DegenBacktester import config # Paramètres à tester PARAM_GRID = { 'leverage': [50, 75, 100, 150, 200], 'timeframe': ['1m', '3m', '5m'], 'min_score': [0.75, 0.80, 0.85, 0.90], 'min_confluences': [2, 3, 4], 'rsi_period': [5, 7, 9, 11], 'strategy': ['SCALPING_RSI_VOLUME', 'MACD_MOMENTUM', 'BOLLINGER_BOUNCE'] } class StrategyOptimizer: """Optimiseur de stratégies""" def __init__(self, symbol: str = 'BTCUSDT', days: int = 7): self.logger = logging.getLogger(self.__class__.__name__) self.symbol = symbol self.days = days self.client = ApolloDegenClient() self.results = [] self.logger.info(f"🔧 Optimiseur initialisé: {symbol} {days} jours") def download_data(self, timeframe: str): """Télécharge les données pour un timeframe""" # Calculer nombre de chandeliers nécessaires candles_per_day = { '1m': 1440, '3m': 480, '5m': 288, '15m': 96, '1h': 24 } total_candles = self.days * candles_per_day.get(timeframe, 1440) limit = min(1500, total_candles) # Max 1500 par requête self.logger.debug(f"Téléchargement {limit} chandeliers {timeframe}...") df = self.client.get_klines(self.symbol, timeframe, limit=limit) if df.empty: self.logger.error(f"❌ Données vides pour {timeframe}") return None self.logger.debug(f"✅ {len(df)} chandeliers récupérés") return df def apply_params(self, params): """Applique temporairement des paramètres à la config""" # Sauvegarder config originale original = { 'RSI_PERIOD': config.RSI_PERIOD, 'MIN_SIGNAL_SCORE_AGGRESSIVE': config.MIN_SIGNAL_SCORE_AGGRESSIVE, 'MIN_CONFLUENCES_AGGRESSIVE': config.MIN_CONFLUENCES_AGGRESSIVE, 'TIMEFRAME_PRIMARY': config.TIMEFRAME_PRIMARY } # Appliquer nouveaux params config.RSI_PERIOD = params['rsi_period'] config.MIN_SIGNAL_SCORE_AGGRESSIVE = params['min_score'] config.MIN_CONFLUENCES_AGGRESSIVE = params['min_confluences'] config.TIMEFRAME_PRIMARY = params['timeframe'] return original def restore_params(self, original): """Restaure la config originale""" config.RSI_PERIOD = original['RSI_PERIOD'] config.MIN_SIGNAL_SCORE_AGGRESSIVE = original['MIN_SIGNAL_SCORE_AGGRESSIVE'] config.MIN_CONFLUENCES_AGGRESSIVE = original['MIN_CONFLUENCES_AGGRESSIVE'] config.TIMEFRAME_PRIMARY = original['TIMEFRAME_PRIMARY'] def test_combination(self, params, df): """Teste une combinaison de paramètres""" # Appliquer params original = self.apply_params(params) try: # Backtester avec leverage backtester = DegenBacktester( initial_capital=1000, leverage=params['leverage'] ) # Tester la stratégie result = backtester.backtest_strategy( df, self.symbol, strategy_name=params['strategy'] ) # Restaurer config self.restore_params(original) if not result: return None # Extraire métriques return { 'leverage': params['leverage'], 'timeframe': params['timeframe'], 'strategy': params['strategy'], 'min_score': params['min_score'], 'min_confluences': params['min_confluences'], 'rsi_period': params['rsi_period'], 'total_return_pct': result.total_return_pct, 'total_trades': result.total_trades, 'win_rate': result.win_rate, 'profit_factor': result.profit_factor, 'max_drawdown_pct': result.max_drawdown_pct, 'sharpe_ratio': result.sharpe_ratio, 'avg_win_pct': result.avg_win_pct, 'avg_loss_pct': result.avg_loss_pct } except Exception as e: self.logger.error(f"Erreur test combinaison: {e}") self.restore_params(original) return None def optimize(self, fast_mode: bool = False): """ Lance l'optimisation Args: fast_mode: Si True, teste moins de combinaisons """ self.logger.info("🚀 Début de l'optimisation...") # Réduire grid en fast mode if fast_mode: param_grid = { 'leverage': [50, 100, 150], 'timeframe': ['1m', '5m'], 'min_score': [0.80, 0.85], 'min_confluences': [2, 3], 'rsi_period': [7, 9], 'strategy': ['SCALPING_RSI_VOLUME', 'MACD_MOMENTUM'] } else: param_grid = PARAM_GRID # Générer toutes les combinaisons keys = param_grid.keys() values = param_grid.values() combinations = [dict(zip(keys, v)) for v in product(*values)] total_combinations = len(combinations) self.logger.info(f"📊 {total_combinations} combinaisons à tester") # Grouper par timeframe pour éviter de re-télécharger timeframe_groups = {} for combo in combinations: tf = combo['timeframe'] if tf not in timeframe_groups: timeframe_groups[tf] = [] timeframe_groups[tf].append(combo) tested = 0 valid_results = 0 # Tester par timeframe for timeframe, combos in timeframe_groups.items(): self.logger.info(f"\n📈 Timeframe {timeframe}: {len(combos)} combinaisons") # Télécharger données une fois df = self.download_data(timeframe) if df is None or len(df) < 100: self.logger.warning(f"⚠️ Données insuffisantes pour {timeframe}, skip") continue # Tester toutes les combos pour ce timeframe for combo in combos: tested += 1 # Progress if tested % 10 == 0: progress = (tested / total_combinations) * 100 self.logger.info(f"Progress: {tested}/{total_combinations} ({progress:.1f}%)") # Tester result = self.test_combination(combo, df) if result: self.results.append(result) valid_results += 1 # Log si profitable if result['total_return_pct'] > 0: self.logger.info( f"✅ {result['strategy']} {result['leverage']}x {result['timeframe']}: " f"{result['total_return_pct']:+.2f}% " f"(WR: {result['win_rate']*100:.0f}% PF: {result['profit_factor']:.2f})" ) self.logger.info(f"\n✅ Optimisation terminée: {valid_results}/{tested} résultats valides") def get_best_results(self, top_n: int = 10): """Retourne les N meilleurs résultats""" if not self.results: return [] # Trier par return décroissant sorted_results = sorted( self.results, key=lambda x: x['total_return_pct'], reverse=True ) return sorted_results[:top_n] def get_results_by_metric(self, metric: str, top_n: int = 5): """Retourne les meilleurs résultats selon une métrique""" if not self.results: return [] # Filtrer résultats valides valid = [r for r in self.results if r['total_trades'] >= 5] if metric == 'win_rate': sorted_results = sorted(valid, key=lambda x: x['win_rate'], reverse=True) elif metric == 'profit_factor': sorted_results = sorted(valid, key=lambda x: x['profit_factor'], reverse=True) elif metric == 'sharpe': sorted_results = sorted(valid, key=lambda x: x['sharpe_ratio'], reverse=True) elif metric == 'min_drawdown': sorted_results = sorted(valid, key=lambda x: x['max_drawdown_pct']) else: sorted_results = sorted(valid, key=lambda x: x['total_return_pct'], reverse=True) return sorted_results[:top_n] def print_summary(self): """Affiche le résumé de l'optimisation""" if not self.results: self.logger.warning("Aucun résultat à afficher") return self.logger.info(f"\n{'='*100}") self.logger.info("🏆 TOP 10 CONFIGURATIONS PAR RETURN") self.logger.info(f"{'='*100}") # Header self.logger.info( f"{'Strat':<20} {'Lev':>5} {'TF':>4} {'Score':>6} {'Conf':>5} " f"{'RSI':>4} | {'Return':>8} {'Trades':>7} {'WR':>6} {'PF':>6} {'DD':>7}" ) self.logger.info("-" * 100) # Top 10 for result in self.get_best_results(10): self.logger.info( f"{result['strategy']:<20} " f"{result['leverage']:>5}x " f"{result['timeframe']:>4} " f"{result['min_score']:>6.2f} " f"{result['min_confluences']:>5} " f"{result['rsi_period']:>4} | " f"{result['total_return_pct']:>7.2f}% " f"{result['total_trades']:>7} " f"{result['win_rate']*100:>5.1f}% " f"{result['profit_factor']:>6.2f} " f"{result['max_drawdown_pct']:>6.2f}%" ) # Statistiques globales profitable = [r for r in self.results if r['total_return_pct'] > 0] self.logger.info(f"\n📊 STATISTIQUES GLOBALES:") self.logger.info(f"Total testées: {len(self.results)}") self.logger.info(f"Profitables: {len(profitable)} ({len(profitable)/len(self.results)*100:.1f}%)") if profitable: best = max(self.results, key=lambda x: x['total_return_pct']) self.logger.info(f"\n🥇 MEILLEURE CONFIGURATION:") self.logger.info(f" Stratégie: {best['strategy']}") self.logger.info(f" Leverage: {best['leverage']}x") self.logger.info(f" Timeframe: {best['timeframe']}") self.logger.info(f" Min Score: {best['min_score']}") self.logger.info(f" Min Confluences: {best['min_confluences']}") self.logger.info(f" RSI Period: {best['rsi_period']}") self.logger.info(f" Return: {best['total_return_pct']:+.2f}%") self.logger.info(f" Win Rate: {best['win_rate']*100:.1f}%") self.logger.info(f" Profit Factor: {best['profit_factor']:.2f}") self.logger.info(f" Trades: {best['total_trades']}") else: self.logger.warning("⚠️ Aucune configuration profitable trouvée") # Meilleurs par métrique self.logger.info(f"\n{'='*100}") self.logger.info("🎯 MEILLEURS PAR MÉTRIQUE") self.logger.info(f"{'='*100}") # Win Rate self.logger.info("\n🎲 Top 3 Win Rate:") for i, result in enumerate(self.get_results_by_metric('win_rate', 3), 1): self.logger.info( f" {i}. {result['strategy']:<20} {result['leverage']:>3}x {result['timeframe']:>4} | " f"WR: {result['win_rate']*100:>5.1f}% Return: {result['total_return_pct']:>+7.2f}%" ) # Profit Factor self.logger.info("\n💰 Top 3 Profit Factor:") for i, result in enumerate(self.get_results_by_metric('profit_factor', 3), 1): self.logger.info( f" {i}. {result['strategy']:<20} {result['leverage']:>3}x {result['timeframe']:>4} | " f"PF: {result['profit_factor']:>6.2f} Return: {result['total_return_pct']:>+7.2f}%" ) # Sharpe self.logger.info("\n📈 Top 3 Sharpe Ratio:") for i, result in enumerate(self.get_results_by_metric('sharpe', 3), 1): self.logger.info( f" {i}. {result['strategy']:<20} {result['leverage']:>3}x {result['timeframe']:>4} | " f"Sharpe: {result['sharpe_ratio']:>7.2f} Return: {result['total_return_pct']:>+7.2f}%" ) self.logger.info(f"{'='*100}\n") def save_results(self, filename: str = None): """Sauvegarde les résultats""" if not filename: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f'optimization_results_{self.symbol}_{timestamp}.json' with open(filename, 'w') as f: json.dump(self.results, f, indent=2) self.logger.info(f"💾 Résultats sauvegardés: {filename}") # CSV pour analyse csv_filename = filename.replace('.json', '.csv') df = pd.DataFrame(self.results) df.to_csv(csv_filename, index=False) self.logger.info(f"💾 CSV sauvegardé: {csv_filename}") if __name__ == '__main__': logging.basicConfig( level=logging.INFO, format='%(asctime)s - [%(name)s] - %(message)s' ) logger = logging.getLogger(__name__) logger.info("🤖 Apollo Degen Bot - Optimizer\n") print("Mode d'optimisation:") print("1. Fast (48 combinaisons, ~10 min)") print("2. Normal (360 combinaisons, ~1h)") print("3. Exhaustif (tous paramètres, ~3h)") choice = input("\nChoix (1-3): ").strip() if choice == '1': logger.info("🚀 Mode Fast") optimizer = StrategyOptimizer('BTCUSDT', days=3) optimizer.optimize(fast_mode=True) elif choice == '2': logger.info("🚀 Mode Normal") optimizer = StrategyOptimizer('BTCUSDT', days=5) optimizer.optimize(fast_mode=False) elif choice == '3': logger.info("🚀 Mode Exhaustif") optimizer = StrategyOptimizer('BTCUSDT', days=7) optimizer.optimize(fast_mode=False) else: logger.error("Choix invalide") sys.exit(1) # Afficher résultats optimizer.print_summary() # Sauvegarder optimizer.save_results() logger.info("\n✅ Optimisation terminée!")