""" ═══════════════════════════════════════════════════════════════════════════════ AI SELL PREDICTOR - Prédiction IA pour les décisions de vente ═══════════════════════════════════════════════════════════════════════════════ Ce module utilise le modèle LSTM + analyse technique avancée pour prédire le meilleur moment de vendre une position ouverte. PRIORITÉ IA sur les règles fixes: - L'IA peut déclencher une vente AVANT que le SL fixe soit touché - L'IA peut empêcher une vente prématurée si elle prédit une continuation - L'IA ajuste dynamiquement les niveaux SL/TP selon les conditions du marché SÉCURITÉS MAINTENUES: - Stop-loss d'urgence -5% → TOUJOURS prioritaire (safety net absolu) - L'IA ne peut pas empêcher un SL d'urgence ═══════════════════════════════════════════════════════════════════════════════ """ import time import logging import numpy as np from typing import Dict, List, Optional, Tuple from datetime import datetime logger = logging.getLogger(__name__) # ═══════════════════════════════════════════════════════════════════════════════ # IMPORTS CONDITIONNELS # ═══════════════════════════════════════════════════════════════════════════════ TORCH_AVAILABLE = False DEVICE = 'cpu' try: import torch import torch.nn as nn # 🔵 FIX 25/03: Limiter threads CPU pour éviter saturation torch.set_num_threads(2) try: torch.set_num_interop_threads(1) except RuntimeError: pass # Peut être déjà fixé par un autre module importé avant TORCH_AVAILABLE = True if torch.cuda.is_available(): DEVICE = 'cuda' logger.info(f"🔥 AISellPredictor: GPU CUDA détecté ({torch.cuda.get_device_name(0)})") else: logger.info("⚠️ AISellPredictor: Mode CPU (pas de GPU CUDA)") except ImportError: logger.warning("⚠️ PyTorch non disponible - AISellPredictor en mode rule-based uniquement") class AISellPredictor: """ IA prédictive pour les décisions de vente. Combine: 1. Modèle LSTM (prédiction directionnelle court terme) 2. Analyse technique multi-indicateurs avancée 3. Contexte de position (P&L, durée, drawdown depuis peak) 4. Détection de patterns de retournement Produit une recommandation: HOLD / SELL / STRONG_SELL / EXTEND_TP avec un score de confiance et des niveaux SL/TP dynamiques. """ def __init__(self, ai_predictor=None): """ Args: ai_predictor: Instance de AIPredictor (pour accéder au modèle LSTM) """ self.ai_predictor = ai_predictor self.model = None self._load_model() # Cache des prédictions (évite de recalculer trop souvent) self._prediction_cache = {} self._cache_ttl = 15 # 15 secondes # Historique des prédictions pour auto-évaluation self._prediction_history = [] self._correct_predictions = 0 self._total_predictions = 0 # Poids adaptatifs — LSTM dépondéré car biaisé "up" (modèle à ré-entraîner) # Les indicateurs techniques sont la source FIABLE de signal self.weights = { 'lstm_direction': 10, # ⚠️ RÉDUIT (était 25): LSTM biaisé → toujours "up" 'momentum_reversal': 25, # 🔼 PRINCIPAL: retournement de momentum (fiable) 'rsi_divergence': 20, # 🔼 Divergence RSI (signal précoce) 'volume_exhaustion': 10, # Épuisement de volume 'drawdown_from_peak': 20, # 🔼 Chute depuis le pic (protège profits) 'pattern_detection': 15, # Patterns de retournement } # Détection de biais LSTM: si >90% des prédictions sont "up", c'est biaisé self._lstm_up_count = 0 self._lstm_total_count = 0 self._lstm_is_biased = False logger.info(f"✅ AISellPredictor initialisé (LSTM={'Oui' if self.model else 'Non'}, Device={DEVICE})") def _load_model(self): """Charge le modèle LSTM depuis le AIPredictor existant""" if self.ai_predictor and hasattr(self.ai_predictor, 'model') and self.ai_predictor.model: self.model = self.ai_predictor.model logger.info("🧠 AISellPredictor: Modèle LSTM partagé avec AIPredictor") else: logger.info("ℹ️ AISellPredictor: Pas de modèle LSTM - mode analyse technique pure") # ═══════════════════════════════════════════════════════════════════════════ # MÉTHODE PRINCIPALE: PRÉDICTION DE VENTE # ═══════════════════════════════════════════════════════════════════════════ def predict_sell(self, symbol: str, prices: List[float], position: Dict, volumes: List[float] = None) -> Dict: """ Analyse prédictive complète pour décider de vendre ou garder une position. Args: symbol: Symbole de la crypto (ex: BTCUSDT) prices: Liste des prix récents (min 50 valeurs) position: Données de la position {entry_price, quantity, stop_loss, take_profit, timestamp, max_price, max_pnl, ...} volumes: Liste des volumes récents (optionnel) Returns: Dict avec: action: 'HOLD' | 'SELL' | 'STRONG_SELL' | 'EXTEND_TP' confidence: 0-100 sell_score: 0-100 (urgence de vente) predicted_direction: 'up' | 'down' | 'neutral' ai_stop_loss: float or None (suggestion SL dynamique) ai_take_profit: float or None (suggestion TP dynamique) reasons: [str] details: Dict (détails des composants de score) """ # Vérifier le cache cache_key = f"{symbol}_{len(prices)}" now = time.time() # 🔧 FIX 05/04: Purger le cache expiré périodiquement (évite croissance mémoire infinie) if not hasattr(self, '_last_cache_purge'): self._last_cache_purge = 0 if now - self._last_cache_purge > 300: # Toutes les 5 min self._last_cache_purge = now expired = [k for k, (t, _) in self._prediction_cache.items() if now - t > self._cache_ttl * 10] for k in expired: del self._prediction_cache[k] if len(self._prediction_history) > 500: self._prediction_history = self._prediction_history[-500:] if cache_key in self._prediction_cache: cached_time, cached_result = self._prediction_cache[cache_key] if now - cached_time < self._cache_ttl: return cached_result # Vérification minimale if len(prices) < 50: return self._default_hold(symbol, "Données insuffisantes (< 50 prix)") current_price = prices[-1] entry_price = position.get('entry_price', current_price) current_pnl_pct = ((current_price / entry_price) - 1) * 100 if entry_price > 0 else 0 # ═══════════════════════════════════════════════════════════════ # CALCUL DU CONTEXTE DE POSITION # ═══════════════════════════════════════════════════════════════ position_context = self._analyze_position_context(position, current_price, current_pnl_pct) # ═══════════════════════════════════════════════════════════════ # COMPOSANTS DE SCORE # ═══════════════════════════════════════════════════════════════ components = {} reasons = [] # 1. Prédiction LSTM directionnelle (avec détection de biais) lstm_result = self._lstm_predict(prices) components['lstm'] = lstm_result # Tracker le biais LSTM self._lstm_total_count += 1 if lstm_result.get('direction') == 'up': self._lstm_up_count += 1 if self._lstm_total_count >= 10: up_ratio = self._lstm_up_count / self._lstm_total_count self._lstm_is_biased = up_ratio > 0.90 # >90% "up" = biaisé # Si LSTM biaisé → utiliser la prédiction technique comme override if self._lstm_is_biased: tech_direction = self._simple_direction_predict(prices) # Fusionner: garder LSTM mais réduire sa confiance if tech_direction['direction'] != lstm_result['direction']: lstm_result = { 'direction': tech_direction['direction'], 'confidence': tech_direction['confidence'], 'down_probability': tech_direction['down_probability'], 'up_probability': tech_direction['up_probability'], 'model_used': False, 'bias_corrected': True } # 2. Analyse momentum et retournement momentum_result = self._analyze_momentum_reversal(prices) components['momentum'] = momentum_result # 3. Divergence RSI (prix monte mais RSI descend = danger) rsi_result = self._analyze_rsi_divergence(prices) components['rsi'] = rsi_result # 4. Analyse volume (épuisement) volume_result = self._analyze_volume(prices, volumes) components['volume'] = volume_result # 5. Drawdown depuis le pic drawdown_result = self._analyze_drawdown(position, current_price, current_pnl_pct) components['drawdown'] = drawdown_result # 6. Détection de patterns de retournement pattern_result = self._detect_reversal_patterns(prices) components['pattern'] = pattern_result # ═══════════════════════════════════════════════════════════════ # CALCUL DU SELL SCORE PONDÉRÉ # ═══════════════════════════════════════════════════════════════ sell_score = 0 hold_score = 0 # LSTM direction if lstm_result['direction'] == 'down': sell_score += self.weights['lstm_direction'] * lstm_result['confidence'] / 100 reasons.append(f"LSTM prédit baisse ({lstm_result['confidence']:.0f}%)") elif lstm_result['direction'] == 'up': hold_score += self.weights['lstm_direction'] * lstm_result['confidence'] / 100 if lstm_result['confidence'] > 65: reasons.append(f"LSTM prédit hausse ({lstm_result['confidence']:.0f}%)") # Momentum reversal if momentum_result['reversing_down']: sell_score += self.weights['momentum_reversal'] * momentum_result['strength'] / 100 reasons.append(f"Momentum retourne ({momentum_result['detail']})") elif momentum_result['accelerating_up']: hold_score += self.weights['momentum_reversal'] * momentum_result['strength'] / 100 # RSI divergence if rsi_result['bearish_divergence']: sell_score += self.weights['rsi_divergence'] * rsi_result['strength'] / 100 reasons.append(f"Divergence RSI baissière (RSI={rsi_result['rsi']:.0f})") if rsi_result['rsi_overbought']: sell_score += 5 reasons.append(f"RSI en surachat ({rsi_result['rsi']:.0f})") if rsi_result['rsi_oversold']: sell_score += 8 reasons.append(f"RSI effondré ({rsi_result['rsi']:.0f})") # Volume exhaustion if volume_result['exhaustion']: sell_score += self.weights['volume_exhaustion'] * volume_result['strength'] / 100 reasons.append(f"Volume en épuisement") # Drawdown depuis le pic if drawdown_result['significant_drawdown']: sell_score += self.weights['drawdown_from_peak'] * drawdown_result['strength'] / 100 reasons.append(f"Chute {drawdown_result['drawdown_pct']:.1f}% depuis le pic") # Patterns de retournement if pattern_result['bearish_pattern']: sell_score += self.weights['pattern_detection'] * pattern_result['strength'] / 100 reasons.append(f"Pattern: {pattern_result['pattern_name']}") elif pattern_result['bullish_continuation']: hold_score += self.weights['pattern_detection'] * pattern_result['strength'] / 100 # ═══════════════════════════════════════════════════════════════ # AJUSTEMENT SELON LE CONTEXTE DE POSITION # ═══════════════════════════════════════════════════════════════ # Plus on est en profit, plus on protège agressivement if current_pnl_pct >= 3.0: sell_score *= 1.3 # +30% urgence de vente si gros profit reasons.append(f"Protection profit +{current_pnl_pct:.1f}%") elif current_pnl_pct >= 1.5: sell_score *= 1.15 # En perte, l'IA est plus prudente pour couper (laisse le SL gérer) # SAUF si signaux très forts if current_pnl_pct < -1.0 and sell_score < 60: sell_score *= 0.8 # Réduire légèrement, le SL est là pour ça # Position très ancienne sans progrès = fatigue time_held = position_context['time_held_minutes'] if time_held > 120 and abs(current_pnl_pct) < 0.5: sell_score += 8 reasons.append(f"Position stagnante ({time_held:.0f} min)") # ═══════════════════════════════════════════════════════════════ # DÉCISION FINALE # ═══════════════════════════════════════════════════════════════ sell_score = min(100, max(0, sell_score)) hold_score = min(100, max(0, hold_score)) net_score = sell_score - hold_score # Seuils dynamiques selon le P&L sell_threshold = self._get_dynamic_threshold(current_pnl_pct) if net_score >= sell_threshold + 15: action = 'STRONG_SELL' confidence = min(95, 55 + sell_score * 0.4) # Base 55% + proportionnel au sell_score elif net_score >= sell_threshold: action = 'SELL' confidence = min(90, 50 + sell_score * 0.35) elif hold_score > sell_score + 10 and not momentum_result.get('reversing_down'): # EXTEND_TP: ne plus dépendre de LSTM seul (biaisé) # Condition: hold domine ET pas de retournement momentum action = 'EXTEND_TP' hold_dominance = hold_score - sell_score confidence = min(85, 40 + hold_dominance * 1.5) reasons.append(f"IA prédit continuation haussière") else: action = 'HOLD' confidence = max(0, 100 - net_score) # Calculer SL/TP dynamiques suggérés ai_sl, ai_tp = self._compute_dynamic_levels( prices, position, current_price, current_pnl_pct, lstm_result, momentum_result, action ) predicted_direction = lstm_result['direction'] result = { 'action': action, 'confidence': round(confidence, 1), 'sell_score': round(sell_score, 1), 'hold_score': round(hold_score, 1), 'net_score': round(net_score, 1), 'threshold': sell_threshold, 'predicted_direction': predicted_direction, 'ai_stop_loss': ai_sl, 'ai_take_profit': ai_tp, 'reasons': reasons[:5], # Top 5 raisons 'details': { 'lstm': lstm_result, 'momentum': momentum_result, 'rsi': rsi_result, 'drawdown': drawdown_result, 'pattern': pattern_result, 'position_context': position_context, }, 'current_pnl_pct': round(current_pnl_pct, 2), 'time_held_min': round(position_context['time_held_minutes'], 1), } # Mettre en cache self._prediction_cache[cache_key] = (now, result) # Logger si action non-HOLD if action != 'HOLD': logger.info( f"🤖 AI_SELL {symbol}: {action} (score={sell_score:.0f} vs seuil={sell_threshold}, " f"conf={confidence:.0f}%, dir={predicted_direction}) | " f"P&L={current_pnl_pct:+.2f}% | {', '.join(reasons[:3])}" ) return result # ═══════════════════════════════════════════════════════════════════════════ # COMPOSANTS D'ANALYSE # ═══════════════════════════════════════════════════════════════════════════ def _lstm_predict(self, prices: List[float]) -> Dict: """Prédiction LSTM directionnelle (réutilise le modèle existant)""" if not self.model or not TORCH_AVAILABLE or len(prices) < 50: # Fallback: analyse de tendance simple return self._simple_direction_predict(prices) try: prices_arr = np.array(prices[-50:], dtype=np.float32) prices_norm = (prices_arr - np.mean(prices_arr)) / (np.std(prices_arr) + 1e-8) # Construire les features (même format que _predict_with_model) X = np.zeros((1, 50, 20), dtype=np.float32) X[0, :, 0] = prices_norm for i in range(50): if i >= 5: X[0, i, 1] = (prices_norm[i] - prices_norm[i-5]) / 5 # Momentum 5 if i >= 10: X[0, i, 2] = (prices_norm[i] - prices_norm[i-10]) / 10 # Momentum 10 # Features additionnelles pour la vente if i >= 3: X[0, i, 3] = (prices_norm[i] - prices_norm[i-3]) / 3 # Momentum 3 (court terme) if i >= 1: X[0, i, 4] = prices_norm[i] - prices_norm[i-1] # Changement immédiat if i >= 20: X[0, i, 5] = (prices_norm[i] - np.mean(prices_norm[i-20:i])) / (np.std(prices_norm[i-20:i]) + 1e-8) # Z-score X_tensor = torch.tensor(X).to(DEVICE) self.model.eval() with torch.no_grad(): prediction = self.model(X_tensor) probs = torch.softmax(prediction, dim=1).cpu().numpy()[0] # probs[0]=down, probs[1]=neutral, probs[2]=up direction_idx = np.argmax(probs) direction = ['down', 'neutral', 'up'][direction_idx] confidence = float(probs[direction_idx] * 100) # Score de baisse (plus il est haut, plus on devrait vendre) down_score = float(probs[0] * 100) up_score = float(probs[2] * 100) return { 'direction': direction, 'confidence': confidence, 'down_probability': round(down_score, 1), 'up_probability': round(up_score, 1), 'model_used': True } except Exception as e: logger.warning(f"⚠️ LSTM prediction error: {e}") return self._simple_direction_predict(prices) def _simple_direction_predict(self, prices: List[float]) -> Dict: """Fallback: prédiction sans LSTM basée sur la tendance récente""" if len(prices) < 20: return {'direction': 'neutral', 'confidence': 30, 'down_probability': 33, 'up_probability': 33, 'model_used': False} # Momentum multi-échelle mom3 = (prices[-1] - prices[-3]) / prices[-3] * 100 mom5 = (prices[-1] - prices[-5]) / prices[-5] * 100 mom10 = (prices[-1] - prices[-10]) / prices[-10] * 100 # EMA slope ema9 = self._ema(prices, 9) ema21 = self._ema(prices, 21) ema_gap = (ema9 - ema21) / ema21 * 100 # Pondérer les signaux up_signals = sum([1 for x in [mom3, mom5, mom10, ema_gap] if x > 0.1]) down_signals = sum([1 for x in [mom3, mom5, mom10, ema_gap] if x < -0.1]) if down_signals >= 3: direction = 'down' confidence = min(75, 40 + down_signals * 10) elif up_signals >= 3: direction = 'up' confidence = min(75, 40 + up_signals * 10) else: direction = 'neutral' confidence = 40 return { 'direction': direction, 'confidence': confidence, 'down_probability': confidence if direction == 'down' else (100 - confidence) / 2, 'up_probability': confidence if direction == 'up' else (100 - confidence) / 2, 'model_used': False } def _analyze_momentum_reversal(self, prices: List[float]) -> Dict: """Détecte les retournements de momentum (signal précoce de vente)""" if len(prices) < 15: return {'reversing_down': False, 'accelerating_up': False, 'strength': 0, 'detail': ''} mom1 = (prices[-1] - prices[-2]) / prices[-2] * 100 mom3 = (prices[-1] - prices[-3]) / prices[-3] * 100 mom5 = (prices[-1] - prices[-5]) / prices[-5] * 100 mom10 = (prices[-1] - prices[-10]) / prices[-10] * 100 # Accélération du momentum (dérivée seconde) mom3_prev = (prices[-3] - prices[-6]) / prices[-6] * 100 if len(prices) >= 6 else 0 mom_acceleration = mom3 - mom3_prev # Négatif = décélération # Momentum 5min il y a 5 bougies mom5_prev = (prices[-5] - prices[-10]) / prices[-10] * 100 if len(prices) >= 10 else 0 reversing_down = False accelerating_up = False strength = 0 detail = '' # RETOURNEMENT: Momentum court terme négatif ALORS QUE moyen terme était positif # 🔧 FIX 16/03: -0.1→-0.4% (était trop sensible: -0.1% = bougie normale en bull) if mom3 < -0.4 and mom5_prev > 0.3: reversing_down = True strength = min(100, abs(mom3) * 80 + abs(mom_acceleration) * 40) detail = f"mom3={mom3:+.2f}% (prev_mom5={mom5_prev:+.2f}%)" # DÉCÉLÉRATION FORTE: momentum positif mais qui s'essouffle rapidement # 🔧 FIX 16/03: -0.3→-0.6 (décélération mineure = normale en bull) if mom_acceleration < -0.6 and mom3 > 0: reversing_down = True strength = max(strength, min(70, abs(mom_acceleration) * 80)) detail = f"décélération={mom_acceleration:+.2f}%" # CRASH immédiat: -1.0%+ en 3 bougies (était -0.5% = trop sensible) # 🔧 FIX 16/03: -0.5→-1.0% (une correction -0.5% est normale en bull) if mom3 < -1.0: reversing_down = True strength = max(strength, min(100, abs(mom3) * 120)) detail = f"crash mom3={mom3:+.2f}%" # ACCÉLÉRATION HAUSSIÈRE: momentum qui s'accélère if mom_acceleration > 0.2 and mom3 > 0.1 and mom5 > 0.1: accelerating_up = True strength = min(80, mom_acceleration * 100) return { 'reversing_down': reversing_down, 'accelerating_up': accelerating_up, 'strength': min(100, strength), 'detail': detail, 'mom3': mom3, 'mom5': mom5, 'acceleration': mom_acceleration } def _analyze_rsi_divergence(self, prices: List[float]) -> Dict: """Détecte divergences RSI (prix monte mais RSI descend = retournement imminent)""" if len(prices) < 30: return {'bearish_divergence': False, 'rsi_overbought': False, 'rsi_oversold': False, 'rsi': 50, 'strength': 0} rsi = self._rsi(prices) # RSI il y a 10 bougies rsi_prev = self._rsi(prices[:-10]) if len(prices) >= 40 else rsi # Prix il y a 10 bougies price_change_10 = (prices[-1] - prices[-10]) / prices[-10] * 100 rsi_change = rsi - rsi_prev bearish_divergence = False strength = 0 # DIVERGENCE BAISSIÈRE: Prix monte (+0.5%+) MAIS RSI descend (-5+) if price_change_10 > 0.3 and rsi_change < -5: bearish_divergence = True strength = min(100, abs(rsi_change) * 5 + price_change_10 * 20) # RSI en chute libre if rsi_change < -15: bearish_divergence = True strength = max(strength, min(100, abs(rsi_change) * 4)) return { 'bearish_divergence': bearish_divergence, 'rsi_overbought': rsi > 75, 'rsi_oversold': rsi < 25, 'rsi': rsi, 'rsi_prev': rsi_prev, 'rsi_change': rsi_change, 'strength': strength } def _analyze_volume(self, prices: List[float], volumes: List[float] = None) -> Dict: """Analyse l'épuisement de volume (volume décroissant en hausse = faiblesse)""" if not volumes or len(volumes) < 10: return {'exhaustion': False, 'strength': 0} recent_vol = np.mean(volumes[-3:]) prev_vol = np.mean(volumes[-10:-3]) vol_change = (recent_vol - prev_vol) / prev_vol * 100 if prev_vol > 0 else 0 # Prix en hausse mais volume en baisse = épuisement price_up = prices[-1] > prices[-5] if len(prices) >= 5 else False vol_declining = vol_change < -20 exhaustion = price_up and vol_declining strength = min(80, abs(vol_change)) if exhaustion else 0 return { 'exhaustion': exhaustion, 'strength': strength, 'vol_change_pct': round(vol_change, 1) } def _analyze_drawdown(self, position: Dict, current_price: float, current_pnl_pct: float) -> Dict: """Analyse la chute depuis le pic (drawdown = perte de profit acquis)""" max_price = position.get('max_price', current_price) max_pnl = position.get('max_pnl', current_pnl_pct) entry_price = position.get('entry_price', current_price) drawdown_from_peak = ((current_price / max_price) - 1) * 100 if max_price > 0 else 0 profit_lost = max_pnl - current_pnl_pct if max_pnl > current_pnl_pct else 0 significant = False strength = 0 # Si on a perdu plus de 50% du profit maximum atteint if max_pnl > 1.0 and profit_lost > max_pnl * 0.5: significant = True strength = min(100, profit_lost / max_pnl * 100) # Chute > 1.5% depuis le pic if drawdown_from_peak < -1.5: significant = True strength = max(strength, min(100, abs(drawdown_from_peak) * 40)) # Chute > 2% depuis le pic ET on était en profit if drawdown_from_peak < -2.0 and max_pnl > 0.5: strength = max(strength, 85) return { 'significant_drawdown': significant, 'drawdown_pct': round(drawdown_from_peak, 2), 'profit_lost_pct': round(profit_lost, 2), 'max_pnl': round(max_pnl, 2), 'strength': strength } def _detect_reversal_patterns(self, prices: List[float]) -> Dict: """Détecte les patterns techniques de retournement""" if len(prices) < 30: return {'bearish_pattern': False, 'bullish_continuation': False, 'pattern_name': '', 'strength': 0} prices_arr = np.array(prices[-30:]) bearish = False bullish = False pattern_name = '' strength = 0 # === DOUBLE TOP (M pattern) === # Chercher 2 pics proches séparés par un creux highs = [] for i in range(2, len(prices_arr) - 2): if prices_arr[i] > prices_arr[i-1] and prices_arr[i] > prices_arr[i-2] and \ prices_arr[i] > prices_arr[i+1] and prices_arr[i] > prices_arr[i+2]: highs.append((i, prices_arr[i])) if len(highs) >= 2: last_two = highs[-2:] peak1, peak2 = last_two[0][1], last_two[1][1] # Pics similaires (< 0.5% de différence) if abs(peak1 - peak2) / peak1 * 100 < 0.5: # Prix actuel sous les pics if prices_arr[-1] < peak2 * 0.995: bearish = True pattern_name = 'Double Top' strength = 70 # === ENGULFING BAISSIER === if len(prices_arr) >= 3: prev_body = prices_arr[-2] - prices_arr[-3] # Bougie précédente curr_body = prices_arr[-1] - prices_arr[-2] # Bougie actuelle if prev_body > 0 and curr_body < 0 and abs(curr_body) > abs(prev_body) * 1.5: bearish = True pattern_name = 'Bearish Engulfing' strength = max(strength, 60) # === EVENING STAR (3 bougies) === if len(prices_arr) >= 4: body1 = prices_arr[-3] - prices_arr[-4] # Grande bougie haussière body2 = abs(prices_arr[-2] - prices_arr[-3]) # Petite bougie (doji) body3 = prices_arr[-1] - prices_arr[-2] # Grande bougie baissière if body1 > 0 and body2 < body1 * 0.3 and body3 < -body1 * 0.5: bearish = True pattern_name = 'Evening Star' strength = max(strength, 75) # === EMA DEATH CROSS === ema9 = self._ema(prices, 9) ema21 = self._ema(prices, 21) ema9_prev = self._ema(prices[:-1], 9) ema21_prev = self._ema(prices[:-1], 21) # EMA9 vient de passer sous EMA21 if ema9_prev > ema21_prev and ema9 < ema21: bearish = True pattern_name = 'Death Cross EMA' strength = max(strength, 65) # === CONTINUATION HAUSSIÈRE === # EMA9 > EMA21 + écart croissant + momentum positif ema_gap = (ema9 - ema21) / ema21 * 100 ema_gap_prev = (ema9_prev - ema21_prev) / ema21_prev * 100 if ema21_prev > 0 else 0 mom5 = (prices[-1] - prices[-5]) / prices[-5] * 100 if len(prices) >= 5 else 0 if ema_gap > 0.1 and ema_gap > ema_gap_prev and mom5 > 0.1: bullish = True if not bearish: pattern_name = 'Bullish Continuation' strength = min(70, ema_gap * 50 + mom5 * 30) return { 'bearish_pattern': bearish, 'bullish_continuation': bullish and not bearish, 'pattern_name': pattern_name, 'strength': strength } # ═══════════════════════════════════════════════════════════════════════════ # UTILITAIRES # ═══════════════════════════════════════════════════════════════════════════ def _analyze_position_context(self, position: Dict, current_price: float, current_pnl_pct: float) -> Dict: """Analyse le contexte de la position (durée, drawdown, etc.)""" # Durée de la position timestamp_str = position.get('timestamp', '') time_held_minutes = 0 if timestamp_str: try: entry_dt = datetime.fromisoformat(timestamp_str) time_held_minutes = (time.time() - entry_dt.timestamp()) / 60 except: pass max_price = position.get('max_price', current_price) max_pnl = position.get('max_pnl', current_pnl_pct) entry_price = position.get('entry_price', current_price) return { 'time_held_minutes': time_held_minutes, 'current_pnl_pct': current_pnl_pct, 'max_pnl_pct': max_pnl, 'drawdown_from_peak_pct': max_pnl - current_pnl_pct if max_pnl > current_pnl_pct else 0, 'price_vs_entry': (current_price / entry_price - 1) * 100 if entry_price > 0 else 0, } def _get_dynamic_threshold(self, pnl_pct: float) -> float: """ Seuil dynamique de vente selon le P&L. Plus on est en profit, plus le seuil est BAS (on protège les gains). En perte, seuil moyen (le SL gère, mais l'IA peut aider). 🔧 FIX 16/03: Seuils relevés — trop de sorties prématurées en marché haussier """ if pnl_pct >= 5.0: return 35 # En gros profit: vendre au moindre doute (était 25) elif pnl_pct >= 3.0: return 42 # (était 30) elif pnl_pct >= 2.0: return 50 # (était 35) elif pnl_pct >= 1.0: return 55 # ← Principal fix: était 40 → +15pts elif pnl_pct >= 0.5: return 60 # (était 50) elif pnl_pct >= 0: return 65 # Breakeven: très prudent (était 55) elif pnl_pct >= -1.0: return 60 # Petite perte: laisser le SL gérer (était 50) else: return 55 # Grosse perte: IA active, mais moins urgente (était 45) def _compute_dynamic_levels(self, prices: List[float], position: Dict, current_price: float, current_pnl_pct: float, lstm_result: Dict, momentum_result: Dict, action: str) -> Tuple[Optional[float], Optional[float]]: """ Calcule des niveaux SL/TP dynamiques suggérés par l'IA. Returns: (ai_stop_loss, ai_take_profit) """ entry_price = position.get('entry_price', current_price) current_sl = position.get('stop_loss', entry_price * 0.975) current_tp = position.get('take_profit', entry_price * 1.025) ai_sl = None ai_tp = None # === SL DYNAMIQUE (basé sur technique, PAS seulement LSTM) === # 1. Momentum retourne → resserrer SL if momentum_result.get('reversing_down') and momentum_result.get('strength', 0) > 40: # Retournement confirmé → protéger if current_pnl_pct > 1.0: protect_sl = current_price * 0.992 # -0.8% du prix elif current_pnl_pct > 0.3: protect_sl = current_price * 0.985 # -1.5% du prix else: protect_sl = current_price * 0.98 # -2% du prix if protect_sl > current_sl: ai_sl = protect_sl # 2. LSTM prédit baisse (si pas biaisé) → resserrer if lstm_result.get('direction') == 'down' and lstm_result.get('confidence', 0) > 55: tighter_sl = current_price * 0.985 if ai_sl: ai_sl = max(ai_sl, tighter_sl) elif tighter_sl > current_sl: ai_sl = tighter_sl # 3. En profit significatif → toujours protéger if current_pnl_pct >= 2.0: tight_sl = current_price * 0.99 # -1% if ai_sl: ai_sl = max(ai_sl, tight_sl) elif tight_sl > current_sl: ai_sl = tight_sl elif current_pnl_pct >= 1.0: tight_sl = current_price * 0.985 # -1.5% if ai_sl: ai_sl = max(ai_sl, tight_sl) elif tight_sl > current_sl: ai_sl = tight_sl # 4. Action SELL/STRONG_SELL → SL agressif if action in ('SELL', 'STRONG_SELL') and current_pnl_pct > 0: aggressive_sl = current_price * 0.995 # -0.5% if ai_sl: ai_sl = max(ai_sl, aggressive_sl) elif aggressive_sl > current_sl: ai_sl = aggressive_sl # === TP DYNAMIQUE === if action == 'EXTEND_TP': # Extension de TP si continuation prédite extended_tp = current_price * 1.025 # +2.5% du prix actuel (réduit: était 3%) if extended_tp > current_tp: ai_tp = extended_tp return ai_sl, ai_tp def _default_hold(self, symbol: str, reason: str) -> Dict: """Recommandation HOLD par défaut""" return { 'action': 'HOLD', 'confidence': 50, 'sell_score': 0, 'hold_score': 50, 'net_score': 0, 'threshold': 55, 'predicted_direction': 'neutral', 'ai_stop_loss': None, 'ai_take_profit': None, 'reasons': [reason], 'details': {}, 'current_pnl_pct': 0, 'time_held_min': 0, } # ═══════════════════════════════════════════════════════════════════════════ # FONCTIONS TECHNIQUES (CALCUL D'INDICATEURS) # ═══════════════════════════════════════════════════════════════════════════ @staticmethod def _ema(prices, period): """Calcul EMA""" if len(prices) < period: return prices[-1] if prices else 0 prices_arr = np.array(prices) multiplier = 2 / (period + 1) ema = prices_arr[0] for p in prices_arr[1:]: ema = (p - ema) * multiplier + ema return float(ema) @staticmethod def _rsi(prices, period=14): """Calcul RSI""" if len(prices) < period + 1: return 50 prices_arr = np.array(prices) deltas = np.diff(prices_arr) gains = np.maximum(deltas, 0) losses = np.abs(np.minimum(deltas, 0)) avg_gain = np.mean(gains[-period:]) avg_loss = np.mean(losses[-period:]) if avg_loss == 0: return 100 rs = avg_gain / avg_loss return 100 - (100 / (1 + rs)) # ═══════════════════════════════════════════════════════════════════════════ # AUTO-ÉVALUATION (pour amélioration continue) # ═══════════════════════════════════════════════════════════════════════════ def record_outcome(self, symbol: str, prediction_action: str, actual_pnl_change: float): """ Enregistre le résultat réel pour auto-évaluer les prédictions. Appelé quand une position est effectivement fermée. """ self._total_predictions += 1 # SELL/STRONG_SELL était correct si le prix a effectivement baissé if prediction_action in ['SELL', 'STRONG_SELL'] and actual_pnl_change < 0: self._correct_predictions += 1 # HOLD/EXTEND_TP était correct si le prix a effectivement monté elif prediction_action in ['HOLD', 'EXTEND_TP'] and actual_pnl_change > 0: self._correct_predictions += 1 accuracy = (self._correct_predictions / self._total_predictions * 100) \ if self._total_predictions > 0 else 0 if self._total_predictions % 20 == 0: logger.info(f"📊 AI Sell Predictor accuracy: {accuracy:.1f}% " f"({self._correct_predictions}/{self._total_predictions})") def get_stats(self) -> Dict: """Retourne les statistiques de performance""" accuracy = (self._correct_predictions / self._total_predictions * 100) \ if self._total_predictions > 0 else 0 return { 'total_predictions': self._total_predictions, 'correct_predictions': self._correct_predictions, 'accuracy_pct': round(accuracy, 1), 'lstm_available': self.model is not None, 'device': DEVICE, } # ═══════════════════════════════════════════════════════════════════════════════ # SINGLETON # ═══════════════════════════════════════════════════════════════════════════════ _sell_predictor_instance = None def get_sell_predictor(ai_predictor=None) -> AISellPredictor: """Retourne l'instance singleton du AISellPredictor""" global _sell_predictor_instance if _sell_predictor_instance is None: _sell_predictor_instance = AISellPredictor(ai_predictor) return _sell_predictor_instance