""" Analyseur de Performance Automatique Analyse les trades historiques et génère dynamiquement les whitelist de performance S'actualise automatiquement toutes les 50 trades ou 24h """ import json import os from datetime import datetime, timedelta from collections import defaultdict import logging logger = logging.getLogger(__name__) # Configuration MIN_TRADES_FOR_STATS = 10 # Minimum de trades pour être évalué (augmenté 5→10) WHITELIST_WIN_RATE_THRESHOLD = 0.65 # Win rate > 65% → whitelist MIN_PNL_FOR_WHITELIST = 0.20 # P&L minimum pour whitelist UPDATE_INTERVAL_HOURS = 24 # Actualisation toutes les 24h UPDATE_INTERVAL_TRADES = 50 # Ou tous les 50 trades class PerformanceAnalyzer: def __init__(self): self.trade_history_file = 'trade_history.json' self.whitelist_file = 'performance_whitelist.json' self.stats_file = 'performance_stats.json' # Cache pour éviter de recharger à chaque appel self._whitelist_cache = None self._cache_loaded_at = None self._cache_ttl = 300 # 5 minutes def load_trade_history(self): """Charger l'historique des trades""" if not os.path.exists(self.trade_history_file): return [] try: with open(self.trade_history_file, 'r', encoding='utf-8-sig') as f: # Fix UTF-8 BOM return json.load(f) except Exception as e: # Ne logger qu'une fois par minute pour éviter spam if not hasattr(self, '_last_error_log') or (datetime.now() - self._last_error_log).total_seconds() > 60: logger.error(f"Erreur chargement trade_history: {e}") self._last_error_log = datetime.now() return [] def analyze_crypto_performance(self, trades): """Analyser les performances de chaque crypto""" crypto_stats = defaultdict(lambda: { 'trades': 0, 'wins': 0, 'losses': 0, 'total_pnl': 0.0, 'win_rate': 0.0, 'avg_pnl': 0.0, 'last_trades': [] }) for trade in trades: symbol = trade.get('symbol') pnl = trade.get('pnl', 0) if not symbol: continue stats = crypto_stats[symbol] stats['trades'] += 1 stats['total_pnl'] += pnl if pnl > 0: stats['wins'] += 1 else: stats['losses'] += 1 # Garder les 10 derniers trades stats['last_trades'].append({ 'pnl': pnl, 'timestamp': trade.get('timestamp', ''), 'reason': trade.get('reason', '') }) if len(stats['last_trades']) > 10: stats['last_trades'].pop(0) # Calculer win rate et moyenne for symbol, stats in crypto_stats.items(): if stats['trades'] > 0: stats['win_rate'] = stats['wins'] / stats['trades'] stats['avg_pnl'] = stats['total_pnl'] / stats['trades'] return dict(crypto_stats) def generate_whitelist(self, crypto_stats): """Générer la whitelist basée sur les performances""" whitelist = [] scores = {} for symbol, stats in crypto_stats.items(): # Minimum de trades requis if stats['trades'] < MIN_TRADES_FOR_STATS: continue # Critères de whitelist should_whitelist = False bonus_score = 0 # 1. Excellent win rate + P&L positif if stats['win_rate'] >= WHITELIST_WIN_RATE_THRESHOLD and stats['total_pnl'] >= MIN_PNL_FOR_WHITELIST: should_whitelist = True bonus_score = 20 # Bonus supplémentaire pour très bon win rate if stats['win_rate'] >= 0.80: bonus_score = 25 if stats['win_rate'] >= 0.90: bonus_score = 30 # 2. Win rate moyen mais P&L excellent elif stats['win_rate'] >= 0.55 and stats['total_pnl'] >= 0.50: should_whitelist = True bonus_score = 15 # 3. Série de gains récents recent_trades = stats['last_trades'][-5:] if len(recent_trades) >= 5: recent_wins = sum(1 for t in recent_trades if t['pnl'] > 0) if recent_wins >= 4: # 4/5 ou 5/5 gagnants if not should_whitelist: should_whitelist = True bonus_score = 10 else: bonus_score += 5 # Bonus supplémentaire if should_whitelist: whitelist.append(symbol) scores[symbol] = { 'bonus_score': bonus_score, 'win_rate': f"{stats['win_rate']*100:.1f}%", 'total_pnl': f"{stats['total_pnl']:.2f}€", 'trades': stats['trades'], 'avg_pnl': f"{stats['avg_pnl']:.2f}€" } return whitelist, scores def should_update(self): """Vérifier si une mise à jour est nécessaire""" # Vérifier la date de dernière mise à jour if os.path.exists(self.stats_file): try: with open(self.stats_file, 'r') as f: stats = json.load(f) last_update = datetime.fromisoformat(stats.get('last_update', '2000-01-01')) trades_at_update = stats.get('trades_count', 0) # Vérifier intervalle de temps hours_since_update = (datetime.now() - last_update).total_seconds() / 3600 if hours_since_update >= UPDATE_INTERVAL_HOURS: logger.info(f"📊 Mise à jour performance (24h écoulées)") return True # Vérifier nombre de trades current_trades = len(self.load_trade_history()) trades_diff = current_trades - trades_at_update if trades_diff >= UPDATE_INTERVAL_TRADES: logger.info(f"📊 Mise à jour performance ({trades_diff} nouveaux trades)") return True return False except: pass # Première exécution ou erreur return True def update(self): """Mettre à jour les listes de performance""" logger.info("📊 Début analyse de performance...") # Invalider cache self._whitelist_cache = None self._cache_time = None # Fix: utiliser _cache_time pas _cache_loaded_at # Charger les trades trades = self.load_trade_history() if len(trades) < 20: logger.info("📊 Pas assez de trades pour analyse (<20)") return # Analyser les performances crypto_stats = self.analyze_crypto_performance(trades) # Générer whitelist whitelist, whitelist_scores = self.generate_whitelist(crypto_stats) logger.info(f"📊 Whitelist générée: {len(whitelist)} cryptos") # Sauvegarder whitelist with open(self.whitelist_file, 'w', encoding='utf-8') as f: json.dump({ 'whitelist': whitelist, 'scores': whitelist_scores, 'updated': datetime.now().isoformat(), 'trades_analyzed': len(trades) }, f, indent=2, ensure_ascii=False) # Sauvegarder stats de mise à jour with open(self.stats_file, 'w', encoding='utf-8') as f: json.dump({ 'last_update': datetime.now().isoformat(), 'trades_count': len(trades), 'whitelist_count': len(whitelist) }, f, indent=2) logger.info(f"✅ Whitelist: {', '.join(whitelist[:5])}..." if whitelist else "✅ Whitelist: vide") return { 'whitelist': whitelist, 'stats': crypto_stats } def get_whitelist(self): """Récupérer la whitelist actuelle (avec cache)""" # Cache valide ? if self._whitelist_cache and self._cache_time: if (datetime.now() - self._cache_time).total_seconds() < self._cache_ttl: return self._whitelist_cache # Recharger if os.path.exists(self.whitelist_file): try: with open(self.whitelist_file, 'r', encoding='utf-8') as f: data = json.load(f) result = (data.get('whitelist', []), data.get('scores', {})) self._whitelist_cache = result self._cache_time = datetime.now() return result except: pass result = ([], {}) self._whitelist_cache = result self._cache_time = datetime.now() return result # Instance globale _analyzer = None def get_performance_analyzer(): """Récupérer l'instance de l'analyseur""" global _analyzer if _analyzer is None: _analyzer = PerformanceAnalyzer() return _analyzer if __name__ == '__main__': # Test de l'analyseur logging.basicConfig(level=logging.INFO) analyzer = PerformanceAnalyzer() print("📊 Analyse de performance...") result = analyzer.update() if result: print(f"\n✅ Whitelist ({len(result['whitelist'])} cryptos):") for sym in result['whitelist'][:10]: print(f" ⭐ {sym}")