【Python量化系统工程实战 #03】任务跑挂了没人知道？用 60 行代码搭一套「监控 + 邮件 + 桌面推送」告警系统

系列前情：#02 我们搭了一条「自动触发 + 入库 + 去重」的调度流水线。但流水线上线后第 3 天凌晨数据库写挂了、你睡了 8 小时没收到任何提醒——这是量化系统最容易翻车的环节。本篇解决「异常怎么被自动发现 + 主动告知你」。

一、为什么必须做监控告警

自动化流水线跑得越久，越容易出现「沉默失败」：

• 调度任务进程被 OOM Killer 干掉，无人值守
• 证书过期导致拉数据全部失败，库里全是过期 K 线
• API 限流让数据停在 3 天前，策略还在用旧数据发信号
• 数据源某只股票字段缺失，回测结果偏离真实

这些故障不会主动通知你。等你第 2 天打开终端发现 PnL 异常时，已经损失了一整个交易日的决策机会。告警系统不是锦上添花，是量化系统的安全气囊。

二、本文你将得到什么

三、整体架构

run_monitor()
├── fetch_recent(stock) # 拉最近 K 线（含 fallback）
├── check_drawdown(df) # 监控 1：最大回撤
├── check_freshness(df) # 监控 2：数据新鲜度
├── check_heartbeat(db) # 监控 3：任务心跳
└── push(alerts) # 邮件 + 桌面通知

三块检查互不依赖，每块都能独立打开 / 关闭。推送层做「邮件 + 桌面」双通道是因为不同告警需要不同响应速度：邮件适合事后追溯，桌面推送适合立即响应。

四、核心代码（可复制即用）

import sqlite3, time, json, logging, sys, os
from datetime import datetime

from mairui import Client
import pandas as pd

DEMO_LICENCE = "LICENCE-66D8-9F96-0C7F0FBCD073"
STOCK = "600519"
DD_THRESHOLD = 5.0 # 最大回撤 > 5% 告警
FRESH_DAYS = 3 # 数据新鲜度超过 3 个交易日告警
HEARTBEAT_HOURS = 24 # 任务心跳超过 24 小时告警

logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[logging.FileHandler("monitor.log", encoding="utf-8"), logging.StreamHandler()]
)
log = logging.getLogger("monitor")

def fetch_recent(stock: str, licence: str) –> pd.DataFrame:
"""拉最近 60 个交易日（含重试）"""
for attempt in range(2):
try:
client = Client(licence)
raw = client.stock_history(stock, "d", "f")
if raw:
df = pd.DataFrame(raw)
df["date"] = pd.to_datetime(df["t"])
df["daily_ret"] = df["c"] / df["pc"] – 1.0
return df
except Exception as e:
log.warning(f"[retry {attempt+1}] {str(e)[:60]}")
time.sleep(0.5)
return pd.DataFrame()

def check_drawdown(df: pd.DataFrame) –> dict:
"""监控 1：最大回撤"""
if df.empty:
return {"alert": False, "reason": "no data"}
df = df.sort_values("date").reset_index(drop=True)
cum = (1 + df["daily_ret"]).cumprod()
dd = float((cum / cum.cummax() – 1).min() * 100)
alert = bool(dd < –DD_THRESHOLD)
return {"alert": alert, "max_drawdown": round(dd, 2), "threshold": –DD_THRESHOLD}

def check_freshness(df: pd.DataFrame) –> dict:
"""监控 2：数据新鲜度"""
if df.empty:
return {"alert": True, "reason": "empty data", "fresh_days": 999}
last_date = df["date"].max()
fresh_days = (datetime.now() – last_date).days
return {
"alert": bool(fresh_days > FRESH_DAYS),
"last_date": str(last_date.date()),
"fresh_days": fresh_days,
"threshold_days": FRESH_DAYS,
}

def check_heartbeat(db_path: str) –> dict:
"""监控 3：最后任务日志距今"""
if not os.path.exists(db_path):
return {"alert": True, "reason": "db not found"}
conn = sqlite3.connect(db_path)
row = conn.cursor().execute(
"SELECT run_at, status, note FROM task_log ORDER BY id DESC LIMIT 1"
).fetchone()
conn.close()
if not row:
return {"alert": True, "reason": "no task log"}
last_run = datetime.strptime(row[0], "%Y-%m-%d %H:%M:%S")
hours = (datetime.now() – last_run).total_seconds() / 3600
return {
"alert": bool(hours > HEARTBEAT_HOURS),
"last_run_at": row[0],
"hours_since": round(hours, 1),
"threshold_hours": HEARTBEAT_HOURS,
}

def send_desktop(title: str, message: str) –> dict:
"""桌面推送（plyer 不可用时降级到 stderr）"""
try:
from plyer import notification
notification.notify(title=title, message=message, timeout=10)
return {"ok": True, "mode": "plyer"}
except Exception:
log.warning(f"[DESKTOP ALERT] {title}: {message}")
return {"ok": True, "mode": "stderr"}

def push(alerts: list) –> dict:
if not alerts:
return {"pushed": 0, "alerts": []}
title = f"量化监控告警 {len(alerts)} 条"
body = "\\n".join([f"• {a['type']}: {a['detail']}" for a in alerts])
return {"pushed": len(alerts), "desktop": send_desktop(title, body), "alerts": alerts}

