#!/usr/bin/env python3 """ TRADE LOGGER - Système de logs structurés pour analyse de performance ===================================================================== Capture tous les événements de trading avec leurs métriques pour permettre l'analyse et l'optimisation des paramètres dynamiques. Créé le: 03/01/2026 """ import json import os from datetime import datetime from typing import Dict, Any, Optional import logging logger = logging.getLogger("TradeLogger") class TradeLogger: """Enregistre tous les trades et signaux avec métriques complètes""" def __init__(self, log_dir: str = "trade_logs"): self.log_dir = log_dir os.makedirs(log_dir, exist_ok=True) # Fichiers de logs self.signals_file = os.path.join(log_dir, "signals_log.jsonl") self.trades_file = os.path.join(log_dir, "trades_log.jsonl") self.performance_file = os.path.join(log_dir, "performance_log.jsonl") self.daily_summary_file = os.path.join(log_dir, "daily_summary.json") def log_signal(self, signal_data: Dict[str, Any]) -> None: """ Log un signal d'achat détecté par l'IA Args: signal_data: { 'timestamp': datetime, 'symbol': str, 'signal_type': 'BUY'|'SELL', 'ai_score': float, 'pattern': str, 'smart_signal': str, 'smart_eligible': bool, 'features': { 'rsi': float, 'ema_9': float, 'ema_21': float, 'ema_diff': float, 'momentum_3': float, 'momentum_5': float, 'momentum_10': float, 'bb_position': float, 'volatility_score': float }, 'dynamic_sltp': { 'sl_pct': float, 'tp_pct': float, 'rr_ratio': float, 'base_sl': float, 'base_tp': float, 'volatility_adjustment': float, 'confidence_adjustment': float }, 'market_conditions': { 'regime': str, 'btc_trend': str, 'volatility_index': float }, 'executed': bool, 'execution_reason': str (si executed=False) } """ log_entry = { 'timestamp': signal_data.get('timestamp', datetime.now()).isoformat(), 'type': 'SIGNAL', **signal_data } self._append_to_jsonl(self.signals_file, log_entry) logging.getLogger("TradeLogger").debug(f"📝 Signal logged: {signal_data.get('symbol')} - Score={signal_data.get('ai_score')}") def log_trade_open(self, trade_data: Dict[str, Any]) -> None: """ Log l'ouverture d'une position Args: trade_data: { 'timestamp': datetime, 'symbol': str, 'side': 'BUY', 'entry_price': float, 'quantity': float, 'order_size_usd': float, 'stop_loss': float, 'take_profit': float, 'sl_pct': float, 'tp_pct': float, 'rr_ratio': float, 'is_dynamic_sltp': bool, 'ai_score': float, 'pattern': str, 'reason': str, 'features': dict, 'market_conditions': dict } """ log_entry = { 'timestamp': trade_data.get('timestamp', datetime.now()).isoformat(), 'type': 'TRADE_OPEN', 'trade_id': f"{trade_data['symbol']}_{int(datetime.now().timestamp())}", **trade_data } self._append_to_jsonl(self.trades_file, log_entry) logging.getLogger("TradeLogger").info(f"💰 Trade ouvert: {trade_data['symbol']} @ {trade_data['entry_price']:.6f} (SL/TP: {trade_data['sl_pct']:.2f}%/{trade_data['tp_pct']:.2f}%)") def log_trade_close(self, close_data: Dict[str, Any]) -> None: """ Log la fermeture d'une position Args: close_data: { 'timestamp': datetime, 'symbol': str, 'side': 'SELL', 'entry_price': float, 'exit_price': float, 'quantity': float, 'duration_minutes': float, 'pnl_pct': float, 'pnl_usd': float, 'exit_reason': 'TP_HIT'|'SL_HIT'|'MANUAL'|'ROTATION'|'TECHNICAL_EXIT', 'was_dynamic_sltp': bool, 'original_sl_pct': float, 'original_tp_pct': float, 'max_price': float, 'max_profit_pct': float, 'min_price': float, 'max_drawdown_pct': float, 'pattern': str, 'ai_score': float } """ log_entry = { 'timestamp': close_data.get('timestamp', datetime.now()).isoformat(), 'type': 'TRADE_CLOSE', 'trade_id': f"{close_data['symbol']}_{close_data.get('open_timestamp', int(datetime.now().timestamp()))}", **close_data } self._append_to_jsonl(self.trades_file, log_entry) # Emoji selon résultat emoji = "✅" if close_data.get('pnl_pct', 0) > 0 else "❌" logging.getLogger("TradeLogger").info(f"{emoji} Trade fermé: {close_data['symbol']} - P&L: {close_data.get('pnl_pct', 0):.2f}% ({close_data.get('exit_reason', 'UNKNOWN')})") def log_performance_metrics(self, metrics: Dict[str, Any]) -> None: """ Log les métriques de performance globales (toutes les heures) Args: metrics: { 'timestamp': datetime, 'total_trades': int, 'winning_trades': int, 'losing_trades': int, 'win_rate': float, 'avg_profit_pct': float, 'avg_loss_pct': float, 'profit_factor': float, 'total_pnl_pct': float, 'total_pnl_usd': float, 'sharpe_ratio': float, 'max_drawdown_pct': float, 'avg_trade_duration_min': float, 'dynamic_sltp_performance': { 'trades_count': int, 'win_rate': float, 'avg_profit': float, 'avg_rr_achieved': float }, 'fixed_sltp_performance': { 'trades_count': int, 'win_rate': float, 'avg_profit': float }, 'best_patterns': list, 'worst_patterns': list, 'best_cryptos': list, 'worst_cryptos': list } """ log_entry = { 'timestamp': metrics.get('timestamp', datetime.now()).isoformat(), 'type': 'PERFORMANCE_SNAPSHOT', **metrics } self._append_to_jsonl(self.performance_file, log_entry) logging.getLogger("TradeLogger").info(f"📊 Performance snapshot: Win rate={metrics.get('win_rate', 0):.1f}% | Total P&L={metrics.get('total_pnl_pct', 0):.2f}%") def update_daily_summary(self, summary: Dict[str, Any]) -> None: """ Met à jour le résumé quotidien (remplace le fichier) Args: summary: { 'date': str, 'total_signals': int, 'signals_executed': int, 'execution_rate': float, 'trades_opened': int, 'trades_closed': int, 'win_rate': float, 'total_pnl_pct': float, 'best_trade': dict, 'worst_trade': dict, 'avg_sl_pct': float, 'avg_tp_pct': float, 'avg_rr_ratio': float, 'dynamic_sltp_usage': float (%) } """ with open(self.daily_summary_file, 'w', encoding='utf-8') as f: json.dump(summary, f, indent=2, ensure_ascii=False, default=str) logging.getLogger("TradeLogger").info(f"📅 Daily summary updated: {summary.get('trades_closed', 0)} trades | P&L: {summary.get('total_pnl_pct', 0):.2f}%") def _append_to_jsonl(self, filepath: str, data: Dict[str, Any]) -> None: """Ajoute une ligne JSON au fichier (JSONL format)""" try: with open(filepath, 'a', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, default=str) f.write('\n') except Exception as e: logging.getLogger("TradeLogger").error(f"❌ Erreur écriture log {filepath}: {e}") def get_recent_signals(self, limit: int = 100) -> list: """Récupère les N derniers signaux""" return self._read_jsonl_tail(self.signals_file, limit) def get_recent_trades(self, limit: int = 100) -> list: """Récupère les N derniers trades""" return self._read_jsonl_tail(self.trades_file, limit) def _read_jsonl_tail(self, filepath: str, limit: int) -> list: """Lit les N dernières lignes d'un fichier JSONL""" if not os.path.exists(filepath): return [] try: with open(filepath, 'r', encoding='utf-8') as f: lines = f.readlines() result = [] for line in lines[-limit:]: try: result.append(json.loads(line.strip())) except: pass return result except Exception as e: logging.getLogger("TradeLogger").error(f"❌ Erreur lecture {filepath}: {e}") return [] def analyze_sltp_performance(self) -> Dict[str, Any]: """ Analyse la performance des SL/TP dynamiques vs fixes Returns: { 'dynamic': {'trades': int, 'win_rate': float, 'avg_profit': float, 'avg_rr': float}, 'fixed': {'trades': int, 'win_rate': float, 'avg_profit': float}, 'improvement': float (%) } """ trades = self.get_recent_trades(1000) dynamic_trades = [t for t in trades if t.get('type') == 'TRADE_CLOSE' and t.get('was_dynamic_sltp')] fixed_trades = [t for t in trades if t.get('type') == 'TRADE_CLOSE' and not t.get('was_dynamic_sltp')] def calc_stats(trade_list): if not trade_list: return {'trades': 0, 'win_rate': 0, 'avg_profit': 0, 'avg_rr': 0} winning = [t for t in trade_list if t.get('pnl_pct', 0) > 0] profits = [t['pnl_pct'] for t in trade_list] return { 'trades': len(trade_list), 'win_rate': len(winning) / len(trade_list) * 100, 'avg_profit': sum(profits) / len(profits), 'avg_rr': sum(t.get('original_tp_pct', 0) / t.get('original_sl_pct', 1) for t in trade_list) / len(trade_list) } dynamic_stats = calc_stats(dynamic_trades) fixed_stats = calc_stats(fixed_trades) 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 } # Instance globale _trade_logger: Optional[TradeLogger] = None def get_trade_logger() -> TradeLogger: """Retourne l'instance globale du trade logger""" global _trade_logger if _trade_logger is None: _trade_logger = TradeLogger() return _trade_logger if __name__ == "__main__": # Test du système de logging logger = get_trade_logger() # Test signal logger.log_signal({ 'symbol': 'BTCUSDT', 'signal_type': 'BUY', 'ai_score': 85, 'pattern': 'SQUEEZE_BREAKOUT', 'smart_signal': 'ACHAT', 'smart_eligible': True, 'features': { 'rsi': 45, 'ema_diff': -0.5, 'momentum_3': 0.8 }, 'dynamic_sltp': { 'sl_pct': 2.5, 'tp_pct': 6.0, 'rr_ratio': 2.4 }, 'executed': True }) # Test trade logger.log_trade_open({ 'symbol': 'BTCUSDT', 'side': 'BUY', 'entry_price': 45000.0, 'quantity': 0.001, 'order_size_usd': 45.0, 'stop_loss': 43875.0, 'take_profit': 47700.0, 'sl_pct': 2.5, 'tp_pct': 6.0, 'rr_ratio': 2.4, 'is_dynamic_sltp': True, 'ai_score': 85, 'pattern': 'SQUEEZE_BREAKOUT', 'reason': 'IA Signal' }) print("✅ Test logging OK - Fichiers créés dans trade_logs/")