【Python 量化取数指南 #05】历史行情数据 API 评测与回测实战
摘要:【Python 量化取数指南 #05】历史行情数据 API 评测与回测实战 系列:《Python 量化取数指南》|连载项目 · 纯 GET 取数 · 仅依赖 requests 数据:由智兔数服提供。更多接口见 智
系列:《Python 量化取数指南》|连载项目 · 纯 GET 取数 · 仅依赖 requests
数据:由智兔数服提供。更多接口见 智兔数服技术博客。
1. 你将得到什么
- 拉指数/个股历史 K 线的完整代码(日线、周线、月线)
- 一个本地计算均线 + 均线交叉信号的回测示例(不依赖任何回测框架)
- 一个离线
run_check(),不填 token 也能验证取数逻辑
2. 本篇取数约定
- 历史 K 线:
/hz/history/fsjy/{code}.{market}/{lvl}(lvl=d/w/m/5/15/30/60) - 指数 MA:
/hz/history/ma/{code}.{market}/{lvl}(直接取均线,省得自己算) - 北交所 MA:
/bj/history/ma/{code}/d/n - 请求:
GET https://api.zhituapi.com<path>?token=<你的智兔token> - 默认未复权,复权与对齐处理见第 14 篇
3. 核心模板(全系列复用)
import time, json, requests
BASE = "https://api.zhituapi.com"
TOKEN = "你的智兔token" # 演示证书(免费版)即可起步
def _get(path, params=None, timeout=15, retry=3, backoff=1.5):
params = dict(params or {})
params["token"] = TOKEN
url = BASE + path
last = None
for i in range(retry):
try:
r = requests.get(url, params=params, timeout=timeout)
if r.status_code != 200:
last = f"HTTP {r.status_code} {r.text[:120]}"
time.sleep(backoff * (i + 1)); continue
try:
return r.json(), None
except ValueError:
last = f"非JSON响应: {r.text[:120]}"
return None, last
except requests.RequestException as e:
last = str(e); time.sleep(backoff * (i + 1))
return None, last
def _hit_key(d, *keys, default=None):
if not isinstance(d, dict):
return default
for k in keys:
if k in d and d[k] not in (None, "", []):
return d[k]
return default
def _to_float(x, default=float("nan")):
try:
return float(x)
except (TypeError, ValueError):
return default
4. 跑通示例:拉 K 线 + 均线交叉回测
def load_kline(code="000001.SH", lvl="d", n=60):
data, err = _get(f"/hz/history/fsjy/{code}/{lvl}")
if err:
return None, err
bars = data if isinstance(data, list) else (data.get("data") or [])
rows = []
for b in bars:
rows.append({
"date": _hit_key(b, "date", "rq"),
"close": _to_float(_hit_key(b, "close", "sp", "收盘")),
})
return rows[-n:], None
def sma(vals, window):
out = []
for i in range(len(vals)):
if i + 1 < window:
out.append(float("nan"))
else:
out.append(sum(vals[i+1-window:i+1]) / window)
return out
def ma_cross_backtest(rows, fast=5, slow=20):
closes = [r["close"] for r in rows if r["close"] == r["close"]]
ma_f = sma(closes, fast)
ma_s = sma(closes, slow)
sig = 0
trades = 0
for i in range(1, len(closes)):
if ma_f[i] > ma_s[i] and ma_f[i-1] <= ma_s[i-1]:
sig = 1; trades += 1
elif ma_f[i] < ma_s[i] and ma_f[i-1] >= ma_s[i-1]:
sig = 0; trades += 1
return trades
def run_check():
synth = [{"date": f"2024{i:03d}", "close": 100 + (i % 10)} for i in range(60)]
t = ma_cross_backtest(synth)
print(f" [run_check] 合成数据信号切换次数: {t}")
if __name__ == "__main__":
rows, err = load_kline("000001.SH", "d", 120)
if err:
print("K线失败:", err)
else:
t = ma_cross_backtest(rows)
print(f" 上证日线 {len(rows)} 根,均线交叉切换 {t} 次")
run_check()
返回字段说明:K 线 list 每项 date/rq(日期)、open/zk(开)、close/sp(收)、high/zg(高)、low/zd(低)、volume/cjl(量)、amount/cje(额)。均线端点 /hz/history/ma/ 返回同样结构但含 ma5/ma20 等字段。
5. 坑与注意事项
- 默认未复权:分红送股会造成「假跳空」,回测前必须复权(见第 14 篇)。
- 停牌日缺根:个股可能缺交易日 K 线,直接用
date做索引会错位,按位置取更安全。 - 均线起点是 NaN:前
window-1根算不出均线,信号判断要i >= window。 - 指数 vs 个股:指数代码
000001.SH,个股600519.SH,别混。 - 回测≠实盘:本示例只演示信号计数,不含手续费/滑点/仓位,真实策略要补。
- 字段名三套:
close/sp/收盘,用_hit_key。
6. 常见报错速查
| 报错 / 现象 | 原因 | 处理 |
|---|---|---|
404 |
代码缺后缀/级别错 | 检查 {code}.{market}/{lvl} |
| 返回空 | 非交易日区间 | 放宽日期/换代码 |
ZeroDivisionError |
窗口>数据量 | 保证 n >= slow |
KeyError |
字段名不符 | print(rows[0]) 看真实 key |
7. 小结与下一篇预告
小结:拉 K 线用 /hz/history/fsjy/,均线可自己算也可走 /hz/history/ma/;回测核心是「去 NaN + 按位置索引 + 信号切换计数」。注意未复权会污染结果。
下一篇计划写 #06《实时行情 API 实测与可用性对比》:横向对比实时行情端点的延迟与可用性,告诉你盘中盯盘该用哪个。
8. 免责声明
本文仅演示公开数据接口的用法,所有代码示例均为演示数据,不构成任何投资建议;实际返回字段以接口文档与你的证书权限为准。数据由 智兔数服 提供,更多接口示例见 技术博客。
免费领取证书 / 查看完整接口文档,可前往 智兔数服官网。
想亲自试一下?免费获取证书