""" Gestionnaire de cooldown pour éviter les fermetures/réouvertures rapides qui génèrent des pertes dues aux frais """ import logging import time from datetime import datetime, timedelta from typing import Dict, Optional, Set import threading class CooldownManager: """Gère les cooldowns pour éviter le overtrading sur les mêmes symboles""" def __init__(self, default_cooldown_minutes: int = 15): self.logger = logging.getLogger("CooldownManager") # Configuration des cooldowns par défaut self.default_cooldown_minutes = default_cooldown_minutes # Cooldowns spécifiques par symbole (en minutes) self.symbol_cooldowns = { 'BTC': 10, # Cooldown plus court pour BTC (très liquide) 'ETH': 10, # Pareil pour ETH 'BNB': 15, # Cooldown standard pour BNB 'SOL': 12, # Légèrement plus court pour SOL 'DOGE': 20, # Plus long pour DOGE (plus volatile) 'DEFAULT': 15 # Par défaut pour les autres } # Stockage des dernières actions self.last_close_times = {} # {symbol: datetime} self.recent_opens = {} # {symbol: [datetime, ...]} self.lock = threading.Lock() # Statistiques self.cooldown_blocks = 0 self.overtrading_prevention = 0 def record_position_close(self, symbol: str, reason: str = ""): """Enregistre la fermeture d'une position""" with self.lock: close_time = datetime.now() self.last_close_times[symbol] = close_time self.logger.info(f"Position {symbol} fermée - Cooldown activé jusqu'à {self._get_cooldown_end(symbol, close_time).strftime('%H:%M:%S')}") if reason: self.logger.debug(f"Raison fermeture {symbol}: {reason}") def record_position_open(self, symbol: str): """Enregistre l'ouverture d'une position""" with self.lock: open_time = datetime.now() if symbol not in self.recent_opens: self.recent_opens[symbol] = [] self.recent_opens[symbol].append(open_time) # Nettoyer les anciennes ouvertures (garder seulement la dernière heure) cutoff_time = open_time - timedelta(hours=1) self.recent_opens[symbol] = [ t for t in self.recent_opens[symbol] if t >= cutoff_time ] self.logger.info(f"Position {symbol} ouverte - {len(self.recent_opens[symbol])} ouvertures dans la dernière heure") def can_open_position(self, symbol: str) -> tuple: """ Vérifie si on peut ouvrir une position sur ce symbole Returns: Tuple: (allowed, reason) """ with self.lock: now = datetime.now() # 1. Vérifier le cooldown après fermeture if symbol in self.last_close_times: last_close = self.last_close_times[symbol] cooldown_end = self._get_cooldown_end(symbol, last_close) if now < cooldown_end: remaining = cooldown_end - now remaining_minutes = int(remaining.total_seconds() / 60) self.cooldown_blocks += 1 return (False, f"Cooldown actif pour {symbol} - {remaining_minutes}min restantes") # 2. Vérifier le overtrading (trop d'ouvertures récentes) if symbol in self.recent_opens: recent_count = len(self.recent_opens[symbol]) # Max 3 ouvertures par heure sur le même symbole if recent_count >= 3: self.overtrading_prevention += 1 return (False, f"Trop d'ouvertures récentes sur {symbol} ({recent_count}/h - max 3)") # Vérifier les ouvertures très récentes (dernières 10 minutes) very_recent = [ t for t in self.recent_opens[symbol] if (now - t).total_seconds() < 600 # 10 minutes ] # Max 1 ouverture toutes les 10 minutes if len(very_recent) >= 1: self.overtrading_prevention += 1 return (False, f"Ouverture trop récente sur {symbol} (dernière il y a {int((now - very_recent[-1]).total_seconds()/60)}min - attendre 10min)") # 3. Vérification spéciale pour les symboles volatiles volatility_cooldown = self._check_volatility_cooldown(symbol, now) if not volatility_cooldown[0]: return volatility_cooldown return (True, f"Ouverture autorisée pour {symbol}") def _get_cooldown_end(self, symbol: str, close_time: datetime) -> datetime: """Calcule la fin du cooldown pour un symbole""" cooldown_minutes = self.symbol_cooldowns.get(symbol, self.symbol_cooldowns['DEFAULT']) return close_time + timedelta(minutes=cooldown_minutes) def _check_volatility_cooldown(self, symbol: str, now: datetime) -> tuple: """Vérifications spéciales pour les symboles volatiles""" # Pour les symboles très volatiles, cooldown plus long le weekend if symbol in ['DOGE', 'SHIB', 'WIF'] and now.weekday() >= 5: # Weekend return (True, "Weekend - pas de restriction supplémentaire pour meme coins") return (True, "Pas de restriction de volatilité") def get_remaining_cooldown(self, symbol: str) -> Optional[int]: """Retourne le temps de cooldown restant en minutes""" with self.lock: if symbol not in self.last_close_times: return None now = datetime.now() last_close = self.last_close_times[symbol] cooldown_end = self._get_cooldown_end(symbol, last_close) if now >= cooldown_end: return None remaining = cooldown_end - now return max(0, int(remaining.total_seconds() / 60)) def get_cooldown_status(self) -> Dict: """Retourne le statut complet des cooldowns""" with self.lock: now = datetime.now() status = { 'active_cooldowns': {}, 'recent_activity': {}, 'statistics': { 'cooldown_blocks': self.cooldown_blocks, 'overtrading_prevention': self.overtrading_prevention } } # Cooldowns actifs for symbol, last_close in self.last_close_times.items(): remaining = self.get_remaining_cooldown(symbol) if remaining and remaining > 0: status['active_cooldowns'][symbol] = { 'remaining_minutes': remaining, 'last_close': last_close.strftime('%H:%M:%S') } # Activité récente for symbol, opens in self.recent_opens.items(): if opens: status['recent_activity'][symbol] = { 'opens_last_hour': len(opens), 'last_open': opens[-1].strftime('%H:%M:%S') if opens else None } return status def cleanup_old_data(self): """Nettoie les anciennes données (appelé périodiquement)""" with self.lock: now = datetime.now() cutoff_time = now - timedelta(hours=24) # Nettoyer les anciennes fermetures old_closes = [ symbol for symbol, close_time in self.last_close_times.items() if close_time < cutoff_time ] for symbol in old_closes: del self.last_close_times[symbol] # Nettoyer les anciennes ouvertures for symbol in list(self.recent_opens.keys()): self.recent_opens[symbol] = [ t for t in self.recent_opens[symbol] if t >= cutoff_time ] if not self.recent_opens[symbol]: del self.recent_opens[symbol] if old_closes: self.logger.info(f"Nettoyage: supprimé {len(old_closes)} anciens cooldowns") def force_reset_cooldown(self, symbol: str) -> bool: """Force le reset du cooldown pour un symbole (usage manuel)""" with self.lock: if symbol in self.last_close_times: del self.last_close_times[symbol] self.logger.warning(f"Cooldown forcé à zéro pour {symbol}") return True return False def set_symbol_cooldown(self, symbol: str, minutes: int): """Modifie le cooldown pour un symbole spécifique""" with self.lock: old_cooldown = self.symbol_cooldowns.get(symbol, self.symbol_cooldowns['DEFAULT']) self.symbol_cooldowns[symbol] = minutes self.logger.info(f"Cooldown {symbol} modifié: {old_cooldown}min → {minutes}min") def test_cooldown_manager(): """Test du gestionnaire de cooldown""" print("=== TEST COOLDOWN MANAGER ===\n") manager = CooldownManager(default_cooldown_minutes=5) # 5min pour le test # Test 1: Ouverture normale print("1. Test ouverture normale:") allowed, reason = manager.can_open_position('BNB') print(f" Résultat: {'OK' if allowed else 'BLOQUE'} - {reason}") if allowed: manager.record_position_open('BNB') # Test 2: Fermeture et cooldown print("\n2. Test fermeture et activation cooldown:") manager.record_position_close('BNB', "Test fermeture") allowed, reason = manager.can_open_position('BNB') print(f" Résultat: {'OK' if allowed else 'BLOQUE'} - {reason}") # Test 3: Overtrading print("\n3. Test protection overtrading:") for i in range(7): manager.record_position_open('ETH') allowed, reason = manager.can_open_position('ETH') print(f" Résultat: {'OK' if allowed else 'BLOQUE'} - {reason}") # Test 4: Statut global print("\n4. Statut global:") status = manager.get_cooldown_status() print(" Cooldowns actifs:") for symbol, data in status['active_cooldowns'].items(): print(f" {symbol}: {data['remaining_minutes']}min (fermé à {data['last_close']})") print(" Activité récente:") for symbol, data in status['recent_activity'].items(): print(f" {symbol}: {data['opens_last_hour']} ouvertures (dernière: {data['last_open']})") print(f" Statistiques: {status['statistics']['cooldown_blocks']} blocs, {status['statistics']['overtrading_prevention']} préventions") return manager if __name__ == "__main__": test_cooldown_manager()