【Python量化实战宝典 #02】历史K线接口全拆解:日线·分钟线·技术指标·复权处理(Python 可跑)
系列:《Python量化实战宝典》|连载项目 · 纯 GET 取数 · 取数层仅依赖 requests
适用:想做择时 / 动量 / 回测,但被「日线从哪取」「周月线有没有独立接口」「复权因子去哪找」绕晕的读者。本篇给沪深 A 股「日 K 线 / 分钟 K 线 / 技术指标(MACD·MA·BOLL·KDJ)/ 分红复权」9 个端点的分组地图、一次跑通的代码、pandas 多周期聚合与手工前复权,所有示例均为演示数据,不构成投资建议。
1. 你将得到什么
读完这一篇,你能拿走四样东西:
- 一张分组地图:9 个历史端点按「日 K / 分钟 K / 技术指标 / 分红」分成 4 组;
- 一次跑通的代码:
/hs/history拿日线,/hs/pro/history拿分钟线,/hs/history/{macd|ma|boll|kdj}直出指标; - 多周期聚合:日线 → 周线 / 月线用 pandas
resample,不必找"周月线专用接口"(官方没有单列); - 手工前复权:用
/hs/gs/jnff分红数据自己算,不再等"复权因子"接口。
代码全部自包含,复制进.py直接能跑,取数层不依赖 numpy / pandas,分析层仅需 pandas。
2. 本篇取数约定
- 全部接口都是GET + 路径参数,token 放在查询串里(
?token=xxx); - 统一基址
https://api.zhituapi.com; - 代码块里的
你的token是占位符,换成你的 token 即可; - 所有接口路径均取自官方已验证文档,跨篇零重复(本篇专注历史序列,与第 1 篇盘中实时不重叠)。
3. 9 个端点分 4 组
先建立地图。历史行情类一共 9 个端点,按用途分:
| 组 | 端点 | 用途 | 周期参数 |
|---|---|---|---|
| 日 K 线 | /hs/history/{code}.{m}/d/{n} | 最近 n 根日线 OHLC | d=日 |
| 日 K 线 | /hs/latest/{code}.{m}/d/{n} | 盘中最新日线(含当日未收盘) | d=日 |
| 分钟 K | /hs/pro/history/{code}.{m}/1/{n} | 最近 n 根一分钟 K | 1=分钟 |
| 分钟 K | /hs/pro/latest/{code}.{m}/1/{n} | 盘中最新分钟 K | 1=分钟 |
| 技术指标 | /hs/history/macd/{code}.{m}/d/{n} | 直出 DIF/DEA/柱 | d=日 |
| 技术指标 | /hs/history/ma/{code}.{m}/d/{n} | 直出均线(多周期) | d=日 |
| 技术指标 | /hs/history/boll/{code}.{m}/d/{n} | 直出布林带上中下轨 | d=日 |
| 技术指标 | /hs/history/kdj/{code}.{m}/d/{n} | 直出 K/D/J | d=日 |
| 复权数据源 | /hs/gs/jnff/{code} | 近年分红(用于手工前复权) | — |
{code}.{m}为带交易所后缀的代码,如600519.SH、000001.SZ;n为条数。周 / 月线没有独立端点:拿到日线后用 pandasresample聚合即可,本文第 5 节给出代码。5/15/30/60 分钟线同理用resample("5min")等聚合分钟 K。
4. 核心模板函数
importrequests,time BASE="https://api.zhituapi.com"TOKEN="你的token"# ---------- 1. 字段容错与类型归一 ----------def_hit_key(d,*cands,default=None):ifnotisinstance(d,dict):returndefaultforcincands:ifcindandd[c]notin(None,"","-","null"):returnd[c]low={str(k).lower():vfork,vind.items()}forcincands:v=low.get(str(c).lower())ifvnotin(None,"","-","null"):returnvreturndefaultdef_to_float(v,default=None):try:ifvin(None,"","-","null","None"):returndefaultreturnfloat(v)except(TypeError,ValueError):returndefault# ---------- 2. 统一请求:重试 + 退避 + 降级 ----------def_get(path,timeout=10,retries=2,backoff=0.6,default=None):q={"token":TOKEN}last=""foriinrange(retries+1):try:r=requests.get(BASE+path,params=q,timeout=timeout)ifr.status_code==200:try:returnr.json()exceptValueError:returndefault last="HTTP %s %s"%(r.status_code,(r.textor"").strip()[:80])exceptExceptionase:last="%s: %s"%(type(e).__name__,e)ifi<retries:time.sleep(backoff*(i+1))return{"_error":last}# ---------- 3. 取数封装 ----------_MOCK_DAILY=Nonedeffetch_daily(code,market="SZ",n=5):if_MOCK_DAILYisnotNone:return_MOCK_DAILYreturn_get("/hs/history/%s.%s/d/%d"%(code,market,n),default=[])deffetch_minute(code,market="SZ",n=240):return_get("/hs/pro/history/%s.%s/1/%d"%(code,market,n),default=[])deffetch_indicator(kind,code,market="SZ",n=5):# kind in {'macd','ma','boll','kdj'}return_get("/hs/history/%s/%s.%s/d/%d"%(kind,code,market,n),default=[])deffetch_dividend(code):return_get("/hs/gs/jnff/%s"%code,default=[])# ---------- 4. 字段归一:统一成标准 OHLC ----------defnorm_kline(rows):"""原始行 -> [{'date','open','high','low','close','volume','amount'}]"""out=[]forrinrowsor[]:out.append({"date":_hit_key(r,"date","t","时间",default="-"),"open":_to_float(_hit_key(r,"open","o")),"high":_to_float(_hit_key(r,"high","h")),"low":_to_float(_hit_key(r,"low","l")),"close":_to_float(_hit_key(r,"close","c")),"volume":_to_float(_hit_key(r,"volume","vol")),"amount":_to_float(_hit_key(r,"amount","amt")),})returnout# ---------- 5. 校验 ----------defrun_check():fake=[{"date":"2026-09-01","open":10.20,"high":10.55,"low":10.18,"close":10.50,"volume":120000,"amount":1.26e8},{"date":"2026-09-02","open":10.50,"high":10.62,"low":10.40,"close":10.45,"volume":98000,"amount":1.02e8},]global_MOCK_DAILY _MOCK_DAILY=fake# 演示数据替换网络调用rows=norm_kline(fetch_daily("600519","SH",5))_MOCK_DAILY=Noneassertlen(rows)==2androws[0]["close"]==10.50# 指标端点可取assertfetch_indicator("macd","600519","SH",5)isnotNone# 分红端点可取assertfetch_dividend("600519")isnotNoneprint("校验通过")if__name__=="__main__":run_check()print("-"*62)code,market="600519","SH"forname,fnin[("日线",lambda:fetch_daily(code,market,5)),("分钟K",lambda:fetch_minute(code,market,240)),("MACD",lambda:fetch_indicator("macd",code,market,5)),("BOLL",lambda:fetch_indicator("boll",code,market,5)),("近年分红",fetch_dividend)]:data=fn()ifname!="近年分红"elsefn(code)ifisinstance(data,dict)and"_error"indata:print("%-8s -> %s"%(name,data["_error"][:60]))else:print("%-8s -> %d 条"%(name,len(data)ifisinstance(data,list)else1))5. 跑通示例:拿到日线后怎么用
把上面的代码跑通后,下面演示「日线 → 周 / 月线聚合」与「用分红数据手工前复权」——这两步是回测前必做的,也是官方没有单列接口、必须自己算的部分。
importpandasaspd# —— 演示数据:真实调用替换为 fetch_daily('600519','SH',N) 的返回并经 norm_kline 归一 ——demo=[("2026-09-01",10.20,10.55,10.18,10.50,120000,1.26e8),("2026-09-02",10.50,10.62,10.40,10.45,98000,1.02e8),("2026-09-03",10.45,10.70,10.43,10.68,150000,1.58e8),("2026-09-04",10.68,10.80,10.60,10.75,132000,1.41e8),("2026-09-07",10.75,10.90,10.70,10.88,110000,1.18e8),]df=pd.DataFrame(demo,columns=["date","open","high","low","close","volume","amount"])df["date"]=pd.to_datetime(df["date"])df=df.set_index("date").sort_index()# 日线 -> 周线 / 月线聚合(官方无周月线独立接口,用 resample)weekly=df.resample("W").agg({"open":"first","high":"max","low":"min","close":"last","volume":"sum","amount":"sum"})monthly=df.resample("ME").agg({"open":"first","high":"max","low":"min","close":"last","volume":"sum","amount":"sum"})print("周线:\n",weekly)print("月线:\n",monthly)# 手工前复权:用近年分红 /hs/gs/jnff 取到的除权事件defforward_adjust(close,events):""" close : pd.Series,index=日期(升序),value=未复权收盘价 events: list of (ex_date:str, div:float, pre_close:float) pre_close = 除权前一交易日收盘价 """s=close.astype(float).copy()forex_date,div,pre_closeinevents:ex=pd.Timestamp(ex_date)factor=(pre_close-div)/pre_close s.loc[s.index<ex]=s.loc[s.index<ex]*factorreturns raw=df["close"].copy()adj=forward_adjust(raw,[("2026-06-10",0.50,12.00)])print(pd.DataFrame({"未复权":raw,"前复权":adj.round(3)}))若你的 K 线端点本身支持复权参数(前复权 / 后复权),直接传参最省事;否则用上面的分红法自行复权。回测全程必须统一口径,不能把复权价和未复权价混在一起算收益。
6. 坑与注意事项
坑 1:周 / 月线没有独立接口,别拼不存在的路径。
官方只提供日线(d)和一分钟线(1),周 / 月线是用日线resample聚合出来的。直接去猜/hs/history/xxx/w/n这类路径会 404。
坑 2:复权因子没有单列接口,要用分红自己算。
找不到"复权因子"端点时,用/hs/gs/jnff的近年分红数据,按"复权因子 =(除权前收盘 − 每股分红)/ 除权前收盘"向前连乘即可(见第 5 节)。
坑 3:技术指标端点返回的是已算好的值,不是原始价。/hs/history/macd等返回 DIF/DEA/柱(或 K/D/J、上中下轨),直接消费即可,不必自己再实现一遍指标,但字段名要先print看一行确认。
坑 4:分钟线量大,聚合前先想清楚范围。
Pro 分钟线一次可能返回几百上千根,做 5/15/30/60 分钟聚合前先限定n,否则内存和限频都吃紧。
坑 5:停牌日缺失,resample 不会补 NaN。resample只会对已有日期聚合,停牌日不会自动补空。需要完整日历的,用交易日历reindex+fillna处理。
7. 小结与下篇预告
本篇把 9 个历史端点分成 4 组,给出日线 / 分钟线 / 技术指标 / 分红的取数封装,并用 pandas 演示了周月线聚合与手工前复权——这两块是官方没单列、必须自己处理的核心。
下一篇预告:《【Python量化实战宝典 #03】指数与板块行情接口全拆解:大盘·行业·概念一键取(Python 可跑)》:用/hz与/hs/list/sectors一组接口,把指数实时、行业板块、概念板块的成分与涨跌一次取全,择时与行业中性化都靠它。
8. 免责声明
本文仅演示沪深 A 股历史行情数据的取数与聚合方法,所有代码示例均为演示数据,未含任何真实行情数值,不构成投资建议,亦不承诺收益。