你已经在电脑前搭好了盯盘脚本，涨幅一到阈值就弹桌面通知。问题是——你不可能 24 小时坐在电脑前。 lunch 时间、通勤路上、睡前刷手机，才是真正需要知道持仓异动的时候。

这篇文章解决一个问题：让 Python 把盘中告警直接推到你手机上。我们会用到一个轻量级方案：本地行情筛选 + webhook 推送。覆盖三种最常见的通道：Server酱（个人微信）、企业微信机器人、钉钉机器人。读完你能拿到一段可直接运行的脚本，替换自己的 key 就能收消息。

一、本文你将得到什么

1. 一段用麦蕊 SDK 拉取自选实时行情的代码（批量接口，10 只 1 次额度）。
2. 一个可扩展的「规则引擎」：大涨、大跌、高振幅任意组合。
3. Server酱 / 企业微信 / 钉钉三种 webhook 推送函数。
4. 本地 mock 验证方法：不用真实 key 也能确认请求格式正确。
5. 一个避坑清单：为什么你的机器人发了消息却没收到。

二、环境准备

本文代码使用 mairui SDK 获取股票数据，安装方法如下：

pip install mairui

运行环境：

• Python 3.9+
• mairui SDK 1.0.0
• requests（用于发 webhook）

本文数据截至 2026-08-07，读者复现时数据可能因行情更新而略有差异。

证书直接写在代码里（文中使用演示证书，演示证书（读者复制即用，无需替换））。

三、三种推送通道怎么选

通道	适用场景	需要准备	消息格式
Server酱	个人微信，最轻量	SendKey	title + desp
企业微信机器人	小团队/工作室	群机器人 Webhook	{"msgtype":"text","text":{"content":"..."}}
钉钉机器人	已用钉钉办公	群机器人 Webhook	同企业微信

三种通道的代码结构几乎一样：构造一段文本，POST 到对应的 URL。差别只在 URL 和 payload。

四、行情拉取：批量接口省额度

我们要监控的自选股如下（10 只）：

WATCHLIST = [
    "000001", "000002", "000858", "002594", "300750",
    "600519", "601318", "600036", "000333", "002415",
]

用 stock_ssjy_more 批量接口一次拉完，只计 1 次额度：

from mairui import Client

def fetch_watchlist(codes):
    api = Client(licence="LICENCE-66D8-9F96-0C7F0FBCD073")  # 演示证书，演示证书（读者复制即用，无需替换）
    return api.stock_ssjy_more(stock_codes=",".join(codes))

records = fetch_watchlist(WATCHLIST)
print(f"返回 {len(records)} 只")

关键解释：

• stock_ssjy_more 一次最多支持 20 只，单次调用只计 1 次额度。
• 返回字段中，pc 是涨跌幅（%），zf 是振幅（%），p 是现价，cje 是成交额。
• 免费版 500 次/日，10 只一次只需要 1 次，足够盘中反复轮询。

五、规则引擎：大涨、大跌、高振幅

把行情记录丢进规则引擎，命中任意一条就产生告警：

RULES = [
    {"name": "大涨", "field": "pc", "op": ">=", "threshold": 2.0},
    {"name": "大跌", "field": "pc", "op": "<=", "threshold": -2.0},
    {"name": "高振幅", "field": "zf", "op": ">=", "threshold": 3.0},
]

def check_alerts(records):
    hits = []
    for r in records:
        for rule in RULES:
            val = r.get(rule["field"])
            if val is None:
                continue
            triggered = (
                (rule["op"] == ">=" and val >= rule["threshold"]) or
                (rule["op"] == "<=" and val <= rule["threshold"])
            )
            if triggered:
                hits.append({
                    "dm": r["dm"], "name": r.get("mc", r["dm"]),
                    "p": r["p"], "pc": r["pc"], "zf": r["zf"],
                    "cje": r["cje"], "rule": rule["name"],
                })
    return hits

关键解释：

• field 可以是 pc（涨跌幅）或 zf（振幅）。
• 阈值写成 2.0 / -2.0 / 3.0，盘中正常波动不会频繁触发；如果你想更敏感，改成 1.0 / -1.0 / 1.5。
• 规则引擎只负责生成告警对象，具体推送到哪由下一节决定。

六、三种推送函数

6.1 Server酱（推送到个人微信）

import requests

def notify_serverchan(title, desp, sendkey="<你的SendKey>"):
    url = f"https://sctapi.ftqq.com/{sendkey}.send"
    return requests.post(url, data={"title": title, "desp": desp}, timeout=10)

6.2 企业微信机器人

def notify_wecom(text, webhook_url="<你的企业微信机器人Webhook>"):
    return requests.post(
        webhook_url,
        json={"msgtype": "text", "text": {"content": text}},
        timeout=10,
    )

6.3 钉钉机器人

def notify_dingtalk(text, webhook_url="<你的钉钉机器人Webhook>"):
    return requests.post(
        webhook_url,
        json={"msgtype": "text", "text": {"content": text}},
        timeout=10,
    )

关键解释：

• 三个函数都只做一件事：把文本 POST 出去。
• sendkey 和 webhook_url 必须替换成你自己的，代码中保持占位符，不要泄露。
• 企业微信/钉钉的 payload 结构几乎一致，只是 URL 不同。

七、完整可运行示例

下面是把拉取、筛选、推送串起来的完整脚本。为了让你不用真的配置 webhook 也能跑通，我在本地起了一个 mock webhook 服务器，验证请求格式是否正确。

