用结构化日志给量化脚本装上「黑匣子」：logging + JSON + 崩溃回溯的十分钟落地

写量化脚本的人大多经历过同一个场景：早上打开电脑，发现昨晚的定时任务没跑完，终端只剩一句 KeyError: 'close'。没有上下文、没有输入、不知道跑到第几只票。于是只能把脚本重跑一遍，祈祷它别再出错。

问题不在于 Python 报错，而在于日志写得不像证据。本文用一个不到 100 行的改造，把普通脚本升级成「出事之后能自证」的状态：结构化字段、崩溃自动留痕、关键步骤耗时可查。

一、为什么 print 不够用

print 有三个致命缺陷：

1. 没有时间戳——你不知道它是 03:00 还是 03:40 打出来的；
2. 没有严重级别——警告和错误混在一起，grep 不出来；
3. 没有结构化字段——处理 600519 完成 这种文本没法被程序消费，想统计"哪只票最慢"只能靠肉眼。

量化场景尤其吃亏：一次复盘要跑几百只标的、几千次网络请求，任何一个环节偶发失败，事后都需要精确定位。

二、十行配置，把 logging 换成 JSON 输出

核心思路：用标准库 logging 打点，用一个自定义 Formatter 把每条日志序列化成一行 JSON（即 JSON Lines，每行一条，方便 jq 和后续入库）。

import json, logging, time, traceback
from datetime import datetime, timezone

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
        }
        # 业务字段通过 extra= 传入，例如 extra={"symbol": "600519"}
        for k, v in record.__dict__.items():
            if k not in logging.LogRecord.__dict__ and not k.startswith("_"):
                payload.setdefault(k, v)
        if record.exc_info:
            payload["traceback"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False)

def setup_logger(name="quant", path="run.log"):
    lg = logging.getLogger(name)
    lg.setLevel(logging.INFO)
    lg.handlers.clear()
    fh = logging.FileHandler(path, encoding="utf-8")
    fh.setFormatter(JsonFormatter())
    lg.addHandler(fh)
    sh = logging.StreamHandler()
    sh.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
    lg.addHandler(sh)
    return lg

关键点在 for k, v in record.__dict__.items() 这一段：凡是通过 extra= 传进来的字段都会被自动带进 JSON，业务代码只需要写：

log.info("fetch ok", extra={"symbol": code, "rows": len(df), "cost_ms": 120})

不需要手工拼字符串，字段天然可被检索。

三、给每个关键步骤装上耗时探针

量化脚本最常问的两个问题是"哪一步慢"和"哪一步挂"。用 contextmanager 包一层就能同时解决：

from contextlib import contextmanager

@contextmanager
def step(log, name, **fields):
    t0 = time.perf_counter()
    log.info("step start", extra={"step": name, **fields})
    try:
        yield
    except Exception:
        cost = int((time.perf_counter() - t0) * 1000)
        log.error("step failed", extra={"step": name, "cost_ms": cost, **fields}, exc_info=True)
        raise
    else:
        cost = int((time.perf_counter() - t0) * 1000)
        log.info("step done", extra={"step": name, "cost_ms": cost, **fields})

用法非常自然：

with step(log, "fetch_kline", symbol=code):
    df = fetch_daily(code)
with step(log, "compute_factor", symbol=code):
    score = factor(df)

一旦某只票失败，日志里会同时留下：失败步骤名、耗时、标的、完整堆栈。不需要重跑。

四、顶层兜底：让崩溃也变成一条日志

最后在 main() 外面套一层，把未捕获异常统一记下来再退出：

def main():
    ...

if __name__ == "__main__":
    log = setup_logger()
    try:
        main()
    except Exception:
        log.critical("run aborted", exc_info=True)
        raise
    finally:
        logging.shutdown()

配套一条排查命令，出事当天直接跑：

# 看今天所有失败步骤及耗时
jq -c 'select(.level=="ERROR") | {ts, step, symbol, cost_ms}' run.log | tail -20

# 统计各步骤平均耗时，找性能瓶颈
jq -r 'select(.step and .cost_ms) | [.step, .cost_ms] | @tsv' run.log \
  | awk '{a[$1]+=$2; c[$1]++} END {for (k in a) printf "%-16s %6.1f ms  (%d 次)\n", k, a[k]/c[k], c[k]}'

五、落地后的三个真实收益

• 故障定位从"重跑一遍"变成"一条命令"：以前定位一次偶发失败要 20 分钟，现在 30 秒能拿到堆栈和上下文；
• 性能瓶颈可见：上面的 awk 统计第一次跑出来，我就发现 90% 的时间花在一个没加缓存的因子计算上，补上缓存后整轮从 8 分钟降到 1 分 40 秒；
• 日志可直接入库：一行一个 JSON，FileBeat / Logstash 或者简单的 pandas.read_json(lines=True) 都能直接吃，后续做监控面板不用二次解析。

六、几个容易踩的坑

• 别在日志里写敏感信息：结构化字段方便检索，也意味着一旦日志外泄，标的、持仓、接口 token 全都暴露。token 之类请走环境变量，并且绝不打进日志；
• 文件句柄要轮转：长期运行的脚本用 RotatingFileHandler(maxBytes=20*1024*1024, backupCount=5)，否则几个月后单文件能涨到几个 G；
• extra= 字段名别撞车：logging 内置了 msg、args、levelname 等保留名，用它们做 key 会抛 KeyError，建议业务字段统一加前缀，比如 sym_、cost_。

小结

这套改造不引入任何第三方依赖，十分钟内可以落到任意一个已有脚本上。它的价值不在于"日志更好看"，而在于把一次不可复现的失败，变成一份可检索的证据。对量化这种"跑完就走、出错就得查"的场景，这可能是投入产出比最高的一次改造。