系列：《A股分钟级高频数据完全手册》｜连载项目 · 纯 GET 取数 · 仅依赖 requests · Pro 版分钟级高频

① 你将得到什么

个股指标之外，指数层面的 MACD 用于判断大盘趋势与择时。指数 Pro 版 MACD 历史 /hz/pro/history/macd/{code}.SH/1 与个股同源，只是前缀 /hz/ + fsjy 路径结构。本篇把它拉通并解析指数 MACD。

本篇产出：用 /hz/pro/history/macd/{code}.SH/1 拉取沪深指数 1 分钟 MACD 历史，解析 DIF/DEA/MACD 并做指标统计，用于指数择时。

读完这篇，你应当能独立把本篇端点数据拉下来、落本地，并跑出一份可用的加工分析表。

② 取数约定

一个端点拉数据，返回里字段名可能中英文混用（不同接口版本、不同数据源习惯不同）。演示用 你的token 占位，代码里 MOCK=True 时回放演示数据，接真实 token 把 MOCK 改成 False 即可，无需改动其他逻辑。

所有请求走 https://api.zhituapi.com + ?token=，纯 GET、仅依赖 requests，不依赖任何付费 SDK。字段口径以接口实际返回为准，本文用 _hit_key 做了大小写/中英文容错；落地前建议先用 print(r[0].keys()) 核对一次你账号下的真实字段名。

落地建议：把每天拉到的数据按 代码 + 日期 做主键写入本地 SQLite/CSV，避免重复插入；历史回看靠本地累积，而非反复调接口。本篇示例代码已包含字段容错映射与本地加工函数，你只需把 print 替换为写入语句。

③ 端点分组地图

本篇涉及字段（落地建表参考）：代码、名称、时间、DIF、DEA、MACD、收盘、成交量。

返回结构为 JSON 数组（或含数组字段的对象），每条记录即一根分钟 bar / 一笔明细；字段映射函数已按上述字段做容错映射，直接复用即可。

④ 核心模板函数（自包含·可跑）

import requests, time, copy
from urllib.parse import quote

BASE = "https://api.zhituapi.com"
# 将 "你的token" 替换成您自己的智兔API token（官网注册后在「控制台 / 接口管理」页获取，免费额度即可起步）。
TOKEN = "你的token"

# ---------- 字段容错与类型归一 ----------
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

# ---------- 统一请求：token 鉴权 + 重试退避（MOCK 开关） ----------
MOCK = True
DEMO_DATA = {}

def _get(path, params=None, timeout=10, retries=2, backoff=0.6, default=None):
    """返回 JSON；MOCK=True 时回放 DEMO_DATA；否则真实 GET 并带 token 鉴权、重试 retries 次"""
    if MOCK:
        return copy.deepcopy(DEMO_DATA.get(path, {} if default is None else default))
    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. 取数封装 ----------
def fetch_rows():
    return _get("/hz/pro/history/macd/000001.SH/1", default=[])

# ---------- 4. 数据加工 ----------
def process(rows):
    out = []
    for r in rows or []:
        out.append({
        "代码": _hit_key(r, "code", "dm"),
        "名称": _hit_key(r, "name", "mc"),
        "时间": _hit_key(r, "t", "time", "datetime"),
        "DIF": _hit_key(r, "dif"),
        "DEA": _hit_key(r, "dea"),
        "MACD": _hit_key(r, "macd", "bar"),
        "收盘": _hit_key(r, "c", "close"),
        "成交量": _hit_key(r, "v", "vol", "volume")
        })
    return out

# ---------- 5. 本地加工分析（演示：多列描述统计 + 主列 Top5 + 异常） ----------
def analysis(data):
    if not data:
        print("无可分析数据，请检查字段映射")
        return
    num_cols = [k for k in data[0].keys() if _to_float(data[0][k]) is not None]
    for col in num_cols:
        vs = [(_to_float(d.get(col)) or 0.0) for d in data]
        if not vs:
            continue
        print("列[%s] n=%d 均值=%2f 最大=%2f 最小=%2f" % (col, len(vs), sum(vs)/len(vs), max(vs), min(vs)))
    col = "MACD"
    vs = [(_to_float(d.get(col)) or 0.0) for d in data]
    vs.sort(reverse=True)
    print("【主数值列 %s】Top5：" % col)
    for v in vs[:5]:
        print("  %s=%2f" % (col, v))
    anom = [v for v in vs if v < 0 or v > (sum(vs)/len(vs)) * 2]
    print("异常扫描：%d 条（负值或超 2 倍均值）" % len(anom))

