技术分享

【Python量化实战 #03】Python实时股票行情接口怎么用?mairui秒级行情与批量订阅

盘中监控、条件选股、告警机器人——都离不开实时行情接口。本文是「Python量化实战」第 3 篇(#03),演示如何用 mairui 在几行代码内拿到单票与批量实时数据。

安装与证书请参考「Python量化实战」#01 或 https://www.mairuiapi.com/sdk

单标的实时行情

from mairui import Client

with Client(licence="你的证书UUID") as api:
    # 多数实时接口使用 6 位代码
    tick = api.stock_real_time("000001")
    print(tick)

    # 五档等(以文档字段为准)
    five = api.stock_real_five("000001")
    print(five)

stock_ssjy 也是常用实时入口,可按业务选择:

quote = api.stock_ssjy("000001")

多代码批量:减少往返

codes = ["000001", "600519", "300750"]

with Client(licence="你的证书UUID") as api:
    # 多码合并请求(若接口支持)
    data = api.stock_ssjy_more(codes)
    print(data)

    # 或用 map 并发单票接口
    ticks = api.map(api.stock_real_time, codes, max_workers=8)
    for c, t in zip(codes, ticks):
        print(c, t)

全市场类接口(如 stock_ssjy_all / stock_real_all)流量与耗时更大,请结合证书限额与业务必要性使用,详情见 API 文档

做一个「简易涨跌监控」

import time
from mairui import Client

WATCH = ["000001", "600519"]
LICENCE = "你的证书UUID"

def last_price(row):
    # 字段名请按实际返回调整,常见可能是 p / price / 最新价 等
    if isinstance(row, dict):
        for k in ("p", "price", "zxj", "last"):
            if k in row:
                return row[k]
    return row

with Client(licence=LICENCE) as api:
    while True:
        for code in WATCH:
            q = api.stock_real_time(code)
            print(time.strftime("%H:%M:%S"), code, last_price(q))
        time.sleep(3)  # 注意频率限制,勿过密轮询

生产环境请加异常捕获、退避与告警;遵守接口 请求频率限制,避免触发限流。

实时 vs 历史:怎么选?

  • 盘中决策 / 盯盘:实时接口
  • 回测 / 复盘:历史 K 线(见「Python量化实战」#02)
  • 不确定字段:先用 Playground 看原始 JSON

限额与稳定性小贴士

  • 免费证有日调用上限,监控脚本请算清「每秒请求 × 标的数 × 时长」。
  • SDK 已对部分 5xx/429 做有限重试,但仍建议业务层做熔断。
  • 网络抖动时设置合理 timeout(构造 Client 时可覆盖默认)。

下一篇预告

第 4 篇(#04):技术指标 API——直接取 MACD、KDJ、均线,少在本地重复造轮子。

关键词:Python 实时股票行情、股票行情 API、盘中监控、stock_real_time、批量行情接口、麦蕊智数。

麦蕊智数 MyData 团队 · Python量化实战 #03