""" ═══════════════════════════════════════════════════════════════════════════════ 🧠 AI COMPATIBILITY SCORER - Pré-analyse des valeurs pour l'IA ═══════════════════════════════════════════════════════════════════════════════ Ce module analyse chaque crypto et calcule un score de compatibilité avec les stratégies IA du bot. Les valeurs sont ensuite triées par ordre d'efficacité. Critères d'analyse: 1. VOLATILITÉ - La crypto bouge-t-elle suffisamment pour générer des profits ? 2. PATTERNS EMA - Les croisements EMA sont-ils clairs et exploitables ? 3. RÉACTIVITÉ MOMENTUM - La crypto réagit-elle bien aux changements de momentum ? 4. BOLLINGER QUALITY - Les bandes BB sont-elles exploitables ? 5. VOLUME/LIQUIDITÉ - Suffisant pour exécuter les ordres ? 6. TENDANCE CLAIRE - Éviter les sideways permanents Auteur: Trading IA Bot Date: Décembre 2024 ═══════════════════════════════════════════════════════════════════════════════ """ import json import logging import os from datetime import datetime, timedelta from typing import Dict, List, Tuple, Optional import numpy as np logger = logging.getLogger(__name__) class AICompatibilityScorer: """ Analyse et classe les cryptos par compatibilité avec les stratégies IA. """ def __init__(self): self.compatibility_cache = {} self.cache_duration = timedelta(hours=1) # Recalculer toutes les heures self.last_update = None self.ranked_symbols = [] # Poids des critères (total = 100) self.weights = { 'volatility': 20, # Volatilité optimale 'ema_quality': 25, # Qualité des patterns EMA 'momentum_reactivity': 20, # Réactivité au momentum 'bb_quality': 15, # Qualité des Bollinger Bands 'trend_clarity': 20 # Clarté de la tendance } # Seuils optimaux pour les stratégies CREUX_EMA et PULLBACK self.optimal_ranges = { 'bb_bandwidth_min': 0.5, # Minimum de volatilité 'bb_bandwidth_max': 4.0, # Pas trop volatile (chaos) 'bb_bandwidth_ideal': 1.5, # Idéal 'ema_diff_range': 0.3, # EMA9-EMA21 doit varier 'momentum_range_min': 0.2, # Momentum doit bouger 'rsi_range_min': 20, # RSI doit varier 'rsi_range_max': 80, } def calculate_compatibility_score(self, symbol: str, features: Dict, historical_features: List[Dict] = None) -> Dict: """ Calcule le score de compatibilité IA pour une crypto. Args: symbol: Symbole de la crypto (ex: BTCUSDT) features: Features actuelles de la crypto historical_features: Historique des features (optionnel) Returns: Dict avec score total et détail par critère """ scores = {} reasons = [] # ══════════════════════════════════════════════════════════════════════ # 1. VOLATILITÉ (20 pts) - La crypto bouge-t-elle bien ? # ══════════════════════════════════════════════════════════════════════ bb_bandwidth = features.get('bb_bandwidth', 1.0) if bb_bandwidth < self.optimal_ranges['bb_bandwidth_min']: # Trop peu volatile - mauvais pour le trading scores['volatility'] = 5 reasons.append(f"⚠️ Faible volatilité ({bb_bandwidth:.2f}%)") elif bb_bandwidth > self.optimal_ranges['bb_bandwidth_max']: # Trop volatile - risqué et imprévisible scores['volatility'] = 10 reasons.append(f"⚠️ Très volatile ({bb_bandwidth:.2f}%)") else: # Zone idéale ideal = self.optimal_ranges['bb_bandwidth_ideal'] distance = abs(bb_bandwidth - ideal) / ideal scores['volatility'] = int(20 * (1 - distance * 0.5)) if scores['volatility'] >= 15: reasons.append(f"✅ Volatilité optimale ({bb_bandwidth:.2f}%)") # ══════════════════════════════════════════════════════════════════════ # 2. QUALITÉ DES PATTERNS EMA (25 pts) # ══════════════════════════════════════════════════════════════════════ ema_diff = features.get('ema_diff', 0) ema_slope = features.get('ema_slope', 0) ema_slope_long = features.get('ema_slope_long', 0) ema21_slope = features.get('ema21_slope', 0) # L'EMA doit montrer des mouvements clairs ema_movement = abs(ema_diff) + abs(ema_slope) * 10 if ema_movement < 0.05: scores['ema_quality'] = 5 reasons.append("❌ EMA trop plate (sideways)") elif ema_movement > 2.0: scores['ema_quality'] = 15 reasons.append(f"⚠️ EMA très volatile") else: # Vérifier si les croisements sont exploitables ema_score = min(25, int(ema_movement * 50)) # Bonus si EMA9 et EMA21 ont des pentes cohérentes if (ema_slope > 0 and ema21_slope > 0) or (ema_slope < 0 and ema21_slope < 0): ema_score = min(25, ema_score + 5) reasons.append("✅ Tendance EMA cohérente") scores['ema_quality'] = ema_score # ══════════════════════════════════════════════════════════════════════ # 3. RÉACTIVITÉ MOMENTUM (20 pts) # ══════════════════════════════════════════════════════════════════════ momentum_3 = features.get('momentum_3', 0) momentum_5 = features.get('momentum_5', 0) momentum_activity = abs(momentum_3) + abs(momentum_5) if momentum_activity < 0.1: scores['momentum_reactivity'] = 5 reasons.append("❌ Momentum plat") elif momentum_activity > 3.0: scores['momentum_reactivity'] = 12 reasons.append(f"⚠️ Momentum excessif ({momentum_activity:.1f}%)") else: scores['momentum_reactivity'] = min(20, int(momentum_activity * 15)) if scores['momentum_reactivity'] >= 15: reasons.append(f"✅ Bonne réactivité momentum") # ══════════════════════════════════════════════════════════════════════ # 4. QUALITÉ BOLLINGER BANDS (15 pts) # ══════════════════════════════════════════════════════════════════════ bb_position = features.get('bb_position', 0.5) # Vérifier que le prix utilise bien les bandes BB # Idéal: bb_position varie entre 0.2 et 0.8 (pas collé aux extrêmes) if bb_position < 0.1 or bb_position > 0.9: scores['bb_quality'] = 8 reasons.append(f"⚠️ Prix aux extrêmes BB ({bb_position:.2f})") elif 0.3 <= bb_position <= 0.7: scores['bb_quality'] = 15 reasons.append("✅ Position BB idéale") else: scores['bb_quality'] = 12 # ══════════════════════════════════════════════════════════════════════ # 5. CLARTÉ DE LA TENDANCE (20 pts) # ══════════════════════════════════════════════════════════════════════ rsi = features.get('rsi', 50) ema_trend_bullish = features.get('ema_trend_bullish', 0) ema_trend_bearish = features.get('ema_trend_bearish', 0) # RSI doit être dans une zone exploitable if 30 <= rsi <= 70: rsi_score = 10 elif 20 <= rsi <= 80: rsi_score = 7 else: rsi_score = 5 # Extrêmes - moins prévisible # Tendance claire (bullish OU bearish, pas les deux) if ema_trend_bullish == 1 and ema_trend_bearish == 0: trend_score = 10 reasons.append("📈 Tendance haussière claire") elif ema_trend_bearish == 1 and ema_trend_bullish == 0: trend_score = 10 reasons.append("📉 Tendance baissière claire") else: trend_score = 5 reasons.append("⚠️ Tendance confuse") scores['trend_clarity'] = rsi_score + trend_score # ══════════════════════════════════════════════════════════════════════ # SCORE TOTAL # ══════════════════════════════════════════════════════════════════════ total_score = sum(scores.values()) max_score = sum(self.weights.values()) # Classification if total_score >= 80: grade = 'A' compatibility = 'EXCELLENT' elif total_score >= 65: grade = 'B' compatibility = 'BON' elif total_score >= 50: grade = 'C' compatibility = 'MOYEN' elif total_score >= 35: grade = 'D' compatibility = 'FAIBLE' else: grade = 'F' compatibility = 'INCOMPATIBLE' return { 'symbol': symbol, 'total_score': total_score, 'max_score': max_score, 'percentage': round(total_score / max_score * 100, 1), 'grade': grade, 'compatibility': compatibility, 'scores': scores, 'reasons': reasons, 'timestamp': datetime.now().isoformat() } def analyze_all_symbols(self, watchlist: List[Dict]) -> List[Dict]: """ Analyse toutes les cryptos et retourne la liste triée par compatibilité. Args: watchlist: Liste des items avec features Returns: Liste triée par score de compatibilité (meilleurs en premier) """ results = [] for item in watchlist: symbol = item.get('symbol', '') features = item.get('features', {}) if not features: continue try: score_result = self.calculate_compatibility_score(symbol, features) # Ajouter les infos de l'item original score_result['original_item'] = item score_result['smart_signal'] = item.get('smart_signal', 'HOLD') score_result['smart_score'] = item.get('smart_score', 0) results.append(score_result) except Exception as e: logger.warning(f"Erreur analyse compatibilité {symbol}: {e}") continue # Trier par score décroissant results.sort(key=lambda x: x['total_score'], reverse=True) # Mettre à jour le cache self.ranked_symbols = results self.last_update = datetime.now() logger.info(f"✅ Analyse compatibilité IA: {len(results)} symboles classés") return results def get_priority_symbols(self, watchlist: List[Dict], min_grade: str = 'C', limit: int = 30) -> List[Dict]: """ Retourne les symboles prioritaires pour le traitement IA. Args: watchlist: Liste complète min_grade: Grade minimum (A, B, C, D, F) limit: Nombre max de symboles Returns: Liste des symboles prioritaires avec leur score """ # Analyser si nécessaire if not self.ranked_symbols or self._cache_expired(): self.analyze_all_symbols(watchlist) # Filtrer par grade grade_order = {'A': 5, 'B': 4, 'C': 3, 'D': 2, 'F': 1} min_grade_value = grade_order.get(min_grade, 3) priority = [ r for r in self.ranked_symbols if grade_order.get(r['grade'], 0) >= min_grade_value ] return priority[:limit] def _cache_expired(self) -> bool: """Vérifie si le cache a expiré.""" if not self.last_update: return True return datetime.now() - self.last_update > self.cache_duration def get_compatibility_summary(self) -> Dict: """Retourne un résumé de la compatibilité globale.""" if not self.ranked_symbols: return {'error': 'Aucune analyse disponible'} grades = {'A': 0, 'B': 0, 'C': 0, 'D': 0, 'F': 0} for r in self.ranked_symbols: grades[r['grade']] = grades.get(r['grade'], 0) + 1 avg_score = sum(r['total_score'] for r in self.ranked_symbols) / len(self.ranked_symbols) return { 'total_analyzed': len(self.ranked_symbols), 'average_score': round(avg_score, 1), 'grades': grades, 'top_5': [ {'symbol': r['symbol'], 'score': r['total_score'], 'grade': r['grade']} for r in self.ranked_symbols[:5] ], 'bottom_5': [ {'symbol': r['symbol'], 'score': r['total_score'], 'grade': r['grade']} for r in self.ranked_symbols[-5:] ], 'last_update': self.last_update.isoformat() if self.last_update else None } # Instance globale _compatibility_scorer = None def get_compatibility_scorer() -> AICompatibilityScorer: """Retourne l'instance globale du scorer.""" global _compatibility_scorer if _compatibility_scorer is None: _compatibility_scorer = AICompatibilityScorer() return _compatibility_scorer def prioritize_watchlist(watchlist: List[Dict], min_grade: str = 'C') -> List[Dict]: """ Fonction utilitaire pour trier une watchlist par compatibilité IA. Args: watchlist: Liste des items avec features min_grade: Grade minimum requis Returns: Watchlist triée avec scores de compatibilité ajoutés """ scorer = get_compatibility_scorer() ranked = scorer.analyze_all_symbols(watchlist) # Enrichir les items originaux avec le score de compatibilité result = [] for r in ranked: item = r['original_item'].copy() item['ai_compatibility'] = { 'score': r['total_score'], 'percentage': r['percentage'], 'grade': r['grade'], 'compatibility': r['compatibility'], 'reasons': r['reasons'] } result.append(item) return result if __name__ == '__main__': # Test logging.basicConfig(level=logging.INFO) test_features = { 'bb_bandwidth': 1.5, 'ema_diff': -0.15, 'ema_slope': -0.1, 'ema_slope_long': -0.3, 'ema21_slope': -0.2, 'momentum_3': 0.3, 'momentum_5': 0.5, 'bb_position': 0.4, 'rsi': 35, 'ema_trend_bullish': 0, 'ema_trend_bearish': 1 } scorer = AICompatibilityScorer() result = scorer.calculate_compatibility_score('BTCUSDT', test_features) print(f"\n{'='*60}") print(f"🧠 Analyse Compatibilité IA: {result['symbol']}") print(f"{'='*60}") print(f"Score: {result['total_score']}/{result['max_score']} ({result['percentage']}%)") print(f"Grade: {result['grade']} - {result['compatibility']}") print(f"\nDétail:") for criteria, score in result['scores'].items(): print(f" {criteria}: {score} pts") print(f"\nRaisons:") for reason in result['reasons']: print(f" {reason}")