免费领取证书

← 返回博客列表

【Python量化实战 #19】盘中盯盘脚本：实时行情监控与涨跌提醒实现

2026年08月05日 09:32 · 内容龙虾

本文是「Python量化实战」系列第 19 篇（工程化模块）。前面我们把数据批量拉下来、存进 SQLite、画成 K 线——但这些都是"盘后复盘"。真正想盘中抓住机会，需要让脚本盯着行情跑：自选股涨到/跌到阈值就自动提醒。本文用 60 行核心代码搭一个可运行的盘中盯盘器，实时轮询 + 涨跌幅阈值 + 本地告警日志，生产环境还能直接接到钉钉/企微 webhook。

本文你将得到什么

  1. 一个完整可运行的盘中盯盘脚本（轮询、阈值、告警）
  2. 用 stock_real_time 拉取实时行情的真实代码与返回字段解读
  3. 告警通道的最小可复现版本：本地 JSONL 日志 + webhook 结构
  4. 实盘部署的 4 个注意点（频率、异常、日志、去重）
  5. 下篇预告：#20 把"稳定获取"这件事讲透——重试、缓存、防封 IP

一、在线体验

想先在线试试接口效果？打开 API Playground 即可直接调用测试： https://mairuiapi.com/playground

本文用到 stock_real_time 单票实时行情接口，返回最新价、涨跌幅、成交量等字段，盘中监控主要靠它。

二、环境准备

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

pip install mairui

SDK 的完整接口文档与使用说明请查阅 GitHub 仓库：https://github.com/MaiRuiApi/mairui

接口的详细参数说明请查阅官网 API 文档：https://mairuiapi.com/hsdata

运行环境： - Python 3.9+ - mairui SDK 1.0.0 - pandas 2.2+（可选，本文未强制使用）

import os
import mairui

# 证书从环境变量读取（官网注册后获取），不要写死在代码里
api = mairui.Client(os.environ["MAIRUI_LICENCE"])

本文数据截至 2026-07-31 盘中，读者复现时行情会不同，但代码逻辑不变。

三、为什么需要盘中盯盘脚本

手动刷行情有三大问题：

痛点 后果
盯盘耗时 策略信号来了，你还在刷其他页面
情绪干扰 涨跌一刺激就临时改计划
反应慢 涨停/跌停发生在几秒之内，人工来不及

Python 盯盘脚本的价值是把规则前置：提前设好阈值，让程序 7×24 秒级监控，触发才通知你。你只在"有情况"时看一眼，而不是一直盯着屏幕。

四、实时行情接口 stock_real_time

先单独跑一次接口，看返回什么：

import os, mairui

api = mairui.Client(os.environ["MAIRUI_LICENCE"])
snap = api.stock_real_time("600519")
print(snap)

真实输出（2026-07-31 盘中）：

{
  'pe': 15.25,
  'ud': -33.59,
  'pc': -2.4667,
  'zf': 0.7512,
  'tr': 0.17,
  'pb_ratio': 7.04,
  'p': 1328.17,
  'o': 1330.03,
  'h': 1336,
  'l': 1325.77,
  'yc': 1361.76,
  'cje': 2875668300,
  'v': 21622,
  'pv': 2162161,
  'tv': 5,
  't': '2026-07-31 09:55:26'
}

关键字段： - p：最新价 - pc：涨跌幅（%） - ud：涨跌额 - o/h/l：开盘价/最高价/最低价 - v/cje：成交量/成交额 - t：数据时间戳

监控逻辑只需要两件事：定时轮询 + 比较 pc 与阈值。

五、核心代码：自选股池 + 阈值告警

import os, sys, time, json, logging
from datetime import datetime
from pathlib import Path

import mairui

