#!/usr/bin/env python3 """ Détecteur de conditions de marché Identifie: Trending/Ranging, Volatile/Calme, Bullish/Bearish """ import pandas as pd import numpy as np from dataclasses import dataclass from enum import Enum import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class TrendType(Enum): """Type de tendance""" STRONG_UPTREND = "strong_uptrend" UPTREND = "uptrend" RANGING = "ranging" DOWNTREND = "downtrend" STRONG_DOWNTREND = "strong_downtrend" class VolatilityType(Enum): """Type de volatilité""" EXTREME = "extreme" HIGH = "high" MEDIUM = "medium" LOW = "low" @dataclass class MarketConditions: """Conditions de marché détectées""" trend_type: TrendType trend_strength: float # 0-100 volatility_type: VolatilityType volatility_value: float # ATR relatif is_bullish: bool adx: float atr_percent: float bb_width_percent: float ema_alignment: str # 'bullish', 'bearish', 'mixed' def __str__(self): return ( f"Trend: {self.trend_type.value} ({self.trend_strength:.1f}) | " f"Vol: {self.volatility_type.value} ({self.atr_percent:.2f}%) | " f"{'Bullish' if self.is_bullish else 'Bearish'}" ) class MarketConditionDetector: """Détecte les conditions de marché""" def __init__(self): # Périodes des indicateurs self.adx_period = 14 self.atr_period = 14 self.bb_period = 20 self.ema_fast = 9 self.ema_mid = 21 self.ema_slow = 50 def detect(self, df: pd.DataFrame) -> MarketConditions: """ Détecte les conditions actuelles du marché Args: df: DataFrame avec colonnes: timestamp, open, high, low, close, volume Returns: MarketConditions object """ # Calculer les indicateurs adx = self._calculate_adx(df) atr_percent = self._calculate_atr_percent(df) bb_width = self._calculate_bb_width(df) ema_alignment = self._calculate_ema_alignment(df) # Détecter le type de tendance trend_type, trend_strength = self._detect_trend(df, adx, ema_alignment) # Détecter le type de volatilité volatility_type = self._detect_volatility(atr_percent, bb_width) # Détecter la direction (bullish/bearish) is_bullish = self._is_bullish(df, ema_alignment) return MarketConditions( trend_type=trend_type, trend_strength=trend_strength, volatility_type=volatility_type, volatility_value=atr_percent, is_bullish=is_bullish, adx=adx, atr_percent=atr_percent, bb_width_percent=bb_width, ema_alignment=ema_alignment ) def _calculate_adx(self, df: pd.DataFrame) -> float: """Calcule l'ADX (Average Directional Index)""" high = df['high'] low = df['low'] close = df['close'] # True Range tr1 = high - low tr2 = abs(high - close.shift()) tr3 = abs(low - close.shift()) tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) # Directional Movement up_move = high - high.shift() down_move = low.shift() - low plus_dm = np.where((up_move > down_move) & (up_move > 0), up_move, 0) minus_dm = np.where((down_move > up_move) & (down_move > 0), down_move, 0) # Smooth TR et DM atr = tr.ewm(span=self.adx_period, adjust=False).mean() plus_di = 100 * pd.Series(plus_dm).ewm(span=self.adx_period, adjust=False).mean() / atr minus_di = 100 * pd.Series(minus_dm).ewm(span=self.adx_period, adjust=False).mean() / atr # ADX dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di) adx = dx.ewm(span=self.adx_period, adjust=False).mean() return adx.iloc[-1] if not adx.empty else 0 def _calculate_atr_percent(self, df: pd.DataFrame) -> float: """Calcule l'ATR en % du prix""" high = df['high'] low = df['low'] close = df['close'] tr1 = high - low tr2 = abs(high - close.shift()) tr3 = abs(low - close.shift()) tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) atr = tr.ewm(span=self.atr_period, adjust=False).mean() atr_percent = (atr / close) * 100 return atr_percent.iloc[-1] if not atr_percent.empty else 0 def _calculate_bb_width(self, df: pd.DataFrame) -> float: """Calcule la largeur des Bollinger Bands en %""" close = df['close'] sma = close.rolling(window=self.bb_period).mean() std = close.rolling(window=self.bb_period).std() upper = sma + (2 * std) lower = sma - (2 * std) bb_width = ((upper - lower) / sma) * 100 return bb_width.iloc[-1] if not bb_width.empty else 0 def _calculate_ema_alignment(self, df: pd.DataFrame) -> str: """Calcule l'alignement des EMAs""" close = df['close'] ema_fast = close.ewm(span=self.ema_fast, adjust=False).mean().iloc[-1] ema_mid = close.ewm(span=self.ema_mid, adjust=False).mean().iloc[-1] ema_slow = close.ewm(span=self.ema_slow, adjust=False).mean().iloc[-1] if ema_fast > ema_mid > ema_slow: return 'bullish' elif ema_fast < ema_mid < ema_slow: return 'bearish' else: return 'mixed' def _detect_trend(self, df: pd.DataFrame, adx: float, ema_alignment: str) -> tuple: """Détecte le type et la force de la tendance""" close = df['close'] # Pente de la SMA50 sma50 = close.rolling(window=50).mean() if len(sma50) > 10: slope = (sma50.iloc[-1] - sma50.iloc[-10]) / sma50.iloc[-10] * 100 else: slope = 0 # Force de la tendance basée sur ADX et pente trend_strength = min(adx * abs(slope) * 2, 100) # Type de tendance if adx > 25: # Tendance forte if ema_alignment == 'bullish' and slope > 0.5: trend_type = TrendType.STRONG_UPTREND elif ema_alignment == 'bearish' and slope < -0.5: trend_type = TrendType.STRONG_DOWNTREND elif slope > 0: trend_type = TrendType.UPTREND else: trend_type = TrendType.DOWNTREND else: # Ranging trend_type = TrendType.RANGING return trend_type, trend_strength def _detect_volatility(self, atr_percent: float, bb_width: float) -> VolatilityType: """Détecte le type de volatilité""" # Seuils empiriques if atr_percent > 2.0 or bb_width > 8.0: return VolatilityType.EXTREME elif atr_percent > 1.0 or bb_width > 5.0: return VolatilityType.HIGH elif atr_percent > 0.5 or bb_width > 3.0: return VolatilityType.MEDIUM else: return VolatilityType.LOW def _is_bullish(self, df: pd.DataFrame, ema_alignment: str) -> bool: """Détermine si le marché est bullish""" close = df['close'].iloc[-1] ema21 = df['close'].ewm(span=21, adjust=False).mean().iloc[-1] # Bullish si prix > EMA21 ET alignement bullish ou mixed return close > ema21 and ema_alignment in ['bullish', 'mixed'] if __name__ == '__main__': # Test du détecteur import sys sys.path.append('.') detector = MarketConditionDetector() # Charger les données 5m df = pd.read_csv('historical_data/BTCUSDT_5m_9000candles.csv') df['timestamp'] = pd.to_datetime(df['timestamp']) logger.info("\n🔍 TEST DU DÉTECTEUR DE CONDITIONS DE MARCHÉ\n") logger.info("="*80) # Tester sur les 100 dernières bougies for i in range(-10, 0): df_slice = df.iloc[:len(df)+i].copy() conditions = detector.detect(df_slice) timestamp = df_slice['timestamp'].iloc[-1] price = df_slice['close'].iloc[-1] logger.info(f"\n{timestamp} | ${price:,.2f}") logger.info(f" {conditions}")