# ---------- 6. 校验 ----------
DEMO_DATA = {
    "/hz/pro/history/macd/000001.SH/1": [{'code': '000001.SH', 'name': '上证指数', 't': '2024-01-08 09:31:00', 'dif': '2.10', 'dea': '1.80', 'macd': '0.60', 'c': '2885.10', 'v': '1400000'}, {'code': '399001.SZ', 'name': '深证成指', 't': '2024-01-08 09:31:00', 'dif': '8.50', 'dea': '7.20', 'macd': '2.60', 'c': '9170.80', 'v': '1700000'}, {'code': '000001.SH', 'name': '上证指数', 't': '2024-01-08 09:31:00', 'dif': '1.54', 'dea': '1.32', 'macd': '0.44', 'c': '2115.74', 'v': '1.02667e+06'}, {'code': '399001.SZ', 'name': '深证成指', 't': '2024-01-08 09:31:00', 'dif': '6.8', 'dea': '5.76', 'macd': '2.08', 'c': '7336.64', 'v': '1.36e+06'}, {'code': '000001.SH', 'name': '上证指数', 't': '2024-01-08 09:31:00', 'dif': '1.82', 'dea': '1.56', 'macd': '0.52', 'c': '2500.42', 'v': '1.21333e+06'}, {'code': '399001.SZ', 'name': '深证成指', 't': '2024-01-08 09:31:00', 'dif': '7.93', 'dea': '6.72', 'macd': '2.43', 'c': '8559.41', 'v': '1.58667e+06'}, {'code': '000001.SH', 'name': '上证指数', 't': '2024-01-08 09:31:00', 'dif': '2.1', 'dea': '1.8', 'macd': '0.6', 'c': '2885.1', 'v': '1.4e+06'}, {'code': '399001.SZ', 'name': '深证成指', 't': '2024-01-08 09:31:00', 'dif': '9.07', 'dea': '7.68', 'macd': '2.77', 'c': '9782.19', 'v': '1.81333e+06'}, {'code': '000001.SH', 'name': '上证指数', 't': '2024-01-08 09:31:00', 'dif': '2.38', 'dea': '2.04', 'macd': '0.68', 'c': '3269.78', 'v': '1.58667e+06'}, {'code': '399001.SZ', 'name': '深证成指', 't': '2024-01-08 09:31:00', 'dif': '10.2', 'dea': '8.64', 'macd': '3.12', 'c': '11005', 'v': '2.04e+06'}, {'code': '000001.SH', 'name': '上证指数', 't': '2024-01-08 09:31:00', 'dif': '2.66', 'dea': '2.28', 'macd': '0.76', 'c': '3654.46', 'v': '1.77333e+06'}, {'code': '399001.SZ', 'name': '深证成指', 't': '2024-01-08 09:31:00', 'dif': '11.33', 'dea': '9.6', 'macd': '3.47', 'c': '12227.7', 'v': '2.26667e+06'}],
}

def run_check():
    assert _hit_key({"Code": "000001", "Name": "平安银行"}, "code", "dm") == "000001"
    assert _to_float("-") is None and _to_float("12.5") == 12.5
    rows = fetch_rows()
    assert isinstance(rows, list) and len(rows) >= 2
    data = process(rows)
    assert len(data) == len(rows)
    assert _hit_key(rows[0], "code", "dm") not in (None, "", "-")
    print("校验通过")

if __name__ == "__main__":
    run_check()
    print("-" * 62)
    data = _get("/hz/pro/history/macd/000001.SH/1", default=[])
    if isinstance(data, dict) and "_error" in data:
        print("接口异常: %s" % data["_error"][:60])
    else:
        rows = data if isinstance(data, list) else []
        print("样例条数: %d" % len(rows))

代码讲解：① SCAFFOLD 提供字段容错 _hit_key 与带 token 鉴权、重试退避的请求 _get，MOCK 开关让你无需 token 就能验证逻辑；② 取数封装把端点参数化，process() 把接口返回的混乱字段统一映射成中文列；③ analysis() 在本地做多列描述统计 / 主列 Top5 / 异常扫描，不消耗接口额度；④ run_check() 用断言自检，可放进 CI 持续校验。

⑤ 跑通示例（演示数据）

把上面代码保存为 hf_20.py，MOCK=True 直接跑（Python3 + requests 即可，无需其他依赖）。

预期输出：先打印 校验通过（字段容错与取数链路自检），随后 analysis() 打印本篇样本的多列统计、主数值列 Top5 与异常扫描；__main__ 末尾按代码打印样例条数。这三步通了，说明 ①字段能正确取到 ②取数链路（MOCK 回放）可用 ③数据加工能产出可用结论。关掉 MOCK、填入真实 token，把 print 换成 INSERT 即生产脚本——本篇端点数据你已能独立拉下来、落本地并跑出分析表。

⑥ 编号坑

• 指数 MACD 走 /hz/pro/history/macd + 指数代码，别套用个股 /hs/ 路径。
• 指数点位量级远大于个股，DIF/DEA 绝对数值也大，看的是相对穿越而非绝对值。
• 指数 MACD 用于择时（金叉看多/死叉看空大盘），与个股信号分层使用。
• 指标字段名 dif/dea/macd 容错复用。
• 落库去重按 (代码,时间)。
• 复权：指数本身不计复权，但成分调整有跳变，长周期注意。
• MOCK 演示数据虚构，实盘核对真实点位指标。
• 指数趋势信号滞后，配合量能/市场宽度。

⑦ 小结与下篇预告

指数 Pro MACD 与个股同构，是择时大盘的核心。下一篇 #21 用指数 Pro MA 看均线系统。

本篇是《A股分钟级高频数据完全手册》第 20/42 篇。建议按系列顺序读：主推篇把体系与地基搭好，后面的单点篇（如 #02 起）都是换端点、加加工。

下一篇预告：【A股分钟级高频数据完全手册 #21】指数Pro版MA接口：沪深指数1分钟均线

⑧ 免责声明

本文代码仅用于接口调用与本地数据落库的技术演示，所有示例数据均为演示数据（MOCK 开关开启），不构成任何投资建议。股市有风险，入市需谨慎；实盘使用前请在小资金下充分验证接口稳定性与字段口径。

免费领取证书
数据来自 智兔数服（www.zhituapi.com）：零 SDK、纯 GET、免费版即可起步。

领取路径：进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」（邮箱验证 3 步即可拿到 token）。

把代码里的 你的token 换成你拿到的真实 token，上面的脚本就能直接拉取 Pro 版分钟级行情、实时盘口与逐笔成交数据，照着本系列 42 篇即可从零搭好属于你自己的分钟级高频数据底座。