#!/usr/bin/env python3 """ Stratégie Adaptative Multi-Conditions S'adapte automatiquement aux conditions de marché """ import pandas as pd import numpy as np from dataclasses import dataclass from typing import Optional, Dict import logging from market_conditions import ( MarketConditionDetector, MarketConditions, TrendType, VolatilityType ) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class AdaptiveSignal: """Signal de trading adaptatif""" symbol: str side: str # 'LONG' ou 'SHORT' confidence: float # 0-100 entry_price: float stop_loss: float take_profit: float leverage: int reason: str market_conditions: MarketConditions strategy_used: str # 'trending', 'ranging', 'volatile', 'breakout' class AdaptiveStrategy: """ Stratégie qui s'adapte aux conditions de marché 4 sous-stratégies: 1. TRENDING: Suit la tendance avec des stops larges 2. RANGING: Mean reversion avec des stops serrés 3. VOLATILE: Breakouts avec des stops très larges 4. CALM: Scalping avec des stops ultra-serrés """ def __init__(self, leverage: int = 225): self.leverage = leverage self.detector = MarketConditionDetector() # Paramètres par condition de marché self.params = { # TRENDING: Suit la tendance 'trending': { 'min_trend_strength': 25.0, 'tp_multiplier': 3.0, # TP = 3x SL 'sl_atr_multiplier': 1.5, # SL basé sur ATR 'use_trailing': True, 'min_confidence': 70, }, # RANGING: Mean reversion 'ranging': { 'max_trend_strength': 20.0, 'tp_atr_multiplier': 3.0, # Augmenté 1.0 → 3.0 pour meilleur ratio 'sl_atr_multiplier': 0.8, 'use_bb_reversal': True, 'min_confidence': 75, }, # VOLATILE: Breakouts 'volatile': { 'min_atr_percent': 0.8, 'tp_multiplier': 4.0, 'sl_atr_multiplier': 2.0, 'min_volume_spike': 2.0, 'min_confidence': 65, }, # CALM: Scalping 'calm': { 'max_atr_percent': 0.5, 'tp_multiplier': 7.0, # Comme config optimisée 'sl_percent': 0.00037, # 0.037% 'min_confidence': 75, } } def analyze(self, symbol: str, df: pd.DataFrame) -> Optional[AdaptiveSignal]: """ Analyse et génère un signal adaptatif Args: symbol: Symbole du token df: DataFrame avec OHLCV Returns: AdaptiveSignal ou None """ if len(df) < 100: return None # 1. Détecter conditions de marché conditions = self.detector.detect(df) # 2. Choisir la stratégie adaptée strategy_type = self._select_strategy(conditions) # 3. Générer signal selon la stratégie if strategy_type == 'trending': signal = self._trending_strategy(symbol, df, conditions) elif strategy_type == 'ranging': signal = self._ranging_strategy(symbol, df, conditions) elif strategy_type == 'volatile': signal = self._volatile_strategy(symbol, df, conditions) elif strategy_type == 'calm': signal = self._calm_strategy(symbol, df, conditions) else: return None if signal: signal.strategy_used = strategy_type return signal def _select_strategy(self, conditions: MarketConditions) -> str: """Sélectionne la stratégie appropriée""" # Priority 1: Volatilité extrême -> Volatile strategy if conditions.volatility_type in [VolatilityType.EXTREME, VolatilityType.HIGH]: if conditions.atr_percent > 0.8: return 'volatile' # Priority 2: TRENDING DÉSACTIVÉ (trop risqué avec 225x leverage) # if conditions.trend_strength > 25: # return 'trending' # Priority 3: Ranging -> Ranging strategy (MEILLEURE STRATÉGIE!) if conditions.trend_type == TrendType.RANGING or conditions.trend_strength < 30: return 'ranging' # Default: Calm strategy (scalping) return 'calm' def _trending_strategy( self, symbol: str, df: pd.DataFrame, conditions: MarketConditions ) -> Optional[AdaptiveSignal]: """Stratégie de suivi de tendance""" params = self.params['trending'] # Ne trader que si tendance forte if conditions.trend_strength < params['min_trend_strength']: return None close = df['close'].iloc[-1] ema9 = df['close'].ewm(span=9, adjust=False).mean().iloc[-1] ema21 = df['close'].ewm(span=21, adjust=False).mean().iloc[-1] # ATR pour stops atr = self._calculate_atr(df) sl_distance = atr * params['sl_atr_multiplier'] # LONG si tendance haussière if conditions.trend_type in [TrendType.UPTREND, TrendType.STRONG_UPTREND]: if close > ema9 > ema21: return AdaptiveSignal( symbol=symbol, side='LONG', confidence=min(conditions.trend_strength + 25, 95), entry_price=close, stop_loss=close - sl_distance, take_profit=close + (sl_distance * params['tp_multiplier']), leverage=self.leverage, reason=f"Trending UP | Strength: {conditions.trend_strength:.1f} | ADX: {conditions.adx:.1f}", market_conditions=conditions, strategy_used='trending' ) # SHORT si tendance baissière elif conditions.trend_type in [TrendType.DOWNTREND, TrendType.STRONG_DOWNTREND]: if close < ema9 < ema21: return AdaptiveSignal( symbol=symbol, side='SHORT', confidence=min(conditions.trend_strength + 25, 95), entry_price=close, stop_loss=close + sl_distance, take_profit=close - (sl_distance * params['tp_multiplier']), leverage=self.leverage, reason=f"Trending DOWN | Strength: {conditions.trend_strength:.1f} | ADX: {conditions.adx:.1f}", market_conditions=conditions, strategy_used='trending' ) return None def _ranging_strategy( self, symbol: str, df: pd.DataFrame, conditions: MarketConditions ) -> Optional[AdaptiveSignal]: """Stratégie mean reversion pour marché ranging""" params = self.params['ranging'] close = df['close'].iloc[-1] # Bollinger Bands sma20 = df['close'].rolling(window=20).mean().iloc[-1] std = df['close'].rolling(window=20).std().iloc[-1] upper = sma20 + (2 * std) lower = sma20 - (2 * std) # RSI rsi = self._calculate_rsi(df, period=14) # ATR pour stops atr = self._calculate_atr(df) sl_distance = atr * params['sl_atr_multiplier'] tp_distance = atr * params['tp_atr_multiplier'] # LONG si oversold près de la bande basse if close <= lower * 1.001 and rsi < 30: return AdaptiveSignal( symbol=symbol, side='LONG', confidence=min(80 + (30 - rsi), 95), entry_price=close, stop_loss=close - sl_distance, take_profit=close + tp_distance, leverage=self.leverage, reason=f"Mean Reversion LONG | RSI: {rsi:.1f} | BB Lower", market_conditions=conditions, strategy_used='ranging' ) # SHORT si overbought près de la bande haute elif close >= upper * 0.999 and rsi > 70: return AdaptiveSignal( symbol=symbol, side='SHORT', confidence=min(80 + (rsi - 70), 95), entry_price=close, stop_loss=close + sl_distance, take_profit=close - tp_distance, leverage=self.leverage, reason=f"Mean Reversion SHORT | RSI: {rsi:.1f} | BB Upper", market_conditions=conditions, strategy_used='ranging' ) return None def _volatile_strategy( self, symbol: str, df: pd.DataFrame, conditions: MarketConditions ) -> Optional[AdaptiveSignal]: """Stratégie de breakout pour marché volatile""" params = self.params['volatile'] close = df['close'].iloc[-1] volume = df['volume'].iloc[-1] avg_volume = df['volume'].rolling(window=20).mean().iloc[-1] # Breakout avec volume if volume < avg_volume * params['min_volume_spike']: return None # ATR pour stops larges atr = self._calculate_atr(df) sl_distance = atr * params['sl_atr_multiplier'] # Détection breakout high_20 = df['high'].rolling(window=20).max().iloc[-2] low_20 = df['low'].rolling(window=20).min().iloc[-2] # LONG breakout if close > high_20 and conditions.is_bullish: return AdaptiveSignal( symbol=symbol, side='LONG', confidence=min(75 + (volume/avg_volume * 10), 95), entry_price=close, stop_loss=close - sl_distance, take_profit=close + (sl_distance * params['tp_multiplier']), leverage=self.leverage, reason=f"Volatile Breakout UP | Vol: {volume/avg_volume:.1f}x | ATR: {conditions.atr_percent:.2f}%", market_conditions=conditions, strategy_used='volatile' ) # SHORT breakout elif close < low_20 and not conditions.is_bullish: return AdaptiveSignal( symbol=symbol, side='SHORT', confidence=min(75 + (volume/avg_volume * 10), 95), entry_price=close, stop_loss=close + sl_distance, take_profit=close - (sl_distance * params['tp_multiplier']), leverage=self.leverage, reason=f"Volatile Breakout DOWN | Vol: {volume/avg_volume:.1f}x | ATR: {conditions.atr_percent:.2f}%", market_conditions=conditions, strategy_used='volatile' ) return None def _calm_strategy( self, symbol: str, df: pd.DataFrame, conditions: MarketConditions ) -> Optional[AdaptiveSignal]: """Stratégie de scalping pour marché calme""" params = self.params['calm'] # Seulement si vraiment calme if conditions.atr_percent > params['max_atr_percent']: return None close = df['close'].iloc[-1] # Indicateurs rapides rsi = self._calculate_rsi(df, period=7) macd, signal, hist = self._calculate_macd(df, fast=5, slow=13, signal_period=5) # Volume volume = df['volume'].iloc[-1] avg_volume = df['volume'].rolling(window=20).mean().iloc[-1] volume_ratio = volume / avg_volume if avg_volume > 0 else 0 # Bollinger Bands sma20 = df['close'].rolling(window=20).mean().iloc[-1] std = df['close'].rolling(window=20).std().iloc[-1] upper = sma20 + (2 * std) lower = sma20 - (2 * std) bb_position = (close - lower) / (upper - lower) if (upper - lower) > 0 else 0.5 # Stops fixes (comme optimisé) sl_distance = close * params['sl_percent'] tp_distance = sl_distance * params['tp_multiplier'] # Conditions pour LONG if (bb_position < 0.1 and rsi < 25 and volume_ratio > 2.0 and macd > signal): return AdaptiveSignal( symbol=symbol, side='LONG', confidence=min(80 + (30 - rsi) + (volume_ratio * 5), 95), entry_price=close, stop_loss=close - sl_distance, take_profit=close + tp_distance, leverage=self.leverage, reason=f"Scalp LONG | RSI: {rsi:.1f} | BB: {bb_position*100:.0f}% | Vol: {volume_ratio:.1f}x", market_conditions=conditions, strategy_used='calm' ) # Conditions pour SHORT elif (bb_position > 0.9 and rsi > 75 and volume_ratio > 2.0 and macd < signal): return AdaptiveSignal( symbol=symbol, side='SHORT', confidence=min(80 + (rsi - 70) + (volume_ratio * 5), 95), entry_price=close, stop_loss=close + sl_distance, take_profit=close - tp_distance, leverage=self.leverage, reason=f"Scalp SHORT | RSI: {rsi:.1f} | BB: {bb_position*100:.0f}% | Vol: {volume_ratio:.1f}x", market_conditions=conditions, strategy_used='calm' ) return None def _calculate_atr(self, df: pd.DataFrame, period: int = 14) -> float: """Calcule l'ATR""" 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=period, adjust=False).mean() return atr.iloc[-1] def _calculate_rsi(self, df: pd.DataFrame, period: int = 14) -> float: """Calcule le RSI""" delta = df['close'].diff() gain = (delta.where(delta > 0, 0)).ewm(span=period, adjust=False).mean() loss = (-delta.where(delta < 0, 0)).ewm(span=period, adjust=False).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi.iloc[-1] def _calculate_macd(self, df: pd.DataFrame, fast=12, slow=26, signal_period=9): """Calcule le MACD""" close = df['close'] ema_fast = close.ewm(span=fast, adjust=False).mean() ema_slow = close.ewm(span=slow, adjust=False).mean() macd = ema_fast - ema_slow signal = macd.ewm(span=signal_period, adjust=False).mean() hist = macd - signal return macd.iloc[-1], signal.iloc[-1], hist.iloc[-1] if __name__ == '__main__': # Test de la stratégie logger.info("\n🎯 TEST STRATÉGIE ADAPTATIVE\n") strategy = AdaptiveStrategy(leverage=225) # Charger données 5m df = pd.read_csv('historical_data/BTCUSDT_5m_9000candles.csv') df['timestamp'] = pd.to_datetime(df['timestamp']) # Tester sur dernières bougies signals_found = 0 for i in range(-50, 0): df_slice = df.iloc[:len(df)+i].copy() signal = strategy.analyze('BTCUSD_DEGEN', df_slice) if signal: signals_found += 1 logger.info(f"\n✅ SIGNAL #{signals_found}") logger.info(f" {signal.side} @ ${signal.entry_price:,.2f}") logger.info(f" Confiance: {signal.confidence:.1f}%") logger.info(f" Stratégie: {signal.strategy_used.upper()}") logger.info(f" Conditions: {signal.market_conditions}") logger.info(f" Raison: {signal.reason}") logger.info(f"\n📊 {signals_found}/50 bougies ont généré un signal ({signals_found/50*100:.1f}%)")