Deepvest/ProfilingAI

1# src/analysis/enhanced_backtest.py2 3import pandas as pd4import numpy as np5from typing import Dict, List, Optional6from datetime import datetime, timedelta7from dataclasses import dataclass8 9from src.models.signals import StrategySignal10 11@dataclass12class EnhancedBacktestResult:13    """Résultats détaillés du backtest"""14    returns: pd.Series15    positions: pd.DataFrame16    trades: pd.DataFrame17    signals: pd.DataFrame18    performance_metrics: Dict[str, float]19    risk_metrics: Dict[str, float]20    alternative_contribution: Dict[str, float]21    signal_quality: Dict[str, float]22 23class EnhancedBacktestEngine:24    """Moteur de backtest avancé pour la stratégie améliorée"""25    26    def __init__(self, 27                 enhanced_strategy,28                 risk_manager,29                 data_fetcher,30                 initial_capital: float = 1_000_000):31        self.strategy = enhanced_strategy32        self.risk_manager = risk_manager33        self.data_fetcher = data_fetcher34        self.initial_capital = initial_capital35        36    async def run_enhanced_backtest(self,37                                  symbols: List[str],38                                  start_date: datetime,39                                  end_date: datetime,40                                  use_alternative_data: bool = True) -> EnhancedBacktestResult:41        """Exécute un backtest complet avec données alternatives"""42        try:43            # Récupération des données44            market_data = await self.data_fetcher.fetch_market_data(symbols, start_date, end_date)45            46            # Initialisation des structures de résultats47            results = self._initialize_results(market_data.index, symbols)48            portfolio_value = self.initial_capital49            current_positions = {symbol: 0 for symbol in symbols}50            51            # Boucle principale du backtest52            for date in market_data.index:53                try:54                    # 1. Récupération des données alternatives pour cette date55                    if use_alternative_data:56                        alternative_data = await self._get_historical_alternative_data(57                            symbols, date58                        )59                    else:60                        alternative_data = {}61                    62                    # 2. Génération des signaux63                    signals = await self.strategy.generate_trade_signals(64                        portfolio=current_positions,65                        market_data=market_data.loc[:date],66                        alternative_data=alternative_data67                    )68                    69                    # 3. Exécution des trades70                    trades = self._execute_signals(71                        signals,72                        current_positions,73                        portfolio_value,74                        market_data.loc[date]75                    )76                    77                    # 4. Mise à jour du portefeuille78                    portfolio_value, current_positions = self._update_portfolio(79                        portfolio_value,80                        current_positions,81                        trades,82                        market_data.loc[date]83                    )84                    85                    # 5. Enregistrement des résultats86                    self._record_results(87                        results,88                        date,89                        portfolio_value,90                        current_positions,91                        trades,92                        signals93                    )94                    95                except Exception as e:96                    print(f"Erreur pendant le backtest à la date {date}: {e}")97                    continue98            99            # Calcul des métriques finales100            return self._calculate_final_metrics(results, market_data)101            102        except Exception as e:103            print(f"Erreur dans le backtest: {e}")104            raise105 106    def _initialize_results(self, dates: pd.DatetimeIndex, symbols: List[str]) -> Dict:107        """Initialise les structures de données pour les résultats"""108        return {109            'portfolio_value': pd.Series(index=dates, dtype=float),110            'positions': pd.DataFrame(index=dates, columns=symbols, dtype=float),111            'trades': pd.DataFrame(columns=['symbol', 'type', 'size', 'price', 'cost']),112            'signals': pd.DataFrame(columns=['symbol', 'direction', 'confidence', 'size']),113            'alternative_signals': pd.DataFrame(columns=['symbol', 'signal_type', 'value'])114        }115 116    async def _get_historical_alternative_data(self,117                                             symbols: List[str],118                                             date: datetime) -> Dict:119        """Récupère les données alternatives historiques"""120        try:121            # Satellite data122            satellite_data = await self.data_fetcher.fetch_historical_satellite_data(123                symbols, date124            )125            126            # Social media data127            social_data = await self.data_fetcher.fetch_historical_social_data(128                symbols, date129            )130            131            # Web traffic data132            traffic_data = await self.data_fetcher.fetch_historical_traffic_data(133                symbols, date134            )135            136            return {137                'satellite': satellite_data,138                'social_media': social_data,139                'web_traffic': traffic_data140            }141            142        except Exception as e:143            print(f"Erreur récupération données alternatives: {e}")144            return {}145 146    def _execute_signals(self,147                        signals: List[StrategySignal],148                        current_positions: Dict[str, float],149                        portfolio_value: float,150                        market_data: pd.Series) -> List[Dict]:151        """Exécute les signaux de trading"""152        trades = []153        154        for signal in signals:155            try:156                if signal.direction == 'buy':157                    size = self._calculate_buy_size(158                        signal, portfolio_value, current_positions159                    )160                    if size > 0:161                        trades.append({162                            'symbol': signal.symbol,163                            'type': 'buy',164                            'size': size,165                            'price': market_data[signal.symbol],166                            'cost': size * market_data[signal.symbol] * 0.001  # 0.1% de coût167                        })168                        169                elif signal.direction == 'sell':170                    size = self._calculate_sell_size(171                        signal, current_positions172                    )173                    if size > 0:174                        trades.append({175                            'symbol': signal.symbol,176                            'type': 'sell',177                            'size': size,178                            'price': market_data[signal.symbol],179                            'cost': size * market_data[signal.symbol] * 0.001180                        })181                        182            except Exception as e:183                print(f"Erreur exécution signal {signal.symbol}: {e}")184                continue185                186        return trades187 188    def _calculate_final_metrics(self, results: Dict, market_data: pd.DataFrame) -> EnhancedBacktestResult:189        """Calcule les métriques finales du backtest"""190        try:191            # Calcul des rendements192            portfolio_returns = results['portfolio_value'].pct_change().dropna()193            194            # Métriques de performance195            performance_metrics = {196                'total_return': (results['portfolio_value'].iloc[-1] / self.initial_capital) - 1,197                'annual_return': self._calculate_annual_return(portfolio_returns),198                'sharpe_ratio': self._calculate_sharpe_ratio(portfolio_returns),199                'max_drawdown': self._calculate_max_drawdown(portfolio_returns)200            }201            202            # Métriques de risque203            risk_metrics = self.risk_manager.calculate_risk_metrics(portfolio_returns)204            205            # Contribution des données alternatives206            alternative_contribution = self._calculate_alternative_contribution(207                results['signals'], results['alternative_signals']208            )209            210            # Qualité des signaux211            signal_quality = self._evaluate_signal_quality(212                results['signals'], portfolio_returns213            )214            215            return EnhancedBacktestResult(216                returns=portfolio_returns,217                positions=results['positions'],218                trades=results['trades'],219                signals=results['signals'],220                performance_metrics=performance_metrics,221                risk_metrics=risk_metrics,222                alternative_contribution=alternative_contribution,223                signal_quality=signal_quality224            )225            226        except Exception as e:227            print(f"Erreur calcul métriques finales: {e}")228            raise