Deepvest/ProfilingAI

1# src/analysis/strategy_optimizer.py2 3import numpy as np4import pandas as pd5from typing import Dict, List, Tuple6from dataclasses import dataclass7import optuna8import matplotlib.pyplot as plt9import seaborn as sns10from concurrent.futures import ProcessPoolExecutor11from datetime import datetime, timedelta12 13from src.models.backtest_results import EnhancedBacktestResult14 15 16@dataclass17class OptimizationResult:18    """Résultats de l'optimisation de la stratégie"""19    best_params: Dict[str, float]20    performance_metrics: Dict[str, float]21    optimization_path: pd.DataFrame22    parameter_importance: Dict[str, float]23 24class StrategyOptimizer:25    """Optimiseur de stratégie utilisant Optuna"""26    27    def __init__(self, enhanced_backtest_engine, n_trials: int = 100):28        self.backtest_engine = enhanced_backtest_engine29        self.n_trials = n_trials30        self.study = None31        32    async def optimize_strategy(self,33                              symbols: List[str],34                              start_date: pd.Timestamp,35                              end_date: pd.Timestamp) -> OptimizationResult:36        """Optimise les paramètres de la stratégie"""37        38        # Création de l'étude Optuna39        self.study = optuna.create_study(40            direction="maximize",41            study_name="strategy_optimization",42            sampler=optuna.samplers.TPESampler(seed=42)43        )44        45        # Exécution de l'optimisation46        self.study.optimize(47            lambda trial: self._objective(trial, symbols, start_date, end_date),48            n_trials=self.n_trials,49            show_progress_bar=True50        )51        52        # Analyse des résultats53        optimization_results = self._analyze_optimization_results()54        55        # Visualisation des résultats56        self._plot_optimization_results()57        58        return optimization_results59    60    async def _objective(self, 61                  trial: optuna.Trial,62                  symbols: List[str],63                  start_date: pd.Timestamp,64                  end_date: pd.Timestamp) -> float:65        """Fonction objectif pour l'optimisation"""66        67        # Paramètres à optimiser68        params = {69            'technical_weight': trial.suggest_float('technical_weight', 0.2, 0.6),70            'alternative_weight': trial.suggest_float('alternative_weight', 0.1, 0.4),71            'social_weight': trial.suggest_float('social_weight', 0.1, 0.4),72            'signal_threshold': trial.suggest_float('signal_threshold', 0.3, 0.7),73            'position_size_factor': trial.suggest_float('position_size_factor', 0.5, 2.0),74            'stop_loss': trial.suggest_float('stop_loss', 0.02, 0.10),75            'take_profit': trial.suggest_float('take_profit', 0.03, 0.15)76        }77        78        # Exécution du backtest avec les paramètres actuels79        results = await self.backtest_engine.run_enhanced_backtest(80            symbols=symbols,81            start_date=start_date,82            end_date=end_date,83            strategy_params=params84        )85        86        # Calcul du score d'optimisation87        optimization_score = self._calculate_optimization_score(results)88        89        return optimization_score90    91    def _calculate_optimization_score(self, results: Dict) -> float:92        """Calcule le score pour l'optimisation"""93        # Extraction des métriques94        sharpe_ratio = results.performance_metrics['sharpe_ratio']95        max_drawdown = abs(results.performance_metrics['max_drawdown'])96        return_risk_ratio = results.performance_metrics['annual_return'] / max_drawdown97        98        # Combinaison pondérée des métriques99        score = (0.4 * sharpe_ratio + 100                0.3 * return_risk_ratio + 101                0.3 * (1 / (1 + max_drawdown)))102        103        return score104 105    def _analyze_optimization_results(self) -> OptimizationResult:106        """Analyse des résultats de l'optimisation"""107        # Meilleurs paramètres108        best_params = self.study.best_params109        110        # Chemin d'optimisation111        optimization_path = pd.DataFrame(112            [t.params for t in self.study.trials],113            index=[t.number for t in self.study.trials]114        )115        116        # Importance des paramètres117        parameter_importance = optuna.importance.get_param_importances(self.study)118        119        return OptimizationResult(120            best_params=best_params,121            performance_metrics=self.study.best_value,122            optimization_path=optimization_path,123            parameter_importance=parameter_importance124        )125 126    def _plot_optimization_results(self):127        """Visualisation des résultats d'optimisation"""128        # Configuration du style129        plt.style.use('seaborn')130        fig = plt.figure(figsize=(15, 10))131        132        # 1. Évolution de l'optimisation133        plt.subplot(221)134        optuna.visualization.matplotlib.plot_optimization_history(self.study)135        plt.title('Progression de l\'optimisation')136        137        # 2. Importance des paramètres138        plt.subplot(222)139        optuna.visualization.matplotlib.plot_param_importances(self.study)140        plt.title('Importance des paramètres')141        142        # 3. Corrélations entre paramètres143        plt.subplot(223)144        optuna.visualization.matplotlib.plot_parallel_coordinate(self.study)145        plt.title('Corrélations des paramètres')146        147        # 4. Distribution des meilleurs paramètres148        plt.subplot(224)149        data = pd.DataFrame(150            [t.params for t in self.study.trials],151            columns=self.study.best_params.keys()152        )153        sns.boxplot(data=data)154        plt.xticks(rotation=45)155        plt.title('Distribution des paramètres')156        157        plt.tight_layout()158        plt.show()159 160# Visualisation détaillée des résultats de backtest161class BacktestVisualizer:162    """Visualisation détaillée des résultats de backtest"""163    164    @staticmethod165    def create_performance_dashboard(results: EnhancedBacktestResult):166        """Création d'un dashboard de performance complet"""167        plt.style.use('seaborn')168        fig = plt.figure(figsize=(20, 12))169        170        # 1. Courbe de croissance du portefeuille171        plt.subplot(331)172        cumulative_returns = (1 + results.returns).cumprod()173        plt.plot(cumulative_returns.index, cumulative_returns.values)174        plt.title('Performance du portefeuille')175        176        # 2. Drawdowns177        plt.subplot(332)178        drawdowns = BacktestVisualizer._calculate_drawdowns(results.returns)179        plt.fill_between(drawdowns.index, drawdowns.values, 0, color='red', alpha=0.3)180        plt.title('Drawdowns')181        182        # 3. Distribution des rendements183        plt.subplot(333)184        sns.histplot(results.returns, kde=True)185        plt.title('Distribution des rendements')186        187        # 4. Heat map des positions188        plt.subplot(334)189        sns.heatmap(results.positions.T, cmap='RdYlGn', center=0)190        plt.title('Évolution des positions')191        192        # 5. Contribution par signal193        plt.subplot(335)194        signal_returns = BacktestVisualizer._calculate_signal_returns(results)195        sns.barplot(data=signal_returns)196        plt.title('Performance par type de signal')197        198        # 6. Métriques de risque199        plt.subplot(336)200        risk_metrics = pd.Series(results.risk_metrics)201        sns.barplot(x=risk_metrics.index, y=risk_metrics.values)202        plt.xticks(rotation=45)203        plt.title('Métriques de risque')204        205        # 7. Impact des données alternatives206        plt.subplot(337)207        alt_contribution = pd.Series(results.alternative_contribution)208        sns.barplot(x=alt_contribution.index, y=alt_contribution.values)209        plt.xticks(rotation=45)210        plt.title('Contribution des données alternatives')211        212        plt.tight_layout()213        plt.show()214 215    @staticmethod216    def _calculate_drawdowns(returns: pd.Series) -> pd.Series:217        """Calcul des drawdowns"""218        cumulative = (1 + returns).cumprod()219        running_max = cumulative.expanding().max()220        drawdowns = cumulative / running_max - 1221        return drawdowns222 223    @staticmethod224    def _calculate_signal_returns(results: EnhancedBacktestResult) -> pd.DataFrame:225        """Calcul des rendements par type de signal"""226        signal_returns = pd.DataFrame()227        for signal_type in results.signals['direction'].unique():228            mask = results.signals['direction'] == signal_type229            signal_returns[signal_type] = results.returns[mask].mean()230        return signal_returns