#!/usr/bin/env python3 """ Optimisation RAPIDE de paramètres pour la stratégie Degen Scalping Utilise random search au lieu de grid search pour être beaucoup plus rapide """ 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 import json import random logging.basicConfig( level=logging.WARNING, format='%(message)s' ) logger = logging.getLogger(__name__) class FastOptimizer: """Optimisation rapide par random search""" def __init__(self, df: pd.DataFrame): self.df = df self.results = [] self.best_result = None def optimize(self, n_iterations: int = 200): """ Lance l'optimisation avec random search Args: n_iterations: Nombre de combinaisons aléatoires à tester """ # Plages de paramètres param_ranges = { 'stop_loss_pct': (0.0002, 0.0005), # 0.02% à 0.05% 'take_profit_pct': (0.0008, 0.003), # 0.08% à 0.3% 'min_confidence': (75, 95), 'max_volatility_5m': (0.003, 0.006), # 0.3% à 0.6% 'min_volume_multiplier': (1.2, 3.0) } logger.warning(f"\n{'='*80}") logger.warning(f"⚡ OPTIMISATION RAPIDE (Random Search)") logger.warning(f"{'='*80}") logger.warning(f"Combinaisons à tester: {n_iterations}") logger.warning(f"Cela devrait prendre ~{n_iterations*6//60} minutes...\n") profitable_count = 0 # Tester N combinaisons aléatoires for idx in range(1, n_iterations + 1): # Générer paramètres aléatoires params = { 'stop_loss_pct': random.uniform(*param_ranges['stop_loss_pct']), 'take_profit_pct': random.uniform(*param_ranges['take_profit_pct']), 'min_confidence': random.randint(param_ranges['min_confidence'][0], param_ranges['min_confidence'][1]), 'max_volatility_5m': random.uniform(*param_ranges['max_volatility_5m']), 'min_volume_multiplier': random.uniform(*param_ranges['min_volume_multiplier']) } # Assurer que TP > SL if params['take_profit_pct'] < params['stop_loss_pct'] * 2: params['take_profit_pct'] = params['stop_loss_pct'] * random.uniform(2.5, 5.0) # Créer stratégie avec ces paramètres (150x leverage) strategy = DegenScalpingStrategy(leverage=150) 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) # Tracker le meilleur if result['total_pnl_pct'] > 0: profitable_count += 1 if self.best_result is None or result['total_pnl_pct'] > self.best_result['total_pnl_pct']: self.best_result = result logger.warning(f"🎯 Nouveau meilleur: PnL {result['total_pnl_pct']:+.1f}% | WR {result['win_rate']:.1f}% | Trades {result['total_trades']} | #{idx}") # Afficher progression toutes les 20 itérations if idx % 20 == 0: logger.warning(f" [{idx}/{n_iterations}] - {profitable_count} configurations profitables trouvées") logger.warning(f"\n✅ Optimisation terminée: {len(self.results)} configurations testées") logger.warning(f" Configurations profitables: {profitable_count}/{len(self.results)}\n") def get_best_results(self, n: int = 10, sort_by: str = 'total_pnl_pct') -> pd.DataFrame: """Retourne les N meilleures configurations""" 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!") return df.nlargest(n, sort_by) 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:.3f}% | 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:.2f}% | Vol mult: {params['min_volume_multiplier']:.2f}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 = "fast_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() 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 12 # 12h par défaut (plus rapide) n_iter = int(sys.argv[2]) if len(sys.argv) > 2 else 200 # 200 combinaisons 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 = FastOptimizer(df) optimizer.optimize(n_iterations=n_iter) # Show best results optimizer.print_best_results(n=10) # Save results optimizer.save_results("fast_optimization_results.json") logger.warning("\n✅ Optimisation complète!")