from __future__ import annotations
import json, os, sys, threading, time
from http.server import BaseHTTPRequestHandler, HTTPServer


import requests
from mairui import Client

WATCHLIST = [
    "000001", "000002", "000858", "002594", "300750",
    "600519", "601318", "600036", "000333", "002415",
]

RULES = [
    {"name": "大涨", "field": "pc", "op": ">=", "threshold": 2.0},
    {"name": "大跌", "field": "pc", "op": "<=", "threshold": -2.0},
    {"name": "高振幅", "field": "zf", "op": ">=", "threshold": 3.0},
]

# ── 本地 mock webhook 服务器 ──
class MockHandler(BaseHTTPRequestHandler):
    def log_message(self, fmt, *args): pass
    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        self.server.last_body = self.rfile.read(length).decode("utf-8")
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"ok")

def start_mock_server(port=18081):
    server = HTTPServer(("127.0.0.1", port), MockHandler)
    server.last_body = ""
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server

# ── 行情与规则 ──
def fetch_watchlist(codes):
    api = Client(licence="LICENCE-66D8-9F96-0C7F0FBCD073")  # 演示证书，演示证书（读者复制即用，无需替换）
    return api.stock_ssjy_more(stock_codes=",".join(codes))

def check_alerts(records):
    hits = []
    for r in records:
        for rule in RULES:
            val = r.get(rule["field"])
            if val is None: continue
            ok = (rule["op"] == ">=" and val >= rule["threshold"]) or \
                 (rule["op"] == "<=" and val <= rule["threshold"])
            if ok:
                hits.append(r | {"rule": rule["name"]})
    return hits

def format_msg(hit):
    direction = "涨" if hit["pc"] >= 0 else "跌"
    return (
        f"【{hit['rule']}】{hit.get('mc', hit['dm'])}({hit['dm']})\n"
        f"现价：{hit['p']}  {direction}{hit['pc']:.2f}%  振幅{hit['zf']:.2f}%\n"
        f"成交额：{hit['cje']/1e8:.2f}亿"
    )

# ── 推送函数（占位符 key/url） ──
def notify_serverchan(title, desp, sendkey="<你的SendKey>"):
    url = f"https://sctapi.ftqq.com/{sendkey}.send"
    return requests.post(url, data={"title": title, "desp": desp}, timeout=10)

def notify_wecom(text, webhook_url="<你的企业微信机器人Webhook>"):
    return requests.post(webhook_url, json={
        "msgtype": "text", "text": {"content": text}
    }, timeout=10)

def notify_dingtalk(text, webhook_url="<你的钉钉机器人Webhook>"):
    return requests.post(webhook_url, json={
        "msgtype": "text", "text": {"content": text}
    }, timeout=10)

# ── 主流程 ──
if __name__ == "__main__":
    server = start_mock_server()
    records = fetch_watchlist(WATCHLIST)
    hits = check_alerts(records)
    print(f"命中 {len(hits)} 条告警")
    for h in hits[:3]:
        msg = format_msg(h)
        notify_wecom(msg, webhook_url="http://127.0.0.1:18081/webhook")
        time.sleep(0.05)
        body = json.loads(server.last_body)
        print(body["text"]["content"][:60], "...")

运行结果示例：

命中 14 条告警
【大跌】平安银行(000001) 现价：11.17  跌-0.89%  振幅1.06% ...
【高振幅】万科A(000002) 现价：3.20  跌-1.84%  振幅2.45% ...

注：本地 mock 只验证请求格式；真实使用时把 webhook_url 换成企业微信/钉钉机器人地址，把 sendkey 换成 Server酱 key。

八、常见坑与进阶

1. 为什么发了消息手机没收到？

• 企业微信/钉钉机器人需要在群里 @机器人 或开启「自定义关键词」白名单。
• Server酱的 SendKey 分免费版与付费版，免费版有每日条数限制。
• 网络超时常见：把 timeout 设成 10 秒以上，并加 try/except。

2. 阈值设多少合适？

• 自选股盯盘：涨跌幅 ±2%、振幅 3% 起步，避免消息轰炸。
• 短线异动：涨跌幅 ±3%、振幅 5%。
• 持仓告警：针对你自己的持仓设单边阈值，比如跌破 -2% 提醒止损。

3. 盘中一直轮询会不会很耗额度？

• 用 stock_ssjy_more 批量接口，20 只一次只计 1 次额度。
• 每 60 秒轮询一次，4 小时交易时段 = 240 次，免费版 500 次/日足够。
• 把轮询放到服务器或本地常驻脚本里，见本系列 #16。

4. 能不能推图片/K线？

• 企业微信/钉钉支持 markdown、图片、link 等 msgtype，把 text 换成对应格式即可。
• 图片需要先上传到图床或机器人支持的媒体接口。

九、小结 + 下篇预告

本文把「行情拉取 → 规则筛选 → 手机推送」串成了一条可运行的链路。核心就三步：

1. 用 stock_ssjy_more 批量拉取自选行情，省额度。
2. 用规则引擎筛选大涨/大跌/高振幅。
3. 把告警文本通过 Server酱/企业微信/钉钉 webhook 发到手机。

下篇 #12 盘中指数快照：沪指深指创业板异动监控，我们会把监控对象从个股切换到指数，一屏看全大盘温度。

---

本文为技术演示，不构成投资建议。文中示例数据仅供接口用法参考，不代表实时行情。