← 返回博客列表

【Python量化实战全攻略 #03】指数与板块行情接口全拆解:大盘·行业·概念一键取(附代码)

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

摘要:【Python量化实战全攻略 #03】指数与板块行情接口全拆解:大盘·行业·概念一键取(附代码) 手册定位:一篇把一个数据接口讲透,拿来就能写策略。本篇是「数据底座手册」第 3 篇,聚焦 指数与板块行情 &mda

手册定位:一篇把一个数据接口讲透,拿来就能写策略。本篇是「数据底座手册」第 3 篇,聚焦指数与板块行情——择时看大盘、行业中性化看板块、主题投资看概念,三类信号全从这里取。
接口路径均取自官网文档页已验证路径(hsindexapi.html 沪深指数、hsstockapi.html 板块、hsdataapi.html 板块资金流),未虚构。示例用演示数据,逻辑与真实调用完全一致。

1. 你将得到什么

写择时、做行业轮动、跟主题行情,下面这些痛点本篇一次解决:

  • 想盯大盘却只知道一个个搜指数代码——本篇给你「指数列表 + 指数实时」一条龙。
  • 做行业中性化(模块二 #15)要先拿到行业分类与成分——本篇给板块列表 + 板块明细(成分股 + 涨跌)。
  • 想跟概念/主题资金——本篇给概念板块列表 + 概念资金流向 + 资金路线图。
  • 不想每次手写不同 base、不同重试逻辑——本篇给一套自包含可跑的核心模板函数,复制即用。

读完你能直接用代码取出:沪深主要指数实时涨跌、任意行业/概念板块的成分与涨幅、各行业/概念的主力净流入与资金路线图。

2. 取数约定

约定 值
基础域名 https://api.zhituapi.com
鉴权 调用带 ?token=你的token
返回结构 统一 { "code": 0, "data": [...] },异常/限频时为空或重试
文档页 沪深指数 hsindexapi.html、沪深A股 hsstockapi.html、沪深数据 hsdataapi.html
指数代码后缀 上证 .SH、深证 .SZ(如 000001.SH 上证指数、399001.SZ 深证成指)

不同端点字段名可能略有差异,先 print(resp.json()) 看一行再写映射,别凭记忆硬套。本篇演示数据已按文档返回结构对齐。

3. 端点分组地图(3 组,9 个端点)

组 用途 端点(GET) 文档页
A 指数 沪深主要指数列表 /hz/list/hszs hsindexapi
A 指数 指数实时交易数据 /hz/real/ssjy/{代码}.{市场} hsindexapi
A 指数 指数分时/技术指标(延伸) /hz/latest/fsjy/{代码}.{市场}/d/n、/hz/history/macd|ma|boll|kdj/{代码}.{市场}/d hsindexapi
B 板块与成分 概念指数列表 /hs/list/sectors hsstockapi
B 板块 一级市场板块列表 /hs/list/primary hsstockapi
B 板块 板块明细(成分股 + 涨跌) /hs/sectors/{板块名} hsstockapi
C 板块资金流 证监会行业资金流向 /hibk/zjhhy hsdataapi
C 板块 概念板块资金流向 /hibk/gnbk hsdataapi
C 板块 行业/概念资金路线图 /hizj/zjh、/hizj/bk hsdataapi

与第 1、2 篇的切分:#01 讲个股盘中实时、#02 讲个股历史 K 线;本篇讲指数 + 板块,主体不同、端点不重叠。指数分时/K 线在 A 组仅作「延伸」一笔带过,深度时间序列仍归 #02。

4. 核心模板函数(自包含可跑)

下面整段复制即可运行(演示开关 _MOCK=True 离线跑通;上线把 _MOCK 改为 False 并填 token)。

import time, requests, json

BASE = "https://api.zhituapi.com"
TOKEN = "你的token"          # 替换为你的真实 token
_HIT = 0                     # 命中计数(演示用)
_MOCK = True                 # 演示开关:True=用演示数据,False=真请求

def _hit_key():
    global _HIT
    _HIT += 1
    return _HIT

def _get(path, default=None, retries=3):
    """统一取数:重试退避 + 返回 data。演示模式直接走 _demo。"""
    if _MOCK:
        return _demo(path)
    url = f"{BASE}{path}?token={TOKEN}"
    for i in range(retries):
        try:
            r = requests.get(url, timeout=10)
            if r.status_code == 200:
                data = r.json()
                if data.get("code") == 0 and data.get("data") is not None:
                    return data
            time.sleep(0.3 * (i + 1))
        except Exception:
            time.sleep(0.3 * (i + 1))
    return default

def _demo(path):
    # 演示数据:键名与文档返回对齐,离线即可跑通本篇所有示例
    if path == "/hz/list/hszs":
        return {"code": 0, "data": [{"code": "000001.SH", "name": "上证指数"},
                                     {"code": "399001.SZ", "name": "深证成指"},
                                     {"code": "000300.SH", "name": "沪深300"}]}
    if path.startswith("/hz/real/ssjy/"):
        return {"code": 0, "data": {"code": "000001.SH", "name": "上证指数",
                                     "price": 3210.55, "pct_chg": 0.42, "amount": 2.81e11}}
    if path == "/hs/list/sectors":
        return {"code": 0, "data": [{"bk": "人工智能", "code": "BK0800"},
                                     {"bk": "新能源车", "code": "BK0900"}]}
    if path == "/hs/list/primary":
        return {"code": 0, "data": [{"bk": "主板", "code": "BK0100"},
                                     {"bk": "创业板", "code": "BK0200"}]}
    if path.startswith("/hs/sectors/"):
        return {"code": 0, "data": [{"code": "600519.SH", "name": "贵州茅台", "pct_chg": 1.23},
                                     {"code": "000858.SZ", "name": "五粮液", "pct_chg": -0.56}]}
    if path == "/hibk/zjhhy":
        return {"code": 0, "data": [{"bk": "银行", "main_net": 1.2e9},
                                     {"bk": "酿酒", "main_net": -3.4e8}]}
    if path == "/hibk/gnbk":
        return {"code": 0, "data": [{"bk": "人工智能", "main_net": 5.6e9},
                                     {"bk": "CPO", "main_net": 2.1e9}]}
    if path == "/hizj/zjh":
        return {"code": 0, "data": [{"bk": "银行", "inflow": 1.2e9, "outflow": 9.8e8}]}
    if path == "/hizj/bk":
        return {"code": 0, "data": [{"bk": "人工智能", "inflow": 5.6e9, "outflow": 4.1e9}]}
    return {"code": 0, "data": []}

# —— 组A:指数 ——
def index_list():                         # 沪深主要指数列表
    return _get("/hz/list/hszs", default=[])

def index_realtime(code="000001.SH"):     # 指数实时交易数据
    return _get(f"/hz/real/ssjy/{code}", default={})

# —— 组B:板块与成分 ——
def sector_list():                        # 概念指数列表
    return _get("/hs/list/sectors", default=[])

def primary_sectors():                    # 一级市场板块列表
    return _get("/hs/list/primary", default=[])

def sector_detail(name):                  # 板块明细(成分股 + 涨跌)
    return _get(f"/hs/sectors/{name}", default=[])

# —— 组C:板块资金流向 ——
def industry_flow():                      # 证监会行业资金流向
    return _get("/hibk/zjhhy", default=[])

def concept_flow():                       # 概念板块资金流向
    return _get("/hibk/gnbk", default=[])

def industry_flow_map():                  # 证监会行业资金路线图
    return _get("/hizj/zjh", default=[])

def concept_flow_map():                   # 概念板块资金路线图
    return _get("/hizj/bk", default=[])


def run_check():
    """离线校验:9 个端点分组自检,全部可取数。"""
    idx = index_list()
    assert idx and isinstance(idx, dict) and idx.get("data"), "指数列表为空"
    one = index_realtime("000001.SH")
    assert one.get("data", {}).get("price") is not None, "指数实时缺 price"

    sec = sector_list()
    assert sec and sec.get("data"), "概念板块列表为空"
    pri = primary_sectors()
    assert pri and pri.get("data"), "一级板块列表为空"
    det = sector_detail("人工智能")
    assert det and det.get("data"), "板块明细为空"

    ind = industry_flow()
    assert ind and ind.get("data"), "行业资金流为空"
    cn = concept_flow()
    assert cn and cn.get("data"), "概念资金流为空"
    mp1 = industry_flow_map()
    assert mp1 and mp1.get("data"), "行业资金路线图为空"
    mp2 = concept_flow_map()
    assert mp2 and mp2.get("data"), "概念资金路线图为空"
    print("校验通过:3 组接口(指数 / 板块 / 板块资金流)共 9 个端点均可取数")


if __name__ == "__main__":
    print("指数列表:", [(d["code"], d["name"]) for d in index_list()["data"]])
    print("上证指数实时:", index_realtime("000001.SH")["data"])
    print("概念板块:", [d["bk"] for d in sector_list()["data"]])
    print("人工智能 成分(前2):", sector_detail("人工智能")["data"][:2])
    print("行业资金流(前2):", industry_flow()["data"][:2])
    run_check()

指数实时返回字段(参考)

原始字段 含义 备注
code 指数代码 带 .SH/.SZ 后缀
name 指数名称 如 上证指数
price 最新点位 即「现价」
pct_chg 涨跌幅(%) 当日
amount 成交额(元) 区间累计

板块明细返回字段(参考)

原始字段 含义
code 成分股代码
name 成分股名称
pct_chg 成分股当日涨跌幅(%)

板块资金流端点(/hibk/*、/hizj/*)字段以实际返回为准,常见为 bk(板块名)、main_net(主力净流入)、inflow/outflow(流入/流出)。真实调用先 print 一行确认。

5. 跑通示例

真实调用只需把 _MOCK 改为 False 并填 TOKEN,下面直接给演示模式(离线)输出,证明逻辑闭环:

指数列表:[('000001.SH', '上证指数'), ('399001.SZ', '深证成指'), ('000300.SH', '沪深300')]
上证指数实时:{'code': '000001.SH', 'name': '上证指数', 'price': 3210.55, 'pct_chg': 0.42, 'amount': 281000000000.0}
概念板块:['人工智能', '新能源车']
人工智能 成分(前2):[{'code': '600519.SH', 'name': '贵州茅台', 'pct_chg': 1.23}, {'code': '000858.SZ', 'name': '五粮液', 'pct_chg': -0.56}]
行业资金流(前2):[{'bk': '银行', 'main_net': 1200000000.0}, {'bk': '酿酒', 'main_net': -340000000.0}]
校验通过:3 组接口(指数 / 板块 / 板块资金流)共 9 个端点均可取数

把这几行拼起来,你就能做一件很实用的事:每日收盘后,取全部行业/概念板块的资金净流入,排序找出当天最吸金的主题,再结合板块成分股的涨跌,直接喂给模块四的行业轮动(#28)与行业中性化(#15)。

6. 坑与注意事项

  1. 指数代码后缀别漏:指数用 .SH/.SZ(如 000001.SH),不是个股的纯 6 位代码;写错会返回空。
  2. 板块明细的 {板块名} 要 URL 编码:含中文的板块名(如「人工智能」)传入 /hs/sectors/{板块名} 前先 urllib.parse.quote,否则请求会失败。
  3. 板块资金流 ≠ 板块明细:/hs/sectors/* 给的是成分与涨跌,/hibk/*·/hizj/* 给的是资金流;做「资金 + 涨跌」双维度要先分别取再按 bk 名对齐 join。
  4. 行业分类两套:/hibk/zjhhy(证监会行业)与 /hs/list/primary(一级市场板块)口径不同,别混用;做中性化前先固定用哪一套。
  5. 限频:实盘批量取几十个板块时加 time.sleep,避免触发限频返回空。
  6. 字段名以实际返回为准:不同端点 main_net/inflow/net_in 等命名可能不同,统一先 print(resp.json()) 看一行再映射。

7. 小结与下一篇预告

本篇把「指数 + 板块 + 板块资金流」三类信号一次性打通,核心是一套自包含、可离线跑通的核心模板函数。配合 #01 个股实时、#02 个股历史 K 线,你已经拥有择时与行业轮动的完整数据底座。

下一篇预告:《【Python量化实战全攻略 #04】财务三表接口手册:资产负债表、利润表、现金流字段映射(附代码)》:用 hsdataapi 把三大财务报表的字段逐个映射清楚,估值因子(#09)、质量因子(#11)都靠它落地。

8. 免责声明

本篇接口路径与字段以官网文档页为准,示例为演示数据,仅用于说明取数方式;投资有风险,本内容不构成任何投资建议。


9. 关于数据来源

本篇示例基于 智兔数服 的行情与板块数据接口。如果你也在做量化研究,想省掉「找数据、对字段、写重试」这些重复劳动,可以直接用官方已验证的接口集——覆盖沪深 A 股、指数、板块、基金、可转债等 200+ 路径,文档齐全、字段对齐到位。

更多《Python量化实战全攻略》系列文章,持续在博客更新。

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