Deepvest/ProfilingAI

1# src/analysis/scenario_manager.py2 3import numpy as np4import pandas as pd5from typing import Dict, List, Optional6from dataclasses import dataclass7from datetime import datetime, timedelta8from enum import Enum9 10class MarketScenario(Enum):11    """Types de scénarios de marché"""12    BULL_MARKET = "bull_market"13    BEAR_MARKET = "bear_market"14    HIGH_VOLATILITY = "high_volatility"15    LOW_VOLATILITY = "low_volatility"16    CRISIS = "crisis"17    RECOVERY = "recovery"18    SIDEWAYS = "sideways"19    HIGH_INFLATION = "high_inflation"20    STAGFLATION = "stagflation"21 22@dataclass23class ScenarioConfig:24    """Configuration d'un scénario"""25    scenario_type: MarketScenario26    duration: int  # en jours27    severity: float  # 0 à 128    parameters: Dict[str, float]29    macro_conditions: Dict[str, float]30 31class ScenarioManager:32    """Gestionnaire de scénarios de marché"""33    34    def __init__(self, data_fetcher, market_analyzer, risk_manager):35        self.data_fetcher = data_fetcher36        self.market_analyzer = market_analyzer37        self.risk_manager = risk_manager38        39    def generate_scenario(self, base_data: pd.DataFrame, config: ScenarioConfig) -> pd.DataFrame:40        """Génère un scénario de marché spécifique"""41        try:42            # Cloner les données de base43            scenario_data = base_data.copy()44            45            # Appliquer les modifications selon le scénario46            if config.scenario_type == MarketScenario.BULL_MARKET:47                scenario_data = self._generate_bull_market(scenario_data, config)48            elif config.scenario_type == MarketScenario.BEAR_MARKET:49                scenario_data = self._generate_bear_market(scenario_data, config)50            elif config.scenario_type == MarketScenario.CRISIS:51                scenario_data = self._generate_crisis(scenario_data, config)52            # ... autres scénarios53            54            # Ajouter les conditions macro-économiques55            scenario_data = self._add_macro_conditions(scenario_data, config.macro_conditions)56            57            return scenario_data58            59        except Exception as e:60            print(f"Erreur dans la génération du scénario: {e}")61            return base_data62 63    async def test_strategy_under_scenarios(self,64                                          strategy,65                                          scenarios: List[ScenarioConfig],66                                          base_portfolio: Dict[str, float]) -> Dict:67        """Teste la stratégie sous différents scénarios"""68        results = {}69        70        for scenario_config in scenarios:71            try:72                # Générer les données du scénario73                scenario_data = self.generate_scenario(74                    base_data=await self._get_base_data(),75                    config=scenario_config76                )77                78                # Tester la stratégie79                scenario_results = await self._backtest_scenario(80                    strategy=strategy,81                    scenario_data=scenario_data,82                    base_portfolio=base_portfolio,83                    config=scenario_config84                )85                86                results[scenario_config.scenario_type.value] = scenario_results87                88            except Exception as e:89                print(f"Erreur test scénario {scenario_config.scenario_type}: {e}")90                91        return results92 93    def generate_stress_scenarios(self) -> List[ScenarioConfig]:94        """Génère une série de scénarios de stress"""95        return [96            ScenarioConfig(97                scenario_type=MarketScenario.CRISIS,98                duration=63,  # ~3 mois99                severity=0.8,100                parameters={101                    'volatility_increase': 2.5,102                    'correlation_increase': 0.4,103                    'liquidity_decrease': 0.6104                },105                macro_conditions={106                    'gdp_growth': -0.03,107                    'inflation_rate': 0.06,108                    'interest_rate': 0.04109                }110            ),111            # ... autres scénarios de stress112        ]113 114    def _generate_bull_market(self, data: pd.DataFrame, config: ScenarioConfig) -> pd.DataFrame:115        """Génère un scénario de marché haussier"""116        try:117            # Calculer les rendements de base118            returns = data.pct_change()119            120            # Ajuster les rendements pour le marché haussier121            bull_adjustment = 0.0002 * config.severity  # +20bp par jour en moyenne122            adjusted_returns = returns + bull_adjustment123            124            # Ajuster la volatilité125            vol_adjustment = 0.8  # Réduction de la volatilité126            adjusted_returns = adjusted_returns * vol_adjustment127            128            # Reconstruire les prix129            return (1 + adjusted_returns).cumprod() * data.iloc[0]130            131        except Exception as e:132            print(f"Erreur génération bull market: {e}")133            return data134 135    def _generate_crisis(self, data: pd.DataFrame, config: ScenarioConfig) -> pd.DataFrame:136        """Génère un scénario de crise"""137        try:138            # Phase 1: Chute initiale rapide139            initial_drop = -0.15 * config.severity140            data_crisis = data.copy()141            drop_period = int(config.duration * 0.3)142            143            # Générer la chute144            drop_returns = np.linspace(0, initial_drop, drop_period)145            data_crisis.iloc[:drop_period] *= (1 + drop_returns).cumprod()146            147            # Phase 2: Haute volatilité148            vol_period = data_crisis.iloc[drop_period:]149            increased_vol = vol_period.pct_change() * (1 + config.parameters['volatility_increase'])150            data_crisis.iloc[drop_period:] = (1 + increased_vol).cumprod() * vol_period.iloc[0]151            152            return data_crisis153            154        except Exception as e:155            print(f"Erreur génération crise: {e}")156            return data157 158    def _add_macro_conditions(self, data: pd.DataFrame, macro_conditions: Dict[str, float]) -> pd.DataFrame:159        """Ajoute les conditions macro-économiques aux données"""160        data_with_macro = data.copy()161        162        for condition, value in macro_conditions.items():163            data_with_macro[f'macro_{condition}'] = value164            165        return data_with_macro166 167    async def _backtest_scenario(self,168                               strategy,169                               scenario_data: pd.DataFrame,170                               base_portfolio: Dict[str, float],171                               config: ScenarioConfig) -> Dict:172        """Exécute un backtest sur un scénario spécifique"""173        try:174            # Ajuster les paramètres du risk manager pour le scénario175            self.risk_manager.adjust_for_scenario(config)176            177            # Exécuter le backtest178            results = await strategy.backtest(179                data=scenario_data,180                initial_portfolio=base_portfolio,181                risk_manager=self.risk_manager182            )183            184            # Analyser les résultats185            analysis = self._analyze_scenario_results(results, config)186            187            return {188                'config': config,189                'results': results,190                'analysis': analysis191            }192            193        except Exception as e:194            print(f"Erreur backtest scénario: {e}")195            return {}196 197    def _analyze_scenario_results(self, results: Dict, config: ScenarioConfig) -> Dict:198        """Analyse les résultats d'un scénario"""199        return {200            'max_drawdown': self._calculate_max_drawdown(results['returns']),201            'recovery_time': self._calculate_recovery_time(results['returns']),202            'risk_adjusted_return': self._calculate_risk_adjusted_return(results['returns']),203            'strategy_adaptation': self._evaluate_strategy_adaptation(results, config),204            'risk_management_effectiveness': self._evaluate_risk_management(results)205        }