""" Détecteur de régime de marché par symbole Adapte les paramètres de trading selon les conditions actuelles """ import logging import numpy as np from typing import Dict, Tuple, List from datetime import datetime class MarketRegimeDetector: """Détecte le régime de marché pour chaque symbole individuellement""" def __init__(self, logger=None): self.logger = logger or logging.getLogger("MarketRegimeDetector") # Paramètres de détection self.regimes = { 'BULL_STABLE': { 'conditions': 'change_24h > 2% et volatilité < 3%', 'score_multiplier': 1.2, 'position_multiplier': 1.0, 'leverage_multiplier': 1.0, 'cooldown_multiplier': 0.8, # Cooldown réduit 'long_bias': 1.3, 'short_bias': 0.7 }, 'BULL_VOLATILE': { 'conditions': 'change_24h > 2% et volatilité >= 3%', 'score_multiplier': 0.9, 'position_multiplier': 0.7, # Réduit taille 'leverage_multiplier': 0.8, 'cooldown_multiplier': 1.2, 'long_bias': 1.2, 'short_bias': 0.8 }, 'BEAR_STABLE': { 'conditions': '-8% < change_24h < -2% et volatilité < 3%', 'score_multiplier': 1.1, 'position_multiplier': 1.0, 'leverage_multiplier': 1.0, 'cooldown_multiplier': 0.9, 'long_bias': 0.7, 'short_bias': 1.3 # SHORT bias }, 'BEAR_VOLATILE': { 'conditions': '-8% < change_24h < -2% et volatilité >= 3%', 'score_multiplier': 0.7, # Très prudent 'position_multiplier': 0.5, # Petites positions 'leverage_multiplier': 0.6, 'cooldown_multiplier': 1.5, 'long_bias': 0.5, 'short_bias': 1.2 }, 'DUMP': { 'conditions': 'change_24h < -8%', 'score_multiplier': 0.8, # On trade mais prudemment 'position_multiplier': 0.6, # Positions réduites 'leverage_multiplier': 0.8, # Leverage modéré 'cooldown_multiplier': 1.0, 'long_bias': 0.0, # AUCUN LONG en dump 'short_bias': 1.5 # SHORT fortement favorisé }, 'CONSOLIDATION': { 'conditions': '-2% <= change_24h <= 2%', 'score_multiplier': 0.95, 'position_multiplier': 0.9, 'leverage_multiplier': 1.0, 'cooldown_multiplier': 1.0, 'long_bias': 1.0, 'short_bias': 1.0 } } def detect_regime(self, closes: List[float], volumes: List[float], current_price: float) -> Tuple[str, Dict]: """ Détecte le régime actuel pour un symbole Args: closes: Liste des prix de clôture (au moins 24 périodes) volumes: Liste des volumes current_price: Prix actuel Returns: Tuple (regime_name, regime_params) """ try: if len(closes) < 25: self.logger.warning(f"Pas assez de données pour détecter régime: {len(closes)} périodes") return 'CONSOLIDATION', self.regimes['CONSOLIDATION'] # Calculer le changement 24h price_24h_ago = closes[-25] change_24h = ((current_price - price_24h_ago) / price_24h_ago) * 100 # Calculer la volatilité (std des returns sur 24 périodes) recent_closes = np.array(closes[-25:]) # 25 pour avoir 24 returns returns = np.diff(recent_closes) / recent_closes[:-1] volatility_pct = np.std(returns) * 100 # Détecter le régime if change_24h < -8: regime = 'DUMP' elif change_24h < -2 and volatility_pct >= 3: regime = 'BEAR_VOLATILE' elif change_24h < -2 and volatility_pct < 3: regime = 'BEAR_STABLE' elif change_24h > 2 and volatility_pct >= 3: regime = 'BULL_VOLATILE' elif change_24h > 2 and volatility_pct < 3: regime = 'BULL_STABLE' else: regime = 'CONSOLIDATION' params = self.regimes[regime].copy() params['change_24h'] = change_24h params['volatility_pct'] = volatility_pct return regime, params except Exception as e: self.logger.error(f"Erreur détection régime: {e}") return 'CONSOLIDATION', self.regimes['CONSOLIDATION'] def adjust_signal_score(self, signal: Dict, regime: str, regime_params: Dict) -> float: """ Ajuste le score d'un signal selon le régime Args: signal: Signal de trading avec 'score' et 'side' regime: Nom du régime regime_params: Paramètres du régime Returns: Score ajusté """ try: base_score = signal.get('score', 0) side = signal.get('side', '') # Multiplicateur de base du régime multiplier = regime_params.get('score_multiplier', 1.0) # Appliquer le bias LONG/SHORT if side == 'LONG': multiplier *= regime_params.get('long_bias', 1.0) elif side == 'SHORT': multiplier *= regime_params.get('short_bias', 1.0) adjusted_score = base_score * multiplier # Log si changement significatif if abs(adjusted_score - base_score) > 0.05: self.logger.debug( f"{signal['symbol']} {side}: Score {base_score:.3f} → {adjusted_score:.3f} " f"(régime: {regime}, mult: {multiplier:.2f})" ) return adjusted_score except Exception as e: self.logger.error(f"Erreur ajustement score: {e}") return signal.get('score', 0) def should_trade_in_regime(self, regime: str, side: str = None) -> bool: """ Vérifie si on peut trader dans ce régime Args: regime: Le régime détecté side: LONG ou SHORT (si fourni, vérifie la compatibilité) """ # En DUMP, autoriser seulement les SHORT if regime == 'DUMP': return side == 'SHORT' if side else False # Autres régimes: toujours autoriser return True def get_position_size_multiplier(self, regime: str) -> float: """Retourne le multiplicateur de taille de position pour ce régime""" return self.regimes[regime].get('position_multiplier', 1.0) def get_leverage_multiplier(self, regime: str) -> float: """Retourne le multiplicateur de leverage pour ce régime""" return self.regimes[regime].get('leverage_multiplier', 1.0) def get_cooldown_multiplier(self, regime: str) -> float: """Retourne le multiplicateur de cooldown pour ce régime""" return self.regimes[regime].get('cooldown_multiplier', 1.0) def log_regime_detection(self, symbol: str, regime: str, regime_params: Dict): """Log la détection de régime""" change = regime_params.get('change_24h', 0) vol = regime_params.get('volatility_pct', 0) emoji = { 'BULL_STABLE': '🟢', 'BULL_VOLATILE': '🟡', 'BEAR_STABLE': '🔴', 'BEAR_VOLATILE': '🔴', 'DUMP': '⚠️', 'CONSOLIDATION': '⚪' }.get(regime, '❓') self.logger.info( f"{emoji} {symbol}: {regime} | Δ24h: {change:+.2f}% | Vol: {vol:.2f}% | " f"ScoreMult: {regime_params['score_multiplier']:.2f}x" ) def test_regime_detector(): """Test du détecteur de régime""" import logging logging.basicConfig(level=logging.INFO) detector = MarketRegimeDetector() # Test 1: BULL_STABLE print("\n=== TEST 1: BULL_STABLE ===") closes_bull = [100 + i*0.3 for i in range(30)] # +3% sur 24h, faible vol volumes = [1000000] * 30 regime, params = detector.detect_regime(closes_bull, volumes, closes_bull[-1]) print(f"Régime détecté: {regime}") print(f"Paramètres: {params}") # Test 2: DUMP print("\n=== TEST 2: DUMP ===") closes_dump = [100] * 10 + [100 - i*1.5 for i in range(20)] # -10% rapide regime, params = detector.detect_regime(closes_dump, volumes, closes_dump[-1]) print(f"Régime détecté: {regime}") print(f"Can trade: {detector.should_trade_in_regime(regime)}") # Test 3: BEAR_VOLATILE print("\n=== TEST 3: BEAR_VOLATILE ===") closes_bear = [100] for i in range(29): # Baisse avec forte volatilité closes_bear.append(closes_bear[-1] * (0.998 + np.random.uniform(-0.01, 0.005))) regime, params = detector.detect_regime(closes_bear, volumes, closes_bear[-1]) print(f"Régime détecté: {regime}") print(f"Position multiplier: {detector.get_position_size_multiplier(regime)}") # Test 4: Ajustement de score print("\n=== TEST 4: Ajustement scores ===") signal_long = {'symbol': 'BTC', 'side': 'LONG', 'score': 0.45} signal_short = {'symbol': 'BTC', 'side': 'SHORT', 'score': 0.45} for regime_name in ['BULL_STABLE', 'BEAR_VOLATILE', 'DUMP']: regime_params = detector.regimes[regime_name] adjusted_long = detector.adjust_signal_score(signal_long, regime_name, regime_params) adjusted_short = detector.adjust_signal_score(signal_short, regime_name, regime_params) print(f"\n{regime_name}:") print(f" LONG: 0.45 → {adjusted_long:.3f}") print(f" SHORT: 0.45 → {adjusted_short:.3f}") if __name__ == "__main__": test_regime_detector()