# ⚠️ Framework de Risk Management - Trading AI Secure ## 📋 Table des Matières 1. [Philosophie du Risk Management](#philosophie-du-risk-management) 2. [Architecture Multi-Niveaux](#architecture-multi-niveaux) 3. [Limites et Contraintes](#limites-et-contraintes) 4. [Risk Manager Core](#risk-manager-core) 5. [Validation Pré-Trade](#validation-pré-trade) 6. [Circuit Breakers](#circuit-breakers) 7. [Métriques de Risque](#métriques-de-risque) 8. [Implémentation Technique](#implémentation-technique) --- ## 🎯 Philosophie du Risk Management ### Principe Fondamental : "Survivre d'abord, Profiter ensuite" Le risk management est **intégré à tous les niveaux** du système, pas une couche ajoutée après coup. ``` ┌─────────────────────────────────────────────────────────┐ │ HIÉRARCHIE DES PRIORITÉS │ ├─────────────────────────────────────────────────────────┤ │ │ │ 1. 🛡️ PROTECTION DU CAPITAL (Priorité absolue) │ │ 2. 📉 LIMITATION DES PERTES (Drawdown < 10%) │ │ 3. 🎯 CONSISTANCE (Sharpe > volatilité) │ │ 4. 💰 PROFITABILITÉ (Objectif final) │ │ │ └─────────────────────────────────────────────────────────┘ ``` ### Règles d'Or 1. **Jamais de trade sans stop-loss** ❌ 2. **Jamais plus de 2% du capital en risque total** ❌ 3. **Jamais plus de 10% de drawdown** ❌ 4. **Toujours vérifier la corrélation** ✅ 5. **Toujours valider la liquidité** ✅ --- ## 🏗️ Architecture Multi-Niveaux ### 5 Niveaux de Protection ``` ┌──────────────────────────────────────────────────────────┐ │ NIVEAU 5: GLOBAL PORTFOLIO RISK │ │ ├─ Max total risk: 2% du capital │ │ ├─ Max drawdown: 10% │ │ └─ Daily loss limit: 3% │ ├──────────────────────────────────────────────────────────┤ │ NIVEAU 4: STRATEGY ALLOCATION │ │ ├─ Max per strategy: 33% du capital │ │ ├─ Correlation limit: 0.7 entre stratégies │ │ └─ Min diversification: 3 stratégies actives │ ├──────────────────────────────────────────────────────────┤ │ NIVEAU 3: POSITION SIZING │ │ ├─ Kelly Criterion adaptatif │ │ ├─ Max position: 5% du capital │ │ └─ Volatility-adjusted sizing │ ├──────────────────────────────────────────────────────────┤ │ NIVEAU 2: TRADE VALIDATION │ │ ├─ Stop-loss obligatoire │ │ ├─ Risk/Reward > 1.5 │ │ ├─ Liquidity check │ │ └─ Margin verification │ ├──────────────────────────────────────────────────────────┤ │ NIVEAU 1: CIRCUIT BREAKERS │ │ ├─ Emergency stop │ │ ├─ Volatility spike detection │ │ ├─ Flash crash protection │ │ └─ API failure handling │ └──────────────────────────────────────────────────────────┘ ``` --- ## 📊 Limites et Contraintes ### Limites Globales (Portfolio) ```yaml global_limits: # Capital max_portfolio_risk: 0.02 # 2% du capital total en risque max_position_size: 0.05 # 5% max par position max_total_exposure: 1.0 # 100% du capital (pas de levier) # Drawdown max_drawdown: 0.10 # 10% drawdown maximum max_daily_loss: 0.03 # 3% perte journalière max max_weekly_loss: 0.07 # 7% perte hebdomadaire max # Corrélation max_correlation: 0.7 # Corrélation max entre positions min_diversification: 3 # Minimum 3 positions non-corrélées # Liquidité min_daily_volume: 1000000 # 1M$ volume quotidien minimum max_position_vs_volume: 0.01 # Max 1% du volume quotidien # Concentration max_sector_exposure: 0.30 # 30% max par secteur max_asset_class_exposure: 0.50 # 50% max par classe d'actif ``` ### Limites par Stratégie ```yaml strategy_limits: scalping: max_trades_per_day: 50 risk_per_trade: 0.005 # 0.5% par trade max_holding_time: 1800 # 30 minutes max_slippage: 0.001 # 0.1% slippage max min_profit_target: 0.003 # 0.3% profit minimum intraday: max_trades_per_day: 10 risk_per_trade: 0.015 # 1.5% par trade max_holding_time: 86400 # 1 jour max_slippage: 0.002 # 0.2% slippage max min_profit_target: 0.01 # 1% profit minimum swing: max_trades_per_week: 5 risk_per_trade: 0.025 # 2.5% par trade max_holding_time: 432000 # 5 jours max_slippage: 0.003 # 0.3% slippage max min_profit_target: 0.03 # 3% profit minimum ``` ### Limites Dynamiques (Ajustées selon conditions) ```python class DynamicLimits: """ Limites ajustées selon conditions de marché """ def adjust_limits(self, market_conditions: Dict) -> Dict: """ Ajuste limites selon volatilité, drawdown, etc. """ base_limits = self.get_base_limits() # Réduire limites si haute volatilité if market_conditions['volatility'] > 0.03: # > 3% vol quotidienne base_limits['max_position_size'] *= 0.5 base_limits['max_portfolio_risk'] *= 0.7 # Réduire limites si drawdown élevé if market_conditions['current_drawdown'] > 0.05: # > 5% DD reduction_factor = 1 - (market_conditions['current_drawdown'] / 0.10) base_limits['max_position_size'] *= reduction_factor # Réduire limites si losing streak if market_conditions['consecutive_losses'] > 3: base_limits['max_trades_per_day'] //= 2 base_limits['risk_per_trade'] *= 0.5 return base_limits ``` --- ## 🛡️ Risk Manager Core ### Singleton Pattern ```python # src/core/risk_manager.py from typing import Dict, List, Optional from dataclasses import dataclass from datetime import datetime, timedelta import threading import numpy as np @dataclass class Position: """Représente une position ouverte""" symbol: str quantity: float entry_price: float current_price: float stop_loss: float take_profit: float strategy: str entry_time: datetime unrealized_pnl: float risk_amount: float @dataclass class RiskMetrics: """Métriques de risque en temps réel""" total_risk: float current_drawdown: float daily_pnl: float weekly_pnl: float portfolio_var: float portfolio_cvar: float correlation_matrix: np.ndarray largest_position: float num_positions: int class RiskManager: """ Risk Manager Central (Singleton) Responsabilités: - Validation pré-trade - Monitoring positions - Circuit breakers - Calcul métriques risque """ _instance = None _lock = threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def __init__(self): if not hasattr(self, 'initialized'): self.initialized = True # Configuration self.config = self._load_config() # État self.positions: Dict[str, Position] = {} self.daily_trades: List[Dict] = [] self.portfolio_value: float = 100000.0 # Initial capital self.peak_value: float = 100000.0 # Historique self.pnl_history: List[float] = [] self.drawdown_history: List[float] = [] # Circuit breakers self.trading_halted: bool = False self.halt_reason: Optional[str] = None def validate_trade( self, symbol: str, quantity: float, price: float, stop_loss: float, take_profit: float, strategy: str ) -> tuple[bool, Optional[str]]: """ Valide un trade avant exécution Returns: (is_valid, error_message) """ # 1. Vérifier si trading halted if self.trading_halted: return False, f"Trading halted: {self.halt_reason}" # 2. Vérifier stop-loss obligatoire if stop_loss is None or stop_loss == 0: return False, "Stop-loss is mandatory" # 3. Calculer risque du trade risk_amount = abs(price - stop_loss) * quantity risk_pct = risk_amount / self.portfolio_value # 4. Vérifier limites par trade max_risk_per_trade = self.config['strategy_limits'][strategy]['risk_per_trade'] if risk_pct > max_risk_per_trade: return False, f"Risk per trade ({risk_pct:.2%}) exceeds limit ({max_risk_per_trade:.2%})" # 5. Vérifier risque total portfolio total_risk = self._calculate_total_risk() + risk_amount max_portfolio_risk = self.config['global_limits']['max_portfolio_risk'] * self.portfolio_value if total_risk > max_portfolio_risk: return False, f"Total portfolio risk ({total_risk:.2f}) exceeds limit ({max_portfolio_risk:.2f})" # 6. Vérifier taille position position_value = price * quantity position_pct = position_value / self.portfolio_value max_position_size = self.config['global_limits']['max_position_size'] if position_pct > max_position_size: return False, f"Position size ({position_pct:.2%}) exceeds limit ({max_position_size:.2%})" # 7. Vérifier corrélation if not self._check_correlation(symbol, strategy): return False, "Correlation with existing positions too high" # 8. Vérifier nombre de trades quotidiens strategy_trades_today = len([t for t in self.daily_trades if t['strategy'] == strategy]) max_trades = self.config['strategy_limits'][strategy]['max_trades_per_day'] if strategy_trades_today >= max_trades: return False, f"Max daily trades for {strategy} reached ({max_trades})" # 9. Vérifier Risk/Reward ratio risk = abs(price - stop_loss) reward = abs(take_profit - price) rr_ratio = reward / risk if risk > 0 else 0 if rr_ratio < 1.5: return False, f"Risk/Reward ratio ({rr_ratio:.2f}) below minimum (1.5)" # 10. Vérifier drawdown actuel current_dd = self._calculate_current_drawdown() max_dd = self.config['global_limits']['max_drawdown'] if current_dd >= max_dd: return False, f"Max drawdown reached ({current_dd:.2%})" # Toutes validations passées return True, None def add_position(self, position: Position): """Ajoute une position au portfolio""" self.positions[position.symbol] = position # Enregistrer trade self.daily_trades.append({ 'symbol': position.symbol, 'strategy': position.strategy, 'time': position.entry_time, 'risk': position.risk_amount }) def update_position(self, symbol: str, current_price: float): """Met à jour prix d'une position""" if symbol in self.positions: position = self.positions[symbol] position.current_price = current_price position.unrealized_pnl = (current_price - position.entry_price) * position.quantity # Vérifier stop-loss / take-profit self._check_exit_conditions(position) def close_position(self, symbol: str, exit_price: float) -> float: """Ferme une position et retourne P&L""" if symbol not in self.positions: return 0.0 position = self.positions[symbol] pnl = (exit_price - position.entry_price) * position.quantity # Mettre à jour portfolio self.portfolio_value += pnl self.pnl_history.append(pnl) # Mettre à jour peak if self.portfolio_value > self.peak_value: self.peak_value = self.portfolio_value # Supprimer position del self.positions[symbol] return pnl def get_risk_metrics(self) -> RiskMetrics: """Calcule métriques de risque en temps réel""" return RiskMetrics( total_risk=self._calculate_total_risk(), current_drawdown=self._calculate_current_drawdown(), daily_pnl=self._calculate_daily_pnl(), weekly_pnl=self._calculate_weekly_pnl(), portfolio_var=self._calculate_var(), portfolio_cvar=self._calculate_cvar(), correlation_matrix=self._calculate_correlation_matrix(), largest_position=self._get_largest_position(), num_positions=len(self.positions) ) def _calculate_total_risk(self) -> float: """Calcule risque total du portfolio""" return sum(pos.risk_amount for pos in self.positions.values()) def _calculate_current_drawdown(self) -> float: """Calcule drawdown actuel""" if self.peak_value == 0: return 0.0 return (self.peak_value - self.portfolio_value) / self.peak_value def _calculate_daily_pnl(self) -> float: """Calcule P&L du jour""" today = datetime.now().date() daily_pnl = sum( pnl for pnl, time in zip(self.pnl_history, self.daily_trades) if time['time'].date() == today ) # Ajouter unrealized P&L unrealized = sum(pos.unrealized_pnl for pos in self.positions.values()) return daily_pnl + unrealized def _calculate_var(self, confidence=0.95) -> float: """ Calcule Value at Risk (VaR) VaR = perte maximale avec X% de confiance """ if len(self.pnl_history) < 30: return 0.0 returns = np.array(self.pnl_history[-30:]) / self.portfolio_value var = np.percentile(returns, (1 - confidence) * 100) return abs(var * self.portfolio_value) def _calculate_cvar(self, confidence=0.95) -> float: """ Calcule Conditional Value at Risk (CVaR) CVaR = perte moyenne au-delà du VaR """ if len(self.pnl_history) < 30: return 0.0 returns = np.array(self.pnl_history[-30:]) / self.portfolio_value var_threshold = np.percentile(returns, (1 - confidence) * 100) # Moyenne des pertes au-delà du VaR tail_losses = returns[returns <= var_threshold] cvar = np.mean(tail_losses) if len(tail_losses) > 0 else 0 return abs(cvar * self.portfolio_value) def _check_correlation(self, symbol: str, strategy: str) -> bool: """ Vérifie corrélation avec positions existantes """ if len(self.positions) == 0: return True # Simplification: vérifier si même stratégie # En production: calculer corrélation réelle des returns same_strategy_positions = [ pos for pos in self.positions.values() if pos.strategy == strategy ] max_correlation = self.config['global_limits']['max_correlation'] # Si trop de positions de même stratégie, corrélation trop haute if len(same_strategy_positions) >= 3: return False return True def _check_exit_conditions(self, position: Position): """Vérifie conditions de sortie (stop-loss / take-profit)""" # Stop-loss hit if position.current_price <= position.stop_loss: self.close_position(position.symbol, position.stop_loss) logger.warning(f"Stop-loss hit for {position.symbol}") # Take-profit hit elif position.current_price >= position.take_profit: self.close_position(position.symbol, position.take_profit) logger.info(f"Take-profit hit for {position.symbol}") def check_circuit_breakers(self): """ Vérifie conditions d'arrêt automatique """ # 1. Drawdown excessif current_dd = self._calculate_current_drawdown() if current_dd >= self.config['global_limits']['max_drawdown']: self.halt_trading(f"Max drawdown reached: {current_dd:.2%}") return # 2. Perte journalière excessive daily_pnl_pct = self._calculate_daily_pnl() / self.portfolio_value if daily_pnl_pct <= -self.config['global_limits']['max_daily_loss']: self.halt_trading(f"Max daily loss reached: {daily_pnl_pct:.2%}") return # 3. Volatilité extrême if self._detect_volatility_spike(): self.halt_trading("Extreme volatility detected") return def halt_trading(self, reason: str): """Arrête le trading""" self.trading_halted = True self.halt_reason = reason logger.critical(f"TRADING HALTED: {reason}") # Fermer toutes positions (optionnel) # self._close_all_positions() # Envoyer alertes self._send_emergency_alert(reason) def resume_trading(self): """Reprend le trading (manuel uniquement)""" self.trading_halted = False self.halt_reason = None logger.info("Trading resumed") def _detect_volatility_spike(self) -> bool: """Détecte spike de volatilité anormal""" if len(self.pnl_history) < 20: return False recent_vol = np.std(self.pnl_history[-5:]) baseline_vol = np.std(self.pnl_history[-20:-5]) # Spike si volatilité > 3x baseline return recent_vol > 3 * baseline_vol def _send_emergency_alert(self, reason: str): """Envoie alerte d'urgence""" # TODO: Implémenter notifications (Telegram, SMS, Email) pass def _load_config(self) -> Dict: """Charge configuration depuis YAML""" import yaml with open('config/risk_limits.yaml', 'r') as f: return yaml.safe_load(f) ``` --- ## ✅ Validation Pré-Trade ### Checklist Complète ```python class PreTradeValidator: """ Validation exhaustive avant chaque trade """ def __init__(self, risk_manager: RiskManager): self.risk_manager = risk_manager def validate(self, trade_request: Dict) -> tuple[bool, List[str]]: """ Exécute toutes validations Returns: (is_valid, list_of_errors) """ errors = [] # 1. Validation basique errors.extend(self._validate_basic(trade_request)) # 2. Validation risque errors.extend(self._validate_risk(trade_request)) # 3. Validation liquidité errors.extend(self._validate_liquidity(trade_request)) # 4. Validation margin errors.extend(self._validate_margin(trade_request)) # 5. Validation technique errors.extend(self._validate_technical(trade_request)) return len(errors) == 0, errors def _validate_basic(self, trade: Dict) -> List[str]: """Validations basiques""" errors = [] # Stop-loss obligatoire if 'stop_loss' not in trade or trade['stop_loss'] is None: errors.append("Stop-loss is mandatory") # Take-profit obligatoire if 'take_profit' not in trade or trade['take_profit'] is None: errors.append("Take-profit is mandatory") # Quantité positive if trade.get('quantity', 0) <= 0: errors.append("Quantity must be positive") # Prix valide if trade.get('price', 0) <= 0: errors.append("Price must be positive") return errors def _validate_risk(self, trade: Dict) -> List[str]: """Validations risque""" errors = [] # Risk/Reward ratio risk = abs(trade['price'] - trade['stop_loss']) reward = abs(trade['take_profit'] - trade['price']) if risk > 0: rr_ratio = reward / risk if rr_ratio < 1.5: errors.append(f"Risk/Reward ratio {rr_ratio:.2f} below minimum 1.5") # Taille position position_value = trade['price'] * trade['quantity'] position_pct = position_value / self.risk_manager.portfolio_value if position_pct > 0.05: # 5% max errors.append(f"Position size {position_pct:.2%} exceeds 5%") return errors def _validate_liquidity(self, trade: Dict) -> List[str]: """Validations liquidité""" errors = [] # TODO: Vérifier volume quotidien # TODO: Vérifier spread bid/ask # TODO: Vérifier market depth return errors def _validate_margin(self, trade: Dict) -> List[str]: """Validations margin""" errors = [] # TODO: Vérifier margin disponible # TODO: Calculer margin requis # TODO: Vérifier margin call risk return errors def _validate_technical(self, trade: Dict) -> List[str]: """Validations techniques""" errors = [] # Vérifier que stop-loss est du bon côté if trade['quantity'] > 0: # Long if trade['stop_loss'] >= trade['price']: errors.append("Stop-loss must be below entry price for long") if trade['take_profit'] <= trade['price']: errors.append("Take-profit must be above entry price for long") else: # Short if trade['stop_loss'] <= trade['price']: errors.append("Stop-loss must be above entry price for short") if trade['take_profit'] >= trade['price']: errors.append("Take-profit must be below entry price for short") return errors ``` --- ## 🚨 Circuit Breakers ### Types de Circuit Breakers ```python class CircuitBreaker: """ Système d'arrêt automatique multi-niveaux """ def __init__(self): self.breakers = { 'drawdown': DrawdownBreaker(), 'daily_loss': DailyLossBreaker(), 'volatility': VolatilityBreaker(), 'flash_crash': FlashCrashBreaker(), 'api_failure': APIFailureBreaker(), 'correlation': CorrelationBreaker(), } def check_all(self, market_data: Dict, portfolio_state: Dict) -> Optional[str]: """ Vérifie tous circuit breakers Returns: Reason for halt, or None if all OK """ for name, breaker in self.breakers.items(): if breaker.should_halt(market_data, portfolio_state): return f"{name}: {breaker.get_reason()}" return None class DrawdownBreaker: """Arrêt si drawdown excessif""" def __init__(self, max_drawdown=0.10): self.max_drawdown = max_drawdown self.reason = "" def should_halt(self, market_data: Dict, portfolio: Dict) -> bool: current_dd = portfolio['current_drawdown'] if current_dd >= self.max_drawdown: self.reason = f"Drawdown {current_dd:.2%} >= {self.max_drawdown:.2%}" return True return False def get_reason(self) -> str: return self.reason class VolatilityBreaker: """Arrêt si volatilité extrême""" def __init__(self, spike_threshold=3.0): self.spike_threshold = spike_threshold self.reason = "" def should_halt(self, market_data: Dict, portfolio: Dict) -> bool: current_vol = market_data.get('current_volatility', 0) baseline_vol = market_data.get('baseline_volatility', 0) if baseline_vol > 0 and current_vol > self.spike_threshold * baseline_vol: self.reason = f"Volatility spike: {current_vol/baseline_vol:.1f}x baseline" return True return False def get_reason(self) -> str: return self.reason class FlashCrashBreaker: """Arrêt si mouvement de prix extrême""" def __init__(self, max_move_pct=0.05): self.max_move_pct = max_move_pct self.reason = "" def should_halt(self, market_data: Dict, portfolio: Dict) -> bool: price_change = market_data.get('price_change_1min', 0) if abs(price_change) > self.max_move_pct: self.reason = f"Flash crash detected: {price_change:.2%} move in 1 minute" return True return False def get_reason(self) -> str: return self.reason ``` --- ## 📊 Métriques de Risque ### Calculs Avancés ```python class RiskMetricsCalculator: """ Calcule métriques de risque avancées """ @staticmethod def calculate_var(returns: np.ndarray, confidence=0.95) -> float: """Value at Risk""" return np.percentile(returns, (1 - confidence) * 100) @staticmethod def calculate_cvar(returns: np.ndarray, confidence=0.95) -> float: """Conditional Value at Risk (Expected Shortfall)""" var = RiskMetricsCalculator.calculate_var(returns, confidence) return returns[returns <= var].mean() @staticmethod def calculate_sharpe_ratio(returns: np.ndarray, risk_free_rate=0.02) -> float: """Sharpe Ratio""" excess_returns = returns - risk_free_rate / 252 # Daily return np.mean(excess_returns) / np.std(excess_returns) * np.sqrt(252) @staticmethod def calculate_sortino_ratio(returns: np.ndarray, risk_free_rate=0.02) -> float: """Sortino Ratio (downside deviation)""" excess_returns = returns - risk_free_rate / 252 downside_returns = returns[returns < 0] downside_std = np.std(downside_returns) return np.mean(excess_returns) / downside_std * np.sqrt(252) @staticmethod def calculate_max_drawdown(equity_curve: np.ndarray) -> float: """Maximum Drawdown""" peak = np.maximum.accumulate(equity_curve) drawdown = (equity_curve - peak) / peak return np.min(drawdown) @staticmethod def calculate_calmar_ratio(returns: np.ndarray, equity_curve: np.ndarray) -> float: """Calmar Ratio (Return / Max Drawdown)""" annual_return = np.mean(returns) * 252 max_dd = abs(RiskMetricsCalculator.calculate_max_drawdown(equity_curve)) return annual_return / max_dd if max_dd > 0 else 0 ``` --- ## 🔔 Système d'Alertes ### Configuration Alertes ```yaml # config/alerts.yaml alerts: risk_threshold_breach: channels: ['telegram', 'email'] priority: high conditions: - total_risk > max_portfolio_risk - current_drawdown > 0.08 # 80% du max - daily_loss > 0.025 # 83% du max position_alerts: channels: ['telegram'] priority: medium conditions: - position_size > 0.04 # 80% du max - correlation > 0.6 # 85% du max circuit_breaker: channels: ['telegram', 'sms', 'email'] priority: critical conditions: - trading_halted == true ``` --- **Suite dans le prochain fichier...**