#!/usr/bin/env python3 """ Market Regime Detector - Détection avancée du régime de marché Identifie 5 régimes distincts et adapte la stratégie en conséquence """ import numpy as np from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple import logging logger = logging.getLogger("MarketRegimeDetector") class MarketRegimeDetector: """Détecteur de régime de marché pour adaptation de stratégie""" REGIMES = { 'BULL_STRONG': { 'name': 'Bull Market Fort', 'btc_trend_min': 5.0, # BTC +5% sur période 'volatility': 'low', 'strategy': { 'min_score': 45, # Moins strict 'max_positions': 30, # Maximum de positions 'aggressive': True, 'risk_multiplier': 1.2 } }, 'BULL_WEAK': { 'name': 'Bull Market Faible', 'btc_trend_min': 1.0, 'btc_trend_max': 5.0, 'volatility': 'medium', 'strategy': { 'min_score': 55, 'max_positions': 20, 'aggressive': False, 'risk_multiplier': 1.0 } }, 'SIDEWAYS': { 'name': 'Marché Latéral', 'btc_trend_min': -1.0, 'btc_trend_max': 1.0, 'volatility': 'low', 'strategy': { 'min_score': 60, # Plus sélectif 'max_positions': 15, 'aggressive': False, 'risk_multiplier': 0.8 } }, 'BEAR_WEAK': { 'name': 'Bear Market Faible', 'btc_trend_min': -5.0, 'btc_trend_max': -1.0, 'volatility': 'medium', 'strategy': { 'min_score': 70, # Très sélectif 'max_positions': 10, 'aggressive': False, 'risk_multiplier': 0.6 } }, 'BEAR_STRONG': { 'name': 'Bear Market Fort', 'btc_trend_max': -5.0, 'volatility': 'high', 'strategy': { 'min_score': 80, # Extrêmement sélectif 'max_positions': 5, # Minimal 'aggressive': False, 'risk_multiplier': 0.4 } } } def __init__(self): self.current_regime = 'SIDEWAYS' self.regime_history = [] self.last_update = None self.klines_fetcher = None logger.info("✅ Market Regime Detector initialisé") def set_klines_fetcher(self, fetcher): """Injecte le fetcher de klines""" self.klines_fetcher = fetcher def detect_regime(self, btc_prices: List[float] = None) -> Dict: """ Détecte le régime de marché actuel basé sur BTC Args: btc_prices: Prix BTC (si None, fetch automatiquement) Returns: Dict avec régime détecté et stratégie adaptée """ try: # Fetch BTC data si non fourni if btc_prices is None and self.klines_fetcher: klines = self.klines_fetcher('BTCUSDT', interval='1h', limit=100) if klines: btc_prices = [float(k[4]) for k in klines] # Close prices if not btc_prices or len(btc_prices) < 50: logger.warning("⚠️ Données BTC insuffisantes pour détecter régime") return self._get_default_regime() btc_array = np.array(btc_prices) # Calculer tendance BTC (sur 24h = 24 bougies 1h) btc_24h_change = ((btc_array[-1] - btc_array[-24]) / btc_array[-24]) * 100 # Calculer tendance 7j pour confirmation if len(btc_array) >= 168: # 7j × 24h btc_7d_change = ((btc_array[-1] - btc_array[-168]) / btc_array[-168]) * 100 else: btc_7d_change = btc_24h_change # Calculer volatilité (ATR-like sur 14 périodes) volatility = self._calculate_volatility(btc_array) # Déterminer régime regime = self._classify_regime(btc_24h_change, btc_7d_change, volatility) self.current_regime = regime self.last_update = datetime.now() # Ajouter à l'historique self.regime_history.append({ 'regime': regime, 'timestamp': self.last_update, 'btc_24h_change': btc_24h_change, 'btc_7d_change': btc_7d_change, 'volatility': volatility }) # Garder seulement les 100 derniers if len(self.regime_history) > 100: self.regime_history = self.regime_history[-100:] result = { 'success': True, 'regime': regime, 'regime_name': self.REGIMES[regime]['name'], 'btc_24h_change': btc_24h_change, 'btc_7d_change': btc_7d_change, 'volatility': volatility, 'volatility_level': self._classify_volatility(volatility), 'strategy': self.REGIMES[regime]['strategy'], 'last_update': self.last_update.isoformat() } logger.info(f"📊 Régime détecté: {result['regime_name']} (BTC 24h: {btc_24h_change:+.2f}%, Vol: {volatility:.2f}%)") return result except Exception as e: logger.error(f"Erreur détection régime: {e}") return self._get_default_regime() def _classify_regime(self, btc_24h: float, btc_7d: float, volatility: float) -> str: """ Classifie le régime basé sur les métriques Args: btc_24h: Variation BTC 24h (%) btc_7d: Variation BTC 7j (%) volatility: Volatilité (%) """ # Pondérer 24h et 7j (24h plus important pour réactivité) combined_trend = (btc_24h * 0.6 + btc_7d * 0.4) # Bear Strong if combined_trend < -5.0 or (btc_24h < -3.0 and volatility > 4.0): return 'BEAR_STRONG' # Bear Weak elif combined_trend < -1.0: return 'BEAR_WEAK' # Bull Strong elif combined_trend > 5.0 or (btc_24h > 3.0 and volatility < 3.0): return 'BULL_STRONG' # Bull Weak elif combined_trend > 1.0: return 'BULL_WEAK' # Sideways else: return 'SIDEWAYS' def _calculate_volatility(self, prices: np.ndarray, period: int = 14) -> float: """ Calcule la volatilité (ATR-like en pourcentage) Returns: Volatilité en % du prix """ if len(prices) < period + 1: return 2.0 # Valeur par défaut # Calculer les variations absolues changes = np.abs(np.diff(prices[-period-1:])) avg_change = np.mean(changes) # En pourcentage du prix actuel volatility_pct = (avg_change / prices[-1]) * 100 return volatility_pct def _classify_volatility(self, volatility: float) -> str: """Classifie le niveau de volatilité""" if volatility < 2.0: return 'low' elif volatility < 4.0: return 'medium' else: return 'high' def _get_default_regime(self) -> Dict: """Retourne un régime par défaut (neutre)""" return { 'success': False, 'regime': 'SIDEWAYS', 'regime_name': 'Marché Latéral (défaut)', 'btc_24h_change': 0, 'btc_7d_change': 0, 'volatility': 2.0, 'volatility_level': 'medium', 'strategy': self.REGIMES['SIDEWAYS']['strategy'], 'last_update': datetime.now().isoformat() } def get_adapted_strategy(self) -> Dict: """ Retourne la stratégie adaptée au régime actuel Returns: Dict avec paramètres de stratégie """ if self.current_regime in self.REGIMES: return self.REGIMES[self.current_regime]['strategy'] return self.REGIMES['SIDEWAYS']['strategy'] def should_be_aggressive(self) -> bool: """Indique si on doit être agressif dans le trading""" strategy = self.get_adapted_strategy() return strategy.get('aggressive', False) def get_min_score_threshold(self) -> float: """Retourne le score minimum adapté au régime""" strategy = self.get_adapted_strategy() return strategy.get('min_score', 60) def get_max_positions(self) -> int: """Retourne le nombre max de positions adapté au régime""" strategy = self.get_adapted_strategy() return strategy.get('max_positions', 15) def get_risk_multiplier(self) -> float: """Retourne le multiplicateur de risque adapté au régime""" strategy = self.get_adapted_strategy() return strategy.get('risk_multiplier', 1.0) def get_regime_info(self) -> str: """Retourne une description textuelle du régime actuel""" if self.current_regime in self.REGIMES: regime_data = self.REGIMES[self.current_regime] return f"{regime_data['name']} - Min Score: {regime_data['strategy']['min_score']}, Max Pos: {regime_data['strategy']['max_positions']}" return "Régime inconnu" # Instance globale _regime_detector = None def get_regime_detector() -> MarketRegimeDetector: """Retourne l'instance globale du détecteur de régime""" global _regime_detector if _regime_detector is None: _regime_detector = MarketRegimeDetector() return _regime_detector