【Python量化实战全攻略 #03】指数与板块行情接口全拆解:大盘·行业·概念一键取(附代码)
摘要:【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(流入/流出)。真实调用先
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. 坑与注意事项
- 指数代码后缀别漏:指数用
.SH/.SZ(如000001.SH),不是个股的纯 6 位代码;写错会返回空。 - 板块明细的
{板块名}要 URL 编码:含中文的板块名(如「人工智能」)传入/hs/sectors/{板块名}前先urllib.parse.quote,否则请求会失败。 - 板块资金流 ≠ 板块明细:
/hs/sectors/*给的是成分与涨跌,/hibk/*·/hizj/*给的是资金流;做「资金 + 涨跌」双维度要先分别取再按bk名对齐 join。 - 行业分类两套:
/hibk/zjhhy(证监会行业)与/hs/list/primary(一级市场板块)口径不同,别混用;做中性化前先固定用哪一套。 - 限频:实盘批量取几十个板块时加
time.sleep,避免触发限频返回空。 - 字段名以实际返回为准:不同端点
main_net/inflow/net_in等命名可能不同,统一先print(resp.json())看一行再映射。
7. 小结与下一篇预告
本篇把「指数 + 板块 + 板块资金流」三类信号一次性打通,核心是一套自包含、可离线跑通的核心模板函数。配合 #01 个股实时、#02 个股历史 K 线,你已经拥有择时与行业轮动的完整数据底座。
下一篇预告:《【Python量化实战全攻略 #04】财务三表接口手册:资产负债表、利润表、现金流字段映射(附代码)》:用 hsdataapi 把三大财务报表的字段逐个映射清楚,估值因子(#09)、质量因子(#11)都靠它落地。
8. 免责声明
本篇接口路径与字段以官网文档页为准,示例为演示数据,仅用于说明取数方式;投资有风险,本内容不构成任何投资建议。
9. 关于数据来源
本篇示例基于 智兔数服 的行情与板块数据接口。如果你也在做量化研究,想省掉「找数据、对字段、写重试」这些重复劳动,可以直接用官方已验证的接口集——覆盖沪深 A 股、指数、板块、基金、可转债等 200+ 路径,文档齐全、字段对齐到位。
- 官方文档:沪深指数 API | 沪深 A 股 API | 沪深数据 API
- 技术博客合集:智兔数服博客
更多《Python量化实战全攻略》系列文章,持续在博客更新。