""" Pattern Manager - Gestion centralisée des patterns de trading ================================================================= Ce module centralise toute la logique des patterns pour : - Activer/désactiver des patterns individuellement - Gérer les statistiques par pattern - Optimiser les seuils de détection - Filtrer les patterns non performants - Faciliter la maintenance et l'optimisation Author: Trading Bot Team Date: 2026-01-23 """ import json import os import sys from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple from collections import defaultdict # Forcer UTF-8 pour Windows if sys.platform == 'win32': import codecs if sys.stdout.encoding != 'utf-8': sys.stdout = codecs.getwriter('utf-8')(sys.stdout.buffer, 'strict') if sys.stderr.encoding != 'utf-8': sys.stderr = codecs.getwriter('utf-8')(sys.stderr.buffer, 'strict') class PatternConfig: """Configuration d'un pattern de trading""" def __init__(self, name: str, enabled: bool = True, min_score: int = 35, win_rate_threshold: float = 30.0, description: str = ""): self.name = name self.enabled = enabled self.min_score = min_score # Score IA minimum pour ce pattern self.win_rate_threshold = win_rate_threshold # Win rate minimum acceptable self.description = description self.total_signals = 0 self.total_trades = 0 self.wins = 0 self.losses = 0 self.total_pnl = 0.0 self.last_used = None @property def win_rate(self) -> float: """Calcule le win rate du pattern""" if self.total_trades == 0: return 0.0 return (self.wins / self.total_trades) * 100 @property def avg_pnl(self) -> float: """Calcule le P&L moyen""" if self.total_trades == 0: return 0.0 return self.total_pnl / self.total_trades @property def is_profitable(self) -> bool: """Vérifie si le pattern est profitable""" return self.win_rate >= self.win_rate_threshold and self.total_trades >= 5 @property def needs_review(self) -> bool: """Indique si le pattern nécessite une révision""" return self.total_trades >= 10 and self.win_rate < self.win_rate_threshold def to_dict(self) -> dict: """Convertit en dictionnaire pour sauvegarde""" return { 'name': self.name, 'enabled': self.enabled, 'min_score': self.min_score, 'win_rate_threshold': self.win_rate_threshold, 'description': self.description, 'stats': { 'total_signals': self.total_signals, 'total_trades': self.total_trades, 'wins': self.wins, 'losses': self.losses, 'total_pnl': self.total_pnl, 'win_rate': round(self.win_rate, 1), 'avg_pnl': round(self.avg_pnl, 3), 'last_used': self.last_used.isoformat() if self.last_used else None } } @classmethod def from_dict(cls, data: dict): """Crée depuis un dictionnaire""" config = cls( name=data['name'], enabled=data.get('enabled', True), min_score=data.get('min_score', 35), win_rate_threshold=data.get('win_rate_threshold', 30.0), description=data.get('description', '') ) stats = data.get('stats', {}) config.total_signals = stats.get('total_signals', 0) config.total_trades = stats.get('total_trades', 0) config.wins = stats.get('wins', 0) config.losses = stats.get('losses', 0) config.total_pnl = stats.get('total_pnl', 0.0) if stats.get('last_used'): config.last_used = datetime.fromisoformat(stats['last_used']) return config class PatternManager: """ Gestionnaire centralisé des patterns de trading Responsabilités: - Enregistrer et gérer tous les patterns disponibles - Filtrer les patterns actifs/inactifs - Valider si un pattern doit être utilisé pour un achat - Mettre à jour les statistiques par pattern - Désactiver automatiquement les patterns non performants """ # Patterns disponibles dans le système AVAILABLE_PATTERNS = { # Patterns DANGEREUX - À surveiller de près 'UNKNOWN': PatternConfig('UNKNOWN', enabled=True, min_score=40, win_rate_threshold=35.0, description="Pattern non identifié - Requiert surveillance"), 'DEAD_CAT_BOUNCE': PatternConfig('DEAD_CAT_BOUNCE', enabled=False, min_score=50, win_rate_threshold=40.0, description="Rebond temporaire dans tendance baissière - DANGEREUX"), 'PROLONGED_DOWNTREND': PatternConfig('PROLONGED_DOWNTREND', enabled=False, min_score=60, win_rate_threshold=45.0, description="Tendance baissière prolongée - TRÈS RISQUÉ"), 'END_OF_CYCLE': PatternConfig('END_OF_CYCLE', enabled=False, min_score=100, win_rate_threshold=50.0, description="Fin de cycle haussier - BLOCAGE ABSOLU"), 'RSI_TRAP': PatternConfig('RSI_TRAP', enabled=False, min_score=55, win_rate_threshold=40.0, description="Piège RSI en tendance baissière - RISQUÉ"), 'STRONG_DOWNTREND': PatternConfig('STRONG_DOWNTREND', enabled=False, min_score=65, win_rate_threshold=45.0, description="Tendance baissière forte - ÉVITER"), 'ACTIVE_CRASH': PatternConfig('ACTIVE_CRASH', enabled=False, min_score=70, win_rate_threshold=50.0, description="Crash actif en cours - BLOCAGE"), # Patterns NEUTRES - Performance variable 'STABLECOIN_REJECTED': PatternConfig('STABLECOIN_REJECTED', enabled=False, min_score=45, win_rate_threshold=35.0, description="Stablecoin rejeté - Peu d'intérêt"), 'HOLD': PatternConfig('HOLD', enabled=True, min_score=40, win_rate_threshold=30.0, description="Signal HOLD - Neutre"), 'POSSIBLE': PatternConfig('POSSIBLE', enabled=True, min_score=35, win_rate_threshold=30.0, description="Achat possible - Conditions moyennes"), 'ACHAT': PatternConfig('ACHAT', enabled=True, min_score=35, win_rate_threshold=30.0, description="Signal achat générique"), # Patterns PROMETTEURS - À privilégier 'CREUX_REBOUND': PatternConfig('CREUX_REBOUND', enabled=True, min_score=50, win_rate_threshold=40.0, description="Rebond depuis un creux - BON POTENTIEL"), 'EARLY_BREAKOUT': PatternConfig('EARLY_BREAKOUT', enabled=True, min_score=45, win_rate_threshold=45.0, description="Cassure précoce - EXCELLENT"), 'CONSOLIDATION_BREAKOUT': PatternConfig('CONSOLIDATION_BREAKOUT', enabled=True, min_score=45, win_rate_threshold=45.0, description="Sortie de consolidation - BON"), 'EMA_BULLISH': PatternConfig('EMA_BULLISH', enabled=True, min_score=40, win_rate_threshold=40.0, description="Configuration EMA haussière - BON"), 'CROSSOVER_IMMINENT': PatternConfig('CROSSOVER_IMMINENT', enabled=True, min_score=40, win_rate_threshold=40.0, description="Croisement EMA imminent - BON"), 'SQUEEZE_BREAKOUT': PatternConfig('SQUEEZE_BREAKOUT', enabled=False, min_score=60, win_rate_threshold=45.0, description="Sortie de squeeze - DÉSACTIVÉ (18% WR rétro 582 trades, -24.55€)"), 'SQUEEZE_WAITING': PatternConfig('SQUEEZE_WAITING', enabled=True, min_score=35, win_rate_threshold=35.0, description="En attente de squeeze - MOYEN"), 'VOLUME_REVERSAL': PatternConfig('VOLUME_REVERSAL', enabled=True, min_score=45, win_rate_threshold=40.0, description="Retournement volume - BON"), 'RSI_REVERSAL': PatternConfig('RSI_REVERSAL', enabled=True, min_score=45, win_rate_threshold=40.0, description="Retournement RSI - BON"), 'STRONG_UPTREND': PatternConfig('STRONG_UPTREND', enabled=True, min_score=40, win_rate_threshold=40.0, description="Tendance haussière forte - EXCELLENT"), # 🔴 FIX 08/02: Patterns ajoutés/corrigés 'TREND_CONTINUATION': PatternConfig('TREND_CONTINUATION', enabled=False, min_score=65, win_rate_threshold=45.0, description="Continuation tendance - DÉSACTIVÉ (0% WR récent, 31% all-time)"), 'PULLBACK': PatternConfig('PULLBACK', enabled=False, min_score=55, win_rate_threshold=40.0, description="Pullback dans tendance haussière - DÉSACTIVÉ (22% WR rétro 582 trades, -26.12€)"), # Patterns SPÉCIAUX 'HIGH_SCORE_OVERRIDE': PatternConfig('HIGH_SCORE_OVERRIDE', enabled=True, min_score=60, win_rate_threshold=45.0, description="Score IA très élevé - Override des règles"), 'OTHER': PatternConfig('OTHER', enabled=True, min_score=40, win_rate_threshold=30.0, description="Autre pattern"), } def __init__(self, config_file: str = 'pattern_config.json'): """ Initialise le gestionnaire de patterns Args: config_file: Fichier de configuration des patterns """ self.config_file = config_file self.patterns: Dict[str, PatternConfig] = {} self.blacklisted_patterns: set = set() self.load_config() def load_config(self): """Charge la configuration depuis le fichier ou initialise avec les défauts""" if os.path.exists(self.config_file): try: with open(self.config_file, 'r', encoding='utf-8') as f: data = json.load(f) for pattern_name, pattern_data in data.get('patterns', {}).items(): self.patterns[pattern_name] = PatternConfig.from_dict(pattern_data) self.blacklisted_patterns = set(data.get('blacklisted', [])) print(f"[OK] Pattern config chargée: {len(self.patterns)} patterns") except Exception as e: print(f"[WARN] Erreur chargement config patterns: {e}") self._init_default_patterns() else: self._init_default_patterns() self.save_config() def _init_default_patterns(self): """Initialise avec les patterns par défaut""" self.patterns = {name: config for name, config in self.AVAILABLE_PATTERNS.items()} # Blacklister les patterns dangereux par défaut self.blacklisted_patterns = { 'END_OF_CYCLE', 'DEAD_CAT_BOUNCE', 'PROLONGED_DOWNTREND', 'ACTIVE_CRASH', 'STRONG_DOWNTREND' } print(f"[OK] Patterns initialisés: {len(self.patterns)} patterns, {len(self.blacklisted_patterns)} blacklistés") def save_config(self): """Sauvegarde la configuration""" try: data = { 'patterns': {name: config.to_dict() for name, config in self.patterns.items()}, 'blacklisted': list(self.blacklisted_patterns), 'last_updated': datetime.now().isoformat() } with open(self.config_file, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) except Exception as e: print(f"[WARN] Erreur sauvegarde config patterns: {e}") def is_pattern_allowed(self, pattern_name: str, ai_score: int = 0) -> Tuple[bool, str]: """ Vérifie si un pattern est autorisé pour un achat Args: pattern_name: Nom du pattern ai_score: Score IA du signal Returns: (allowed, reason): True si autorisé, False sinon + raison """ # 🔄 Rechargement automatique de la blacklist depuis le fichier # Permet de modifier la config sans redémarrer le bot try: if os.path.exists(self.config_file): with open(self.config_file, 'r', encoding='utf-8') as f: data = json.load(f) self.blacklisted_patterns = set(data.get('blacklisted', [])) except: pass # En cas d'erreur, garder la config en mémoire # Pattern inconnu if pattern_name not in self.patterns: return False, f"Pattern '{pattern_name}' inconnu" config = self.patterns[pattern_name] # Pattern blacklisté if pattern_name in self.blacklisted_patterns: return False, f"Pattern '{pattern_name}' blacklisté (performance insuffisante)" # Pattern désactivé if not config.enabled: return False, f"Pattern '{pattern_name}' désactivé" # Score IA insuffisant if ai_score < config.min_score: return False, f"Score IA {ai_score} < minimum requis ({config.min_score}) pour '{pattern_name}'" # Pattern nécessite révision # 🔴 FIX 02/03: Les patterns PROTÉGÉS (CREUX_REBOUND) ne sont PAS bloqués par needs_review # CREUX_REBOUND est le pattern principal du LSTM BiLSTM+Attention pour détecter les retournements PROTECTED_PATTERNS = {'CREUX_REBOUND'} if config.needs_review and pattern_name not in PROTECTED_PATTERNS: return False, f"Pattern '{pattern_name}' nécessite révision (WR={config.win_rate:.1f}% < {config.win_rate_threshold}%)" elif config.needs_review and pattern_name in PROTECTED_PATTERNS: # Pattern protégé: autorisé malgré WR bas (le LSTM en dépend) pass return True, f"Pattern '{pattern_name}' validé (WR={config.win_rate:.1f}%, AI≥{config.min_score})" def record_signal(self, pattern_name: str): """Enregistre qu'un signal a été généré pour ce pattern""" if pattern_name in self.patterns: self.patterns[pattern_name].total_signals += 1 self.patterns[pattern_name].last_used = datetime.now() def record_trade(self, pattern_name: str, pnl_pct: float): """ Enregistre le résultat d'un trade pour ce pattern Args: pattern_name: Nom du pattern pnl_pct: P&L en pourcentage """ if pattern_name not in self.patterns: print(f"[WARN] Pattern '{pattern_name}' inconnu - Création automatique") self.patterns[pattern_name] = PatternConfig(pattern_name, enabled=True) config = self.patterns[pattern_name] config.total_trades += 1 config.total_pnl += pnl_pct if pnl_pct > 0: config.wins += 1 else: config.losses += 1 # Auto-blacklist si performance catastrophique # 🔧 FIX 02/03: CREUX_REBOUND protégé - c'est le pattern PRINCIPAL de retournement # utilisé par le LSTM BiLSTM+Attention. Le blacklister tue le système de reversal. PROTECTED_PATTERNS = {'CREUX_REBOUND'} # Patterns critiques qui ne peuvent pas être auto-blacklistés if config.total_trades >= 20 and config.win_rate < 25.0 and pattern_name not in PROTECTED_PATTERNS: self.blacklisted_patterns.add(pattern_name) print(f"[BLACKLIST] Pattern '{pattern_name}' AUTO-BLACKLISTÉ (WR={config.win_rate:.1f}% sur {config.total_trades} trades)") elif config.total_trades >= 20 and config.win_rate < 25.0 and pattern_name in PROTECTED_PATTERNS: print(f"[WARN] Pattern '{pattern_name}' WR={config.win_rate:.1f}% mais PROTÉGÉ (pattern critique)") # Sauvegarder après chaque trade self.save_config() def get_enabled_patterns(self) -> List[str]: """Retourne la liste des patterns actifs""" return [name for name, config in self.patterns.items() if config.enabled and name not in self.blacklisted_patterns] def get_pattern_stats(self, pattern_name: str) -> Optional[dict]: """Retourne les statistiques d'un pattern""" if pattern_name in self.patterns: return self.patterns[pattern_name].to_dict() return None def get_all_stats(self) -> dict: """Retourne les statistiques de tous les patterns""" stats = { 'patterns': {}, 'summary': { 'total_patterns': len(self.patterns), 'enabled': len(self.get_enabled_patterns()), 'blacklisted': len(self.blacklisted_patterns), 'needs_review': len([p for p in self.patterns.values() if p.needs_review]) } } for name, config in self.patterns.items(): stats['patterns'][name] = config.to_dict() return stats def enable_pattern(self, pattern_name: str) -> bool: """Active un pattern""" if pattern_name in self.patterns: self.patterns[pattern_name].enabled = True if pattern_name in self.blacklisted_patterns: self.blacklisted_patterns.remove(pattern_name) self.save_config() print(f"[OK] Pattern '{pattern_name}' activé") return True return False def disable_pattern(self, pattern_name: str) -> bool: """Désactive un pattern""" if pattern_name in self.patterns: self.patterns[pattern_name].enabled = False self.save_config() print(f"[OFF] Pattern '{pattern_name}' désactivé") return True return False def blacklist_pattern(self, pattern_name: str) -> bool: """Blackliste un pattern""" if pattern_name in self.patterns: self.blacklisted_patterns.add(pattern_name) self.save_config() print(f"[BLACKLIST] Pattern '{pattern_name}' blacklisté") return True return False def update_min_score(self, pattern_name: str, new_score: int) -> bool: """Met à jour le score minimum d'un pattern""" if pattern_name in self.patterns: self.patterns[pattern_name].min_score = new_score self.save_config() print(f"[OK] Pattern '{pattern_name}' - Score min mis à jour: {new_score}") return True return False def optimize_patterns(self) -> dict: """ Optimise automatiquement les patterns selon leurs performances Returns: dict: Résumé des optimisations effectuées """ optimizations = { 'disabled': [], 'blacklisted': [], 'score_increased': [], 'score_decreased': [] } for name, config in self.patterns.items(): if config.total_trades < 5: continue # Pas assez de données # Désactiver si mauvaise performance if config.win_rate < 25.0 and config.total_trades >= 10: if config.enabled: config.enabled = False optimizations['disabled'].append(name) # Blacklister si très mauvaise performance if config.win_rate < 20.0 and config.total_trades >= 20: if name not in self.blacklisted_patterns: self.blacklisted_patterns.add(name) optimizations['blacklisted'].append(name) # Augmenter score si excellente performance if config.win_rate > 60.0 and config.total_trades >= 15: if config.min_score < 40: config.min_score = min(config.min_score + 5, 40) optimizations['score_decreased'].append(name) # Diminuer score si très bonne performance if config.win_rate > 70.0 and config.total_trades >= 20: if config.min_score > 25: config.min_score = max(config.min_score - 5, 25) optimizations['score_decreased'].append(name) if any(optimizations.values()): self.save_config() print(f"[OPTIM] Optimisations automatiques: {sum(len(v) for v in optimizations.values())} patterns modifiés") return optimizations def get_performance_report(self) -> str: """Génère un rapport de performance des patterns""" report = [ "\n" + "="*70, " RAPPORT DE PERFORMANCE DES PATTERNS", "="*70, "" ] # Trier par win rate sorted_patterns = sorted( [(name, config) for name, config in self.patterns.items() if config.total_trades > 0], key=lambda x: x[1].win_rate, reverse=True ) for name, config in sorted_patterns: status = "[OK]" if config.is_profitable else "[WARN]" if config.needs_review else "[OFF]" enabled = "ON" if config.enabled and name not in self.blacklisted_patterns else "OFF" report.append( f"{status} {name:25s} [{enabled:3s}] WR={config.win_rate:5.1f}% " f"Trades={config.total_trades:3d} PNL avg={config.avg_pnl:+.3f}%" ) report.extend([ "", f"Total patterns: {len(self.patterns)}", f"Actifs: {len(self.get_enabled_patterns())}", f"Blacklistés: {len(self.blacklisted_patterns)}", f"À réviser: {len([p for p in self.patterns.values() if p.needs_review])}", "="*70 ]) return "\n".join(report) # Instance globale (singleton) _pattern_manager_instance = None def get_pattern_manager() -> PatternManager: """Retourne l'instance globale du gestionnaire de patterns""" global _pattern_manager_instance if _pattern_manager_instance is None: _pattern_manager_instance = PatternManager() return _pattern_manager_instance if __name__ == "__main__": # Test du module pm = get_pattern_manager() print(pm.get_performance_report())