【零依赖量化数据实战 #31】沪深实时盘口全景:逐笔·全盘实时·最新价·历史逐笔
摘要:【零依赖量化数据实战 #31】沪深实时盘口全景:逐笔·全盘实时·最新价·历史逐笔 系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK
系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK
适用:想用 Python 把沪深的逐笔交易、全盘实时、最新价、历史逐笔与停价等盘口数据一次拉齐的量化爱好者;数据由智兔数服提供,不依赖任何行情终端。
1. 你将得到什么
- 10 个官方接口的最小可用封装,分四组:
- 单只盘口(4,带
{股票代码}):/hs/real/zbjy/{code}逐笔交易、/hs/indicators/{code}指标汇总、/hs/instrument/{code}合约信息、/hs/stopprice/history/{code}停价历史。 - 全盘实时(4,无参):
/hs/public/realall、/hs/public/ssjymore、/hs/custom/realall、/hs/custom/ssjymore。 - 历史逐笔(1,带
{股票代码}):/hs/history/transaction/{code}。 - 最新价(1,带
{代码.市场}/{分时级别}/{除权方式}):/hs/latest/{code}.{market}/{level}/{right}。 - 一个对字段名不敏感的排名函数
rank_by:按候选键(如最新价/price/last)降序取前 N。
2. 端点语义表
GET https://api.zhituapi.com/hs/real/zbjy/000001.SZ?token=你的智兔token -> 单只逐笔交易
GET https://api.zhituapi.com/hs/public/realall?token=你的智兔token -> 全盘实时(公开)
GET https://api.zhituapi.com/hs/public/ssjymore?token=你的智兔token -> 全盘实时更多(公开)
GET https://api.zhituapi.com/hs/custom/realall?token=你的智兔token -> 全盘实时(自定义)
GET https://api.zhituapi.com/hs/custom/ssjymore?token=你的智兔token -> 全盘实时更多(自定义)
GET https://api.zhituapi.com/hs/history/transaction/000001.SZ?token=你的智兔token -> 历史逐笔
GET https://api.zhituapi.com/hs/latest/000001.SZ/SZ/d/n?token=你的智兔token -> 最新价(代码.市场/级别/除权)
GET https://api.zhituapi.com/hs/stopprice/history/000001.SZ?token=你的智兔token -> 停价历史
GET https://api.zhituapi.com/hs/indicators/000001.SZ?token=你的智兔token -> 指标汇总
GET https://api.zhituapi.com/hs/instrument/000001.SZ?token=你的智兔token -> 合约信息
鉴权:token 走查询参数;{股票代码}、{代码.市场}/{分时级别}/{除权方式} 是路径参数。数据来自 智兔数服(www.zhituapi.com)。
3. 字段名不固定?用候选键命中
盘口返回的「最新价」可能叫 最新价 / price / last。统一候选键命中:
def _hit_key(d, keys):
if not isinstance(d, dict):
return None
for k in keys:
if k in d and d[k] is not None:
return d[k]
low = {str(x).lower(): x for x in d.keys()}
for k in keys:
kl = k.lower()
if kl in low:
return d[low[kl]]
return None
4. 核心模板函数
import sys, requests
BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token" # 占位,换成你申请的真实 token
def _hit_key(d, keys):
if not isinstance(d, dict):
return None
for k in keys:
if k in d and d[k] is not None:
return d[k]
low = {str(x).lower(): x for x in d.keys()}
for k in keys:
kl = k.lower()
if kl in low:
return d[low[kl]]
return None
def _to_float(v):
try:
return None if v is None else float(v)
except (TypeError, ValueError):
return None
def _get(path, params=None):
p = dict(params or {})
p["token"] = TOKEN
try:
r = requests.get(f"{BASE}{path}", params=p, timeout=10)
except Exception as e:
return None, f"网络异常:{e}"
if r.status_code != 200:
return None, f"{r.status_code} {r.text.strip()[:140]}"
try:
return r.json(), None
except Exception:
return None, f"非 JSON:{r.text.strip()[:140]}"
# 单只逐笔交易
def fetch_zbjy(code):
return _get(f"/hs/real/zbjy/{code}")
# 全盘实时(realall / ssjymore)
def fetch_public(kind):
return _get(f"/hs/public/{kind}")
# 自定义全盘实时(realall / ssjymore)
def fetch_custom(kind):
return _get(f"/hs/custom/{kind}")
# 历史逐笔
def fetch_hist_tx(code):
return _get(f"/hs/history/transaction/{code}")
# 最新价(代码.市场/分时级别/除权方式)
def fetch_latest(code, market, level, right):
return _get(f"/hs/latest/{code}.{market}/{level}/{right}")
# 停价历史
def fetch_stopprice(code):
return _get(f"/hs/stopprice/history/{code}")
# 技术指标汇总
def fetch_indicators(code):
return _get(f"/hs/indicators/{code}")
# 合约/工具信息
def fetch_instrument(code):
return _get(f"/hs/instrument/{code}")
def rank_by(rows, keys, descending=True, topn=None):
if not isinstance(rows, list):
return rows
def sc(x):
return _to_float(_hit_key(x, keys)) or 0.0
out = sorted(rows, key=sc, reverse=descending)
return out[:topn] if topn else out
def selftest():
# 合成数据仅逻辑自验,非真实行情
rows = [
{"code": "000001.SZ", "最新价": 12.3},
{"code": "600000.SH", "price": 7.8},
{"code": "300750.SZ", "last": 200.0},
]
top = rank_by(rows, ["最新价", "price", "last"], topn=2)
assert [x["code"] for x in top] == ["300750.SZ", "000001.SZ"], top
for kind in ("realall", "ssjymore"):
assert kind in ("realall", "ssjymore")
print("selftest PASS")
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "--selftest":
selftest()
else:
print("zbjy ->", fetch_zbjy("000001.SZ"))
for kind in ("realall", "ssjymore"):
print(f"public.{kind} ->", fetch_public(kind))
print(f"custom.{kind} ->", fetch_custom(kind))
print("hist_tx ->", fetch_hist_tx("000001.SZ"))
print("latest ->", fetch_latest("000001.SZ", "SZ", "d", "n"))
print("stopprice ->", fetch_stopprice("000001.SZ"))
print("indicators ->", fetch_indicators("000001.SZ"))
print("instrument ->", fetch_instrument("000001.SZ"))
5. 代码自验结果
离线 selftest(合成数据,仅验证逻辑,不含任何真实行情):
selftest PASS
联网实测(占位 token,真实返回):
--- 联网实测(占位 token,预期 404 102:Licence证书不存在)---
zbjy -> (None, '404 102:Licence证书(你的智兔token)不存在')
public.realall -> (None, '404 102:Licence证书(你的智兔token)不存在')
custom.realall -> (None, '404 102:Licence证书(你的智兔token)不存在')
public.ssjymore -> (None, '404 102:Licence证书(你的智兔token)不存在')
custom.ssjymore -> (None, '404 102:Licence证书(你的智兔token)不存在')
hist_tx -> (None, '404 102:Licence证书(你的智兔token)不存在')
latest -> (None, '404 102:Licence证书(你的智兔token)不存在')
stopprice -> (None, '404 102:Licence证书(你的智兔token)不存在')
indicators -> (None, '404 102:Licence证书(你的智兔token)不存在')
instrument -> (None, '404 102:Licence证书(你的智兔token)不存在')
把
TOKEN = "你的智兔token"换成你申请的真实 token,上述函数即可打印沪深实时盘口数据。本文未编造任何真实数值。
6. 坑与注意事项
- 102 不代表路径对:
404 102是「证书不存在」(鉴权先于路由),路径合法与否要靠客户端白名单自查。 - 代码带市场后缀:
zbjy/history/transaction/indicators/instrument/stopprice的{股票代码}要带市场(如000001.SZ);latest是{代码.市场}/{级别}/{除权}三段路径参数。 - public vs custom:
/hs/public/*是公开全盘实时,/hs/custom/*是自定义范围全盘实时,返回结构相近但范围不同。 - 字段名中英文混用:「最新价」可能叫
最新价/price/last,务必候选键命中。
7. 小结与下篇预告
本篇把「沪深实时盘口」拧成了 10 个零依赖接口的最小封装,重点解决了代码带市场后缀、latest 三段路径参数、public/custom 全盘实时区分三个坑,配 rank_by 候选键排名即可一行出榜。
下一篇计划写 #32《沪深公司面补充:财务股东·股本·经营范围》:讲解如何用官方接口拉取沪深公司面剩余字段——流通股东(/hs/fin/flowholder)、户均(/hs/fin/hm)、上市天数(/hs/gs/sszs)、业绩预告(/hs/gs/yjyg)、股本变化(/hs/gs/gdbh)、经营范围(/hs/gs/jyfw)数据。
8. 免责声明
本文仅演示公开数据接口的用法,所有代码示例均以占位 token 自验,未含任何真实数据;文中合成数据仅为逻辑自验用途,不构成投资建议,亦不承诺收益。
免费领取证书
数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。
领取路径:进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」(邮箱验证 3 步即可拿到 token)。
把代码里的 你的智兔token 换成你拿到的真实 token,上面的脚本就能直接打印沪深实时盘口数据。