【跨市场数据实战 #04】沪深A股异动排名与指数历史K线:23个接口覆盖连涨连跌、估值排名与BOLL/KDJ/MA/MACD
摘要:【跨市场数据实战 #04】沪深A股异动排名与指数历史K线:23个接口覆盖连涨连跌、估值排名与BOLL/KDJ/MA/MACD 系列:《跨市场数据实战》|连载项目 · 纯 GET 取数 · 仅依赖 requests
系列:《跨市场数据实战》|连载项目 · 纯 GET 取数 · 仅依赖 requests
适用:想做「异动监控 + 技术指标叠加」、但被一堆排名接口和 K 线参数绕晕的读者;数据由智兔数服提供。本篇给沪深A股异动/排名线(/himk,15 个端点)与指数行情/技术指标线(/hz,8 个端点)共 23 个端点的分组地图、一套字段容错归一化代码、以及一个把「连涨名单 + 创30日新高」和「BOLL/KDJ/MA/MACD」叠在一起的实战模板,全部只依赖 requests,所有示例均为演示数据,不构成收益承诺。
1. 你将得到什么
读完这一篇,你能拿走四样东西:
- 一张分组地图:/himk(沪深A股异动与排名,15 端点)+ /hz(指数行情与技术指标,8 端点),知道「连涨连跌 / 创30日新高新低 / 放量缩量 / 估值排名」和「实时、分时、历史 K 线、MACD/MA/BOLL/KDJ」分别敲哪个门;
- 一套字段容错代码:排名接口字段名各异,本篇用
_hit_key带候选键兜底; - 一个叠加模板:用
scan_anomaly拉异动名单,用overlay_indicators给某只指数叠上四类技术指标; - 五个真实踩坑点,尤其是
/hz/*那个必须先取指数代码的路径参数。
代码全部自包含,复制进 .py 直接能跑,不依赖 numpy / pandas。
2. 本篇取数约定
- 全部接口都是 GET + query 参数,token 放在查询串里(
?token=xxx),不放 header; - 统一基址
https://api.zhituapi.com; - 代码块里的
你的智兔token是占位符,换成你的 token 即可; - 所有接口路径均取自官方文档。
- 数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。
3. 23 个端点分两组
先建立地图。
A 组:/himk 沪深A股异动与排名(15 端点)
| 端点 | 用途 |
|---|---|
/himk/jdzgzd |
阶段(近5/10/20/60日)最高最低价及涨跌幅 |
/himk/pzxg |
今日创近30日新高的股票 |
/himk/pzxd |
今日创近30日新低的股票 |
/himk/cjzz |
今日成交量较上日大幅增加的股票 |
/himk/cjzj |
今日成交量较上日大幅减少的股票 |
/himk/lxfl |
成交量连续放大的股票 |
/himk/lxsl |
成交量连续缩小的股票 |
/himk/lxsz |
连续上涨的股票 |
/himk/lxxd |
连续下跌的股票 |
/himk/zzd |
本周涨跌幅排名(复权) |
/himk/yzd |
本月涨跌幅排名(复权) |
/himk/ltszph |
流通市值排名 |
/himk/syl |
静态/动态市盈率排名 |
/himk/sjl |
市净率排名 |
/himk/roe |
ROE 排名(倒序) |
B 组:/hz 指数行情与技术指标(8 端点)
| 端点 | 用途 | 路径参数 |
|---|---|---|
/hz/list/hszs |
指数列表(取代码用) | 无 |
/hz/real/ssjy/{指数代码.市场} |
指数实时交易(日线最新) | 如 000001.SH |
/hz/latest/fsjy/{指数代码.市场}/{分时级别} |
最新分时 | 级别见下 |
/hz/history/fsjy/{指数代码.市场}/{分时级别} |
历史分时 | 级别见下 |
/hz/history/macd/{指数代码.市场}/{分时级别} |
历史 MACD | 级别见下 |
/hz/history/ma/{指数代码.市场}/{分时级别} |
历史 MA | 级别见下 |
/hz/history/boll/{指数代码.市场}/{分时级别} |
历史 BOLL | 级别见下 |
/hz/history/kdj/{指数代码.市场}/{分时级别} |
历史 KDJ | 级别见下 |
/hz/* 的分时级别统一为:5 / 15 / 30 / 60 / d / w / m / y(5/15/30/60 分钟,d 日线,w 周,m 月,y 年)。指数代码取自 /hz/list/hszs,格式 代码.市场,如 000001.SH(上证指数)。
4. 核心模板函数
import requests, time
BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token"
# ---------- 1. 字段容错与类型归一 ----------
def _hit_key(d, *cands, default=None):
if not isinstance(d, dict):
return default
for c in cands:
if c in d and d[c] not in (None, "", "-", "null"):
return d[c]
low = {str(k).lower(): v for k, v in d.items()}
for c in cands:
v = low.get(str(c).lower())
if v not in (None, "", "-", "null"):
return v
return default
def _to_float(v, default=None):
try:
if v in (None, "", "-", "null", "None"):
return default
return float(v)
except (TypeError, ValueError):
return default
# ---------- 2. 统一请求 ----------
def _get(path, params=None, timeout=10, retries=2, backoff=0.6, default=None):
q = {"token": TOKEN}
if params:
q.update(params)
last = ""
for i in range(retries + 1):
try:
r = requests.get(BASE + path, params=q, timeout=timeout)
if r.status_code == 200:
try:
return r.json()
except ValueError:
return default
last = "HTTP %s %s" % (r.status_code, (r.text or "").strip()[:80])
except Exception as e:
last = "%s: %s" % (type(e).__name__, e)
if i < retries:
time.sleep(backoff * (i + 1))
return {"_error": last}
# ---------- 3. 排名类归一 (/himk/*) ----------
def norm_rank(rows, *valkeys):
"""沪深A股各类排名 -> [{代码,名称,值}]"""
out = []
for r in rows or []:
out.append({
"代码": _hit_key(r, "dm", "code", default="-"),
"名称": _hit_key(r, "mc", "name", default="-"),
"值": _to_float(_hit_key(r, *valkeys, "val", "v", "zf")),
})
return out
# ---------- 4. K线类归一 (/hz/history/fsjy 等) ----------
def norm_kline(rows):
out = []
for r in rows or []:
out.append({
"时间": _hit_key(r, "t", "date", "time", default="-"),
"开": _to_float(_hit_key(r, "o", "open")),
"高": _to_float(_hit_key(r, "h", "high")),
"低": _to_float(_hit_key(r, "l", "low")),
"收": _to_float(_hit_key(r, "c", "close")),
"量": _to_float(_hit_key(r, "v", "vol", "volume")),
})
return out
# ---------- 5. 技术指标类归一 (MACD/MA/BOLL/KDJ) ----------
def norm_indicator(rows):
out = []
for r in rows or []:
out.append({
"时间": _hit_key(r, "t", "date", "time", default="-"),
"值": _to_float(_hit_key(r, "v", "val", "k", "d", "j",
"macd", "diff", "dea", "up", "mid", "down", "ma")),
})
return out
# ---------- 6. 取数封装 ----------
def fetch_lxsz(): return _get("/himk/lxsz", default=[])
def fetch_pzxg(): return _get("/himk/pzxg", default=[])
def fetch_ltszph(): return _get("/himk/ltszph", default=[])
def fetch_syl(): return _get("/himk/syl", default=[])
def fetch_roe(): return _get("/himk/roe", default=[])
def fetch_index_list(): return _get("/hz/list/hszs", default=[])
def fetch_kline(idx="000001.SH", k="d"): return _get("/hz/history/fsjy/%s/%s" % (idx, k), default=[])
def fetch_kdj(idx="000001.SH", k="d"): return _get("/hz/history/kdj/%s/%s" % (idx, k), default=[])
def fetch_boll(idx="000001.SH", k="d"): return _get("/hz/history/boll/%s/%s" % (idx, k), default=[])
def fetch_ma(idx="000001.SH", k="d"): return _get("/hz/history/ma/%s/%s" % (idx, k), default=[])
def fetch_macd(idx="000001.SH", k="d"): return _get("/hz/history/macd/%s/%s" % (idx, k), default=[])
def fetch_real(idx="000001.SH"): return _get("/hz/real/ssjy/%s" % idx, default=[])
# ---------- 7. 实战:异动扫描 ----------
def scan_anomaly(top=10):
"""连涨名单 + 创30日新高名单"""
lx = norm_rank(fetch_lxsz(), "days", "lzs", "zf")
hi = norm_rank(fetch_pzxg(), "zf", "pct", "val")
return {"连涨": lx[:top], "创30日新高": hi[:top]}
# ---------- 8. 实战:技术指标叠加 ----------
def overlay_indicators(idx="000001.SH", k="d"):
"""给某指数叠上 K线 + KDJ/BOLL/MA/MACD 最近 5 根"""
return {
"kline": norm_kline(fetch_kline(idx, k))[-5:],
"kdj": norm_indicator(fetch_kdj(idx, k))[-5:],
"boll": norm_indicator(fetch_boll(idx, k))[-5:],
"ma": norm_indicator(fetch_ma(idx, k))[-5:],
"macd": norm_indicator(fetch_macd(idx, k))[-5:],
}
# ---------- 9. 校验 ----------
def run_check():
assert _hit_key({"DM": "000001", "mc": "平安"}, "dm") == "000001"
assert _to_float("-") is None and _to_float("3.2") == 3.2
r = [{"dm": "600519", "mc": "茅台", "days": "5", "zf": "3.2"}]
nr = norm_rank(r, "days", "zf")
assert nr[0]["值"] == 5.0 and nr[0]["名称"] == "茅台"
kl = [{"t": "2026-01-02", "o": "10", "h": "11", "l": "9.5", "c": "10.8", "v": "1000"}]
nk = norm_kline(kl)
assert nk[0]["收"] == 10.8 and nk[0]["量"] == 1000.0
ind = [{"t": "2026-01-02", "v": "0.5", "k": "1.2", "d": "0.8"}]
ni = norm_indicator(ind)
assert ni[0]["值"] == 0.5
a = scan_anomaly(3)
assert "连涨" in a and "创30日新高" in a
o = overlay_indicators()
assert set(o.keys()) == {"kline", "kdj", "boll", "ma", "macd"}
print("校验通过")
if __name__ == "__main__":
run_check()
print("-" * 62)
for name, path in [("连涨", "/himk/lxsz"),
("创30日新高", "/himk/pzxg"),
("流通市值排名", "/himk/ltszph"),
("ROE排名", "/himk/roe"),
("指数列表", "/hz/list/hszs"),
("上证日线", "/hz/history/fsjy/000001.SH/d"),
("上证KDJ", "/hz/history/kdj/000001.SH/d"),
("上证BOLL", "/hz/history/boll/000001.SH/d")]:
data = _get(path, default=[])
if isinstance(data, dict) and "_error" in data:
print("%-12s %-30s -> %s" % (name, path, data["_error"][:56]))
else:
print("%-12s %-30s -> %d 条" % (name, path, len(data)))
5. 跑通示例
把上面的代码复制到本地,填入你的 token 即可直接运行:它会请求对应接口、拉取真实数据,并输出归一化后的结构化字典(各字段含义见前文各小节)。
6. 坑与注意事项
坑 1:/hz/* 的 {指数代码.市场} 必须先从 /hz/list/hszs 取。
你不能直接猜 000001.SH 一定存在——先拉 /hz/list/hszs 拿到合法指数代码清单,再拼后面的路径。本文示例用 000001.SH(上证指数)只是演示格式。
坑 2:分时级别是路径参数不是查询参数,且只认那 8 个值。/hz/history/kdj/000001.SH/d 才对;写成 /hz/history/kdj?idx=000001.SH&k=d 会 404。级别只认 5/15/30/60/d/w/m/y。
坑 3:/himk/jdzgzd 是「阶段」窗口,不是单日。
它返回近 5/10/20/60 日的最高最低价与涨跌幅,具体取哪一档看上游实现,别把它当成「今日涨跌」。
坑 4:排名接口的单位/字段各不相同。lxsz 的值是「连涨天数」,ltszph 是「流通市值(元或亿视上游)」,syl 是「市盈率倍数」,roe 是「百分比」。norm_rank 只做容错抽取,单位换算要按你自己的口径在调用层做,别直接相加。
坑 5:/hz/real/ssjy 是「日线最新」不是逐笔。
它返回的是实时交易数据(可以理解为日线的最新一根),不是 tick 级逐笔。要做真正的高频,文档里没有对应端点,别硬套。
7. 小结与下篇预告
本篇把沪深A股异动/排名线(/himk,15 端点)与指数行情/技术指标线(/hz,8 端点)打通,给出 norm_rank / norm_kline / norm_indicator 三套归一化,并用 scan_anomaly + overlay_indicators 把「异动名单」和「BOLL/KDJ/MA/MACD」叠成可落地的监控模板。/hz/* 必须先取指数代码、分时级别走路径参数,是这套接口最容易踩错的地方。
下一篇计划写 #05《港股财报全景:9个接口从业绩预告到现金流量表》:用 /hicw(盈利能力、运营能力、成长能力、偿债能力、现金流量、业绩报表/快报/预告、利润细分)批量拉港股年报关键指标做横向对比。
8. 免责声明
本文仅演示沪深A股异动与指数技术指标的取数与归一化方法,所有代码示例均为演示数据,未含任何真实行情数值,不构成投资建议,亦不承诺收益。
免费领取证书
数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。
领取路径:进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」(邮箱验证 3 步即可拿到 token)。
把代码里的 你的智兔token 换成你拿到的真实 token,上面的脚本就能直接打印沪深A股异动名单与指数 K 线技术指标。