【智兔数服|A股分钟级高频数据完全手册 #02】Python调用股票1分钟数据接口:Pro版最新分时取数与字段解析
摘要:【智兔数服|A股分钟级高频数据完全手册 #02】Python调用股票1分钟数据接口:Pro版最新分时取数与字段解析 系列:《A股分钟级高频数据完全手册》|连载项目 · 纯 GET 取数 · 仅依赖 request
系列:《A股分钟级高频数据完全手册》|连载项目 · 纯 GET 取数 · 仅依赖 requests · Pro 版分钟级高频
① 你将得到什么
分钟级 K 线是高频策略的原子数据——任何均线、MACD、量价因子都得先在分钟线上算。Pro 版最新 1 分钟分时接口直接给你某只票最近 N 条 1 分钟 bar(开高低收、量、额),比自己从 tick 聚合省事得多,也比普通分时更细。本篇就把这个端点拉通:从鉴权调用到字段解析,再到本地做一版成交量 Top/均值统计。
本篇产出:拉取单只股票最近的 N 条 Pro 版 1 分钟分时数据,解析开/收/高/低/成交量/成交额字段,本地做成交量均值与 TopN 加工,得到一份可直接落库的分钟 bar 表。
读完这篇,你应当能独立把本篇端点数据拉下来、落本地,并跑出一份可用的加工分析表。
② 取数约定
一个端点拉数据,返回里字段名可能中英文混用(不同接口版本、不同数据源习惯不同)。演示用 你的token 占位,代码里 MOCK=True 时回放演示数据,接真实 token 把 MOCK 改成 False 即可,无需改动其他逻辑。
所有请求走 https://api.zhituapi.com + ?token=,纯 GET、仅依赖 requests,不依赖任何付费 SDK。字段口径以接口实际返回为准,本文用 _hit_key 做了大小写/中英文容错;落地前建议先用 print(r[0].keys()) 核对一次你账号下的真实字段名。
落地建议:把每天拉到的数据按 代码 + 日期 做主键写入本地 SQLite/CSV,避免重复插入;历史回看靠本地累积,而非反复调接口。本篇示例代码已包含字段容错映射与本地加工函数,你只需把 print 替换为写入语句。
③ 端点分组地图
| 端点 | 用途 |
|---|---|
/hs/pro/latest/{code}.SZ/1/n |
Pro 版最新 1 分钟分时:开高低收/量/额,末尾 n 为返回条数 |
本篇涉及字段(落地建表参考):代码、名称、时间、开、收、高、低、成交量、成交额。
返回结构为 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("/hs/pro/latest/000001.SZ/1/5", 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"),
"开": _hit_key(r, "o", "open"),
"收": _hit_key(r, "c", "close"),
"高": _hit_key(r, "h", "high"),
"低": _hit_key(r, "l", "low"),
"成交量": _hit_key(r, "v", "vol", "volume"),
"成交额": _hit_key(r, "a", "amount")
})
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 = "成交量"
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 = {
"/hs/pro/latest/000001.SZ/1/5": [{'code': '000001.SZ', 'name': '平安银行', 't': '2024-01-10 09:31:00', 'o': '10.20', 'c': '10.25', 'h': '10.26', 'l': '10.19', 'v': '12000', 'a': '1224000'}, {'code': '600519.SH', 'name': '贵州茅台', 't': '2024-01-10 09:31:00', 'o': '1685.0', 'c': '1688.5', 'h': '1689.0', 'l': '1684.5', 'v': '3500', 'a': '5899750'}, {'code': '000001.SZ', 'name': '平安银行', 't': '2024-01-10 09:31:00', 'o': '7.48', 'c': '7.52', 'h': '7.52', 'l': '7.47', 'v': '8800', 'a': '897600'}, {'code': '600519.SH', 'name': '贵州茅台', 't': '2024-01-10 09:31:00', 'o': '1348', 'c': '1350.8', 'h': '1351.2', 'l': '1347.6', 'v': '2800', 'a': '4.7198e+06'}, {'code': '000001.SZ', 'name': '平安银行', 't': '2024-01-10 09:31:00', 'o': '8.84', 'c': '8.88', 'h': '8.89', 'l': '8.83', 'v': '10400', 'a': '1.0608e+06'}, {'code': '600519.SH', 'name': '贵州茅台', 't': '2024-01-10 09:31:00', 'o': '1572.67', 'c': '1575.93', 'h': '1576.4', 'l': '1572.2', 'v': '3266.67', 'a': '5.50643e+06'}, {'code': '000001.SZ', 'name': '平安银行', 't': '2024-01-10 09:31:00', 'o': '10.2', 'c': '10.25', 'h': '10.26', 'l': '10.19', 'v': '12000', 'a': '1.224e+06'}, {'code': '600519.SH', 'name': '贵州茅台', 't': '2024-01-10 09:31:00', 'o': '1797.33', 'c': '1801.07', 'h': '1801.6', 'l': '1796.8', 'v': '3733.33', 'a': '6.29307e+06'}, {'code': '000001.SZ', 'name': '平安银行', 't': '2024-01-10 09:31:00', 'o': '11.56', 'c': '11.62', 'h': '11.63', 'l': '11.55', 'v': '13600', 'a': '1.3872e+06'}, {'code': '600519.SH', 'name': '贵州茅台', 't': '2024-01-10 09:31:00', 'o': '2022', 'c': '2026.2', 'h': '2026.8', 'l': '2021.4', 'v': '4200', 'a': '7.0797e+06'}, {'code': '000001.SZ', 'name': '平安银行', 't': '2024-01-10 09:31:00', 'o': '12.92', 'c': '12.98', 'h': '13', 'l': '12.91', 'v': '15200', 'a': '1.5504e+06'}, {'code': '600519.SH', 'name': '贵州茅台', 't': '2024-01-10 09:31:00', 'o': '2246.67', 'c': '2251.33', 'h': '2252', 'l': '2246', 'v': '4666.67', 'a': '7.86633e+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("/hs/pro/latest/000001.SZ/1/5", 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_02.py,MOCK=True 直接跑(Python3 + requests 即可,无需其他依赖)。
预期输出:先打印 校验通过(字段容错与取数链路自检),随后 analysis() 打印本篇样本的多列统计、主数值列 Top5 与异常扫描;__main__ 末尾按代码打印样例条数。这三步通了,说明 ①字段能正确取到 ②取数链路(MOCK 回放)可用 ③数据加工能产出可用结论。关掉 MOCK、填入真实 token,把 print 换成 INSERT 即生产脚本——本篇端点数据你已能独立拉下来、落本地并跑出分析表。
⑥ 编号坑
- 路径里 {code}.SZ 的交易所后缀必须带(.SZ/.SH/.BJ),且大小写敏感,写错返回空或报错。
- 路径第 3 段 1 是周期标记(1=1 分钟),不要误填成天数;末尾 n 是返回条数,不是日期。
- 返回的是最新 N 条快照而非历史区间,要做历史必须每天按日期落库累积,别指望它回放过去某天。
- 成交量单位通常是手、成交额是元,跨接口汇总前先统一单位,避免量级错配。
- 时间字段格式可能是字符串或时间戳,统一转 datetime 再入库,别直接存字符串导致排序/聚合出错。
- 复权:最新分时默认不含复权,做长周期回测要另行处理复权因子,别把不复权的价格直接喂给策略。
- 字段名可能中英文混用(o/open、v/vol),用 _hit_key 容错,直接 r['o'] 可能 KeyError。
- MOCK 回放数据为演示构造(数值虚构),实盘前务必用真 token 跑一次核对字段名与量级。
⑦ 小结与下篇预告
Pro 版最新 1 分钟分时是最直接的原子数据入口,拉通后你就有了做任意分钟级因子的原料。下一篇 #03 我们用 Pro 版历史 1 分钟 K 线接口批量回放更长区间,并讲断点续传雏形。
本篇是《A股分钟级高频数据完全手册》第 02/42 篇。建议按系列顺序读:主推篇把体系与地基搭好,后面的单点篇(如 #02 起)都是换端点、加加工。
下一篇预告:【A股分钟级高频数据完全手册 #03】股票历史1分钟K线数据接口怎么用:Pro版批量拉取与本地缓存
⑧ 免责声明
本文代码仅用于接口调用与本地数据落库的技术演示,所有示例数据均为演示数据(MOCK 开关开启),不构成任何投资建议。股市有风险,入市需谨慎;实盘使用前请在小资金下充分验证接口稳定性与字段口径。
免费领取证书
数据来自 智兔数服(www.zhituapi.com):零 SDK、纯 GET、免费版即可起步。
领取路径:进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」(邮箱验证 3 步即可拿到 token)。
把代码里的 你的token 换成你拿到的真实 token,上面的脚本就能直接拉取 Pro 版分钟级行情、实时盘口与逐笔成交数据,照着本系列 42 篇即可从零搭好属于你自己的分钟级高频数据底座。