64 lines
2.2 KiB
Python
64 lines
2.2 KiB
Python
"""统一日志模块。
|
||
|
||
提供 setup_logger 函数，所有模块共用统一格式和输出目标：
|
||
- 格式：2025-12-20 18:05:30 [INFO] data_loader.py:42 - 消息内容
|
||
- 输出：同时打印到 stdout 和写入 results/logs/app.log
|
||
"""
|
||
import logging
|
||
import os
|
||
from logging.handlers import RotatingFileHandler
|
||
|
||
_LOGGER_CACHE: dict[str, logging.Logger] = {}
|
||
|
||
|
||
class DataLoaderFilter(logging.Filter):
|
||
"""过滤 data_loader 的 WARNING 和 ERROR 日志，不在控制台显示。"""
|
||
|
||
def filter(self, record: logging.LogRecord) -> bool:
|
||
# 如果是 data_loader 模块的警告或错误，不在控制台显示
|
||
if "data_loader" in record.filename and record.levelno >= logging.WARNING:
|
||
return False
|
||
return True
|
||
|
||
|
||
def setup_logger(name: str) -> logging.Logger:
|
||
"""创建或获取指定名称的 Logger 实例。
|
||
|
||
所有 logger：
|
||
- 级别：INFO
|
||
- 格式：时间 + 级别 + 文件名 + 行号 + 消息
|
||
- 输出：stdout + results/logs/app.log
|
||
"""
|
||
if name in _LOGGER_CACHE:
|
||
return _LOGGER_CACHE[name]
|
||
|
||
logger = logging.getLogger(name)
|
||
logger.setLevel(logging.INFO)
|
||
logger.propagate = False
|
||
|
||
if not logger.handlers:
|
||
# 日志目录位于 strategy_backtest/results/logs
|
||
log_dir = os.path.join("results", "logs")
|
||
os.makedirs(log_dir, exist_ok=True)
|
||
log_path = os.path.join(log_dir, "app.log")
|
||
|
||
fmt = "%(asctime)s [%(levelname)s] %(filename)s:%(lineno)d - %(message)s"
|
||
datefmt = "%Y-%m-%d %H:%M:%S"
|
||
formatter = logging.Formatter(fmt=fmt, datefmt=datefmt)
|
||
|
||
# 控制台输出（过滤 data_loader 的警告和错误）
|
||
ch = logging.StreamHandler()
|
||
ch.setLevel(logging.INFO)
|
||
ch.setFormatter(formatter)
|
||
ch.addFilter(DataLoaderFilter()) # 添加过滤器
|
||
logger.addHandler(ch)
|
||
|
||
# 文件输出（带滚动）
|
||
fh = RotatingFileHandler(log_path, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8")
|
||
fh.setLevel(logging.INFO)
|
||
fh.setFormatter(formatter)
|
||
logger.addHandler(fh)
|
||
|
||
_LOGGER_CACHE[name] = logger
|
||
return logger
|