#!/usr/bin/env python3 """ Outlier Detection - Détection des données aberrantes Inspiré de FreqAI - Évite les faux signaux sur pumps/dumps artificiels Méthodes: - Isolation Forest (sklearn) - Z-Score (statistique) - IQR (Interquartile Range) """ import numpy as np from typing import Dict, List, Tuple, Optional import logging from dataclasses import dataclass logging.basicConfig(level=logging.INFO) logger = logging.getLogger("OutlierDetector") # Import sklearn si disponible SKLEARN_AVAILABLE = False try: from sklearn.ensemble import IsolationForest SKLEARN_AVAILABLE = True logger.info("✅ Sklearn disponible - Isolation Forest activé") except ImportError: logger.warning("⚠️ Sklearn non disponible - utilisation méthodes statistiques uniquement") @dataclass class OutlierAnalysis: """Résultat de l'analyse d'outliers""" is_outlier: bool method: str confidence: float # 0-1 reason: str score: float # Score d'anomalie class OutlierDetector: """ Détecteur d'outliers multi-méthodes Détecte: - Pumps artificiels (volume explosif + price spike) - Flash crashes (chute brutale isolée) - Données corrompues (NaN, valeurs aberrantes) """ def __init__(self, contamination: float = 0.05, z_score_threshold: float = 3.0, volume_spike_threshold: float = 5.0): """ Args: contamination: % d'outliers attendus (Isolation Forest) z_score_threshold: Seuil Z-Score pour outliers volume_spike_threshold: Multiplicateur volume pour spike """ self.contamination = contamination self.z_threshold = z_score_threshold self.volume_threshold = volume_spike_threshold # Isolation Forest if SKLEARN_AVAILABLE: self.isolation_forest = IsolationForest( contamination=contamination, random_state=42, n_estimators=100 ) else: self.isolation_forest = None logger.info("✅ Outlier Detector initialisé") logger.info(f" • Contamination: {contamination*100}%") logger.info(f" • Z-Score threshold: {z_score_threshold}") logger.info(f" • Volume spike threshold: {volume_spike_threshold}x") def detect_outlier(self, symbol: str, prices: List[float], volumes: List[float], rsi: Optional[float] = None, bb_position: Optional[float] = None) -> OutlierAnalysis: """ Détecter si les données actuelles sont des outliers Args: symbol: Symbole crypto prices: Historique des prix (dernier = actuel) volumes: Historique des volumes rsi: RSI actuel (optionnel) bb_position: Position Bollinger (0-1, optionnel) Returns: OutlierAnalysis avec résultat """ if len(prices) < 20 or len(volumes) < 20: return OutlierAnalysis( is_outlier=False, method="INSUFFICIENT_DATA", confidence=0.0, reason="Pas assez de données historiques", score=0.0 ) # Vérifier NaN if np.isnan(prices[-1]) or np.isnan(volumes[-1]): return OutlierAnalysis( is_outlier=True, method="NAN_DETECTION", confidence=1.0, reason="Données NaN détectées", score=1.0 ) # Méthode 1: Z-Score (prix) price_zscore = self._calculate_zscore(prices) if abs(price_zscore) > self.z_threshold: return OutlierAnalysis( is_outlier=True, method="Z_SCORE_PRICE", confidence=min(abs(price_zscore) / self.z_threshold, 1.0), reason=f"Prix aberrant (Z-Score={price_zscore:.2f})", score=abs(price_zscore) ) # Méthode 2: Volume Spike volume_ratio = self._calculate_volume_spike(volumes) if volume_ratio > self.volume_threshold: return OutlierAnalysis( is_outlier=True, method="VOLUME_SPIKE", confidence=min(volume_ratio / self.volume_threshold, 1.0), reason=f"Volume explosif ({volume_ratio:.1f}x normal)", score=volume_ratio ) # Méthode 3: Isolation Forest (si sklearn disponible) if SKLEARN_AVAILABLE and rsi is not None and bb_position is not None: features = np.array([[ prices[-1], volumes[-1], rsi, bb_position, self._calculate_momentum(prices, periods=3), self._calculate_momentum(prices, periods=5) ]]) # Fit sur données historiques historical_features = [] for i in range(-20, -1): if i >= -len(prices): historical_features.append([ prices[i], volumes[i], rsi, # Simplifié - devrait calculer RSI historique bb_position, self._calculate_momentum(prices[:i+1], periods=3), self._calculate_momentum(prices[:i+1], periods=5) ]) if len(historical_features) >= 10: self.isolation_forest.fit(historical_features) prediction = self.isolation_forest.predict(features)[0] if prediction == -1: # Outlier détecté # Score d'anomalie (plus négatif = plus anormal) anomaly_score = self.isolation_forest.score_samples(features)[0] outlier_confidence = abs(anomaly_score) # 🔧 FIX 16/03: Seuil minimum de confiance — éviter faux positifs # DOTUSDT bloqué à 64.67% alors que score IA=90 CREUX_REBOUND (signal légitime) # Seulement bloquer si l'anomalie est vraiment significative (≥70%) if outlier_confidence >= 0.70: return OutlierAnalysis( is_outlier=True, method="ISOLATION_FOREST", confidence=outlier_confidence, reason="Pattern anormal détecté par Isolation Forest", score=outlier_confidence ) # else: confiance insuffisante, ne pas bloquer # Méthode 4: Price Spike (changement brutal) price_change_pct = ((prices[-1] - prices[-2]) / prices[-2]) * 100 if len(prices) >= 2 else 0 if abs(price_change_pct) > 10: # +/-10% en 1 bougie return OutlierAnalysis( is_outlier=True, method="PRICE_SPIKE", confidence=min(abs(price_change_pct) / 10, 1.0), reason=f"Spike brutal ({price_change_pct:+.1f}% en 1 bougie)", score=abs(price_change_pct) ) # Méthode 5: Flash Crash Detection if self._is_flash_crash(prices): return OutlierAnalysis( is_outlier=True, method="FLASH_CRASH", confidence=0.9, reason="Flash crash détecté (chute brutale + rebond)", score=10.0 ) # Pas d'outlier détecté return OutlierAnalysis( is_outlier=False, method="NORMAL", confidence=0.0, reason="Données normales", score=0.0 ) def _calculate_zscore(self, values: List[float]) -> float: """Calculer le Z-Score de la dernière valeur""" if len(values) < 2: return 0.0 mean = np.mean(values[:-1]) std = np.std(values[:-1]) if std == 0: return 0.0 return (values[-1] - mean) / std def _calculate_volume_spike(self, volumes: List[float]) -> float: """Calculer le ratio volume actuel / volume moyen""" if len(volumes) < 20: return 0.0 avg_volume = np.mean(volumes[-20:-1]) if avg_volume == 0: return 0.0 return volumes[-1] / avg_volume def _calculate_momentum(self, prices: List[float], periods: int = 3) -> float: """Calculer momentum sur N périodes""" if len(prices) < periods + 1: return 0.0 return ((prices[-1] - prices[-periods-1]) / prices[-periods-1]) * 100 def _is_flash_crash(self, prices: List[float], threshold: float = -5.0) -> bool: """ Détecter un flash crash Critères: Chute > 5% suivie d'un rebond > 50% de la chute """ if len(prices) < 5: return False # Chercher pattern: prix_1 → prix_2 (chute) → prix_3 (rebond) for i in range(-3, -1): if i + 2 >= 0: continue drop_pct = ((prices[i+1] - prices[i]) / prices[i]) * 100 recovery_pct = ((prices[i+2] - prices[i+1]) / prices[i+1]) * 100 # Flash crash: chute brutale + rebond rapide if drop_pct < threshold and recovery_pct > abs(drop_pct) * 0.5: return True return False def detect_pump_dump(self, symbol: str, prices: List[float], volumes: List[float], lookback: int = 10) -> Tuple[bool, str]: """ Détecter un pump&dump artificiel Critères: - Volume explosif (>5x normal) - Price spike >15% - Suivi d'une chute rapide Returns: (is_pump_dump, reason) """ if len(prices) < lookback or len(volumes) < lookback: return False, "Données insuffisantes" recent_prices = prices[-lookback:] recent_volumes = volumes[-lookback:] # 1. Volume explosif avg_volume = np.mean(volumes[-30:-lookback]) max_recent_volume = max(recent_volumes) volume_ratio = max_recent_volume / avg_volume if avg_volume > 0 else 0 # 2. Price spike max_price = max(recent_prices) min_price = min(recent_prices[:lookback//2]) price_spike = ((max_price - min_price) / min_price) * 100 # 3. Retombée rapide current_price = recent_prices[-1] price_drop = ((current_price - max_price) / max_price) * 100 # Détection pump&dump if (volume_ratio > 5.0 and price_spike > 15 and price_drop < -10): return True, f"Pump&Dump détecté: +{price_spike:.1f}% puis {price_drop:.1f}% (vol x{volume_ratio:.1f})" return False, "Normal" # Singleton global _outlier_detector_instance = None def get_outlier_detector() -> OutlierDetector: """Obtenir l'instance singleton du détecteur""" global _outlier_detector_instance if _outlier_detector_instance is None: _outlier_detector_instance = OutlierDetector() return _outlier_detector_instance if __name__ == "__main__": """Test du détecteur d'outliers""" print("="*60) print("🧪 TEST OUTLIER DETECTOR") print("="*60) detector = get_outlier_detector() # Test 1: Prix normal print("\n1️⃣ Test prix normal:") normal_prices = [100, 101, 100.5, 102, 101.5, 103, 102.5, 104] normal_volumes = [1000, 1100, 950, 1050, 1000, 1200, 1100, 1000] result = detector.detect_outlier("TESTUSDT", normal_prices, normal_volumes, rsi=50, bb_position=0.5) print(f" Outlier: {result.is_outlier}") print(f" Raison: {result.reason}") # Test 2: Volume spike print("\n2️⃣ Test volume spike:") spike_volumes = [1000, 1100, 950, 1050, 1000, 1200, 1100, 10000] result = detector.detect_outlier("TESTUSDT", normal_prices, spike_volumes, rsi=50, bb_position=0.5) print(f" Outlier: {result.is_outlier}") print(f" Méthode: {result.method}") print(f" Raison: {result.reason}") # Test 3: Price spike print("\n3️⃣ Test price spike:") spike_prices = [100, 101, 100.5, 102, 101.5, 103, 102.5, 150] result = detector.detect_outlier("TESTUSDT", spike_prices, normal_volumes, rsi=50, bb_position=0.5) print(f" Outlier: {result.is_outlier}") print(f" Méthode: {result.method}") print(f" Raison: {result.reason}") # Test 4: Pump & Dump print("\n4️⃣ Test pump&dump:") pump_prices = [100, 105, 115, 135, 140, 130, 110, 105, 102, 100] pump_volumes = [1000, 2000, 5000, 10000, 12000, 8000, 3000, 2000, 1500, 1000] is_pump, reason = detector.detect_pump_dump("TESTUSDT", pump_prices, pump_volumes) print(f" Pump&Dump: {is_pump}") print(f" Raison: {reason}")