#!/usr/bin/env python3 """ ANALYSEUR DE LOGS DE TRADING ============================ Analyse les logs structurés pour optimiser les paramètres dynamiques Usage: python analyze_trade_logs.py python analyze_trade_logs.py --period 24h python analyze_trade_logs.py --export report.json """ import json import os import argparse from datetime import datetime, timedelta from typing import Dict, List, Any from collections import defaultdict import statistics class TradeAnalyzer: """Analyse les performances des trades pour optimisation""" def __init__(self, log_dir: str = "trade_logs"): self.log_dir = log_dir self.signals_file = os.path.join(log_dir, "signals_log.jsonl") self.trades_file = os.path.join(log_dir, "trades_log.jsonl") def load_logs(self, hours_back: int = 24) -> Dict[str, List]: """Charge les logs récents""" cutoff_time = datetime.now() - timedelta(hours=hours_back) signals = [] trades_open = [] trades_close = [] # Charger signaux if os.path.exists(self.signals_file): with open(self.signals_file, 'r', encoding='utf-8') as f: for line in f: try: entry = json.loads(line.strip()) ts = datetime.fromisoformat(entry['timestamp']) if ts >= cutoff_time: signals.append(entry) except: pass # Charger TOUS les TRADE_OPEN pour mapping par symbole + timestamp # Le bug: trade_id du CLOSE utilise timestamp de fermeture, pas d'ouverture trade_open_by_symbol = defaultdict(list) # Map symbol -> [TRADE_OPEN entries] if os.path.exists(self.trades_file): # Premier passage: charger TOUS les TRADE_OPEN pour mapping with open(self.trades_file, 'r', encoding='utf-8') as f: for line in f: try: entry = json.loads(line.strip()) if entry['type'] == 'TRADE_OPEN': symbol = entry.get('symbol', 'UNKNOWN') entry['_ts_obj'] = datetime.fromisoformat(entry['timestamp']) trade_open_by_symbol[symbol].append(entry) except: pass # Trier par timestamp pour chaque symbole for symbol in trade_open_by_symbol: trade_open_by_symbol[symbol].sort(key=lambda x: x['_ts_obj']) # Deuxième passage: charger les trades de la période avec enrichissement with open(self.trades_file, 'r', encoding='utf-8') as f: for line in f: try: entry = json.loads(line.strip()) ts = datetime.fromisoformat(entry['timestamp']) if entry['type'] == 'TRADE_OPEN' and ts >= cutoff_time: trades_open.append(entry) elif entry['type'] == 'TRADE_CLOSE' and ts >= cutoff_time: # Matcher avec le TRADE_OPEN le plus proche dans le temps (avant le CLOSE) symbol = entry.get('symbol', 'UNKNOWN') matched_open = None if symbol in trade_open_by_symbol: # Chercher le dernier TRADE_OPEN avant ce TRADE_CLOSE for open_entry in reversed(trade_open_by_symbol[symbol]): if open_entry['_ts_obj'] < ts: # Vérifier que ce n'est pas trop vieux (max 24h de différence) time_diff = (ts - open_entry['_ts_obj']).total_seconds() if time_diff < 86400: # 24h max matched_open = open_entry break if matched_open: # Copier le pattern et ai_score depuis TRADE_OPEN entry['pattern'] = matched_open.get('pattern', 'UNKNOWN') entry['ai_score'] = matched_open.get('ai_score', 0) entry['was_dynamic_sltp'] = matched_open.get('is_dynamic_sltp', False) else: # Pas de TRADE_OPEN trouvé, marquer pattern comme UNKNOWN entry['pattern'] = entry.get('pattern', 'UNKNOWN') trades_close.append(entry) except: pass return { 'signals': signals, 'trades_open': trades_open, 'trades_close': trades_close } def analyze_sltp_performance(self, data: Dict) -> Dict: """Analyse performance SL/TP dynamiques vs fixes""" closed_trades = data['trades_close'] dynamic_trades = [t for t in closed_trades if t.get('was_dynamic_sltp')] fixed_trades = [t for t in closed_trades if not t.get('was_dynamic_sltp')] def calc_stats(trades): if not trades: return { 'count': 0, 'win_rate': 0, 'avg_profit': 0, 'avg_duration_min': 0, 'best_profit': 0, 'worst_loss': 0, 'profit_factor': 0, 'avg_sl': 0, 'avg_tp': 0 } winning = [t for t in trades if t['pnl_pct'] > 0] losing = [t for t in trades if t['pnl_pct'] < 0] total_profit = sum(t['pnl_pct'] for t in winning) if winning else 0 total_loss = abs(sum(t['pnl_pct'] for t in losing)) if losing else 1 return { 'count': len(trades), 'win_rate': len(winning) / len(trades) * 100, 'avg_profit': sum(t['pnl_pct'] for t in trades) / len(trades), 'avg_duration_min': sum(t.get('duration_seconds', 0) / 60 for t in trades) / len(trades), 'best_profit': max(t['pnl_pct'] for t in trades), 'worst_loss': min(t['pnl_pct'] for t in trades), 'profit_factor': total_profit / total_loss if total_loss > 0 else 0, 'avg_sl': sum(t.get('sl_pct', 0) for t in trades) / len(trades), 'avg_tp': sum(t.get('tp_pct', 0) for t in trades) / len(trades) } dynamic_stats = calc_stats(dynamic_trades) fixed_stats = calc_stats(fixed_trades) # Calcul amélioration improvement = 0 if fixed_stats['avg_profit'] != 0: improvement = ((dynamic_stats['avg_profit'] - fixed_stats['avg_profit']) / abs(fixed_stats['avg_profit'])) * 100 return { 'dynamic': dynamic_stats, 'fixed': fixed_stats, 'improvement_pct': improvement, 'recommendation': self._generate_sltp_recommendation(dynamic_stats, fixed_stats) } def analyze_patterns(self, data: Dict) -> Dict: """Analyse performance par pattern""" closed_trades = data['trades_close'] by_pattern = defaultdict(list) for trade in closed_trades: pattern = trade.get('pattern', 'UNKNOWN') by_pattern[pattern].append(trade) pattern_stats = {} for pattern, trades in by_pattern.items(): winning = [t for t in trades if t['pnl_pct'] > 0] pattern_stats[pattern] = { 'count': len(trades), 'win_rate': len(winning) / len(trades) * 100 if trades else 0, 'avg_profit': sum(t['pnl_pct'] for t in trades) / len(trades) if trades else 0, 'total_pnl': sum(t['pnl_pct'] for t in trades) } # Trier par performance sorted_patterns = sorted(pattern_stats.items(), key=lambda x: x[1]['avg_profit'], reverse=True) # Créer la liste top_patterns pour le dashboard top_patterns = [] for pattern_name, stats in sorted_patterns: top_patterns.append({ 'pattern': pattern_name, 'count': stats['count'], 'win_rate': stats['win_rate'], 'avg_profit_pct': stats['avg_profit'], 'total_pnl_pct': stats['total_pnl'] }) return { 'by_pattern': dict(sorted_patterns), 'best_pattern': sorted_patterns[0] if sorted_patterns else ('NONE', {}), 'worst_pattern': sorted_patterns[-1] if sorted_patterns else ('NONE', {}), 'top_patterns': top_patterns # Ajout pour le dashboard } def analyze_execution_rate(self, data: Dict) -> Dict: """Analyse taux d'exécution des signaux""" signals = data['signals'] trades_open = data['trades_open'] total_signals = len(signals) executed_signals = len([s for s in signals if s.get('executed')]) # Raisons de non-exécution not_executed = [s for s in signals if not s.get('executed')] reasons = defaultdict(int) for sig in not_executed: reason = sig.get('execution_reason', 'UNKNOWN') reasons[reason] += 1 return { 'total_signals': total_signals, 'executed': executed_signals, 'execution_rate': executed_signals / total_signals * 100 if total_signals > 0 else 0, 'not_executed_reasons': dict(reasons) } def analyze_crypto_performance(self, data: Dict) -> Dict: """Analyse performance par crypto""" closed_trades = data['trades_close'] by_crypto = defaultdict(list) for trade in closed_trades: symbol = trade.get('symbol', 'UNKNOWN') by_crypto[symbol].append(trade) crypto_stats = [] for symbol, trades in by_crypto.items(): winning = [t for t in trades if t['pnl_pct'] > 0] crypto_stats.append({ 'symbol': symbol, 'count': len(trades), 'win_rate': len(winning) / len(trades) * 100 if trades else 0, 'avg_profit': sum(t['pnl_pct'] for t in trades) / len(trades) if trades else 0, 'total_pnl_pct': sum(t['pnl_pct'] for t in trades) }) # Trier par total P&L sorted_cryptos = sorted(crypto_stats, key=lambda x: x['total_pnl_pct'], reverse=True) # Convertir aussi en dict pour compatibilité by_crypto_dict = {} for crypto in sorted_cryptos[:10]: by_crypto_dict[crypto['symbol']] = { 'count': crypto['count'], 'win_rate': crypto['win_rate'], 'avg_profit': crypto['avg_profit'], 'total_pnl': crypto['total_pnl_pct'] } return { 'top_performers': sorted_cryptos, # Pour le dashboard 'by_crypto': by_crypto_dict, # Pour compatibilité 'best_crypto': (sorted_cryptos[0]['symbol'], sorted_cryptos[0]) if sorted_cryptos else ('NONE', {}), 'worst_crypto': (sorted_cryptos[-1]['symbol'], sorted_cryptos[-1]) if sorted_cryptos else ('NONE', {}) } def _generate_sltp_recommendation(self, dynamic_stats, fixed_stats) -> str: """Génère recommandation basée sur stats""" if dynamic_stats['count'] < 10: return "⏳ Pas assez de données (min 10 trades)" if dynamic_stats['avg_profit'] > fixed_stats['avg_profit']: improvement = ((dynamic_stats['avg_profit'] - fixed_stats['avg_profit']) / abs(fixed_stats['avg_profit'])) * 100 return f"✅ DYNAMIQUE MEILLEUR (+{improvement:.1f}%) - Continuer" else: decline = ((fixed_stats['avg_profit'] - dynamic_stats['avg_profit']) / abs(fixed_stats['avg_profit'])) * 100 if decline > 20: return f"❌ FIXE MEILLEUR (+{decline:.1f}%) - Revoir paramètres dynamiques" else: return f"⚠️ Performance similaire - Continuer observation" def analyze_rejections(self, rejected_signals: List[Dict]) -> Dict: """Analyse les raisons de rejet des signaux""" if not rejected_signals: return { 'total': 0, 'by_reason': {}, 'top_rejected_cryptos': [], 'recommendation': 'Aucun signal rejeté' } # Grouper par raison by_reason = defaultdict(int) by_crypto = defaultdict(int) for signal in rejected_signals: reason = signal.get('rejection_reason', 'UNKNOWN') by_reason[reason] += 1 by_crypto[signal.get('symbol', 'UNKNOWN')] += 1 # Top cryptos rejetés top_rejected = sorted(by_crypto.items(), key=lambda x: x[1], reverse=True)[:5] # Générer recommandation total = len(rejected_signals) main_reason = max(by_reason.items(), key=lambda x: x[1])[0] if by_reason else 'UNKNOWN' main_count = by_reason[main_reason] main_pct = (main_count / total * 100) if total > 0 else 0 recommendation = "" if 'MAX_POSITIONS' in main_reason and main_pct > 50: recommendation = f"⚠️ {main_pct:.0f}% rejetés car limite atteinte - Considérer augmenter max_positions ou améliorer scoring" elif 'BEARISH_TREND' in main_reason and main_pct > 30: recommendation = f"📉 {main_pct:.0f}% rejetés car baissiers - Filtres anti-baisse fonctionnent bien" elif 'COOLDOWN' in main_reason and main_pct > 40: recommendation = f"🕒 {main_pct:.0f}% rejetés car cooldown - Réduire cooldown ou améliorer timing" elif 'SCORE_TOO_LOW' in main_reason and main_pct > 30: recommendation = f"📊 {main_pct:.0f}% rejetés car score faible - Scoring fonctionne correctement" else: recommendation = f"ℹ️ Distribution variée des rejets" return { 'total': total, 'by_reason': dict(by_reason), 'top_rejected_cryptos': top_rejected, 'recommendation': recommendation } def generate_report(self, hours_back: int = 24) -> Dict: """Génère rapport complet""" data = self.load_logs(hours_back) # Séparer signaux exécutés et rejetés executed_signals = [s for s in data['signals'] if s.get('executed', True)] rejected_signals = [s for s in data['signals'] if not s.get('executed', True)] # Calculer métriques pour overview closed_trades = data['trades_close'] winning_trades = [t for t in closed_trades if t.get('pnl_pct', 0) > 0] losing_trades = [t for t in closed_trades if t.get('pnl_pct', 0) < 0] win_rate = (len(winning_trades) / len(closed_trades) * 100) if closed_trades else 0 avg_profit_pct = sum(t.get('pnl_pct', 0) for t in closed_trades) / len(closed_trades) if closed_trades else 0 avg_duration_min = sum(t.get('duration_seconds', 0) / 60 for t in closed_trades) / len(closed_trades) if closed_trades else 0 report = { 'period': f"Dernières {hours_back}h", 'generated_at': datetime.now().isoformat(), 'overview': { 'total_signals': len(data['signals']), 'signals_detected': len(data['signals']), 'executed_signals': len(executed_signals), 'signals_executed': len(executed_signals), 'rejected_signals': len(rejected_signals), 'execution_rate': len(executed_signals) / len(data['signals']) * 100 if data['signals'] else 0, 'trades_opened': len(data['trades_open']), 'trades_closed': len(data['trades_close']), 'currently_open': len(data['trades_open']) - len(data['trades_close']), 'wins': len(winning_trades), 'losses': len(losing_trades), 'win_rate': win_rate, 'avg_profit_pct': avg_profit_pct, 'avg_duration_min': avg_duration_min }, 'rejection_analysis': self.analyze_rejections(rejected_signals), 'sltp_performance': self.analyze_sltp_performance(data), 'pattern_analysis': self.analyze_patterns(data), 'execution_analysis': self.analyze_execution_rate(data), 'crypto_performance': self.analyze_crypto_performance(data) } return report def print_report(self, report: Dict): """Affiche rapport formaté""" print("=" * 80) print(f"📊 RAPPORT D'ANALYSE - {report['period']}") print(f"🕐 Généré: {report['generated_at']}") print("=" * 80) print() # Vue d'ensemble ov = report['overview'] print("📈 VUE D'ENSEMBLE:") print(f" Signaux détectés: {ov['total_signals']}") print(f" Signaux exécutés: {ov['executed_signals']} ({ov['execution_rate']:.1f}%)") print(f" Signaux rejetés: {ov['rejected_signals']}") print(f" Trades ouverts: {ov['trades_opened']}") print(f" Trades fermés: {ov['trades_closed']}") print(f" Positions ouvertes: {ov['currently_open']}") print() # Analyse des rejets if ov['rejected_signals'] > 0: rej = report['rejection_analysis'] print("🚫 ANALYSE DES REJETS:") print(f" Total rejetés: {rej['total']}") print(" Raisons:") for reason, count in sorted(rej['by_reason'].items(), key=lambda x: x[1], reverse=True): pct = (count / rej['total'] * 100) if rej['total'] > 0 else 0 print(f" • {reason}: {count} ({pct:.0f}%)") if rej['top_rejected_cryptos']: print(" Top cryptos rejetés:") for symbol, count in rej['top_rejected_cryptos']: print(f" • {symbol}: {count} rejets") print(f" 💡 {rej['recommendation']}") print() # Performance SL/TP sltp = report['sltp_performance'] print("🎯 PERFORMANCE SL/TP:") print(f" DYNAMIQUE: {sltp['dynamic']['count']} trades | Win rate: {sltp['dynamic']['win_rate']:.1f}% | Profit moyen: {sltp['dynamic']['avg_profit']:.2f}%") print(f" SL moyen: {sltp['dynamic']['avg_sl']:.2f}% | TP moyen: {sltp['dynamic']['avg_tp']:.2f}%") print(f" FIXE: {sltp['fixed']['count']} trades | Win rate: {sltp['fixed']['win_rate']:.1f}% | Profit moyen: {sltp['fixed']['avg_profit']:.2f}%") print(f" SL moyen: {sltp['fixed']['avg_sl']:.2f}% | TP moyen: {sltp['fixed']['avg_tp']:.2f}%") print(f" AMÉLIORATION: {sltp['improvement_pct']:+.1f}%") print(f" 📌 {sltp['recommendation']}") print() # Patterns patterns = report['pattern_analysis'] print("🔍 MEILLEURS PATTERNS:") for i, (pattern, stats) in enumerate(list(patterns['by_pattern'].items())[:5]): print(f" {i+1}. {pattern}: {stats['count']} trades | Win: {stats['win_rate']:.0f}% | Avg: {stats['avg_profit']:+.2f}%") print() # Cryptos cryptos = report['crypto_performance'] print("💰 TOP CRYPTOS:") for i, (symbol, stats) in enumerate(list(cryptos['by_crypto'].items())[:5]): print(f" {i+1}. {symbol}: {stats['count']} trades | Win: {stats['win_rate']:.0f}% | Total: {stats['total_pnl']:+.2f}%") print() # Exécution exec_stats = report['execution_analysis'] print("⚡ TAUX D'EXÉCUTION:") print(f" Signaux: {exec_stats['total_signals']} | Exécutés: {exec_stats['executed']} ({exec_stats['execution_rate']:.1f}%)") if exec_stats['not_executed_reasons']: print(" Raisons de non-exécution:") for reason, count in exec_stats['not_executed_reasons'].items(): print(f" - {reason}: {count}") print() print("=" * 80) def main(): parser = argparse.ArgumentParser(description='Analyse des logs de trading') parser.add_argument('--period', default='24h', help='Période d\'analyse (ex: 24h, 168h pour 7j)') parser.add_argument('--export', help='Exporter en JSON vers fichier') args = parser.parse_args() # Parser période hours = int(args.period.rstrip('h')) analyzer = TradeAnalyzer() report = analyzer.generate_report(hours) # Afficher rapport analyzer.print_report(report) # Export si demandé if args.export: with open(args.export, 'w', encoding='utf-8') as f: json.dump(report, f, indent=2, ensure_ascii=False) print(f"✅ Rapport exporté: {args.export}") if __name__ == "__main__": main()