diegobeyl/backtesting

1"""2Sistema de Logging y Monitoreo para AriasCampos Backtester3Fase 5: Registro detallado de operaciones, análisis, y optimizaciones4 5Características:6- Logs estructurados en JSON7- Persistencia en /logs/8- Métricas de rendimiento9- Trazabilidad completa de operaciones10- Dashboards en Streamlit11"""12 13import logging14import json15import os16from datetime import datetime17from pathlib import Path18from typing import Dict, Any, Optional, List19import sys20from enum import Enum21 22 23class LogLevel(Enum):24    """Niveles de logging"""25    DEBUG = logging.DEBUG26    INFO = logging.INFO27    WARNING = logging.WARNING28    ERROR = logging.ERROR29    CRITICAL = logging.CRITICAL30 31 32class OperationType(Enum):33    """Tipos de operaciones a registrar"""34    DOWNLOAD_DATA = "download_data"35    BACKTEST = "backtest"36    ANALYSIS = "analysis"37    OPTIMIZATION = "optimization"38    CACHE_HIT = "cache_hit"39    CACHE_MISS = "cache_miss"40    VALIDATION = "validation"41    ERROR = "error"42 43 44class PerformanceMonitor:45    """Monitor de rendimiento para operaciones"""46    47    def __init__(self, operation_type: OperationType):48        self.operation_type = operation_type49        self.start_time = datetime.now()50        self.metrics: Dict[str, Any] = {}51        self.status = "in_progress"52    53    def record_metric(self, key: str, value: Any):54        """Registra una métrica"""55        self.metrics[key] = value56    57    def finish(self, status: str = "success", error: Optional[str] = None):58        """Marca la operación como finalizada"""59        self.status = status60        duration = (datetime.now() - self.start_time).total_seconds()61        self.metrics['duration_seconds'] = duration62        63        if error:64            self.metrics['error'] = error65    66    def to_dict(self) -> Dict[str, Any]:67        """Convierte a diccionario"""68        return {69            'operation_type': self.operation_type.value,70            'timestamp': self.start_time.isoformat(),71            'status': self.status,72            'metrics': self.metrics73        }74 75 76class StructuredLogger:77    """Logger estructurado con soporte JSON - Singleton"""78    79    _instances = {}80    _initialized = set()81    82    def __new__(cls, name: str = "backtesting_app", logs_dir: str = "logs"):83        # Crear una instancia por nombre84        if name not in cls._instances:85            cls._instances[name] = super().__new__(cls)86        return cls._instances[name]87    88    def __init__(self, name: str = "backtesting_app", logs_dir: str = "logs"):89        # Solo inicializar una vez por nombre90        if name in StructuredLogger._initialized:91            return92        93        self.name = name94        self.logs_dir = Path(logs_dir)95        self.logs_dir.mkdir(exist_ok=True, parents=True)96        97        # Logger principal98        self.logger = logging.getLogger(name)99        self.logger.setLevel(logging.DEBUG)100        self.logger.propagate = False101        102        # Limpiar handlers previos completamente103        for handler in self.logger.handlers[:]:104            handler.close()105            self.logger.removeHandler(handler)106        107        # Handler para archivo JSON (estructurado)108        self.json_file = self.logs_dir / f"app_{datetime.now().strftime('%Y%m%d')}.json"109        self.json_logs: List[Dict[str, Any]] = []110        self._load_existing_logs()111        112        # Handler para archivo de texto (legible)113        self.text_file = self.logs_dir / f"app_{datetime.now().strftime('%Y%m%d')}.log"114        text_handler = logging.FileHandler(self.text_file)115        text_handler.setLevel(logging.DEBUG)116        text_formatter = logging.Formatter(117            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'118        )119        text_handler.setFormatter(text_formatter)120        self.logger.addHandler(text_handler)121        122        # Handler para consola123        console_handler = logging.StreamHandler(sys.stdout)124        console_handler.setLevel(logging.INFO)125        console_formatter = logging.Formatter(126            '%(levelname)s: %(message)s'127        )128        console_handler.setFormatter(console_formatter)129        self.logger.addHandler(console_handler)130        131        # Marcar como inicializado132        StructuredLogger._initialized.add(name)133    134    def _load_existing_logs(self):135        """Carga logs JSON existentes"""136        if self.json_file.exists():137            try:138                with open(self.json_file, 'r') as f:139                    self.json_logs = json.load(f)140            except (json.JSONDecodeError, IOError):141                self.json_logs = []142    143    def _save_json_logs(self):144        """Guarda logs JSON"""145        with open(self.json_file, 'w') as f:146            json.dump(self.json_logs, f, indent=2)147    148    def log_operation(self, monitor: PerformanceMonitor):149        """Registra una operación completada"""150        log_entry = {151            'timestamp': datetime.now().isoformat(),152            **monitor.to_dict()153        }154        self.json_logs.append(log_entry)155        self._save_json_logs()156        157        # Log de texto158        status_emoji = "✅" if monitor.status == "success" else "❌"159        duration = monitor.metrics.get('duration_seconds', 0)160        self.logger.info(161            f"{status_emoji} {monitor.operation_type.value}: {monitor.status} "162            f"({duration:.2f}s)"163        )164    165    def log_event(166        self, 167        level: LogLevel, 168        message: str, 169        operation_type: Optional[OperationType] = None,170        **kwargs171    ):172        """Registra un evento"""173        log_entry = {174            'timestamp': datetime.now().isoformat(),175            'level': level.name,176            'message': message,177            'operation_type': operation_type.value if operation_type else None,178            'context': kwargs179        }180        self.json_logs.append(log_entry)181        self._save_json_logs()182        183        # Log de texto184        if level == LogLevel.DEBUG:185            self.logger.debug(message)186        elif level == LogLevel.INFO:187            self.logger.info(message)188        elif level == LogLevel.WARNING:189            self.logger.warning(message)190        elif level == LogLevel.ERROR:191            self.logger.error(message)192        elif level == LogLevel.CRITICAL:193            self.logger.critical(message)194    195    def get_operations_summary(196        self, 197        operation_type: Optional[OperationType] = None,198        days: int = 1199    ) -> Dict[str, Any]:200        """Obtiene resumen de operaciones"""201        cutoff_time = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)202        203        relevant_logs = [204            log for log in self.json_logs205            if 'operation_type' in log and (206                operation_type is None or 207                log.get('operation_type') == operation_type.value208            ) and datetime.fromisoformat(log['timestamp']) > cutoff_time209        ]210        211        summary = {212            'total_operations': len(relevant_logs),213            'successful': len([l for l in relevant_logs if l.get('status') == 'success']),214            'failed': len([l for l in relevant_logs if l.get('status') == 'error']),215            'total_duration': sum(216                l.get('metrics', {}).get('duration_seconds', 0) 217                for l in relevant_logs218            ),219            'operations': relevant_logs220        }221        222        if summary['total_operations'] > 0:223            summary['avg_duration'] = (224                summary['total_duration'] / summary['total_operations']225            )226        227        return summary228    229    def get_performance_stats(self) -> Dict[str, Any]:230        """Obtiene estadísticas de rendimiento globales"""231        stats = {}232        233        # Agrupar por tipo de operación234        operation_types = set()235        for log in self.json_logs:236            if 'operation_type' in log:237                operation_types.add(log['operation_type'])238        239        for op_type in operation_types:240            ops = [241                log for log in self.json_logs242                if log.get('operation_type') == op_type243            ]244            245            durations = [246                o.get('metrics', {}).get('duration_seconds', 0)247                for o in ops248            ]249            250            if durations:251                stats[op_type] = {252                    'count': len(ops),253                    'min_duration': min(durations),254                    'max_duration': max(durations),255                    'avg_duration': sum(durations) / len(durations),256                    'success_rate': (257                        len([o for o in ops if o.get('status') == 'success'])258                        / len(ops) * 100259                    )260                }261        262        return stats263    264    def clear_old_logs(self, days: int = 30):265        """Limpia logs más antiguos que X días"""266        cutoff_time = datetime.now().timestamp() - (days * 24 * 3600)267        268        new_logs = []269        for log in self.json_logs:270            try:271                log_time = datetime.fromisoformat(log['timestamp']).timestamp()272                if log_time > cutoff_time:273                    new_logs.append(log)274            except (KeyError, ValueError):275                new_logs.append(log)276        277        self.json_logs = new_logs278        self._save_json_logs()279        280        deleted_count = len(self.json_logs) - len(new_logs)281        if deleted_count > 0:282            self.logger.info(f"Deleted {deleted_count} old log entries")283 284 285# Instancia global286_logger_instance: Optional[StructuredLogger] = None287 288 289def get_logger(name: str = "backtesting_app", logs_dir: str = "logs") -> StructuredLogger:290    """Obtiene la instancia global del logger"""291    global _logger_instance292    if _logger_instance is None:293        _logger_instance = StructuredLogger(name, logs_dir)294    return _logger_instance295