← 返回博客列表

【Python量化实战全攻略 #01】实时行情接口全拆解:盘中快照·五档盘口·逐笔一次跑通(附代码)

2026年10月12日 08:37 · 智兔数服 · Python量化实战全攻略

摘要:【Python量化实战全攻略 #01】实时行情接口全拆解:盘中快照·五档盘口·逐笔一次跑通(附代码) 系列:《Python量化实战全攻略》|连载项目 · 纯 GET 取数 · 仅依赖

系列:《Python量化实战全攻略》|连载项目 · 纯 GET 取数 · 仅依赖 requests
适用:想做盘中策略 / 盘口监控,但实时接口字段嵌套、五档顺序总搞反、全市场快照一拉就限频的读者;数据由智兔数服提供。本篇给沪深 A 股「实时交易 / 五档盘口 / 逐笔交易 / 全市场快照」9 个端点的分组地图、一次跑通的代码、五档与逐笔的解法、三个真实踩坑点,全部只依赖 requests,所有示例均为演示数据,不构成投资建议。

1. 你将得到什么

读完这一篇,你能拿走四样东西:

  1. 一张分组地图:9 个实时端点按「单票实时 / 全市场快照」分成 2 组,知道什么数据该敲哪个门;
  2. 一次跑通的代码:/hs/real/ssjy 拿单票快照,/hs/real/five 拿五档盘口;
  3. 五档盘口的解法:买卖五档是嵌套数组,要按档位展开,别把买一卖一顺序搞反;
  4. 三个真实踩坑点,都是第一次用几乎一定会踩的。

代码全部自包含,复制进 .py 直接能跑,不依赖 numpy / pandas。

2. 本篇取数约定

  • 全部接口都是 GET + 路径参数,token 放在查询串里(?token=xxx);
  • 统一基址 https://api.zhituapi.com;
  • 代码块里的 你的智兔token 是占位符,换成你的 token 即可;
  • 所有接口路径均取自官方已验证文档,跨篇零重复。
  • 数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。

3. 9 个端点分 2 组

先建立地图。实时交易类一共 9 个端点,按用途分:

组 端点 用途 更新频率
单票实时 /hs/real/ssjy/{code} 单票实时交易(价 / 量 / 涨跌) 盘中实时
单票实时 /hs/real/zbjy/{code} 单票当天逐笔交易(主买 / 主卖) 盘中实时
单票实时 /hs/real/zbjy/{date}/{code} 指定日期的逐笔交易回放 盘后
单票实时 /hs/real/time/{code} 券商数据源实时交易(字段更细) 盘中实时
单票实时 /hs/real/five/{code} 买卖五档盘口 盘中实时
全市场 /hs/public/realall 全市场实时快照(公开源) 盘中实时
全市场 /hs/public/ssjymore 批量指定代码取快照(公开源) 盘中实时
全市场 /hs/custom/realall 全市场实时快照(券商源) 盘中实时
全市场 /hs/custom/ssjymore 批量指定代码取快照(券商源) 盘中实时

code 为 6 位证券代码,如 000001;date 为 YYYYMMDD,如 20260813。特色市场(北交所 /bj/...、指数 /hz/...)有同名接口,本篇先聚焦沪深 A 股,详见后续手册。

4. 核心模板函数

import requests, time

BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token"

# ---------- 1. 字段容错与类型归一 ----------
def _hit_key(d, *cands, default=None):
    """字段容错:接口偶发大小写/中英文混用时,按顺序取第一个非空值"""
    if not isinstance(d, dict):
        return default
    for c in cands:
        if c in d and d[c] not in (None, "", "-", "null"):
            return d[c]
    low = {str(k).lower(): v for k, v in d.items()}
    for c in cands:
        v = low.get(str(c).lower())
        if v not in (None, "", "-", "null"):
            return v
    return default


def _to_float(v, default=None):
    try:
        if v in (None, "", "-", "null", "None"):
            return default
        return float(v)
    except (TypeError, ValueError):
        return default


# ---------- 2. 统一请求:重试 + 退避 + 降级 ----------
def _get(path, timeout=10, retries=2, backoff=0.6, default=None):
    """返回 JSON;失败重试 retries 次仍失败则返回 {'_error': 原因}"""
    q = {"token": TOKEN}
    last = ""
    for i in range(retries + 1):
        try:
            r = requests.get(BASE + path, params=q, timeout=timeout)
            if r.status_code == 200:
                try:
                    return r.json()
                except ValueError:
                    return default
            last = "HTTP %s %s" % (r.status_code, (r.text or "").strip()[:80])
        except Exception as e:
            last = "%s: %s" % (type(e).__name__, e)
        if i < retries:
            time.sleep(backoff * (i + 1))
    return {"_error": last}


# ---------- 3. 取数封装 ----------
def fetch_tick(code):       return _get("/hs/real/ssjy/%s" % code, default={})
_MOCK_FIVE = None
def fetch_five(code):
    if _MOCK_FIVE is not None:
        return _MOCK_FIVE
    return _get("/hs/real/five/%s" % code, default={})