# 自选股池：每只股票可设不同阈值
WATCHLIST = [
    {"code": "600519", "name": "贵州茅台", "alarm_pct": 2.0},
    {"code": "000001", "name": "平安银行", "alarm_pct": 1.5},
    {"code": "000333", "name": "美的集团", "alarm_pct": 1.5},
    {"code": "002594", "name": "比亚迪",   "alarm_pct": 2.5},
]

POLL_INTERVAL = 5   # 轮询间隔（秒），实盘建议 15~60 秒

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)

ALERT_LOG = Path("alerts.jsonl")


def send_alert(stock, snapshot):
    """把告警写入本地 JSONL；生产环境可改为钉钉/企微 webhook。"""
    msg = {
        "time": datetime.now().isoformat(),
        "code": stock["code"],
        "name": stock["name"],
        "price": round(snapshot.get("p"), 2),
        "pct": snapshot.get("pc"),
        "text": f"{stock['name']}({stock['code']}) 异动：现价 ...，涨跌幅 {snapshot.get('pc'):+.2f}%",
    }
    with ALERT_LOG.open("a", encoding="utf-8") as f:
        f.write(json.dumps(msg, ensure_ascii=False) + "\n")
    logging.warning(msg["text"])


def main():
    api = mairui.Client(os.environ["MAIRUI_LICENCE"])
    while True:
        for stock in WATCHLIST:
            snap = api.stock_real_time(stock["code"])
            pct = snap.get("pc", 0.0)
            logging.info(f"{stock['name']} 价={snap.get('p')} 涨跌幅={pct:+.2f}%")
            if abs(pct) >= stock["alarm_pct"]:
                send_alert(stock, snap)
        time.sleep(POLL_INTERVAL)


if __name__ == "__main__":
    main()

关键解释： - WATCHLIST 是配置区，股票代码、名称、阈值一目了然，方便增减。 - abs(pct) >= alarm_pct 同时监控大涨和大跌。 - send_alert() 目前写本地 alerts.jsonl，避免演示依赖外部 webhook；接入钉钉/企微只需把 msg post 到对应的 webhook URL。

六、完整可运行示例

把上面拼成一个端到端脚本，并限制演示跑 4 轮后退出：

# -*- coding: utf-8 -*-
"""盘中盯盘脚本：自选股实时行情监控 + 涨跌幅阈值提醒。"""
import os
import sys
import time
import json
import logging
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "_common"))
import mairui
from mairui_helper import load_licence

WATCHLIST = [
    {"code": "600519", "name": "贵州茅台", "alarm_pct": 2.0},
    {"code": "000001", "name": "平安银行", "alarm_pct": 1.5},
    {"code": "000333", "name": "美的集团", "alarm_pct": 1.5},
    {"code": "002594", "name": "比亚迪",   "alarm_pct": 2.5},
]
POLL_INTERVAL = 5
MAX_ROUNDS = 4

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
ALERT_LOG = Path("alerts.jsonl")


def send_alert(stock, snapshot):
    msg = {
        "time": datetime.now().isoformat(),
        "code": stock["code"],
        "name": stock["name"],
        "price": round(snapshot.get("p"), 2),
        "pct": snapshot.get("pc"),
        "threshold": stock["alarm_pct"],
        "text": (f"{stock['name']}({stock['code']}) 异动：现价 {snapshot.get('p')}，"
                 f"涨跌幅 {snapshot.get('pc'):+.2f}%，超过阈值 ±{stock['alarm_pct']}%"),
    }
    with ALERT_LOG.open("a", encoding="utf-8") as f:
        f.write(json.dumps(msg, ensure_ascii=False) + "\n")
    logging.warning(msg["text"])


def main():
    api = mairui.Client(load_licence())
    logging.info(f"盯盘启动，监控 {len(WATCHLIST)} 只股票")
    for _ in range(MAX_ROUNDS):
        for stock in WATCHLIST:
            snap = api.stock_real_time(stock["code"])
            pct = snap.get("pc", 0.0)
            logging.info(f"{stock['name']}({stock['code']}) 价={round(snap.get('p'), 2)} 涨跌幅={pct:+.2f}%")
            if abs(pct) >= stock["alarm_pct"]:
                send_alert(stock, snap)
        time.sleep(POLL_INTERVAL)


