← 返回博客列表

【零依赖量化数据实战 #02】沪深全量股票清单:批量拉实时做自选监控

2026年08月20日 08:04 · 智兔数服 · 零依赖量化数据实战

摘要:【零依赖量化数据实战 #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()

复制即跑:换 你的智兔tokenpython 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 自验,未含任何实时行情数据;不构成投资建议,亦不承诺收益。投资决策请基于你自己的判断与风险承受力。

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