← 返回博客列表

【跨市场数据实战 #04】沪深A股异动排名与指数历史K线:23个接口覆盖连涨连跌、估值排名与BOLL/KDJ/MA/MACD

2026年09月16日 08:18 · 智兔数服 · 跨市场数据实战

摘要:【跨市场数据实战 #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. 你将得到什么

读完这一篇,你能拿走四样东西:

  1. 一张分组地图:/himk(沪深A股异动与排名,15 端点)+ /hz(指数行情与技术指标,8 端点),知道「连涨连跌 / 创30日新高新低 / 放量缩量 / 估值排名」和「实时、分时、历史 K 线、MACD/MA/BOLL/KDJ」分别敲哪个门;
  2. 一套字段容错代码:排名接口字段名各异,本篇用 _hit_key 带候选键兜底;
  3. 一个叠加模板:用 scan_anomaly 拉异动名单,用 overlay_indicators 给某只指数叠上四类技术指标;
  4. 五个真实踩坑点,尤其是 /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 线技术指标。

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