""" Parameter Optimizer - Optimisation des Paramètres avec Optuna. Optimise automatiquement les paramètres des stratégies en utilisant Optuna pour éviter l'overfitting: - Bayesian optimization (TPE) - Walk-forward validation out-of-sample - Vraie simulation signal→SL/TP (plus de random) - Pruning pour accélérer """ from typing import Dict, List, Optional import pandas as pd import numpy as np from datetime import datetime import logging try: import optuna optuna.logging.set_verbosity(optuna.logging.WARNING) OPTUNA_AVAILABLE = True except ImportError: OPTUNA_AVAILABLE = False logging.warning("optuna non installé. Installer avec : pip install optuna") logger = logging.getLogger(__name__) # Barres max par trial (compromis vitesse / précision) _BACKTEST_BARS = 1200 class ParameterOptimizer: """ Optimiseur de paramètres utilisant Optuna. Évalue chaque combinaison de paramètres via une vraie simulation signal→SL/TP sur données historiques (pas de PnL aléatoire). Usage: optimizer = ParameterOptimizer(ScalpingStrategy, df) result = optimizer.optimize(n_trials=50) best_params = result['best_params'] """ def __init__( self, strategy_class, data: pd.DataFrame, initial_capital: float = 10000.0, ): """ Initialise l'optimiseur. Args: strategy_class: Classe de la stratégie à optimiser data: Données historiques OHLCV (index datetime) initial_capital: Capital initial pour la simulation """ if not OPTUNA_AVAILABLE: logger.error("Optuna non disponible !") return self.strategy_class = strategy_class # Subset pour la vitesse : dernières _BACKTEST_BARS barres self.data = ( data.iloc[-_BACKTEST_BARS:].copy() if len(data) > _BACKTEST_BARS else data.copy() ) self.initial_capital = initial_capital self.primary_metric = 'sharpe_ratio' self.constraints = { 'min_sharpe': 0.0, # Positif suffit (filtre walk-forward après) 'max_drawdown': 0.20, 'min_win_rate': 0.40, 'min_trades': 5, } logger.info( f"ParameterOptimizer initialisé pour {strategy_class.__name__} " f"({len(self.data)} barres)" ) # ------------------------------------------------------------------ # Interface publique # ------------------------------------------------------------------ def optimize( self, n_trials: int = 50, timeout: Optional[int] = None, n_jobs: int = 1, ) -> Dict: """ Lance l'optimisation Optuna. Args: n_trials: Nombre de trials Optuna timeout: Timeout en secondes (None = pas de limite) n_jobs: Parallélisme (1 = séquentiel recommandé) Returns: {best_params, best_value, walk_forward_results, n_trials_done} """ if not OPTUNA_AVAILABLE: return {} logger.info("=" * 60) logger.info("DÉMARRAGE OPTIMISATION PARAMÈTRES") logger.info("=" * 60) logger.info(f"Stratégie : {self.strategy_class.__name__}") logger.info(f"Trials : {n_trials} | Données : {len(self.data)} barres") study = optuna.create_study( direction='maximize', sampler=optuna.samplers.TPESampler(seed=42), pruner=optuna.pruners.MedianPruner(n_warmup_steps=10), ) study.optimize( self._objective, n_trials=n_trials, timeout=timeout, n_jobs=n_jobs, show_progress_bar=False, ) best_params = study.best_params best_value = study.best_value logger.info("=" * 60) logger.info("OPTIMISATION TERMINÉE") logger.info(f"Meilleur {self.primary_metric} : {best_value:.4f}") logger.info(f"Meilleurs paramètres : {best_params}") logger.info("Validation walk-forward en cours…") wf_results = self._walk_forward_validation(best_params) logger.info( f"WF Sharpe moyen : {wf_results['avg_sharpe']:.2f} " f"Stabilité : {wf_results['stability']:.2%}" ) return { 'best_params': best_params, 'best_value': best_value, 'walk_forward_results': wf_results, 'n_trials_done': len(study.trials), } # ------------------------------------------------------------------ # Fonction objectif Optuna # ------------------------------------------------------------------ def _objective(self, trial: 'optuna.Trial') -> float: params = self._suggest_parameters(trial) strategy = self.strategy_class(params) metrics = self._backtest_strategy(strategy, self.data) sharpe = metrics.get('sharpe_ratio', -999.0) trial.report(sharpe, step=0) if trial.should_prune(): raise optuna.exceptions.TrialPruned() # Pénalité sévère uniquement si trop peu de trades (non significatif) if metrics.get('total_trades', 0) < self.constraints['min_trades']: return -999.0 # Pénaliser le drawdown excessif mais retourner le vrai sharpe sinon if metrics.get('max_drawdown', 1.0) > self.constraints['max_drawdown']: return sharpe - 5.0 return sharpe # ------------------------------------------------------------------ # Suggestion de paramètres (aplatis dans le dict config) # ------------------------------------------------------------------ def _suggest_parameters(self, trial: 'optuna.Trial') -> Dict: """ Suggère des paramètres passés directement à strategy_class(config). Les paramètres sont aplatis (pas de clé 'adaptive_params' imbriquée). """ strategy_name = self.strategy_class.__name__.lower() params: Dict = { 'name': strategy_name, 'timeframe': '1h', 'risk_per_trade': trial.suggest_float('risk_per_trade', 0.003, 0.015), 'max_holding_time': 28800, } if 'scalping' in strategy_name: params.update({ 'bb_period': trial.suggest_int('bb_period', 10, 30), 'bb_std': trial.suggest_float('bb_std', 1.5, 3.0), 'rsi_period': trial.suggest_int('rsi_period', 8, 21), 'rsi_oversold': trial.suggest_int('rsi_oversold', 20, 38), 'rsi_overbought': trial.suggest_int('rsi_overbought', 62, 82), 'volume_threshold': trial.suggest_float('volume_threshold', 1.0, 2.5), 'min_confidence': trial.suggest_float('min_confidence', 0.45, 0.80), }) elif 'intraday' in strategy_name: params.update({ 'ema_fast': trial.suggest_int('ema_fast', 5, 15), 'ema_slow': trial.suggest_int('ema_slow', 15, 30), 'ema_trend': trial.suggest_int('ema_trend', 40, 60), 'atr_multiplier': trial.suggest_float('atr_multiplier', 1.5, 3.5), 'volume_confirmation': trial.suggest_float('volume_confirmation', 1.0, 1.5), 'min_confidence': trial.suggest_float('min_confidence', 0.45, 0.75), 'adx_threshold': trial.suggest_int('adx_threshold', 18, 35), }) elif 'swing' in strategy_name: params.update({ 'sma_short': trial.suggest_int('sma_short', 15, 30), 'sma_long': trial.suggest_int('sma_long', 40, 60), 'rsi_period': trial.suggest_int('rsi_period', 10, 20), 'fibonacci_lookback': trial.suggest_int('fibonacci_lookback', 30, 70), 'min_confidence': trial.suggest_float('min_confidence', 0.40, 0.70), 'atr_multiplier': trial.suggest_float('atr_multiplier', 2.0, 4.0), }) return params # ------------------------------------------------------------------ # Simulation réelle signal → SL/TP # ------------------------------------------------------------------ def _backtest_strategy(self, strategy, data: pd.DataFrame) -> Dict: """ Simule la stratégie sur `data` avec vraie logique SL/TP. Une seule position ouverte à la fois. La position est clôturée dès que le prix atteint SL ou TP sur la barre courante. Clôture forcée à la dernière barre si position encore ouverte. Returns: {sharpe_ratio, max_drawdown, win_rate, total_trades, total_return} """ equity = self.initial_capital equity_curve = [equity] trades: List[Dict] = [] in_position = False position = None # {entry, sl, tp, direction, size} for i in range(50, len(data)): bar = data.iloc[i] # --- Gestion position ouverte --- if in_position and position is not None: closed, pnl = self._check_exit(position, bar) if closed: equity += pnl trades.append({'pnl': pnl, 'win': pnl > 0}) in_position = False position = None equity_curve.append(equity) continue # une seule position à la fois # --- Recherche d'un signal --- try: hist = data.iloc[:i + 1] signal = strategy.analyze(hist) if signal is not None: stop_dist = abs(signal.entry_price - signal.stop_loss) if stop_dist > 0: risk_amt = equity * getattr(strategy.config, 'risk_per_trade', 0.005) size = risk_amt / stop_dist in_position = True position = { 'entry': signal.entry_price, 'sl': signal.stop_loss, 'tp': signal.take_profit, 'direction': signal.direction, 'size': size, } except Exception: pass equity_curve.append(equity) # Clôture forcée au dernier close if in_position and position is not None: last_close = data.iloc[-1]['close'] if position['direction'] == 'LONG': pnl = (last_close - position['entry']) * position['size'] else: pnl = (position['entry'] - last_close) * position['size'] equity += pnl trades.append({'pnl': pnl, 'win': pnl > 0}) return self._compute_metrics(equity, equity_curve, trades) def _check_exit(self, position: Dict, bar: pd.Series): """ Vérifie si SL ou TP est atteint sur la barre courante. Retourne (clôturé: bool, pnl: float). """ high = bar['high'] low = bar['low'] entry = position['entry'] sl = position['sl'] tp = position['tp'] size = position['size'] if position['direction'] == 'LONG': if low <= sl: return True, (sl - entry) * size if high >= tp: return True, (tp - entry) * size else: # SHORT if high >= sl: return True, (entry - sl) * size if low <= tp: return True, (entry - tp) * size return False, 0.0 def _compute_metrics( self, final_equity: float, equity_curve: List, trades: List[Dict], ) -> Dict: """Calcule les métriques de performance à partir des trades.""" if len(trades) == 0: return { 'sharpe_ratio': -1.0, 'max_drawdown': 1.0, 'win_rate': 0.0, 'total_trades': 0, 'total_return': 0.0, } returns = pd.Series([t['pnl'] for t in trades]) sharpe = ( float(returns.mean() / returns.std() * np.sqrt(252)) if returns.std() > 0 else 0.0 ) win_rate = float(sum(1 for t in trades if t['win']) / len(trades)) equity_s = pd.Series(equity_curve) running_max = equity_s.expanding().max() max_dd = float(abs(((equity_s - running_max) / running_max).min())) return { 'sharpe_ratio': sharpe, 'max_drawdown': max_dd, 'win_rate': win_rate, 'total_trades': len(trades), 'total_return': float((final_equity - self.initial_capital) / self.initial_capital), } # ------------------------------------------------------------------ # Vérification des contraintes # ------------------------------------------------------------------ def _check_constraints(self, metrics: Dict) -> bool: return ( metrics.get('sharpe_ratio', -999) >= self.constraints['min_sharpe'] and metrics.get('max_drawdown', 1.0) <= self.constraints['max_drawdown'] and metrics.get('win_rate', 0.0) >= self.constraints['min_win_rate'] and metrics.get('total_trades', 0) >= self.constraints['min_trades'] ) # ------------------------------------------------------------------ # Walk-forward validation (out-of-sample) # ------------------------------------------------------------------ def _walk_forward_validation(self, best_params: Dict, n_folds: int = 4) -> Dict: """ Valide les meilleurs paramètres sur des fenêtres out-of-sample. Train sur fold i, test sur fold i+1 (glissant). """ sharpe_ratios = [] fold_size = len(self.data) // n_folds for i in range(n_folds - 1): test_data = self.data.iloc[(i + 1) * fold_size:(i + 2) * fold_size] if len(test_data) < 60: continue try: strategy = self.strategy_class(dict(best_params)) metrics = self._backtest_strategy(strategy, test_data) sharpe_ratios.append(metrics['sharpe_ratio']) except Exception as exc: logger.warning(f"WF fold {i} échoué : {exc}") if not sharpe_ratios: return { 'avg_sharpe': 0.0, 'std_sharpe': 0.0, 'stability': 0.0, 'sharpe_ratios': [], } avg_sharpe = float(np.mean(sharpe_ratios)) std_sharpe = float(np.std(sharpe_ratios)) stability = float( max(0.0, 1.0 - (std_sharpe / avg_sharpe if avg_sharpe > 0 else 1.0)) ) return { 'avg_sharpe': avg_sharpe, 'std_sharpe': std_sharpe, 'stability': stability, 'sharpe_ratios': sharpe_ratios, }