【零依赖量化数据实战 #02】沪深全量股票清单:批量拉实时做自选监控
摘要:【零依赖量化数据实战 #02】沪深全量股票清单:批量拉实时做自选监控 系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK 适用:想先拿到全市场股票代码,再批量拉实时行情做自选
系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK 适用:想先拿到全市场股票代码,再批量拉实时行情做自选股监控 / 预警的开发者
你将得到什么
- 用
GET /hs/list/all一次拿到沪深全市场股票清单(代码 + 名称) - 一个对返回结构「不敏感」的清单解析函数(列表里是字典还是字符串都能吃)
- 批量轮询实时行情的骨架,缺字段安全置
None不崩 - 完整可复制运行代码,换
你的智兔token即跑
一、先拉全量清单
智兔全市场股票清单接口(公开文档 hsdataapi.html 覆盖):
GET https://api.zhituapi.com/hs/list/all
返回可能是「代码列表」或「含名称的对象列表」,两种都兼容:
import requests
TOKEN = "你的智兔token" # ← 换成你的真实智兔 token
BASE = "https://api.zhituapi.com"
def fetch_list(token=TOKEN):
r = requests.get(f"{BASE}/hs/list/all", params={"token": token}, timeout=15)
r.raise_for_status()
return r.json()
def parse_list(raw):
"""清单返回结构不敏感:对象列表取 code,纯字符串列表直接用。"""
if isinstance(raw, list):
out = []
for it in raw:
if isinstance(it, dict):
out.append(it.get("code") or it.get("dm") or it.get("stock_code"))
else:
out.append(it)
return [c for c in out if c]
return []
二、批量拉实时做监控
拿到清单后,筛出自选批量拉实时。下面给一个最小可运行骨架,单只实时拉取复用上篇的 fetch_realtime:
import sys, time, requests
TOKEN = "你的智兔token" # ← 换成你的真实智兔 token
BASE = "https://api.zhituapi.com"
WATCH = ["600519", "000001", "300750", "600036", "002594"] # 自选股,真实场景从 /hs/list/all 筛
def fetch_list(token=TOKEN):
r = requests.get(f"{BASE}/hs/list/all", params={"token": token}, timeout=15)
r.raise_for_status()
return r.json()
def fetch_realtime(code, token=TOKEN):
r = requests.get(f"{BASE}/hs/real/ssjy/{code}", params={"token": token}, timeout=10)
r.raise_for_status()
return r.json()
def parse_list(raw):
if isinstance(raw, list):
out = []
for it in raw:
if isinstance(it, dict):
out.append(it.get("code") or it.get("dm") or it.get("stock_code"))
else:
out.append(it)
return [c for c in out if c]
return []
def snapshot(codes):
rows = []
for code in codes:
try:
d = fetch_realtime(code)
except Exception as e:
rows.append((code, None, None, f"异常:{e}"))
continue
rows.append((code, d.get("name"), d.get("p"), d.get("zf")))
time.sleep(0.2) # 礼貌限速,批量别打太猛
return rows
def selftest():
# 模拟 /hs/list/all 两种返回形态
assert parse_list([{"code": "600519", "name": "贵州茅台"},
{"code": "000001", "name": "平安银行"}]) == ["600519", "000001"]
assert parse_list(["600519", "000001"]) == ["600519", "000001"]
# 模拟实时快照缺字段不崩
d = {"name": "贵州茅台"} # 没有 p / zf
assert (d.get("name"), d.get("p"), d.get("zf")) == ("贵州茅台", None, None)
print("selftest PASS: 清单双形态解析 + 实时缺字段容错均正常")
def main():
if TOKEN == "你的智兔token":
print("[演示] 占位 token,下面请求返回 102 契约;请换真实 token。")
# 真实场景:codes = parse_list(fetch_list()) 再筛出自选
rows = snapshot(WATCH)
for code, name, price, pct in rows:
print(f"{code} {name}: price={price} pct={pct}")
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "--selftest":
selftest()
else:
main()
复制即跑:换 你的智兔token 后 python main.py;只看逻辑不联网跑 python main.py --selftest。
三、常见坑
- 批量要配额:
/hs/list/all一次返回全市场,再逐只拉实时属于批量访问,免费版配额有限;监控几十只以上建议升级到更高配额证书,避免触发限流。 - 清单结构别假设:有的版本返回对象列表、有的返回纯代码列表,
parse_list两种都兼容,别写死索引。 - 限速是礼貌也是防护:批量轮询每只之间
sleep(0.2),别对接口做高压并发。
小结与下篇预告
本篇你拿到了全市场清单并批量拉实时做监控。下一篇(#03)讲怎么把历史 K 线落成 CSV:复权口径与增量更新一次讲清。
免费领取证书
数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。
领取路径:进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」(邮箱验证 3 步即可拿到 token)。
把代码里的 你的智兔token 换成你拿到的真实 token,上面的示例就能直接拉到真实行情。
免责声明
本文仅演示智兔数服公开数据接口的用法,所有代码示例均以占位 token 自验,未含任何实时行情数据;不构成投资建议,亦不承诺收益。投资决策请基于你自己的判断与风险承受力。
想亲自试一下?免费获取证书