#!/usr/bin/env python3 """ SMART OPTIMIZER - Moteur d'optimisation intelligent background Stratégie: 1. Exploration (50% du temps): Random search 2. Exploitation (50% du temps): Mutate best configs 3. Auto-orientation vers setups profitables 4. Sauvegarde continue, évite doublons 5. Objectif: PROFIT MAXIMUM """ import pandas as pd import numpy as np import json import logging import hashlib import time from datetime import datetime from pathlib import Path import random from v10_range_strategy import V10RangeStrategy from backtest_adaptive import AdaptiveBacktest from backtest_v10_range import Signal logging.basicConfig( level=logging.WARNING, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class SmartOptimizer: """Optimiseur intelligent avec auto-orientation""" def __init__(self, df: pd.DataFrame, initial_capital: float = 500.0, results_file: str = 'smart_optimizer_results.json'): self.df = df self.initial_capital = initial_capital self.results_file = results_file self.results = [] self.tested_hashes = set() # Charger résultats existants self._load_results() # Statistiques self.total_tests = 0 self.exploration_tests = 0 self.exploitation_tests = 0 def _load_results(self): """Charge résultats existants""" if Path(self.results_file).exists(): with open(self.results_file, 'r') as f: self.results = json.load(f) # Reconstruire hashes for r in self.results: config_hash = self._hash_config(r['params']) self.tested_hashes.add(config_hash) print(f"📂 Chargé {len(self.results)} résultats existants") def _save_results(self): """Sauvegarde résultats""" with open(self.results_file, 'w') as f: json.dump(self.results, f, indent=2) def _hash_config(self, params: dict) -> str: """Hash une config pour détecter doublons""" # Arrondir les floats pour éviter les faux doublons rounded = { k: round(v, 3) if isinstance(v, float) else v for k, v in params.items() } config_str = json.dumps(rounded, sort_keys=True) return hashlib.md5(config_str.encode()).hexdigest() def _generate_random_config(self) -> dict: """Génère une config complètement aléatoire (EXPLORATION)""" # Paramètres V10 Range Reversal config = { # Entry conditions (assouplir pour plus de trades) 'rsi_oversold': random.randint(20, 35), 'rsi_overbought': random.randint(65, 80), 'range_atr_threshold': round(random.uniform(0.20, 0.50), 2), 'min_confidence': random.choice([60, 65, 70, 75, 80]), # Stop Loss (serré à large) 'sl_atr_mult': round(random.uniform(0.5, 2.0), 2), # Trailing Stop % (clé!) 'trailing_pct': round(random.uniform(0.05, 0.35), 2), # Leverage (focus high leverage scalping) 'leverage': random.choice([100, 125, 150, 175, 200, 225, 250]), } # Valider cohérence if config['rsi_overbought'] - config['rsi_oversold'] < 25: config['rsi_overbought'] = config['rsi_oversold'] + 30 return config def _mutate_config(self, base_config: dict, mutation_rate: float = 0.3) -> dict: """Mute une config existante (EXPLOITATION)""" config = base_config.copy() # Muter chaque param avec probabilité mutation_rate if random.random() < mutation_rate: config['rsi_oversold'] = max(20, min(35, config['rsi_oversold'] + random.randint(-5, 5))) if random.random() < mutation_rate: config['rsi_overbought'] = max(65, min(80, config['rsi_overbought'] + random.randint(-5, 5))) if random.random() < mutation_rate: config['range_atr_threshold'] = max(0.20, min(0.50, round(config['range_atr_threshold'] + random.uniform(-0.10, 0.10), 2))) if random.random() < mutation_rate: config['min_confidence'] = random.choice([60, 65, 70, 75, 80]) if random.random() < mutation_rate: config['sl_atr_mult'] = max(0.5, min(2.0, round(config['sl_atr_mult'] + random.uniform(-0.3, 0.3), 2))) if random.random() < mutation_rate: config['trailing_pct'] = max(0.05, min(0.35, round(config['trailing_pct'] + random.uniform(-0.05, 0.05), 2))) if random.random() < mutation_rate: leverages = [100, 125, 150, 175, 200, 225, 250] current_idx = leverages.index(config['leverage']) if config['leverage'] in leverages else 3 new_idx = max(0, min(len(leverages)-1, current_idx + random.randint(-1, 1))) config['leverage'] = leverages[new_idx] # Valider if config['rsi_overbought'] - config['rsi_oversold'] < 25: config['rsi_overbought'] = config['rsi_oversold'] + 30 return config def _get_best_configs(self, top_n: int = 10) -> list: """Récupère les N meilleures configs""" if not self.results: return [] # Filtrer profitables profitable = [r for r in self.results if r['results']['total_pnl_pct'] > 0 and r['results']['profit_factor'] > 1.0] if not profitable: return [] # Trier par PnL profitable.sort(key=lambda x: x['results']['total_pnl_pct'], reverse=True) return [r['params'] for r in profitable[:top_n]] def _generate_config(self) -> dict: """Génère une config (exploration vs exploitation)""" best_configs = self._get_best_configs(top_n=10) # 50% exploration, 50% exploitation if not best_configs or random.random() < 0.5: # EXPLORATION: Random self.exploration_tests += 1 return self._generate_random_config() else: # EXPLOITATION: Muter un des meilleurs self.exploitation_tests += 1 base = random.choice(best_configs) return self._mutate_config(base) def _test_config(self, params: dict) -> dict: """Teste une config""" try: # Créer stratégie strategy = V10RangeStrategy(leverage=params['leverage']) # Appliquer params strategy.params['rsi_oversold'] = params['rsi_oversold'] strategy.params['rsi_overbought'] = params['rsi_overbought'] strategy.params['range_atr_threshold'] = params['range_atr_threshold'] strategy.params['min_confidence'] = params['min_confidence'] strategy.params['sl_atr_multiplier'] = params['sl_atr_mult'] strategy.params['tp_atr_multiplier'] = 3.0 # Pas utilisé avec trailing # Adapter class CustomAdapter: def __init__(self, strat): self.strategy = strat self.leverage = strat.leverage self.params = { 'ranging': { 'sl_atr_multiplier': strat.params['sl_atr_multiplier'], 'tp_atr_multiplier': strat.params['tp_atr_multiplier'], 'min_confidence': strat.params['min_confidence'], } } def analyze(self, symbol, df): signal_dict = self.strategy.analyze(symbol, df) if not signal_dict: return None return Signal( symbol=signal_dict['symbol'], side=signal_dict['side'], entry_price=signal_dict['entry_price'], stop_loss=signal_dict['stop_loss'], take_profit=signal_dict['take_profit'], leverage=signal_dict['leverage'], risk_pct=signal_dict['risk_pct'], score=signal_dict['confidence'], confluences=signal_dict['confluences'], strategy_used='v10_range_reversal', confidence=signal_dict['confidence'] ) adapter = CustomAdapter(strategy) # Backtest avec TRAILING STOP + EARLY STOPPING backtest = AdaptiveBacktest( adapter, initial_capital=self.initial_capital, use_trailing_stop=True, trailing_pct=params['trailing_pct'], early_stop_on_losses=True, # Si 3 premières positions perdantes min_capital_pct=0.50 # Abort si balance < 50% (250$ sur 500$) ) results = backtest.run(self.df, 'BTCUSDT') if not results or results['total_trades'] == 0: return None return { 'params': params.copy(), 'results': { 'total_trades': results['total_trades'], 'win_rate': results['win_rate'], 'total_pnl_pct': results['total_pnl_pct'], 'final_capital': results['final_capital'], 'profit_factor': results['profit_factor'], 'avg_win': results['avg_win'], 'avg_loss': results['avg_loss'] }, 'timestamp': datetime.now().isoformat() } except Exception as e: logger.error(f"Erreur test: {e}") return None def optimize_forever(self): """Boucle d'optimisation infinie""" print("="*80) print("🚀 SMART OPTIMIZER - Démarrage") print("="*80) print(f"📊 Données: {len(self.df)} bougies 1m") print(f"📅 Période: {self.df['timestamp'].iloc[0]} → {self.df['timestamp'].iloc[-1]}") print(f"💰 Capital: ${self.initial_capital}") print(f"🎯 Objectif: PROFIT MAXIMUM") print(f"♾️ Mode: Background infini") print("="*80) if self.results: best = max(self.results, key=lambda x: x['results']['total_pnl_pct']) print(f"\n📈 Meilleur résultat actuel:") print(f" PnL: {best['results']['total_pnl_pct']:+.2f}%") print(f" PF: {best['results']['profit_factor']:.2f}") print(f" Trades: {best['results']['total_trades']}") print(f"\n🔄 Démarrage de l'optimisation...\n") iteration = 0 start_time = time.time() try: while True: iteration += 1 self.total_tests += 1 # Générer config params = self._generate_config() # Check si déjà testée config_hash = self._hash_config(params) if config_hash in self.tested_hashes: continue # Tester result = self._test_config(params) if result: self.results.append(result) self.tested_hashes.add(config_hash) # Sauvegarder self._save_results() # Afficher res = result['results'] mode = "🔍" if (self.exploration_tests / self.total_tests) > 0.45 else "🎯" if res['total_pnl_pct'] > 0 and res['profit_factor'] > 1.0: print(f"{mode} #{iteration} | " f"PnL: {res['total_pnl_pct']:+.2f}% | " f"PF: {res['profit_factor']:.2f} | " f"Trades: {res['total_trades']} | " f"Trail:{int(params['trailing_pct']*100)}% " f"Lev:{params['leverage']}x " f"RSI:{params['rsi_oversold']}/{params['rsi_overbought']}") else: print(f"❌ #{iteration} | Unprofitable") # Stats toutes les 10 itérations if iteration % 10 == 0: self._print_stats() except KeyboardInterrupt: print("\n\n⏸️ Optimisation interrompue par l'utilisateur") self._print_final_summary() def _print_stats(self): """Affiche stats courantes""" elapsed = time.time() - time.time() # À implémenter proprement profitable = [r for r in self.results if r['results']['total_pnl_pct'] > 0 and r['results']['profit_factor'] > 1.0] if not self.results: return print(f"\n{'='*80}") print(f"📊 STATS (après {self.total_tests} tests)") print(f"{'='*80}") print(f"Total testé: {len(self.results)}") print(f"Profitables: {len(profitable)} ({len(profitable)/len(self.results)*100:.1f}%)") print(f"Exploration: {self.exploration_tests} | Exploitation: {self.exploitation_tests}") if profitable: best = max(profitable, key=lambda x: x['results']['total_pnl_pct']) print(f"\n🏆 MEILLEUR:") print(f" PnL: {best['results']['total_pnl_pct']:+.2f}%") print(f" PF: {best['results']['profit_factor']:.2f}") print(f" Trades: {best['results']['total_trades']}") print(f" Trailing: {int(best['params']['trailing_pct']*100)}%") print(f" Leverage: {best['params']['leverage']}x") print(f"{'='*80}\n") def _print_final_summary(self): """Résumé final""" print(f"\n{'='*80}") print(f"📊 RÉSUMÉ FINAL") print(f"={'='*80}\n") profitable = [r for r in self.results if r['results']['total_pnl_pct'] > 0 and r['results']['profit_factor'] > 1.0] print(f"Tests effectués: {len(self.results)}") print(f"Profitables: {len(profitable)} ({len(profitable)/len(self.results)*100:.1f}%)\n") if profitable: # Top 10 profitable.sort(key=lambda x: x['results']['total_pnl_pct'], reverse=True) print(f"🏆 TOP 10 PAR PnL:\n") for i, r in enumerate(profitable[:10], 1): res = r['results'] p = r['params'] print(f"{i:2}. PnL: {res['total_pnl_pct']:+7.2f}% | " f"PF: {res['profit_factor']:5.2f} | " f"Trades: {res['total_trades']:3} | " f"Trail:{int(p['trailing_pct']*100):2}% " f"Lev:{p['leverage']:3}x " f"RSI:{p['rsi_oversold']:2}/{p['rsi_overbought']:2}") print(f"\n💾 Résultats sauvegardés: {self.results_file}") print(f"{'='*80}") if __name__ == '__main__': # Charger données 1m df = pd.read_csv('historical_data/BTCUSDT_1m_60000candles.csv') df['timestamp'] = pd.to_datetime(df['timestamp']) # Créer optimizer optimizer = SmartOptimizer(df, initial_capital=500.0) # GO! optimizer.optimize_forever()