def main():
df = fetch_recent(STOCK, DEMO_LICENCE)
log.info(f"[fetch] {STOCK} {len(df)} 行")

dd_res = check_drawdown(df)
fresh_res = check_freshness(df)
hb_res = check_heartbeat("kline_pipeline.db")

log.info(f"[check] 回撤: dd={dd_res.get('max_drawdown', 'N/A')}% alert={dd_res['alert']}")
log.info(f"[check] 新鲜度: fresh_days={fresh_res.get('fresh_days')} alert={fresh_res['alert']}")
log.info(f"[check] 心跳: hours={hb_res.get('hours_since')} alert={hb_res['alert']}")

alerts = []
if dd_res["alert"]:
alerts.append({"type": "回撤异常", "detail": f"DD={dd_res['max_drawdown']}% 阈值={dd_res['threshold']}%"})
if fresh_res["alert"]:
alerts.append({"type": "数据过期", "detail": f"fresh_days={fresh_res['fresh_days']} 阈值={FRESH_DAYS}天"})
if hb_res["alert"]:
alerts.append({"type": "任务心跳超时", "detail": f"hours={hb_res['hours_since']} 阈值={HEARTBEAT_HOURS}小时"})

push_res = push(alerts)
log.info(f"[push] {len(alerts)} 条告警 | desktop={push_res['desktop']['mode']}")

summary = {"stock": STOCK, "data_rows": len(df),
"checks": {"drawdown": dd_res, "freshness": fresh_res, "heartbeat": hb_res},
"alerts": alerts, "push_result": push_res}
with open("summary.json", "w", encoding="utf-8") as f:
json.dump(summary, f, ensure_ascii=False, indent=2, default=str)
print("[ok] summary.json saved")

if __name__ == "__main__":
main()

五、三类监控详解

监控 1：最大回撤（drawdown）

回撤超过阈值 → 策略可能进入"持续亏损区间" → 立刻告警让运营介入。

cum = (1 + df["daily_ret"]).cumprod() # 累计净值
dd = float((cum / cum.cummax() – 1).min()) # 历史最大回撤

阈值设定经验：

• 短周期策略（日内 / 5 分钟）：阈值 2%~3%
• 中频策略（日线 / 持仓 1 周）：阈值 5%~8%
• 低频策略（月线 / 持仓 1 月）：阈值 10%~15%

阈值过松会漏报，过严会噪声告警。建议先按"过去 6 个月最大回撤 × 1.2"起步，再根据告警频次调整。

监控 2：数据新鲜度（freshness）

数据没在更新是最常见的"沉默失败"：

• API 限流、证书过期、采集进程被 kill
• 节假日/周末之后忘了重跑
• 数据库锁死导致新数据写不进去

last_date = df["date"].max()
fresh_days = (datetime.now() – last_date).days
alert = fresh_days > FRESH_DAYS

阈值设定：比"调度频率"略宽（如 cron 每 30 分钟跑一次，阈值设 1 天；每天 15:30 跑一次，阈值设 3 个交易日）。

监控 3：任务心跳（heartbeat）

不管数据是否更新，先确认"任务确实在跑"：

SELECT run_at, status FROM task_log ORDER BY id DESC LIMIT 1
hours = (now – last_run).total_seconds() / 3600
alert = hours > HEARTBEAT_HOURS

依赖 #02 的 task_log 表，每跑一次写一行。心跳超时比"数据过期"更早触发，能在数据出问题前就预警调度失败。

六、多通道推送策略

生产环境推荐组合：桌面推送（实时）+ 邮件（汇总）+ 企业微信（团队可见）。demo 演示只用桌面 + stderr fallback——plyer 不是必装依赖，缺失时自动降级到 stderr 不影响监控逻辑。

七、实测：演示证书下的告警触发

[fetch] 600519 50 行
[check] 回撤: dd=-2.81% alert=False # 50 日窗口最大回撤未越线
[check] 新鲜度: fresh_days=282 alert=True # 演示证书最后数据 = 2025-12-01
[check] 心跳: hours=0.1 alert=False # #02 流水线刚跑过
[push] 1 条告警 | desktop=stderr

解读：

• 回撤监控正常（-2.81% < 5%）
• 数据过期告警必然触发 —— 演示证书 K 线截止到 2025-12-01，距今 282 天。生产环境使用付费证书后此告警消失
• 任务心跳正常（#02 demo 0.1 小时前刚跑）

八、常见坑

九、小结

本文给量化系统加了「沉默失败」自动检测能力。三个核心监控项：

• 最大回撤：策略层面的健康度
• 数据新鲜度：基础设施是否在更新
• 任务心跳：调度器是否还在跑

推送通道用「桌面 + 邮件」组合，开发期桌面立刻响应、生产期邮件可追溯。配合 #02 的 task_log 表，整个监控闭环就成型了。

下一篇（#04）我们解决「策略上线前的最后一道关卡」——做完整的回归测试 + 模拟盘验证 + 风险敞口检查，把量化系统从「能跑」带到「敢跑」。

免责声明：本文仅供技术学习交流，不构成任何投资建议。量化策略回测表现不代表未来收益，投资有风险，决策需谨慎。

代码与文档：https://github.com/MaiRuiApi