def fetch_trade(code, date=None):
    path = "/hs/real/zbjy/%s" % date if date else "/hs/real/zbjy"
    return _get(path, default=[])
def fetch_time(code):       return _get("/hs/real/time/%s" % code, default={})
def fetch_all_market():     return _get("/hs/public/realall", default=[])
def fetch_custom():         return _get("/hs/custom/realall", default=[])


# ---------- 4. 五档盘口展开 ----------
def five_levels(code):
    d = fetch_five(code)
    if isinstance(d, dict) and "_error" in d:
        return None, d["_error"]
    bids = _hit_key(d, "bids", "buy", "委买", default=[])
    asks = _hit_key(d, "asks", "sell", "委卖", default=[])
    return {"bids": bids[:5], "asks": asks[:5]}, None


# ---------- 5. 校验 ----------
def run_check():
    # 1) 字段容错:大小写混用也能取到
    assert _hit_key({"Code": "000001", "NAME": "平安银行"}, "code") == "000001"
    assert _to_float("-") is None and _to_float("11.85") == 11.85

    # 2) 五档展开:买卖各取前 5 档(用演示数据替换网络调用)
    global _MOCK_FIVE
    _MOCK_FIVE = {"bids": [[1, 10.10, 1000] for _ in range(5)],
                  "asks": [[1, 10.20, 800] for _ in range(5)]}
    lv, err = five_levels("000001")
    _MOCK_FIVE = None
    assert err is None and len(lv["bids"]) == 5 and len(lv["asks"]) == 5

    # 3) 历史逐笔回放接口可调用
    assert fetch_trade("000001", "20240813") is not None

    print("校验通过")


if __name__ == "__main__":
    run_check()
    print("-" * 62)
    code = "000001"
    for name, fn in [("单票快照", lambda: fetch_tick(code)),
                     ("五档盘口", lambda: fetch_five(code)),
                     ("当天逐笔", lambda: fetch_trade(code)),
                     ("券商实时", lambda: fetch_time(code)),
                     ("全市场快照", fetch_all_market),
                     ("券商全市场", fetch_custom)]:
        data = fn()
        if isinstance(data, dict) and "_error" in data:
            print("%-10s -> %s" % (name, data["_error"][:60]))
        else:
            print("%-10s -> %d 条" % (name, len(data) if isinstance(data, list) else 1))

5. 跑通示例

把上面的代码复制到本地,填入你的 智兔token 即可直接运行:传入股票代码,five_levels 展开买卖五档,fetch_trade 还能拉指定日期的逐笔回放(各字段含义见前文各小节)。

6. 坑与注意事项

坑 1:实时接口盘中才有数据,盘后返回空或昨收快照。
/hs/real/ssjy、/hs/real/five 都是盘中实时推送,收盘后要么返回空、要么返回昨收静态值。做盘后回测别直接调实时接口,要用 /hs/real/zbjy/{date}/{code} 的盘后逐笔回放,或用历史行情接口(下一篇讲)。

坑 2:五档盘口是嵌套数组,档位顺序别反。
/hs/real/five 返回 bids(买盘,价格从低到高)和 asks(卖盘,价格从高到低),每档是 [档位, 价格, 数量]。做买卖价差要确认买一在 bids[0]、卖一在 asks[0],别把数组顺序搞反导致价差算负。

坑 3:全市场快照量大,别每次全量拉。
/hs/public/realall 一次返回全市场几千只股票的实时快照,数据量大、频率高的话容易触发限频。做全市场监控建议用 /hs/custom/realall(券商源子集)或 /hs/public/ssjymore(指定代码列表)按需拉,别无脑全量轮询。

7. 小结与下篇预告

本篇把沪深 A 股 9 个实时端点分成 2 组,five_levels 展开买卖五档,_hit_key 处理盘口字段嵌套。

下一篇预告:《【Python量化实战全攻略 #02】历史K线接口全拆解:日线·分钟线·技术指标·复权处理(附代码)》:用 /hs/history·/hs/pro·/hs/history/{macd|ma|boll|kdj} 一组接口,把日线 / 分钟线 / 技术指标 / 分红复权一次取全,并给出 pandas 多周期聚合与手工前复权代码。

8. 免责声明

本文仅演示沪深 A 股实时行情数据的取数方法,所有代码示例均为演示数据,未含任何真实行情数值,不构成投资建议,亦不承诺收益。


关于数据来源:本文接口来自智兔数服(www.zhituapi.com)官方文档,覆盖沪深A股、指数、基金、可转债等 200+ 已验证接口路径,注册可领免费 token 做研究测试。
更多实战手册见技术博客:zhituapi.com/blog.html。
本系列《Python量化实战全攻略》共 32 篇,下一篇(#02)讲历史 K 线接口:日线、分钟线、技术指标与复权处理。

想亲自试一下?免费获取证书