import logging import logging.handlers import json import os from datetime import datetime from typing import Dict, Any, Optional import yaml try: settings = yaml.safe_load(open("config/settings.yaml")) except: settings = {"telegram": {"enabled": False}} class ScalperLogger: def __init__(self): self.setup_logging() self.trade_logs = [] self.signal_logs = [] self.error_logs = [] def setup_logging(self): os.makedirs("logs", exist_ok=True) self.logger = logging.getLogger('scalper') self.logger.setLevel(logging.DEBUG) self.logger.handlers.clear() console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) console_formatter = logging.Formatter( '%(asctime)s - %(levelname)s - %(message)s' ) console_handler.setFormatter(console_formatter) self.logger.addHandler(console_handler) file_handler = logging.handlers.RotatingFileHandler( 'logs/scalper.log', maxBytes=10*1024*1024, backupCount=5 ) file_handler.setLevel(logging.DEBUG) file_formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s' ) file_handler.setFormatter(file_formatter) self.logger.addHandler(file_handler) trade_handler = logging.handlers.RotatingFileHandler( 'logs/trades.log', maxBytes=5*1024*1024, backupCount=3 ) trade_handler.setLevel(logging.INFO) trade_formatter = logging.Formatter( '%(asctime)s - TRADE - %(message)s' ) trade_handler.setFormatter(trade_formatter) trade_handler.addFilter(self._trade_filter) self.logger.addHandler(trade_handler) error_handler = logging.handlers.RotatingFileHandler( 'logs/errors.log', maxBytes=5*1024*1024, backupCount=3 ) error_handler.setLevel(logging.ERROR) error_formatter = logging.Formatter( '%(asctime)s - ERROR - %(message)s' ) error_handler.setFormatter(error_formatter) self.logger.addHandler(error_handler) def _trade_filter(self, record): return 'TRADE' in record.getMessage() or 'SIGNAL' in record.getMessage() def log_signal(self, symbol: str, signal: str, confidence: float, price: float, strategy_details: Optional[Dict[str, Any]] = None): try: signal_data = { 'timestamp': datetime.now().isoformat(), 'symbol': symbol, 'signal': signal, 'confidence': confidence, 'price': price, 'strategy_details': strategy_details or {} } self.signal_logs.append(signal_data) if len(self.signal_logs) > 1000: self.signal_logs = self.signal_logs[-1000:] self._save_signals_to_file() self.logger.info(f"SIGNAL - {symbol}: {signal} (conf: {confidence:.2f}) at {price}") except Exception as e: self.logger.error(f"Error logging signal: {e}") def log_trade(self, symbol: str, side: str, qty: float, price: float, order_type: str = "MARKET", pnl: Optional[float] = None, reason: str = "signal"): try: trade_data = { 'timestamp': datetime.now().isoformat(), 'symbol': symbol, 'side': side, 'quantity': qty, 'price': price, 'order_type': order_type, 'pnl': pnl, 'reason': reason } self.trade_logs.append(trade_data) if len(self.trade_logs) > 1000: self.trade_logs = self.trade_logs[-1000:] self._save_trades_to_file() pnl_str = f" (PnL: {pnl:.2f})" if pnl is not None else "" self.logger.info(f"TRADE - {symbol}: {side} {qty} at {price}{pnl_str}") except Exception as e: self.logger.error(f"Error logging trade: {e}") def log_error(self, error_type: str, message: str, details: Optional[Dict[str, Any]] = None): try: error_data = { 'timestamp': datetime.now().isoformat(), 'type': error_type, 'message': message, 'details': details or {} } self.error_logs.append(error_data) if len(self.error_logs) > 500: self.error_logs = self.error_logs[-500:] self._save_errors_to_file() self.logger.error(f"{error_type.upper()}: {message}") except Exception as e: print(f"Critical logging error: {e}") def log_performance(self, symbol: str, period: str, metrics: Dict[str, Any]): try: perf_data = { 'timestamp': datetime.now().isoformat(), 'symbol': symbol, 'period': period, 'metrics': metrics } self._save_performance_to_file(perf_data) metrics_str = ", ".join([f"{k}: {v:.2f}" if isinstance(v, (int, float)) else f"{k}: {v}" for k, v in metrics.items()]) self.logger.info(f"PERFORMANCE - {symbol} ({period}): {metrics_str}") except Exception as e: self.logger.error(f"Error logging performance: {e}") def _save_signals_to_file(self): try: with open('logs/signals.json', 'w') as f: json.dump(self.signal_logs, f, indent=2) except Exception as e: self.logger.error(f"Error saving signals: {e}") def _save_trades_to_file(self): try: with open('logs/trades.json', 'w') as f: json.dump(self.trade_logs, f, indent=2) except Exception as e: self.logger.error(f"Error saving trades: {e}") def _save_errors_to_file(self): try: with open('logs/errors.json', 'w') as f: json.dump(self.error_logs, f, indent=2) except Exception as e: print(f"Error saving errors: {e}") def _save_performance_to_file(self, perf_data: Dict[str, Any]): try: with open('logs/performance.jsonl', 'a') as f: json.dump(perf_data, f) f.write('\n') except Exception as e: self.logger.error(f"Error saving performance: {e}") def get_recent_signals(self, limit: int = 50) -> list: return self.signal_logs[-limit:] def get_recent_trades(self, limit: int = 50) -> list: return self.trade_logs[-limit:] def get_error_summary(self) -> Dict[str, int]: summary = {} for error in self.error_logs: error_type = error.get('type', 'unknown') summary[error_type] = summary.get(error_type, 0) + 1 return summary def get_trade_statistics(self) -> Dict[str, Any]: if not self.trade_logs: return {'total_trades': 0, 'win_rate': 0.0, 'total_pnl': 0.0} total_trades = len(self.trade_logs) winning_trades = sum(1 for trade in self.trade_logs if trade.get('pnl', 0) > 0) total_pnl = sum(trade.get('pnl', 0) for trade in self.trade_logs) return { 'total_trades': total_trades, 'win_rate': winning_trades / total_trades if total_trades > 0 else 0.0, 'total_pnl': total_pnl, 'avg_pnl_per_trade': total_pnl / total_trades if total_trades > 0 else 0.0 } logger_instance = ScalperLogger() def log(message: str, level: str = "info"): if level.lower() == "debug": logger_instance.logger.debug(message) elif level.lower() == "info": logger_instance.logger.info(message) elif level.lower() == "warning": logger_instance.logger.warning(message) elif level.lower() == "error": logger_instance.logger.error(message) elif level.lower() == "critical": logger_instance.logger.critical(message) def log_signal(symbol: str, signal: str, confidence: float, price: float, strategy_details: Optional[Dict[str, Any]] = None): logger_instance.log_signal(symbol, signal, confidence, price, strategy_details) def log_trade(symbol: str, side: str, qty: float, price: float, order_type: str = "MARKET", pnl: Optional[float] = None, reason: str = "signal"): logger_instance.log_trade(symbol, side, qty, price, order_type, pnl, reason) def log_error(error_type: str, message: str, details: Optional[Dict[str, Any]] = None): logger_instance.log_error(error_type, message, details) def log_performance(symbol: str, period: str, metrics: Dict[str, Any]): logger_instance.log_performance(symbol, period, metrics)