""" Système de Protection d'Urgence - Blocage automatique des achats =================================================================== Détecte les marchés chaotiques/dangereux et suspend les achats automatiquement """ import json import os from datetime import datetime, timedelta from collections import deque import logging logger = logging.getLogger(__name__) class MarketSafety: """Gestionnaire de sécurité pour bloquer les achats en période dangereuse""" def __init__(self): self.script_dir = os.path.dirname(os.path.abspath(__file__)) self.safety_file = os.path.join(self.script_dir, 'market_safety_status.json') self.signals_log = os.path.join(self.script_dir, 'trade_logs', 'signals_log.jsonl') self.trades_log = os.path.join(self.script_dir, 'trade_logs', 'trades_log.jsonl') # Seuils de sécurité (STRICTS) self.MIN_WIN_RATE = 0.0 # Désactivé pour tests après corrections MOMENTUM self.MIN_TRADES_FOR_ANALYSIS = 1000 # Désactivé: ignorer historique pour tester nouvelles corrections self.MAX_CONSECUTIVE_LOSSES = 100 # Désactivé pour tests self.MAX_REJECTION_RATE = 0.99 # Quasi-désactivé pour tests self.MAX_SIGNALS_PER_HOUR = 5000 # 🔧 FIX 31/03b: 500→5000 — quasi-désactivé, signaux légitimes en BULL_STRONG # État self.is_blocked = False self.block_reason = None self.block_timestamp = None self.last_check = None # Charger l'état existant self.load_status() def load_status(self): """Charger l'état de blocage depuis le fichier""" try: if os.path.exists(self.safety_file): with open(self.safety_file, 'r', encoding='utf-8') as f: data = json.load(f) self.is_blocked = data.get('is_blocked', False) self.block_reason = data.get('block_reason') self.block_timestamp = data.get('block_timestamp') self.last_check = data.get('last_check') except Exception as e: logger.warning(f"Erreur chargement safety status: {e}") def save_status(self): """Sauvegarder l'état de blocage""" try: data = { 'is_blocked': self.is_blocked, 'block_reason': self.block_reason, 'block_timestamp': self.block_timestamp, 'last_check': datetime.now().isoformat(), 'last_update': datetime.now().isoformat() } with open(self.safety_file, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) except Exception as e: logger.error(f"Erreur sauvegarde safety status: {e}") def analyze_recent_trades(self, hours=6): """Analyser les trades récents pour détecter performance catastrophique""" try: if not os.path.exists(self.trades_log): return None # Lire tous les trades récents recent_trades = [] now = datetime.now() cutoff = now - timedelta(hours=hours) with open(self.trades_log, 'r', encoding='utf-8') as f: for line in f: try: trade = json.loads(line) if trade.get('type') != 'TRADE_CLOSE': continue ts_str = trade.get('timestamp', '') ts = datetime.fromisoformat(ts_str) if ts >= cutoff: recent_trades.append(trade) except: continue if len(recent_trades) < self.MIN_TRADES_FOR_ANALYSIS: return None # Pas assez de données # Calculer win rate wins = sum(1 for t in recent_trades if t.get('pnl_pct', 0) > 0) losses = sum(1 for t in recent_trades if t.get('pnl_pct', 0) < 0) total = wins + losses if total == 0: return None win_rate = wins / total # Détecter pertes consécutives consecutive_losses = 0 max_consecutive_losses = 0 for trade in reversed(recent_trades): # Plus récent en premier if trade.get('pnl_pct', 0) < 0: consecutive_losses += 1 max_consecutive_losses = max(max_consecutive_losses, consecutive_losses) else: consecutive_losses = 0 return { 'total_trades': total, 'wins': wins, 'losses': losses, 'win_rate': win_rate, 'consecutive_losses': max_consecutive_losses, 'period_hours': hours } except Exception as e: logger.error(f"Erreur analyse trades: {e}") return None def analyze_signal_saturation(self, hours=1): """Détecter saturation du marché (trop de signaux, trop de rejets)""" try: if not os.path.exists(self.signals_log): return None # Lire signaux récents recent_signals = [] now = datetime.now() cutoff = now - timedelta(hours=hours) with open(self.signals_log, 'r', encoding='utf-8') as f: for line in f: try: sig = json.loads(line) ts_str = sig.get('timestamp', '') ts = datetime.fromisoformat(ts_str) if ts >= cutoff: recent_signals.append(sig) except: continue if not recent_signals: return None # Analyser rejets rejected = [s for s in recent_signals if not s.get('executed')] max_pos_rejections = sum(1 for s in rejected if 'MAX_POSITIONS' in s.get('rejection_reason', '')) rejection_rate = len(rejected) / len(recent_signals) if recent_signals else 0 max_pos_rate = max_pos_rejections / len(rejected) if rejected else 0 signals_per_hour = len(recent_signals) / hours return { 'total_signals': len(recent_signals), 'rejected': len(rejected), 'rejection_rate': rejection_rate, 'max_pos_rejections': max_pos_rejections, 'max_pos_rate': max_pos_rate, 'signals_per_hour': signals_per_hour, 'period_hours': hours } except Exception as e: logger.error(f"Erreur analyse saturation: {e}") return None def check_safety(self, force=False): """ Vérifier si le marché est sûr pour trader Retourne: (is_safe: bool, reason: str) """ # Si déjà bloqué, retourner immédiatement sans recalculer (économie CPU) # Mais recalculer toutes les 5 minutes pour voir si déblocage possible if not force and self.is_blocked and self.last_check: try: last = datetime.fromisoformat(self.last_check) if (datetime.now() - last).total_seconds() < 300: return (False, self.block_reason) except: pass self.last_check = datetime.now().isoformat() # === ANALYSE 1: Performance des trades === trades_analysis = self.analyze_recent_trades(hours=6) if trades_analysis: win_rate = trades_analysis['win_rate'] consecutive_losses = trades_analysis['consecutive_losses'] total_trades = trades_analysis['total_trades'] # CONDITION 1: Win rate catastrophique if win_rate < self.MIN_WIN_RATE and total_trades >= self.MIN_TRADES_FOR_ANALYSIS: reason = f"🚨 WIN RATE CATASTROPHIQUE: {win_rate*100:.1f}% sur {total_trades} trades (min: {self.MIN_WIN_RATE*100}%)" self._block_trading(reason) return (False, reason) # CONDITION 2: Pertes consécutives excessives if consecutive_losses >= self.MAX_CONSECUTIVE_LOSSES: reason = f"🚨 SÉRIE DE PERTES: {consecutive_losses} pertes consécutives (max: {self.MAX_CONSECUTIVE_LOSSES})" self._block_trading(reason) return (False, reason) # === ANALYSE 2: Saturation du marché === saturation = self.analyze_signal_saturation(hours=1) if saturation: max_pos_rate = saturation['max_pos_rate'] signals_per_hour = saturation['signals_per_hour'] # 🔧 FIX 07/02: CONDITION 3 DÉSACTIVÉE — Saturation MAX_POSITIONS # PROBLÈME: Cette condition créait un DEADLOCK: # positions=5/10 → signaux rejetés MAX_POS → safety détecte 100% rejets # → bloque TOUS les achats → impossible d'atteindre 10/10 # Les rejets MAX_POSITIONS sont NORMAUX quand le bot a des slots libres # mais reçoit plus de signaux que de slots → PAS une urgence marché # if max_pos_rate > self.MAX_REJECTION_RATE: # reason = f"MARCHÉ SATURÉ: rejets MAX_POSITIONS" # self._block_trading(reason) # return (False, reason) # CONDITION 4: Trop de signaux = chaos (gardé mais seuil très haut) if signals_per_hour > self.MAX_SIGNALS_PER_HOUR: reason = f"🚨 CHAOS DÉTECTÉ: {signals_per_hour:.0f} signaux/h (max: {self.MAX_SIGNALS_PER_HOUR})" self._block_trading(reason) return (False, reason) # === Marché sûr - débloquer si bloqué === if self.is_blocked: self._unblock_trading() self.save_status() return (True, None) def _block_trading(self, reason): """Bloquer les achats""" if not self.is_blocked: self.is_blocked = True self.block_reason = reason self.block_timestamp = datetime.now().isoformat() logger.critical(f"\n{'='*80}") logger.critical(f"🛑 BLOCAGE AUTOMATIQUE DES ACHATS") logger.critical(f"{'='*80}") logger.critical(reason) logger.critical(f"Timestamp: {self.block_timestamp}") logger.critical(f"{'='*80}\n") print(f"\n{'='*80}") print(f"🛑 PROTECTION ACTIVÉE - ACHATS BLOQUÉS") print(f"{'='*80}") print(reason) print(f"{'='*80}\n") self.save_status() def _unblock_trading(self): """Débloquer les achats""" if self.is_blocked: duration = None if self.block_timestamp: try: blocked_at = datetime.fromisoformat(self.block_timestamp) duration = (datetime.now() - blocked_at).total_seconds() / 60 except: pass logger.info(f"\n{'='*80}") logger.info(f"✅ DÉBLOCAGE AUTOMATIQUE DES ACHATS") logger.info(f"{'='*80}") logger.info(f"Durée du blocage: {duration:.1f} minutes" if duration else "Durée inconnue") logger.info(f"Marché redevenu stable") logger.info(f"{'='*80}\n") print(f"\n✅ Protection désactivée - Achats autorisés\n") self.is_blocked = False self.block_reason = None self.block_timestamp = None self.save_status() def force_unblock(self): """Forcer le déblocage manuellement""" self._unblock_trading() def get_status(self): """Obtenir le statut actuel""" return { 'is_blocked': self.is_blocked, 'reason': self.block_reason, 'blocked_since': self.block_timestamp, 'last_check': self.last_check } # Instance globale _safety_instance = None def get_market_safety(): """Obtenir l'instance singleton de MarketSafety""" global _safety_instance if _safety_instance is None: _safety_instance = MarketSafety() return _safety_instance