#!/usr/bin/env python3 """ FreqAI Integration Module Module d'intégration des améliorations inspirées de FreqAI Fonctionnalités: 1. Auto-retraining adaptatif (toutes les 48h ou si performances faibles) 2. Outlier detection (pumps artificiels, flash crashes) 3. Backtesting adaptatif (simulation réaliste avec retraining) Intégration dans trading_bot.py: from freqai_integration import FreqAIManager freqai = FreqAIManager() # Dans trading_loop(): if freqai.should_check_outliers(symbol, prices, volumes): skip_trade = True # Toutes les heures: freqai.periodic_check() """ import logging from typing import Dict, List, Tuple, Optional from datetime import datetime, timedelta import json import os # Imports des nouveaux modules from outlier_detection import get_outlier_detector, OutlierAnalysis from ai_adaptive_retrainer import get_adaptive_retrainer from backtesting_adaptive import AdaptiveBacktester, compare_adaptive_vs_classic logging.basicConfig(level=logging.INFO) logger = logging.getLogger("FreqAIManager") class FreqAIManager: """ Gestionnaire central des fonctionnalités FreqAI Orchestrate: - Outlier detection - Auto-retraining - Backtesting adaptatif """ def __init__(self, enable_outlier_detection: bool = True, enable_auto_retraining: bool = True, retraining_check_interval_hours: int = 1): """ Args: enable_outlier_detection: Activer la détection d'outliers enable_auto_retraining: Activer le retraining automatique retraining_check_interval_hours: Intervalle entre vérifications retraining """ self.enable_outliers = enable_outlier_detection self.enable_retraining = enable_auto_retraining self.check_interval = retraining_check_interval_hours # Modules self.outlier_detector = get_outlier_detector() if enable_outlier_detection else None self.retrainer = get_adaptive_retrainer() if enable_auto_retraining else None # État self.last_retraining_check = None self.outliers_detected = 0 self.outliers_blocked = 0 logger.info("="*60) logger.info("🚀 FreqAI Manager initialisé") logger.info("="*60) logger.info(f" • Outlier Detection: {'✅ ON' if enable_outlier_detection else '❌ OFF'}") logger.info(f" • Auto-Retraining: {'✅ ON' if enable_auto_retraining else '❌ OFF'}") logger.info(f" • Check Interval: {retraining_check_interval_hours}h") def should_check_outliers(self, symbol: str, prices: List[float], volumes: List[float], rsi: Optional[float] = None, bb_position: Optional[float] = None, block_on_outlier: bool = True) -> Tuple[bool, Optional[OutlierAnalysis]]: """ Vérifier si les données sont des outliers Args: symbol: Symbole crypto prices: Historique des prix volumes: Historique des volumes rsi: RSI actuel bb_position: Position Bollinger block_on_outlier: Si True, retourne True pour bloquer le trade Returns: (should_block_trade, outlier_analysis) """ if not self.enable_outliers or self.outlier_detector is None: return False, None try: analysis = self.outlier_detector.detect_outlier( symbol=symbol, prices=prices, volumes=volumes, rsi=rsi, bb_position=bb_position ) self.outliers_detected += 1 if analysis.is_outlier: self.outliers_blocked += 1 logger.warning(f"🚫 OUTLIER détecté sur {symbol}:") logger.warning(f" • Méthode: {analysis.method}") logger.warning(f" • Raison: {analysis.reason}") logger.warning(f" • Confidence: {analysis.confidence:.2%}") logger.warning(f" • Score: {analysis.score:.2f}") if block_on_outlier: return True, analysis return False, analysis except Exception as e: logger.error(f"❌ Erreur outlier detection sur {symbol}: {e}") return False, None def check_pump_dump(self, symbol: str, prices: List[float], volumes: List[float]) -> bool: """ Détecter un pump&dump Returns: True si pump&dump détecté """ if not self.enable_outliers or self.outlier_detector is None: return False try: is_pump, reason = self.outlier_detector.detect_pump_dump( symbol=symbol, prices=prices, volumes=volumes ) if is_pump: logger.warning(f"🚨 PUMP&DUMP détecté sur {symbol}: {reason}") return is_pump except Exception as e: logger.error(f"❌ Erreur pump&dump detection: {e}") return False def periodic_check(self) -> Optional[Dict]: """ Vérification périodique (retraining) À appeler toutes les heures dans trading_loop() Returns: Résultats du retraining si exécuté, None sinon """ if not self.enable_retraining or self.retrainer is None: return None # Vérifier intervalle now = datetime.now() if self.last_retraining_check is not None: elapsed = now - self.last_retraining_check if elapsed < timedelta(hours=self.check_interval): return None self.last_retraining_check = now try: logger.info("🔍 Vérification retraining périodique...") results = self.retrainer.check_and_retrain() if results is not None: logger.info("="*60) status = results.get('status', 'UNKNOWN') if status == 'FAILED': logger.info(f"⚠️ RETRAINING ÉCHOUÉ: {results.get('reason', 'NO_DATA')}") else: logger.info("✅ RETRAINING EFFECTUÉ") logger.info("="*60) logger.info(f" • Statut: {status}") logger.info(f" • Durée: {results.get('duration', 0):.1f}s") sub_results = results.get('results', {}) models = [k for k, v in sub_results.items() if v] if models: logger.info(f" • Modèles: {', '.join(models)}") logger.info("="*60) return results except Exception as e: logger.error(f"❌ Erreur vérification retraining: {e}") return None def run_backtest(self, start_date: datetime, end_date: datetime, symbols: List[str], initial_capital: float = 10000, retrain_interval_days: int = 7) -> Dict: """ Lancer un backtest adaptatif Args: start_date: Date de début end_date: Date de fin symbols: Liste des symboles à tester initial_capital: Capital initial retrain_interval_days: Intervalle de retraining (jours) Returns: Résultats du backtest """ try: logger.info("="*60) logger.info("🧪 LANCEMENT BACKTEST ADAPTATIF") logger.info("="*60) logger.info(f" • Période: {start_date.date()} → {end_date.date()}") logger.info(f" • Symboles: {len(symbols)}") logger.info(f" • Capital: {initial_capital} USDT") logger.info(f" • Retraining: tous les {retrain_interval_days} jours") logger.info("="*60) backtester = AdaptiveBacktester( initial_capital=initial_capital, retrain_interval_days=retrain_interval_days ) metrics = backtester.run( start_date=start_date, end_date=end_date, symbols=symbols ) logger.info("\n" + "="*60) logger.info("📊 RÉSULTATS BACKTEST") logger.info("="*60) logger.info(f" • Total Trades: {metrics.total_trades}") logger.info(f" • Win Rate: {metrics.win_rate:.2%}") logger.info(f" • Net Profit: {metrics.net_profit:+.2f} USDT ({metrics.net_profit/initial_capital:+.2%})") logger.info(f" • Profit Factor: {metrics.profit_factor:.2f}") logger.info(f" • Max Drawdown: {metrics.max_drawdown:.2%}") logger.info(f" • Sharpe Ratio: {metrics.sharpe_ratio:.2f}") logger.info(f" • Sortino Ratio: {metrics.sortino_ratio:.2f}") logger.info("="*60) return { 'metrics': metrics, 'start_date': start_date.isoformat(), 'end_date': end_date.isoformat(), 'symbols': symbols, 'initial_capital': initial_capital } except Exception as e: logger.error(f"❌ Erreur backtest: {e}") raise def get_stats(self) -> Dict: """ Obtenir les statistiques du manager Returns: Statistiques d'utilisation """ stats = { 'outlier_detection': { 'enabled': self.enable_outliers, 'total_checks': self.outliers_detected, 'outliers_blocked': self.outliers_blocked, 'block_rate': self.outliers_blocked / self.outliers_detected if self.outliers_detected > 0 else 0 }, 'auto_retraining': { 'enabled': self.enable_retraining, 'last_check': self.last_retraining_check.isoformat() if self.last_retraining_check else None, 'check_interval_hours': self.check_interval } } # Stats du retrainer if self.enable_retraining and self.retrainer is not None: stats['auto_retraining'].update({ 'last_retrain': self.retrainer.last_retrain.isoformat() if self.retrainer.last_retrain else None, 'total_retrains': len(self.retrainer.retrain_history) }) return stats def print_stats(self): """Afficher les statistiques""" stats = self.get_stats() print("\n" + "="*60) print("📊 FreqAI Manager - Statistiques") print("="*60) # Outlier Detection od = stats['outlier_detection'] print(f"\n🔍 Outlier Detection:") print(f" • Statut: {'✅ Activé' if od['enabled'] else '❌ Désactivé'}") if od['enabled']: print(f" • Total checks: {od['total_checks']}") print(f" • Outliers bloqués: {od['outliers_blocked']}") print(f" • Taux de blocage: {od['block_rate']:.2%}") # Auto-Retraining ar = stats['auto_retraining'] print(f"\n🔄 Auto-Retraining:") print(f" • Statut: {'✅ Activé' if ar['enabled'] else '❌ Désactivé'}") if ar['enabled']: print(f" • Intervalle: {ar['check_interval_hours']}h") print(f" • Dernière vérif: {ar.get('last_check', 'Jamais')}") print(f" • Dernier retrain: {ar.get('last_retrain', 'Jamais')}") print(f" • Total retrains: {ar.get('total_retrains', 0)}") print("="*60) # Singleton global _freqai_manager_instance = None def get_freqai_manager() -> FreqAIManager: """Obtenir l'instance singleton du manager""" global _freqai_manager_instance if _freqai_manager_instance is None: _freqai_manager_instance = FreqAIManager() return _freqai_manager_instance if __name__ == "__main__": """Test du FreqAI Manager""" print("="*60) print("🧪 TEST FREQAI MANAGER") print("="*60) manager = get_freqai_manager() # Test 1: Outlier detection print("\n1️⃣ Test outlier detection:") normal_prices = [100, 101, 100.5, 102, 101.5, 103, 102.5, 104] normal_volumes = [1000, 1100, 950, 1050, 1000, 1200, 1100, 1000] should_block, analysis = manager.should_check_outliers( symbol="BTCUSDT", prices=normal_prices, volumes=normal_volumes, rsi=50, bb_position=0.5 ) print(f" • Bloquer trade: {should_block}") if analysis: print(f" • Outlier: {analysis.is_outlier}") print(f" • Raison: {analysis.reason}") # Test 2: Volume spike print("\n2️⃣ Test volume spike:") spike_volumes = [1000, 1100, 950, 1050, 1000, 1200, 1100, 10000] should_block, analysis = manager.should_check_outliers( symbol="ETHUSDT", prices=normal_prices, volumes=spike_volumes, rsi=50, bb_position=0.5 ) print(f" • Bloquer trade: {should_block}") if analysis: print(f" • Méthode: {analysis.method}") print(f" • Raison: {analysis.reason}") # Test 3: Statistiques print("\n3️⃣ Statistiques:") manager.print_stats() print("\n✅ Tests terminés")