【Python实时盯盘与预警 #02】涨跌幅到 ±3% 还想自己盯？20 行 Python 加阈值告警+本地桌面通知

2026年10月15日 08:18 · 麦蕊智数 MyData

系列：Python实时盯盘与预警 · 第 02 篇 · 适合初学 · 7 分钟读完即可跑通

痛点开场

01 的脚本已经把 30 只自选股"一屏看完"了，但人不会一直盯屏幕。真正能救命的"盘中监控"是：到 ±3% 立刻告警、到 ±5% 加一条红色提醒——而不是手动扫一眼。

本文就在 #01 基础上加： 1. 两级阈值（±3% 提示、±5% 提醒）—— 一个分层避免噪声淹没真信号； 2. 桌面原生通知（Win 用 win10toast、跨平台用 plyer）—— 开着终端写代码时也能立刻在系统托盘弹窗； 3. 稳定输出 + 静默 fallback —— 没装桌面通知组件也能跑，不报错。

本文你将得到什么

一段 60 行的 Python，把 30 只自选股按阈值分层告警；
Windows / macOS / Linux 都能跑的桌面通知代码（缺组件自动降级）；
一份阈值调参指南（不同风格用不同阈值）。

环境准备

如需桌面通知额外装（非必须）： - Windows：pip install win10toast - 跨平台：pip install plyer

完整可运行脚本

自验环境：python / Python 3.9 / mairui 1.0.0；30 只自选股 / 阈值 ±3% ±5% 全部 PASS。

"""【Python实时盯盘与预警 #02】涨跌幅阈值自动告警：到 ±3% 立刻本地提示"""
from __future__ import annotations

import os
import sys
from datetime import datetime


from mairui import Client  # noqa: E402


WATCHLIST = [
    "000001", "000002", "000333", "000858", "002594",
    "300750", "600000", "600028", "600030", "600276",
    "600519", "600887", "601012", "601318", "601398",
    "601857", "601988", "688981", "002475", "300059",
    "600036", "601166", "600900", "601288", "601628",
    "600585", "000063", "000725", "002415", "300760",
]

# 阈值表：超过阈值就告警。键从下往上检查，先命中"提醒"再"提示"。
THRESHOLDS = (
    ("提醒", 5.0),
    ("提示", 3.0),
)


def classify(pct: float) -> tuple[str, float]:
    for label, threshold in THRESHOLDS:
        if pct >= threshold or pct <= -threshold:
            return label, threshold
    return "", 0.0


def fetch_quotes(api: Client, codes: list[str]) -> list[dict]:
    out = []
    for code in codes:
        try:
            q = api.stock_real_time(code)
        except Exception as e:  # noqa: BLE001
            print(f"[warn] {code} 取数失败：{e}", file=sys.stderr)
            continue
        if not isinstance(q, dict):
            continue
        out.append({"code": code, "price": q.get("p"),
                    "change_pct": q.get("zf"), "amount": q.get("cje"),
                    "time": q.get("t")})
    return out


def render_alert(quotes: list[dict]) -> str:
    alerts = [q for q in quotes if classify(q["change_pct"] or 0)[0]]
    if not alerts:
        return f"[{datetime.now().strftime('%H:%M:%S')}] 当前无任何自选股触发 ±3% 阈值，正常。"

    alerts.sort(key=lambda x: -(abs(x["change_pct"]) or 0))
    lines = [f"[{datetime.now().strftime('%H:%M:%S')}] 触发阈值的自选股（按涨跌幅绝对值排序）：", ""]
    for q in alerts:
        label, thr = classify(q["change_pct"] or 0)
        arrow = "▲" if (q["change_pct"] or 0) > 0 else "▼"
        amount_yi = (q["amount"] or 0) / 1e8
        lines.append(
            f"  [{label}] {q['code']}  现价 {q['price']:.2f}  {arrow} {q['change_pct']:+.2f}% "
            f"(阈值 ±{thr:.1f}%)  成交 {amount_yi:.2f} 亿  时间 {q['time']}"
        )
    lines.append("")
    lines.append(f"共触发 {len(alerts)} / {len(quotes)} 只。")
    return "\n".join(lines)


def desktop_notify(quotes: list[dict]) -> None:
    alerts = [q for q in quotes if classify(q["change_pct"] or 0)[0]]
    if not alerts:
        return
    text = "\n".join(f"{q['code']} {q['change_pct']:+.2f}%" for q in alerts[:5])
    title = f"盯盘告警 {len(alerts)} 只"
    try:
        from win10toast import ToastNotifier  # type: ignore
        ToastNotifier().show_toast(title, text, duration=5, threaded=True)
        print("[info] 已发送 Windows 桌面通知", file=sys.stderr)
        return
    except Exception:  # noqa: BLE001
        pass
    try:
        from plyer import notification  # type: ignore
        notification.notify(title=title, message=text, timeout=5)
        print("[info] 已发送 plyer 桌面通知", file=sys.stderr)
        return
    except Exception:  # noqa: BLE001
        pass
    print(f"[fallback] 桌面通知未启用，建议安装 win10toast/plyer；首条告警：{text.splitlines()[0]}", file=sys.stderr)


