""" Enhanced Signal Engine - Intégration des nouveaux modules Combine stratégies avancées, ML Ensemble et feature engineering """ import logging from typing import List, Dict, Optional, Tuple import numpy as np # Import des nouveaux modules try: from advanced_strategies import AdvancedStrategies ADVANCED_STRATEGIES_AVAILABLE = True except ImportError: ADVANCED_STRATEGIES_AVAILABLE = False logging.warning("⚠️ Module advanced_strategies non disponible") try: from ensemble_ml import get_ensemble_predictor ENSEMBLE_ML_AVAILABLE = True except ImportError: ENSEMBLE_ML_AVAILABLE = False logging.warning("⚠️ Module ensemble_ml non disponible") try: from feature_engineering import FeatureEngineer FEATURE_ENGINEERING_AVAILABLE = True except ImportError: FEATURE_ENGINEERING_AVAILABLE = False logging.warning("⚠️ Module feature_engineering non disponible") # Import de la config from config import ( ENABLE_ADVANCED_STRATEGIES, ENABLE_ENSEMBLE_ML, ENABLE_ADVANCED_FEATURES, ACTIVE_STRATEGIES, STRATEGY_CONSENSUS_THRESHOLD, ML_CONFIDENCE_THRESHOLD ) class EnhancedSignalEngine: """ Moteur de signaux enrichi combinant toutes les nouvelles fonctionnalités """ def __init__(self): self.ml_predictor = None self.strategies_enabled = ENABLE_ADVANCED_STRATEGIES and ADVANCED_STRATEGIES_AVAILABLE self.ml_enabled = ENABLE_ENSEMBLE_ML and ENSEMBLE_ML_AVAILABLE self.features_enabled = ENABLE_ADVANCED_FEATURES and FEATURE_ENGINEERING_AVAILABLE # Initialiser le prédicteur ML si activé if self.ml_enabled: try: self.ml_predictor = get_ensemble_predictor() logging.info("✅ ML Ensemble initialisé") except Exception as e: logging.error(f"❌ Erreur initialisation ML: {e}") self.ml_enabled = False logging.info(f"🎯 Enhanced Signal Engine:") logging.info(f" - Stratégies avancées: {'✅' if self.strategies_enabled else '❌'}") logging.info(f" - ML Ensemble: {'✅' if self.ml_enabled else '❌'}") logging.info(f" - Feature engineering: {'✅' if self.features_enabled else '❌'}") def get_enhanced_signal(self, symbol: str, prices: List[float], highs: Optional[List[float]] = None, lows: Optional[List[float]] = None, opens: Optional[List[float]] = None, closes: Optional[List[float]] = None, volumes: Optional[List[float]] = None, base_signal: Optional[str] = None, base_confidence: float = 0.5) -> Tuple[str, float, Dict]: """ Obtient un signal enrichi en combinant toutes les méthodes Returns: (signal, confidence, details) - signal: "BUY", "SELL", "HOLD" - confidence: 0-1 - details: dict avec les détails de chaque méthode """ details = { 'base_signal': base_signal, 'base_confidence': base_confidence, 'strategies_result': None, 'ml_result': None, 'features': None } signals = [] weights = [] # 1. Signal de base (stratégie existante) if base_signal: signals.append(base_signal) weights.append(base_confidence * 0.3) # 30% de poids # 2. Stratégies avancées if self.strategies_enabled and len(prices) >= 100: try: # Utiliser closes ou prices par défaut if closes is None: closes = prices if highs is None: highs = prices if lows is None: lows = prices if opens is None: opens = prices strategy_results = AdvancedStrategies.evaluate_all_strategies( prices=prices, highs=highs, lows=lows, opens=opens, closes=closes, volumes=volumes ) details['strategies_result'] = strategy_results # Consensus des stratégies consensus = strategy_results['consensus'] strength = strategy_results['consensus_strength'] # Normaliser la force (0-1) normalized_strength = min(strength / 10, 1.0) if consensus in ['BUY', 'SELL']: signals.append(consensus) weights.append(normalized_strength * 0.4) # 40% de poids logging.debug(f" 📊 {symbol} Stratégies: {consensus} (force: {strength})") except Exception as e: logging.error(f"❌ Erreur stratégies avancées: {e}") # 3. ML Ensemble if self.ml_enabled and len(prices) >= 50: try: # Extraire features pour ML features = None if self.ml_predictor: features = self.ml_predictor.extract_features( prices=prices, volumes=volumes ) details['features'] = features # Prédiction ML ml_signal, ml_confidence, ml_details = self.ml_predictor.predict( features, indicators_signal=base_signal, indicators_confidence=base_confidence ) details['ml_result'] = ml_details # Ajouter seulement si confiance suffisante if ml_confidence >= ML_CONFIDENCE_THRESHOLD: signals.append(ml_signal) weights.append(ml_confidence * 0.3) # 30% de poids logging.debug(f" 🧠 {symbol} ML: {ml_signal} (conf: {ml_confidence:.2f})") except Exception as e: logging.error(f"❌ Erreur ML Ensemble: {e}") # 4. Décision finale par vote pondéré if not signals: return "HOLD", 0.5, details # Compter les votes buy_weight = sum(w for s, w in zip(signals, weights) if s == "BUY") sell_weight = sum(w for s, w in zip(signals, weights) if s == "SELL") total_weight = sum(weights) if total_weight == 0: return "HOLD", 0.5, details # Normaliser buy_ratio = buy_weight / total_weight sell_ratio = sell_weight / total_weight # Seuil de décision (60% minimum) threshold = STRATEGY_CONSENSUS_THRESHOLD / 100 if buy_ratio > threshold and buy_ratio > sell_ratio: final_signal = "BUY" final_confidence = buy_ratio elif sell_ratio > threshold and sell_ratio > buy_ratio: final_signal = "SELL" final_confidence = sell_ratio else: final_signal = "HOLD" final_confidence = 0.5 details['final_weights'] = { 'buy': buy_weight, 'sell': sell_weight, 'total': total_weight } return final_signal, final_confidence, details def update_ml_training(self, prices: List[float], volumes: Optional[List[float]] = None, actual_outcome: Optional[str] = None): """ Met à jour l'entraînement ML avec de nouvelles données actual_outcome: "up" si le prix a monté, "down" sinon """ if not self.ml_enabled or not self.ml_predictor: return try: features = self.ml_predictor.extract_features(prices, volumes) if features is not None and actual_outcome: label = 1 if actual_outcome == "up" else 0 self.ml_predictor.add_training_sample(features, label) except Exception as e: logging.error(f"❌ Erreur update ML: {e}") def train_ml_models(self, min_samples: int = 200): """ Entraîne les modèles ML si suffisamment de données """ if not self.ml_enabled or not self.ml_predictor: return False try: return self.ml_predictor.train_models(min_samples) except Exception as e: logging.error(f"❌ Erreur entraînement ML: {e}") return False def save_ml_models(self, filepath: str): """ Sauvegarde les modèles ML """ if not self.ml_enabled or not self.ml_predictor: return False try: return self.ml_predictor.save_models(filepath) except Exception as e: logging.error(f"❌ Erreur sauvegarde ML: {e}") return False def load_ml_models(self, filepath: str): """ Charge les modèles ML """ if not self.ml_enabled or not self.ml_predictor: return False try: return self.ml_predictor.load_models(filepath) except Exception as e: logging.error(f"❌ Erreur chargement ML: {e}") return False # Instance globale _enhanced_signal_engine = None def get_enhanced_signal_engine() -> EnhancedSignalEngine: """Récupère ou crée l'instance du signal engine""" global _enhanced_signal_engine if _enhanced_signal_engine is None: _enhanced_signal_engine = EnhancedSignalEngine() return _enhanced_signal_engine