""" Long-Term Trend Analyzer ======================== Détecte les tendances baissières prolongées (Dead Cat Bounce) et les faux rebonds dans une chute continue Cas typique: FET 12/01 - Tendance baissière 6h+ puis rebond temporaire avant rechute """ import numpy as np from datetime import datetime, timedelta from typing import Dict, List, Tuple, Optional import logging logger = logging.getLogger("LongTermTrend") class LongTermTrendAnalyzer: """Analyse les tendances sur plusieurs heures pour éviter les faux rebonds""" def __init__(self, binance_client=None): self.client = binance_client self.cache = {} self.cache_duration = 300 # 5 minutes def analyze_long_term_trend(self, symbol: str, current_prices: List[float] = None) -> Dict: """ Analyse la tendance sur 4h, 8h et 12h Détecte les "dead cat bounce" (rebonds temporaires dans une chute) Returns: { 'trend_4h': -2.5, # % de variation sur 4h 'trend_8h': -5.2, # % de variation sur 8h 'trend_12h': -7.8, # % de variation sur 12h 'is_prolonged_downtrend': True, # Baisse continue > 2h 'is_dead_cat_bounce': True, # Rebond temporaire dans chute 'downtrend_duration_hours': 6, # Durée de la baisse 'risk_score': 15, # 0-100 (0=très risqué, 100=sûr) 'recommendation': 'AVOID', # AVOID, CAUTION, OK 'reasons': [] } """ # Vérifier cache cache_key = f"{symbol}_longterm" if cache_key in self.cache: cached_data, cached_time = self.cache[cache_key] if (datetime.now() - cached_time).total_seconds() < self.cache_duration: return cached_data result = { 'trend_4h': 0.0, 'trend_8h': 0.0, 'trend_12h': 0.0, 'is_prolonged_downtrend': False, 'is_dead_cat_bounce': False, 'downtrend_duration_hours': 0, 'risk_score': 50, 'recommendation': 'OK', 'reasons': [] } try: # Si pas de client Binance, utiliser les prix fournis if not self.client: if current_prices and len(current_prices) >= 50: # Estimer sur les prix disponibles result = self._analyze_from_prices(current_prices) return result # Récupérer les données historiques (15 min = bougies longues) klines_15m = self.client.get_klines(symbol, interval='15m', limit=100) if not klines_15m or len(klines_15m) < 50: logger.warning(f"⚠️ {symbol}: Pas assez de données 15min") return result prices_15m = [float(k[4]) for k in klines_15m] # Close prices volumes_15m = [float(k[5]) for k in klines_15m] # Calculer les tendances current_price = prices_15m[-1] # 4h = 16 bougies de 15min if len(prices_15m) >= 16: price_4h_ago = prices_15m[-16] result['trend_4h'] = ((current_price - price_4h_ago) / price_4h_ago) * 100 # 8h = 32 bougies de 15min if len(prices_15m) >= 32: price_8h_ago = prices_15m[-32] result['trend_8h'] = ((current_price - price_8h_ago) / price_8h_ago) * 100 # 12h = 48 bougies de 15min if len(prices_15m) >= 48: price_12h_ago = prices_15m[-48] result['trend_12h'] = ((current_price - price_12h_ago) / price_12h_ago) * 100 # ═══════════════════════════════════════════════════════════════════ # DÉTECTION TENDANCE BAISSIÈRE PROLONGÉE # ═══════════════════════════════════════════════════════════════════ # Critère 1: Baisse continue sur 4h+ if result['trend_4h'] < -2.0: # -2% sur 4h result['is_prolonged_downtrend'] = True result['reasons'].append(f"Baisse continue 4h ({result['trend_4h']:.1f}%)") # Critère 2: Baisse continue sur 8h+ if result['trend_8h'] < -3.0: # -3% sur 8h result['is_prolonged_downtrend'] = True result['reasons'].append(f"Baisse continue 8h ({result['trend_8h']:.1f}%)") # Critère 3: Baisse continue sur 12h+ if result['trend_12h'] < -4.0: # -4% sur 12h result['is_prolonged_downtrend'] = True result['reasons'].append(f"Baisse continue 12h ({result['trend_12h']:.1f}%)") # ═══════════════════════════════════════════════════════════════════ # DÉTECTION "DEAD CAT BOUNCE" (rebond temporaire dans chute) # ═══════════════════════════════════════════════════════════════════ # Un dead cat bounce se caractérise par: # 1. Tendance longue (4h+) clairement baissière # 2. Rebond récent (dernière heure) mais faible # 3. Volume du rebond plus faible que la chute if result['is_prolonged_downtrend']: # Analyser les 2 dernières bougies (30min) au lieu de 1h - plus réactif recent_prices = prices_15m[-2:] recent_volumes = volumes_15m[-2:] # Momentum récent (30min) momentum_30m = ((recent_prices[-1] - recent_prices[0]) / recent_prices[0]) * 100 # Volume moyen récent vs volume moyen global avg_volume_recent = np.mean(recent_volumes) avg_volume_global = np.mean(volumes_15m[-20:]) # 5h volume_ratio = avg_volume_recent / avg_volume_global if avg_volume_global > 0 else 1 # 🚨 MODIFIÉ: Dead cat bounce SEULEMENT si: # - Rebond faible (< 2%) # - Volume très faible (< 60% du volume moyen) # - Tendance 4h TRÈS baissière (< -5%) pour éviter de bloquer les vraies reprises if momentum_30m > 0.3 and momentum_30m < 2.0 and volume_ratio < 0.6 and result['trend_4h'] < -5.0: result['is_dead_cat_bounce'] = True result['reasons'].append(f"Dead Cat Bounce: Rebond +{momentum_30m:.1f}% avec volume très faible ({volume_ratio*100:.0f}%)") # ═══════════════════════════════════════════════════════════════════ # CALCULER LA DURÉE DE LA TENDANCE BAISSIÈRE # ═══════════════════════════════════════════════════════════════════ if result['is_prolonged_downtrend']: # Compter combien de bougies consécutives sont en baisse consecutive_down = 0 for i in range(len(prices_15m) - 1, 0, -1): if prices_15m[i] < prices_15m[i-1]: consecutive_down += 1 else: # Tolérer 1-2 bougies vertes (correction mineure) if consecutive_down > 0 and i > 2: # Vérifier si c'était juste une petite correction next_candles = prices_15m[i:min(i+3, len(prices_15m))] if len(next_candles) >= 2 and next_candles[-1] < next_candles[0]: consecutive_down += 1 continue break result['downtrend_duration_hours'] = (consecutive_down * 15) / 60 # En heures # ═══════════════════════════════════════════════════════════════════ # CALCULER LE RISK SCORE (0-100) # ═══════════════════════════════════════════════════════════════════ risk_score = 50 # Neutre par défaut # 🚨 MODIFIÉ: Pénalités assouplies pour permettre les reprises if result['is_dead_cat_bounce']: risk_score -= 25 # Réduit de -40 à -25 elif result['is_prolonged_downtrend']: # Pénalité selon l'intensité de la baisse if result['trend_4h'] < -8.0: risk_score -= 35 # Chute brutale (seuil augmenté de -5 à -8) elif result['trend_4h'] < -5.0: risk_score -= 25 # Baisse forte (seuil augmenté de -3 à -5) elif result['trend_4h'] < -3.0: risk_score -= 15 # Baisse modérée (seuil augmenté de -2 à -3) # Pénalité selon la durée if result['downtrend_duration_hours'] > 8: risk_score -= 15 elif result['downtrend_duration_hours'] > 4: risk_score -= 10 # Bonus si tendance positive if result['trend_4h'] > 2.0: risk_score += 20 elif result['trend_4h'] > 1.0: risk_score += 10 result['risk_score'] = max(0, min(100, risk_score)) # ═══════════════════════════════════════════════════════════════════ # RECOMMANDATION FINALE - MODIFIÉE pour permettre les vraies reprises # ═══════════════════════════════════════════════════════════════════ # 🚨 MODIFIÉ: Ne bloquer Dead Cat Bounce que si le rebond est vraiment faible # Si momentum > 2%, c'est probablement une vraie reprise, pas un dead cat bounce if result['is_dead_cat_bounce'] and result['risk_score'] < 25: result['recommendation'] = 'AVOID' result['reasons'].append("⛔ ÉVITER: Dead Cat Bounce confirmé (rebond faible + volume faible)") elif result['is_prolonged_downtrend'] and result['risk_score'] < 20: result['recommendation'] = 'AVOID' result['reasons'].append(f"⛔ ÉVITER: Tendance baissière sévère ({result['downtrend_duration_hours']:.1f}h)") elif result['risk_score'] < 35: result['recommendation'] = 'CAUTION' result['reasons'].append(f"⚠️ PRUDENCE: Risque modéré (score={result['risk_score']})") else: result['recommendation'] = 'OK' # Mettre en cache self.cache[cache_key] = (result, datetime.now()) return result except Exception as e: logger.error(f"Erreur analyse long terme {symbol}: {e}") return result def _analyze_from_prices(self, prices: List[float]) -> Dict: """Analyse approximative à partir des prix 5min disponibles""" result = { 'trend_4h': 0.0, 'trend_8h': 0.0, 'trend_12h': 0.0, 'is_prolonged_downtrend': False, 'is_dead_cat_bounce': False, 'downtrend_duration_hours': 0, 'risk_score': 50, 'recommendation': 'OK', 'reasons': [] } current_price = prices[-1] # Estimation sur les 100 dernières bougies (5min = 8h20) # 4h = 48 bougies de 5min if len(prices) >= 48: price_4h_ago = prices[-48] result['trend_4h'] = ((current_price - price_4h_ago) / price_4h_ago) * 100 # 8h = 96 bougies de 5min if len(prices) >= 96: price_8h_ago = prices[-96] result['trend_8h'] = ((current_price - price_8h_ago) / price_8h_ago) * 100 # Détection tendance baissière prolongée if result['trend_4h'] < -2.0: result['is_prolonged_downtrend'] = True result['reasons'].append(f"Baisse continue 4h ({result['trend_4h']:.1f}%)") # Détection Dead Cat Bounce # Analyser momentum récent vs tendance globale if len(prices) >= 60: # Momentum 1h (12 bougies de 5min) price_1h_ago = prices[-12] momentum_1h = ((current_price - price_1h_ago) / price_1h_ago) * 100 # Si momentum 1h positif MAIS trend 4h négatif = dead cat bounce if momentum_1h > 0.3 and result['trend_4h'] < -2.5: result['is_dead_cat_bounce'] = True result['reasons'].append(f"Dead Cat Bounce: +{momentum_1h:.1f}% mais trend 4h {result['trend_4h']:.1f}%") # Calculer durée baisse (compter bougies descendantes consécutives) consecutive_down = 0 for i in range(len(prices) - 1, max(0, len(prices) - 50), -1): if i > 0 and prices[i] < prices[i-1]: consecutive_down += 1 elif consecutive_down > 3: # Tolérer 1-2 bougies vertes break if consecutive_down > 10: # Plus de 10 bougies de 5min = 50min+ result['downtrend_duration_hours'] = (consecutive_down * 5) / 60 # Calculer risk score risk_score = 50 if result['is_dead_cat_bounce']: risk_score = 10 # Très risqué result['recommendation'] = 'AVOID' elif result['is_prolonged_downtrend']: if result['trend_4h'] < -5.0: risk_score = 15 result['recommendation'] = 'AVOID' elif result['trend_4h'] < -3.0: risk_score = 25 result['recommendation'] = 'CAUTION' elif result['trend_4h'] < -2.0: risk_score = 35 result['recommendation'] = 'CAUTION' elif result['trend_4h'] > 2.0: risk_score = 70 # Favorable result['risk_score'] = risk_score return result def should_block_buy(self, analysis: Dict) -> Tuple[bool, str]: """ Détermine si l'achat doit être bloqué selon l'analyse long terme Returns: (should_block: bool, reason: str) """ if analysis['recommendation'] == 'AVOID': return True, ' | '.join(analysis['reasons']) if analysis['is_dead_cat_bounce']: return True, "Dead Cat Bounce - Rebond temporaire dans chute prolongée" if analysis['risk_score'] < 20: return True, f"Risque trop élevé (score={analysis['risk_score']})" return False, "" def get_score_penalty(self, analysis: Dict) -> int: """ Calcule la pénalité à appliquer au score IA selon l'analyse long terme Returns: Pénalité (0-50 points) """ if analysis['recommendation'] == 'AVOID': return 50 # Pénalité maximale if analysis['is_dead_cat_bounce']: return 40 if analysis['is_prolonged_downtrend']: # Pénalité proportionnelle à l'intensité if analysis['trend_4h'] < -5.0: return 35 elif analysis['trend_4h'] < -3.0: return 25 elif analysis['trend_4h'] < -2.0: return 15 # Pénalité selon le risk score if analysis['risk_score'] < 30: return 20 elif analysis['risk_score'] < 40: return 10 return 0 # Instance globale _long_term_analyzer = None def get_long_term_analyzer(binance_client=None): """Récupère l'instance globale de l'analyseur de tendance long terme""" global _long_term_analyzer if _long_term_analyzer is None: _long_term_analyzer = LongTermTrendAnalyzer(binance_client) elif binance_client and not _long_term_analyzer.client: _long_term_analyzer.client = binance_client return _long_term_analyzer