#!/usr/bin/env python3 """ Optimiseur de paramètres pour la stratégie Degen Scalping Teste toutes les combinaisons de paramètres pour trouver le setup optimal """ import pandas as pd import numpy as np from datetime import datetime from degen_scalping_strategy import DegenScalpingStrategy from backtest_degen_scalping import DegenScalpingBacktest from price_feed import BinancePriceFeed import logging from itertools import product import json logging.basicConfig( level=logging.WARNING, # Moins verbeux pendant l'optimisation format='%(message)s' ) logger = logging.getLogger(__name__) class StrategyOptimizer: """Optimise les paramètres de la stratégie par grid search""" def __init__(self, df: pd.DataFrame): self.df = df self.results = [] def optimize(self): """Lance l'optimisation sur toutes les combinaisons de paramètres""" # Grille de paramètres à tester param_grid = { 'stop_loss_pct': [0.0002, 0.00025, 0.0003, 0.0004, 0.0005], # 0.02% à 0.05% 'take_profit_pct': [0.0008, 0.001, 0.0015, 0.002, 0.0025, 0.003], # 0.08% à 0.3% 'min_confidence': [75, 80, 85, 90, 95], 'max_volatility_5m': [0.003, 0.004, 0.005, 0.006], # 0.3% à 0.6% 'min_volume_multiplier': [1.2, 1.5, 2.0, 2.5, 3.0] } # Générer toutes les combinaisons keys = param_grid.keys() values = param_grid.values() combinations = list(product(*values)) total = len(combinations) logger.warning(f"\n{'='*80}") logger.warning(f"🔬 OPTIMISATION DE PARAMÈTRES") logger.warning(f"{'='*80}") logger.warning(f"Paramètres à tester: {total:,} combinaisons") logger.warning(f"Cela peut prendre quelques minutes...\n") # Tester chaque combinaison for idx, combo in enumerate(combinations, 1): params = dict(zip(keys, combo)) if idx % 100 == 0: logger.warning(f" Progression: {idx}/{total} ({idx/total*100:.1f}%)") # Créer stratégie avec ces paramètres strategy = DegenScalpingStrategy() strategy.stop_loss_pct = params['stop_loss_pct'] strategy.take_profit_pct = params['take_profit_pct'] strategy.min_confidence = params['min_confidence'] strategy.max_volatility_5m = params['max_volatility_5m'] strategy.min_volume_multiplier = params['min_volume_multiplier'] # Backtest backtest = DegenScalpingBacktest(strategy, initial_capital=100.0) results = backtest.run(self.df, "BTCUSD_DEGEN") if not results or results['total_trades'] == 0: continue # Enregistrer résultats result = { 'params': params, 'total_trades': results['total_trades'], 'win_rate': results['win_rate'], 'total_pnl_pct': results['total_pnl_pct'], 'profit_factor': results['profit_factor'], 'max_drawdown': results['max_drawdown'], 'signals_generated': results['signals_generated'], 'avg_win': results['avg_win'], 'avg_loss': results['avg_loss'], 'final_capital': results['final_capital'] } self.results.append(result) logger.warning(f"\n✅ Optimisation terminée: {len(self.results)} configurations testées\n") def get_best_results(self, n: int = 10, sort_by: str = 'total_pnl_pct') -> pd.DataFrame: """ Retourne les N meilleures configurations Args: n: Nombre de résultats à retourner sort_by: Critère de tri (total_pnl_pct, profit_factor, win_rate, etc.) Returns: DataFrame avec les meilleurs résultats """ if not self.results: return pd.DataFrame() df = pd.DataFrame(self.results) # Filtrer les résultats rentables df_profitable = df[df['total_pnl_pct'] > 0].copy() if len(df_profitable) == 0: logger.warning("⚠️ Aucune configuration profitable trouvée!") # Retourner les moins perdantes return df.nlargest(n, sort_by) # Trier et retourner top N return df_profitable.nlargest(n, sort_by) def print_best_results(self, n: int = 5): """Affiche les N meilleurs résultats""" best = self.get_best_results(n, sort_by='total_pnl_pct') if len(best) == 0: logger.warning("❌ Aucun résultat à afficher") return logger.warning("="*80) logger.warning("🏆 TOP CONFIGURATIONS RENTABLES") logger.warning("="*80) for idx, row in best.iterrows(): params = row['params'] logger.warning(f"\n#{best.index.get_loc(idx) + 1}. PnL: {row['total_pnl_pct']:+.1f}% | WR: {row['win_rate']:.1f}% | Trades: {row['total_trades']}") logger.warning(f" {'─'*76}") logger.warning(f" SL: {params['stop_loss_pct']*100:.3f}% | TP: {params['take_profit_pct']*100:.2f}% | Ratio: {params['take_profit_pct']/params['stop_loss_pct']:.1f}:1") logger.warning(f" Confiance: {params['min_confidence']:.0f}% | Vol max: {params['max_volatility_5m']*100:.1f}% | Vol mult: {params['min_volume_multiplier']:.1f}x") logger.warning(f" Profit Factor: {row['profit_factor']:.2f} | Max DD: {row['max_drawdown']:.1f}%") logger.warning(f" Avg Win: ${row['avg_win']:.2f} | Avg Loss: ${row['avg_loss']:.2f}") logger.warning(f" Capital final: ${row['final_capital']:.2f}") logger.warning("\n" + "="*80) def save_results(self, filename: str = "optimization_results.json"): """Sauvegarde tous les résultats dans un fichier""" if not self.results: return # Convertir en format sérialisable results_serializable = [] for r in self.results: result = r.copy() # Convertir params dict en strings pour les clés result['params'] = {str(k): float(v) for k, v in result['params'].items()} results_serializable.append(result) with open(filename, 'w') as f: json.dump(results_serializable, f, indent=2) logger.warning(f"💾 Résultats sauvegardés dans {filename}") if __name__ == '__main__': import sys # Params hours = int(sys.argv[1]) if len(sys.argv) > 1 else 24 logger.warning(f"\n⏱️ Récupération données historiques: {hours}h (1m timeframe)...") # Get data feed = BinancePriceFeed() limit = min(hours * 60, 1000) df = feed.get_klines("BTCUSDT", "1m", limit) if df is None: logger.error("❌ Impossible de récupérer les données") exit(1) logger.warning(f"✅ {len(df)} bougies récupérées") logger.warning(f" Période: {df['timestamp'].iloc[0]} → {df['timestamp'].iloc[-1]}") logger.warning(f" Prix: ${df['close'].iloc[0]:,.2f} → ${df['close'].iloc[-1]:,.2f}\n") # Run optimization optimizer = StrategyOptimizer(df) optimizer.optimize() # Show best results optimizer.print_best_results(n=10) # Save results optimizer.save_results("optimization_results.json") logger.warning("\n✅ Optimisation complète!")