【Python量化实战全攻略 #01】实时行情接口全拆解:盘中快照·五档盘口·逐笔一次跑通(附代码)
摘要:【Python量化实战全攻略 #01】实时行情接口全拆解:盘中快照·五档盘口·逐笔一次跑通(附代码) 系列:《Python量化实战全攻略》|连载项目 · 纯 GET 取数 · 仅依赖
系列:《Python量化实战全攻略》|连载项目 · 纯 GET 取数 · 仅依赖 requests
适用:想做盘中策略 / 盘口监控,但实时接口字段嵌套、五档顺序总搞反、全市场快照一拉就限频的读者;数据由智兔数服提供。本篇给沪深 A 股「实时交易 / 五档盘口 / 逐笔交易 / 全市场快照」9 个端点的分组地图、一次跑通的代码、五档与逐笔的解法、三个真实踩坑点,全部只依赖 requests,所有示例均为演示数据,不构成投资建议。
1. 你将得到什么
读完这一篇,你能拿走四样东西:
- 一张分组地图:9 个实时端点按「单票实时 / 全市场快照」分成 2 组,知道什么数据该敲哪个门;
- 一次跑通的代码:
/hs/real/ssjy拿单票快照,/hs/real/five拿五档盘口; - 五档盘口的解法:买卖五档是嵌套数组,要按档位展开,别把买一卖一顺序搞反;
- 三个真实踩坑点,都是第一次用几乎一定会踩的。
代码全部自包含,复制进 .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 线接口:日线、分钟线、技术指标与复权处理。