if __name__ == "__main__":
    main()

真实运行输出（2026-07-31 09:57 验证，节选）：

2026-07-31 09:57:18,600 [INFO] 盯盘启动，监控 4 只股票
2026-07-31 09:57:18,678 [INFO] 贵州茅台(600519)  价=1326.93  涨跌幅=-2.56%
2026-07-31 09:57:18,680 [WARNING] 贵州茅台(600519) 异动：现价 1326.93，涨跌幅 -2.56%，超过阈值 ±2.0%
2026-07-31 09:57:18,701 [INFO] 平安银行(000001)  价=11.32  涨跌幅=-2.50%
2026-07-31 09:57:18,701 [WARNING] 平安银行(000001) 异动：现价 11.32，涨跌幅 -2.50%，超过阈值 ±1.5%
2026-07-31 09:57:18,723 [INFO] 美的集团(000333)  价=85.58  涨跌幅=-3.79%
2026-07-31 09:57:18,723 [WARNING] 美的集团(000333) 异动：现价 85.58，涨跌幅 -3.79%，超过阈值 ±1.5%
2026-07-31 09:57:18,744 [INFO] 比亚迪(002594)  价=94.47  涨跌幅=-1.43%

本次演示中，茅台、平安银行、美的集团均触发阈值告警，比亚迪未触发。告警记录写入 alerts.jsonl：

{"time": "2026-07-31T09:57:18.679462", "code": "600519", "name": "贵州茅台", "price": 1326.93, "pct": -2.5577, "threshold": 2.0, "text": "贵州茅台(600519) 异动：现价 1326.93，涨跌幅 -2.56%，超过阈值 ±2.0%"}

七、避坑与进阶

  • 坑 1：轮询太频繁触发限流。实盘建议间隔 ≥15 秒；短周期量化可走专业行情通道或 1 分钟 K 线接口，不要无节制轮询实时接口。
  • 坑 2：盘中异常断线没处理。api.stock_real_time() 可能因网络抖动抛异常，实盘应包 try/except 并记录失败，避免脚本崩溃。
  • 坑 3：同一告警反复轰炸。当前逻辑每轮都会重新告警，生产环境应维护"已告警状态"，同一股票 5 分钟内只发一次。
  • 坑 4：把脚本跑在本地笔记本上。本地电脑休眠/断网会导致漏监控；更稳的姿势是丢到服务器用 systemd/supervisor 常驻。
  • 进阶方向：
  • 接入钉钉/企微 webhook：把 send_alert() 里的写文件改成 requests.post(webhook_url, json=msg)。
  • 加入成交量异动、涨速（最近 N 秒价格变化）等多维度告警。
  • 结合 #16 的并发能力，把自选股池扩展到几十只同时监控。

八、总结与延伸

本文把"盘后看"升级到了"盘中盯"：

  • 用 stock_real_time 获取实时行情
  • 用自选股池 + 阈值实现规则化监控
  • 用 JSONL + webhook 结构演示告警通道

工程化模块还剩最后一篇：#20 专门讲如何稳定、合规地获取股票数据，从爬虫被封 IP 的痛点切入，给出重试、缓存、异常处理、API 方案等完整稳定性实践。

延伸阅读： - 在线体验更多接口：https://mairuiapi.com/playground - 查看完整 API 文档：https://mairuiapi.com/hsdata - SDK 文档与源码：https://github.com/MaiRuiApi/mairui - 关注公众号获取本系列更新

本文为技术演示，所有数据来自公开股票接口，文中监控方案不构成投资建议。

相关阅读

QQ 客服 3826425416 咨询时请提供证书号或订单号，便于快速处理
咨询