【零依赖量化数据实战 #24】指数技术指标实战:MACD·MA·KDJ·BOLL
摘要:【零依赖量化数据实战 #24】指数技术指标实战:MACD·MA·KDJ·BOLL 系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK
系列:《零依赖量化数据实战》|零依赖 · 纯 GET · 不 import 任何 SDK
适用:想用 Python 直接消费智兔已算好的指数技术指标(MACD / MA / KDJ / BOLL),一行出金叉死叉、多空头排列、超买超卖、开口收口信号的量化爱好者;数据由智兔数服提供,不依赖任何行情终端。
1. 你将得到什么
- 5 个官方接口的最小可用封装,覆盖指数技术指标与指数历史分时两类需求:
GET /hz/history/macd/{index}/{period}:指数 MACD(DIF / DEA / MACD 柱)GET /hz/history/ma/{index}/{period}:指数多均线 MAGET /hz/history/kdj/{index}/{period}:指数 KDJGET /hz/history/boll/{index}/{period}:指数 BOLL(上轨 / 中轨 / 下轨)GET /hz/history/fsjy/{index}/{period}:指数历史分时交易数据- 一套对字段名不敏感的信号函数:上游返回里「DIF」可能叫
dif/DIF,「上轨」可能叫up/upper/boll_up,用候选键命中,不写死。 - 用
lt取最近 N 条,即可在本地一行出 MACD 金叉/死叉、MA 多空头排列、KDJ 超买超卖、BOLL 开口收口。
2. 端点语义表
GET https://api.zhituapi.com/hz/history/macd/000001.SH/d?token=你的智兔token<=30
-> 上证指数 日线 MACD,最近 30 条
返回:list,每条含 DIF/DEA/MACD 类字段(字段名随上游变)
GET https://api.zhituapi.com/hz/history/ma/000001.SH/d?token=你的智兔token<=30
-> 上证指数 日线 多均线 MA(ma5/ma10/ma20/ma60...)
GET https://api.zhituapi.com/hz/history/kdj/000001.SH/d?token=你的智兔token<=30
-> 上证指数 日线 KDJ(K/D/J)
GET https://api.zhituapi.com/hz/history/boll/000001.SH/d?token=你的智兔token<=30
-> 上证指数 日线 BOLL(up/mid/low)
GET https://api.zhituapi.com/hz/history/fsjy/000001.SH/d?token=你的智兔token&st=20240101&et=20241231
-> 上证指数 历史分时(无 lt 参数,按 st/et 取区间)
鉴权:token 走查询参数;{index} 为路径参数(必须带市场后缀,如 000001.SH);{period} 为路径参数(d 日线、5/15/30/60 分钟、w/m/y 周/月/年线);lt 取最新 N 条;st/et 格式 YYYYMMDD 或 YYYYMMDDhhmmss。数据来自 智兔数服(www.zhituapi.com)。
3. 字段名不固定?用候选键命中
上游四类指标返回字段名都不统一(中文/英文、大小写混用)。统一用 _hit_key 候选键命中抽取,对字段名不敏感:
def _hit_key(d, keys):
if not isinstance(d, dict):
return None
for k in keys:
if k in d and d[k] is not None:
return d[k]
low = {str(x).lower(): x for x in d.keys()}
for k in keys:
kl = k.lower()
if kl in low:
return d[low[kl]]
return None
4. 核心模板函数
import sys, requests
BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token" # 占位,换成你申请的真实 token
def _hit_key(d, keys):
if not isinstance(d, dict):
return None
for k in keys:
if k in d and d[k] is not None:
return d[k]
low = {str(x).lower(): x for x in d.keys()}
for k in keys:
kl = k.lower()
if kl in low:
return d[low[kl]]
return None
def _get(path, params=None):
p = dict(params or {})
p["token"] = TOKEN
try:
r = requests.get(f"{BASE}{path}", params=p, timeout=10)
except Exception as e:
return None, f"网络异常:{e}"
if r.status_code != 200:
return None, f"{r.status_code} {r.text.strip()[:140]}"
try:
return r.json(), None
except Exception:
return None, f"非 JSON:{r.text.strip()[:140]}"
def fetch_indicator(kind, index="000001.SH", period="d", lt=30):
"""kind: macd|ma|kdj|boll。返回 (list_of_rows, err)。"""
data, err = _get(f"/hz/history/{kind}/{index}/{period}", {"lt": lt})
if err:
return None, err
rows = data
if isinstance(data, dict):
for k in ("data", "list", "rows", "items"):
if isinstance(data.get(k), list):
rows = data[k]; break
else:
rows = [data]
if not isinstance(rows, list):
rows = [rows]
return rows, None
# ===== 信号判断(纯函数,便于离线 selftest)=====
def macd_signal(rows):
if not rows or len(rows) < 2:
return None
a, b = rows[-2], rows[-1]
da = _hit_key(a, ["dif", "DIF"]); db = _hit_key(a, ["dea", "DEA"])
ea = _hit_key(b, ["dif", "DIF"]); eb = _hit_key(b, ["dea", "DEA"])
if None in (da, db, ea, eb):
return None
if da <= db and ea > eb:
return "golden" # 金叉
if da >= db and ea < eb:
return "dead" # 死叉
return "hold"
def ma_alignment(row):
mas = {}
for k, v in row.items():
if v is None:
continue
kl = str(k).lower()
for p in (5, 10, 20, 30, 60, 120, 250):
if kl in (f"ma{p}", f"ma{p}_close", f"ma{p}close"):
try:
mas[p] = float(v)
except Exception:
pass
if len(mas) < 2:
return "mix"
vals = [mas[p] for p in sorted(mas)]
if all(vals[i] <= vals[i+1] for i in range(len(vals)-1)):
return "bull" # 短周期在下、长周期在上 -> 多头排列
if all(vals[i] >= vals[i+1] for i in range(len(vals)-1)):
return "bear" # 空头排列
return "mix"
def kdj_zone(row):
j = _hit_key(row, ["j", "J"]); k = _hit_key(row, ["k", "K"])
if j is None and k is None:
return None
if j is not None:
j = float(j)
if j > 100: return "overbought"
if j < 0: return "oversold"
if k is not None:
k = float(k)
if k > 80: return "overbought"
if k < 20: return "oversold"
return "neutral"
def boll_band(row):
up = _hit_key(row, ["up", "upper", "boll_up", "BOLL_UP"])
mid = _hit_key(row, ["mid", "middle", "BOLL_MID", "mb"])
low = _hit_key(row, ["low", "lower", "boll_low", "BOLL_LOW"])
if None in (up, mid, low):
return None
return float(up), float(mid), float(low)
def summarize(rows, kind):
if not rows:
return " 无数据"
last = rows[-1]
if kind == "macd":
return f" 最新 macd 信号={macd_signal(rows)}; DIF={_hit_key(last, ['dif','DIF'])} DEA={_hit_key(last, ['dea','DEA'])}"
if kind == "ma":
return f" 最新 MA 排列={ma_alignment(last)}"
if kind == "kdj":
return f" 最新 KDJ 区={kdj_zone(last)}; K={_hit_key(last, ['k','K'])} D={_hit_key(last, ['d','D'])} J={_hit_key(last, ['j','J'])}"
if kind == "boll":
b = boll_band(last)
return f" 最新 BOLL up/mid/low={b}" if b else " 最新 BOLL 字段缺失"
return " ok"
def selftest():
# 合成数据仅逻辑自验,非真实行情
assert macd_signal([{"dif": 0.10, "dea": 0.20}, {"dif": 0.30, "dea": 0.25}]) == "golden"
assert macd_signal([{"dif": 0.30, "dea": 0.20}, {"dif": 0.10, "dea": 0.25}]) == "dead"
assert ma_alignment({"ma5": 1.0, "ma10": 1.1, "ma20": 1.2}) == "bull"
assert ma_alignment({"ma5": 1.2, "ma10": 1.1, "ma20": 1.0}) == "bear"
assert kdj_zone({"j": 110, "k": 85}) == "overbought"
assert kdj_zone({"j": -5, "k": 15}) == "oversold"
assert kdj_zone({"j": 50, "k": 50}) == "neutral"
assert boll_band({"up": 1.3, "mid": 1.0, "low": 0.7}) == (1.3, 1.0, 0.7)
assert boll_band({"BOLL_UP": 1.3, "BOLL_MID": 1.0, "BOLL_LOW": 0.7}) == (1.3, 1.0, 0.7)
print("selftest PASS")
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "--selftest":
selftest()
else:
for kind in ("macd", "ma", "kdj", "boll"):
rows, err = fetch_indicator(kind)
print(f"{kind} -> ({rows}, {err})")
if rows:
print(summarize(rows, kind))
data, err = _get("/hz/history/fsjy/000001.SH/d", {"st": "20240101", "et": "20241231"})
print(f"fsjy -> ({data}, {err})")
5. 代码自验结果
离线 selftest(合成数据,仅验证逻辑,不含任何真实行情):
selftest PASS
联网实测(占位 token,真实返回):
--- 联网实测(占位 token,预期 404 102:Licence证书不存在)---
macd -> (None, 404 102:Licence证书(你的智兔token)不存在)
ma -> (None, 404 102:Licence证书(你的智兔token)不存在)
kdj -> (None, 404 102:Licence证书(你的智兔token)不存在)
boll -> (None, 404 102:Licence证书(你的智兔token)不存在)
fsjy -> (None, 404 102:Licence证书(你的智兔token)不存在)
把
TOKEN = "你的智兔token"换成你申请的真实 token,上述函数即可打印指数 MACD、MA、KDJ、BOLL 技术指标数据。本文未编造任何真实数值。
6. 坑与注意事项
- 102 不代表路径对:
404 102是「证书不存在」(鉴权先于路由),路径合法与否要靠客户端白名单自查,不能凭报错反推接口可用。 - 指数代码必须带市场后缀:写
000001会路由错误,必须000001.SH;深证成指399001.SZ、创业板指399006.SZ同理。 lt取最近 N 条:技术指标接口用lt(最新条数)控制返回量,省略则拉全历史;fsjy历史分时没有lt,用st/et区间。- 字段名随上游变:「DIF」可能
dif/DIF,「上轨」可能up/upper/boll_up,务必用候选键命中,别硬写英文键。 - 信号是「本地算」不是「接口给」:接口只给指标序列,金叉/死叉/排列/超买超卖这些判断都在
macd_signal/ma_alignment/kdj_zone里本地完成,可按需改写阈值。
7. 小结与下篇预告
本篇把「指数 MACD / MA / KDJ / BOLL + 历史分时」拧成了 5 个零依赖接口的最小封装,重点解决了指数代码带市场后缀、lt 取最近 N 条、指标字段名中英文混用三个坑,配候选键命中即可一行出信号。
下一篇计划写 #25《北交所技术指标与基本面》:讲解如何用官方接口拉取北交所股票的技术指标(MACD·MA·KDJ·BOLL)与基本面数据(股本、十大股东、流通股东、财务主要指标),把「北交所」也接进零依赖工作流。
8. 免责声明
本文仅演示公开数据接口的用法,所有代码示例均以占位 token 自验,未含任何真实数据;文中合成数据仅为逻辑自验用途,不构成投资建议,亦不承诺收益。
免费领取证书
数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。
领取路径:进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」(邮箱验证 3 步即可拿到 token)。
把代码里的 你的智兔token 换成你拿到的真实 token,上面的脚本就能直接打印指数 MACD、MA、KDJ、BOLL 技术指标数据。