A股量化数据工程：从 REST 接口到策略信号（第 2 篇）：指数分时数据解析与清洗-平芜编程栈

A股量化数据工程：从 REST 接口到策略信号（第 2 篇）：指数分时数据解析与清洗

前言

第 1 篇接好了指数列表与实时行情，但实时只给「当前快照」。要做择时、算分时均线，必须拿到连续的分时序列。本篇用hsindex/latest（最新分时）与hsindex/history（历史分时）拉取序列，并讲解分时数据最常见的清洗动作：缺值补齐、按时间排序、与分时 MA 对齐。

一、接口原始字段梳理

1）指数最新分时接口

接口地址：https://api.biyingapi.com/hsindex/latest/{指数代码.市场}/{级别}/{LICENCE}?lt={数量}

字段	说明
t	时间
o	开
h	高
l	低
c	收（该分钟 / 该级别收盘）
v	成交量
a	成交额（元 ¥）
pc	涨跌幅

2）指数历史分时接口

接口地址：https://api.biyingapi.com/hsindex/history/{指数代码.市场}/{级别}/{LICENCE}?st={start}&et={end}

字段同最新分时，按st/et区间返回。

3）指数分时 MA 接口

接口地址：https://api.biyingapi.com/hsindex/history/ma/{指数代码.市场}/{级别}/{LICENCE}?st={start}&et={end}

字段	说明
t	时间
ma3, ma5, ma10, ma15, ma20, ma30, ma60, ma120, ma200, ma250	各周期均线，缺数据周期为`null`

提示：级别参数level取d（日线级分时）等，三个接口必须一致，否则序列对不齐。

二、自研衍生统计指标

分时缺口识别：相邻两根t间隔异常（如跨午休、停牌），标记为缺口，避免均线计算串段。
MA 回填：接口给的ma*在序列头部为null，用本地滑动窗口重算ma5/ma10补齐前段，保证全序列可用。
量价背离（轻量）：pc创新高但v未放大，记一次背离信号。

以上指标接口不会直接返回，全部基于原始字段业务代码运算。

三、数据表设计

CREATETABLEIFNOTEXISTSindex_minute(dmTEXT,tTEXT,oREAL,hREAL,lREAL,cREAL,vREAL,aREAL,pcREAL,ma5REAL,ma10REAL,PRIMARYKEY(dm,t));

四、完整代码

importrequestsimportsqlite3importtime LICENCE="YOUR_LICENCE"BASE="https://api.biyingapi.com"DB_PATH="quant.db"defbiying_api_get_retry(url,timeout=15,max_retry=3):foriinrange(max_retry):try:r=requests.get(url,timeout=timeout)r.raise_for_status()returnr.json()exceptExceptionase:ifi==max_retry-1:raisetime.sleep(1+i)deffetch_index_latest(code,market,level="d",lt=5):"""最新分时：t/o/h/l/c/v/a/pc。"""url=f"{BASE}/hsindex/latest/{code}.{market}/{level}/{LICENCE}?lt={lt}"data=biying_api_get_retry(url)returndata.get("data",[])ifisinstance(data,dict)else[]deffetch_index_history(code,market,level,st,et):"""历史分时区间：st/et 为 YYYYMMDD。"""url=f"{BASE}/hsindex/history/{code}.{market}/{level}/{LICENCE}?st={st}&et={et}"data=biying_api_get_retry(url)returndata.get("data",[])ifisinstance(data,dict)else[]deffetch_index_ma(code,market,level,st,et):"""分时 MA：t/ma3..ma250。"""url=f"{BASE}/hsindex/history/ma/{code}.{market}/{level}/{LICENCE}?st={st}&et={et}"data=biying_api_get_retry(url)returndata.get("data",[])ifisinstance(data,dict)else[]deffill_ma_local(rows,windows=(5,10)):"""接口 MA 头部为 null 时，本地滑动窗口补齐 ma5/ma10。"""closes=[r.get("c")forrinrows]forwinwindows:key=f"ma{w}"fori,rinenumerate(rows):ifr.get(key)isNoneandi+1>=w:seg=closes[i+1-w:i+1]ifall(xisnotNoneforxinseg):r[key]=round(sum(seg)/w,4)returnrowsdefdetect_gap(rows,max_gap_sec=3600):"""相邻 t 间隔超阈值记为缺口（返回缺口索引列表）。"""gaps=[]foriinrange(1,len(rows)):try:t0=time.mktime(time.strptime(rows[i-1]["t"],"%Y-%m-%d %H:%M:%S"))t1=time.mktime(time.strptime(rows[i]["t"],"%Y-%m-%d %H:%M:%S"))ift1-t0>max_gap_sec:gaps.append(i)exceptException:passreturngapsdefsave_minute(dm,rows):cols=["dm","t","o","h","l","c","v","a","pc","ma5","ma10"]con=sqlite3.connect(DB_PATH)forrinrows:r["dm"]=dm vals=[r.get(c)forcincols]ph=",".join("?"*len(cols))con.execute(f"INSERT OR REPLACE INTO index_minute({','.join(cols)}) VALUES({ph})",vals)con.commit();con.close()if__name__=="__main__":code,market,level="000001","SH","d"# 拉最近 5 根最新分时latest=fetch_index_latest(code,market,level,lt=5)latest=fill_ma_local(latest)save_minute(f"{code}.{market}",latest)print(f"最新分时{len(latest)}根，缺口索引:",detect_gap(latest))print("首根:",latest[0]iflatestelseNone)