

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

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

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

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

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

## 二、本文你将得到什么

1. **3 类核心监控指标**：回撤异常 / 数据新鲜度 / 任务心跳
2. **多通道告警推送**：邮件（dry-run / 真实 SMTP）+ 桌面通知（plyer / stderr fallback）
3. **证书 fallback + 重试**：内部证书失效自动切换演示证书，监控流程不中断
4. **结构化日志输出**：JSON 摘要便于对接 Prometheus / Grafana

## 三、整体架构

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

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

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

```python
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）

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

```python
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
- 节假日/周末之后忘了重跑
- 数据库锁死导致新数据写不进去

```python
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）

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

```python
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` 表，每跑一次写一行。心跳超时比"数据过期"更早触发，能在数据出问题前就预警调度失败。

## 六、多通道推送策略

| 通道 | 适用场景 | 失败降级 |
|------|----------|----------|
| 邮件 | 事后追溯、批量告警合并发送 | SMTP 未配置 → 日志打印 |
| 桌面推送 | 立即响应（开发 / 交易时段） | plyer 未安装 → stderr 高亮 |
| 企业微信机器人 | 团队共享、移动端提醒 | webhook 未配置 → 邮件降级 |
| 钉钉 / Slack | 同上 | 同上 |

**生产环境推荐组合**：桌面推送（实时）+ 邮件（汇总）+ 企业微信（团队可见）。**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 小时前刚跑）

## 八、常见坑

1. **告警风暴**：阈值设太严 → 每分钟都触发 → 运营疲劳后忽略所有告警。建议加 `alert_dedup_key`（如「同一股票同一告警类型 30 分钟内只发一次」）。
2. **监控本身没被监控**：监控脚本也是代码，也会挂。**监控服务的健康度要由第二个独立脚本盯**（如「监控进程心跳写 redis，cron 检查 redis 是否超时」）。
3. **plyer 在 Linux 服务器上不可用**：桌面推送只对 Windows / macOS 开发机有意义。生产 Linux 服务建议改用企业微信机器人或 PagerDuty。
4. **numpy bool 类型被 json 序列化为字符串**：本 demo 已用 `bool()` 强制转 Python 原生类型，避免下游解析报错。

## 九、小结

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

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

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

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

---

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

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


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

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

人工智能

量化系统「沉默失败」监控告警实战：三大核心指标（最大回撤、数据新鲜度、任务心跳）自动检测异常，邮件+桌面双通道推送，证书失效自动降级重试，JSON日志对接Prometheus/Grafana。附可复制Python代码，解决无人值守故障漏报问题。

回归测试

代码解释

数据库

服务器

Windows

Python

Linux

JSON

2026上云采购 | AI焕新·智启新局

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

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

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

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

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