#!/usr/bin/env python3 """ Risk-Adjusted Scorer - Ajustement des scores par le risque Intègre des métriques Sharpe-like pour favoriser le meilleur rapport rendement/risque """ import numpy as np from typing import Dict, List, Optional import logging logger = logging.getLogger("RiskAdjustedScorer") class RiskAdjustedScorer: """Calcule des scores ajustés par le risque (Sharpe-like, Sortino, etc.)""" def __init__(self): # Paramètres self.risk_free_rate = 0.0 # Taux sans risque (0% pour crypto) self.target_sharpe = 2.0 # Sharpe ratio cible # Cache des volatilités historiques self.volatility_cache = {} # {symbol: volatility} logger.info("✅ Risk-Adjusted Scorer initialisé") def calculate_sharpe_ratio(self, returns: List[float], risk_free_rate: float = 0.0) -> float: """ Calcule le Sharpe Ratio Sharpe = (Moyenne des rendements - Taux sans risque) / Écart-type des rendements Args: returns: Liste des rendements (%) risk_free_rate: Taux sans risque (défaut: 0%) Returns: Sharpe Ratio (plus élevé = meilleur rapport rendement/risque) """ if not returns or len(returns) < 2: return 0.0 returns_array = np.array(returns) avg_return = np.mean(returns_array) std_return = np.std(returns_array) if std_return == 0: return 0.0 sharpe = (avg_return - risk_free_rate) / std_return return sharpe def calculate_sortino_ratio(self, returns: List[float], target_return: float = 0.0) -> float: """ Calcule le Sortino Ratio (comme Sharpe mais ne pénalise que la volatilité baissière) Sortino = (Moyenne des rendements - Target) / Downside Deviation Args: returns: Liste des rendements (%) target_return: Rendement cible Returns: Sortino Ratio """ if not returns or len(returns) < 2: return 0.0 returns_array = np.array(returns) avg_return = np.mean(returns_array) # Downside deviation (seulement les rendements négatifs) downside_returns = returns_array[returns_array < target_return] if len(downside_returns) == 0: return 10.0 # Aucun downside = excellent downside_std = np.std(downside_returns) if downside_std == 0: return 0.0 sortino = (avg_return - target_return) / downside_std return sortino def calculate_calmar_ratio(self, returns: List[float], max_drawdown: float) -> float: """ Calcule le Calmar Ratio (rendement annualisé / max drawdown) Args: returns: Liste des rendements max_drawdown: Drawdown maximum (en %) Returns: Calmar Ratio """ if not returns or max_drawdown == 0: return 0.0 avg_return = np.mean(returns) # Annualiser (approximation) annualized_return = avg_return * 365 calmar = annualized_return / abs(max_drawdown) return calmar def calculate_max_drawdown(self, prices: List[float]) -> float: """ Calcule le drawdown maximum Args: prices: Liste de prix Returns: Max drawdown en % """ if not prices or len(prices) < 2: return 0.0 prices_array = np.array(prices) # Calculer les peaks cumulatifs cummax = np.maximum.accumulate(prices_array) # Drawdown à chaque point drawdowns = (prices_array - cummax) / cummax * 100 # Maximum drawdown (plus négatif) max_dd = np.min(drawdowns) return max_dd def analyze_risk_metrics(self, prices: List[float], volumes: List[float] = None) -> Dict: """ Analyse complète des métriques de risque Args: prices: Historique de prix volumes: Historique de volumes (optionnel) Returns: Dict avec toutes les métriques de risque """ if not prices or len(prices) < 10: return { 'sharpe_ratio': 0.0, 'sortino_ratio': 0.0, 'max_drawdown': 0.0, 'volatility': 0.0, 'risk_score': 50, 'risk_adjusted_score': 50 } prices_array = np.array(prices) # Calculer les rendements returns = np.diff(prices_array) / prices_array[:-1] * 100 # Sharpe Ratio sharpe = self.calculate_sharpe_ratio(returns.tolist()) # Sortino Ratio sortino = self.calculate_sortino_ratio(returns.tolist()) # Max Drawdown max_dd = self.calculate_max_drawdown(prices) # Volatilité volatility = np.std(returns) # Calmar Ratio calmar = self.calculate_calmar_ratio(returns.tolist(), max_dd) if max_dd != 0 else 0 # Risk Score (0-100, 100 = faible risque) risk_score = self._calculate_risk_score(sharpe, sortino, max_dd, volatility) # Métriques additionnelles avg_return = np.mean(returns) win_rate = len(returns[returns > 0]) / len(returns) * 100 if len(returns) > 0 else 50 # Value at Risk (VaR) 95% var_95 = np.percentile(returns, 5) if len(returns) >= 20 else 0 metrics = { 'sharpe_ratio': round(sharpe, 2), 'sortino_ratio': round(sortino, 2), 'calmar_ratio': round(calmar, 2), 'max_drawdown': round(max_dd, 2), 'volatility': round(volatility, 2), 'avg_return': round(avg_return, 2), 'win_rate': round(win_rate, 1), 'var_95': round(var_95, 2), 'risk_score': round(risk_score, 1), 'risk_quality': self._classify_risk(risk_score) } return metrics def _calculate_risk_score(self, sharpe: float, sortino: float, max_dd: float, volatility: float) -> float: """ Calcule un score de risque composite (0-100) 100 = Excellent (faible risque, bon rendement) 0 = Terrible (haut risque, mauvais rendement) """ # Sharpe component (0-40 points) if sharpe >= 3: sharpe_points = 40 elif sharpe >= 2: sharpe_points = 30 elif sharpe >= 1: sharpe_points = 20 elif sharpe >= 0: sharpe_points = 10 else: sharpe_points = 0 # Sortino component (0-30 points) if sortino >= 3: sortino_points = 30 elif sortino >= 2: sortino_points = 20 elif sortino >= 1: sortino_points = 10 else: sortino_points = 0 # Max Drawdown component (0-20 points) if max_dd > -10: dd_points = 20 elif max_dd > -20: dd_points = 15 elif max_dd > -30: dd_points = 10 elif max_dd > -40: dd_points = 5 else: dd_points = 0 # Volatility component (0-10 points) if volatility < 2: vol_points = 10 elif volatility < 5: vol_points = 7 elif volatility < 10: vol_points = 4 else: vol_points = 0 total_score = sharpe_points + sortino_points + dd_points + vol_points return min(100, max(0, total_score)) def _classify_risk(self, risk_score: float) -> str: """Classifie le niveau de risque""" if risk_score >= 80: return 'TRÈS FAIBLE' elif risk_score >= 60: return 'FAIBLE' elif risk_score >= 40: return 'MOYEN' elif risk_score >= 20: return 'ÉLEVÉ' else: return 'TRÈS ÉLEVÉ' def adjust_score_by_risk(self, base_score: float, risk_metrics: Dict, symbol: str = None) -> Dict: """ Ajuste un score de base par les métriques de risque Args: base_score: Score de base (0-100) risk_metrics: Métriques de risque (de analyze_risk_metrics) symbol: Symbole de la crypto (pour identifier TOP 20) Returns: Dict avec score ajusté et détails """ # TOP 20 cryptos (moins risquées, pénalités réduites) TOP_20_SYMBOLS = [ "BTCUSDT", "ETHUSDT", "BNBUSDT", "XRPUSDT", "ADAUSDT", "DOGEUSDT", "SOLUSDT", "DOTUSDT", "MATICUSDT", "LTCUSDT", "AVAXUSDT", "LINKUSDT", "ATOMUSDT", "UNIUSDT", "XLMUSDT", "ETCUSDT", "FILUSDT", "APTUSDT", "NEARUSDT", "ARBUSDT" ] is_top20 = symbol in TOP_20_SYMBOLS if symbol else False # Extraire les métriques sharpe = risk_metrics.get('sharpe_ratio', 0) risk_score = risk_metrics.get('risk_score', 50) max_dd = risk_metrics.get('max_drawdown', 0) # Calcul du multiplicateur de risque risk_multiplier = 1.0 # Bonus pour bon Sharpe if sharpe >= 2.5: risk_multiplier += 0.2 # +20% elif sharpe >= 2.0: risk_multiplier += 0.15 # +15% elif sharpe >= 1.5: risk_multiplier += 0.1 # +10% elif sharpe < 0: # Seulement si NÉGATIF (perte systématique) # CORRECTION: Pénalité uniquement si Sharpe négatif penalty = -0.05 if is_top20 else -0.1 # Réduit: -5%/-10% au lieu de -10%/-20% risk_multiplier += penalty # Sharpe 0-0.5 = neutre (pas de pénalité, c'est normal pour crypto volatiles) # Pénalité pour drawdown élevé (RÉDUITE) if max_dd < -50: # Seuil augmenté de -40 à -50 penalty = -0.1 if is_top20 else -0.15 # Réduit de -15%/-30% à -10%/-15% risk_multiplier += penalty elif max_dd < -40: # Seuil augmenté de -30 à -40 penalty = -0.05 if is_top20 else -0.1 # Réduit de -10%/-20% à -5%/-10% risk_multiplier += penalty elif max_dd < -30: # Seuil augmenté de -20 à -30 penalty = -0.03 if is_top20 else -0.05 # Réduit de -5%/-10% à -3%/-5% risk_multiplier += penalty # Drawdown -30% normal en crypto, pas de pénalité # Ajustement par risk_score if risk_score >= 80: risk_multiplier += 0.1 # +10% pour très faible risque elif risk_score < 30: risk_multiplier -= 0.1 # Réduit de -15% à -10% # Limiter le multiplicateur (PLUS TOLÉRANT) risk_multiplier = max(0.7, min(1.5, risk_multiplier)) # 0.5→0.7 (moins de pénalité max) # Score ajusté adjusted_score = base_score * risk_multiplier adjusted_score = max(0, min(100, adjusted_score)) # Bonus/malus absolu absolute_adjustment = adjusted_score - base_score return { 'base_score': round(base_score, 1), 'risk_multiplier': round(risk_multiplier, 2), 'adjusted_score': round(adjusted_score, 1), 'absolute_adjustment': round(absolute_adjustment, 1), 'sharpe_ratio': sharpe, 'risk_quality': risk_metrics.get('risk_quality', 'UNKNOWN'), 'recommendation': self._get_recommendation(risk_multiplier, sharpe) } def _get_recommendation(self, risk_multiplier: float, sharpe: float) -> str: """Génère une recommandation basée sur les métriques""" if risk_multiplier >= 1.2 and sharpe >= 2: return "🔥 Excellent rapport risque/rendement - Opportunité premium" elif risk_multiplier >= 1.1: return "✅ Bon rapport risque/rendement" elif risk_multiplier >= 0.9: return "📊 Risque équilibré" elif risk_multiplier >= 0.7: return "⚠️ Risque élevé - Prudence recommandée" else: return "🚫 Risque très élevé - Éviter" def compare_opportunities(self, opportunities: List[Dict]) -> List[Dict]: """ Compare plusieurs opportunités et les classe par risque ajusté Args: opportunities: Liste de dict avec 'symbol', 'score', 'prices' Returns: Liste triée par score ajusté au risque (décroissant) """ scored_opportunities = [] for opp in opportunities: symbol = opp.get('symbol', 'UNKNOWN') base_score = opp.get('score', 50) prices = opp.get('prices', []) if not prices or len(prices) < 10: # Pas assez de données scored_opportunities.append({ 'symbol': symbol, 'base_score': base_score, 'adjusted_score': base_score, 'risk_quality': 'UNKNOWN' }) continue # Analyser le risque risk_metrics = self.analyze_risk_metrics(prices) # Ajuster le score adjusted = self.adjust_score_by_risk(base_score, risk_metrics) scored_opportunities.append({ 'symbol': symbol, 'base_score': base_score, 'adjusted_score': adjusted['adjusted_score'], 'risk_multiplier': adjusted['risk_multiplier'], 'sharpe_ratio': risk_metrics['sharpe_ratio'], 'sortino_ratio': risk_metrics['sortino_ratio'], 'max_drawdown': risk_metrics['max_drawdown'], 'risk_quality': risk_metrics['risk_quality'], 'recommendation': adjusted['recommendation'] }) # Trier par score ajusté scored_opportunities.sort(key=lambda x: x['adjusted_score'], reverse=True) return scored_opportunities # Instance globale _risk_adjusted_scorer = None def get_risk_adjusted_scorer() -> RiskAdjustedScorer: """Retourne l'instance globale du risk-adjusted scorer""" global _risk_adjusted_scorer if _risk_adjusted_scorer is None: _risk_adjusted_scorer = RiskAdjustedScorer() return _risk_adjusted_scorer