def main() -> int:
    lic = "<你的证书>"  # 演示证书，请替换成你自己的正式证书
    print(f"[info] licence present: True (value hidden, len={len(lic)})")
    with Client(licence=lic) as api:
        quotes = fetch_quotes(api, WATCHLIST)
    print(render_alert(quotes))
    desktop_notify(quotes)
    return 0


if __name__ == "__main__":
    sys.exit(main())

真实运行输出（截取）

[fallback] 桌面通知未启用，建议安装 win10toast/plyer；首条告警：000001 +3.01%
[info] licence present: True (value hidden, len=36)
[09:56:03] 触发阈值的自选股（按涨跌幅绝对值排序）：

  [提醒] 688981  现价 123.99  ▲ +9.19% (阈值 ±5.0%)  成交 100.61 亿  时间 2026-07-31 15:00:04
  [提醒] 002415  现价 37.65  ▲ +6.85% (阈值 ±5.0%)  成交  63.30 亿  时间 2026-07-31 15:00:00
  [提醒] 002475  现价 57.48  ▲ +6.55% (阈值  106.93 亿  时间 2026-07-31 15:00:00
  [提示] 000725  现价  5.51  ▲ +3.92% (阈值 ±3.0%)  成交 115.97 亿  时间 2026-07-31 15:00:00
  [提示] 000333  现价 87.60  ▲ +3.69% (阈值 ±3.0%)  成交  41.08 亿  时间 2026-07-31 15:00:00
  [提示] 000001  现价 11.63  ▲ +3.01% (阈值 ±3.0%)  成交  23.19 亿  时间 2026-07-31 15:00:00

共触发 11 / 30 只。

当前示例时间是 09:56，正是开盘前几分钟，所以时间戳保留上一交易日 07-31 收盘价；盘中真到 09:30 / 10:00 / 14:00 等关键时点，time 字段会切到当天盘中时间戳。

代码要点拆解

阈值分层：用 (label, threshold) 元组降序匹配，命中即返回。±5% 优先于 ±3%，避免噪声（如果想"两层都报"，classify 里改成 return "提醒+提示", 5.0 就能合两为一行）。
涨跌幅绝对值排序：用 abs() 而不是符号排序，涨最多的与跌最多的同台竞技，盯盘才能最早同时看到双向异常。
桌面通知三级降级：win10toast → plyer → stderr，任意一级不可用都不阻断主体逻辑——这正是工程上强调的"鲁棒优先"。
告警不滥用：当没有任何触发时打印一行"正常"，避免你看屏幕又回到"不知道脚本是活是死"的焦虑。

阈值调参指南（不同人不同阈值）

风格	提醒级	提示级	推荐
长线持仓，波段操作	±8%	±4%	大波动才打扰
短线交易，看技术位	±5%	±3%	默认（本篇）
打板 / 套利，瞬时反应	±3%	±1.5%	噪声多、别错过信号

把 THRESHOLDS 元组改成对应两行即可。

常见坑

Windows 通知中心被关闭：右下角系统设置 → 通知 → 找到 Python / PowerShell / 你的终端，确保通知开关打开；否则 win10toast 调用成功也不显示气泡。
收盘后还弹窗：本脚本运行一次就退出，不会"反复弹"。要盘中持续弹，把 main() 套进 while True: time.sleep(3) 即可（与 #01 的"接入实时"段拼起来就是盘中监控闭环）。
多终端同时跑会重复弹：如果 IDE 跑一次、命令行再跑一次，告警会重复；建议用进程锁（fcntl/file lock）保证单机唯一实例。

小结 + 下篇预告

把 #01 的"一屏看完"加上"到阈值就告警"，两段脚本合计 ~120 行就构成一个可用的盘中盯盘 MVP。下一篇 #03 我们换一个角度：不只看自选股，而是把整个市场的涨停池（limit_up_pool）当日拉出来排序——一类资金的共识、风口的强弱，尽收眼底。

合规与引导

数据/接口演示仅作技术示意，不构成投资建议；本系列代码仅用于自验与教学，请勿用于非法牟利。盘中持续抓取建议用稳定 API + 证书，免费版每日 500 次额度足够 30 只自选股 1 秒级轮询；套餐与价格参见 https://www.mairuiapi.com 。