当前位置：首页 > news >正文

多接口联动实战：本地数据引擎如何构建完整量化分析体系

news 2026/7/22 16:51:07

多接口联动实战：本地数据引擎如何构建完整量化分析体系

做量化投资的人都明白，单一数据源远不足以支撑完整的策略分析。你需要实时行情看当前价格，L2指标看资金动向，五档盘口看买卖力量，分时成交看日内趋势，大单交易看机构动向，这些数据必须联动起来才能做出准确的判断。但想把这些数据整合到一起，在线API的痛点比想象中多。后来换成本地落盘方案，才算真正把多接口联动的流程跑通了。今天想聊聊用ig50本地数据引擎构建完整量化分析体系的实战经验。

在线API做多接口联动的痛点很明显。首先是数据延迟不一致，不同接口的延迟从几十毫秒到几小时不等，想把实时行情和L2指标对齐，经常出现时间差。其次是请求频次限制，每个接口都有限频，想同时拉全市场的实时行情、L2指标和五档盘口，根本做不到。还有数据格式不统一，不同接口返回的数据格式千差万别，需要写大量的适配代码。我之前写过一个多接口联动的分析脚本，用在线API跑，光是处理数据对齐和格式转换就花了一半的代码量，还经常因为接口变更导致脚本失效。

后来换成ig50本地数据引擎，才发现多接口联动原来可以这么顺畅。所有数据都按统一的目录结构和JSON格式存储在本地，实时行情在time/real/{股票代码}，L2行情指标在time/real/trace/l2sign/{股票代码}，五档盘口在time/real/trace/level5/{股票代码}，分时成交在time/real/trace/timedeal/{股票代码}，逐笔交易在time/real/trace/onebyone/{股票代码}，大单交易在time/real/trace/bigdeal/{股票代码}。程序直接读本地文件，没有网络延迟，没有请求频次限制，数据格式统一，想怎么联动就怎么联动。

先看多接口联动的基础框架，下面这段代码是我量化分析体系的核心：

importjsonimportpandasaspdimportnumpyasnpfrompathlibimportPathfromdatetimeimportdatetime DATA_DIR=Path('/ig50-data')# 统一的数据加载函数defload_json(path):"""读取JSON文件"""ifnotpath.exists():returnNonewithopen(path,'r',encoding='utf-8')asf:returnjson.load(f)defload_realtime(code):"""读取实时行情"""returnload_json(DATA_DIR/'time'/'real'/code)defload_l2(code):"""读取L2行情指标"""returnload_json(DATA_DIR/'time'/'real'/'trace'/'l2sign'/code)defload_level5(code):"""读取五档盘口"""returnload_json(DATA_DIR/'time'/'real'/'trace'/'level5'/code)defload_timedeal(code):"""读取分时成交"""returnload_json(DATA_DIR/'time'/'real'/'trace'/'timedeal'/code)defload_tick(code):"""读取逐笔交易"""returnload_json(DATA_DIR/'time'/'real'/'trace'/'onebyone'/code)defload_bigdeal(code):"""读取大单交易"""returnload_json(DATA_DIR/'time'/'real'/'trace'/'bigdeal'/code)# 读取股票列表withopen(DATA_DIR/'base'/'gplist','r',encoding='utf-8')asf:stock_list=json.load(f)df_stocks=pd.DataFrame(stock_list)df_stocks=df_stocks[(df_stocks['isSt']==0)&(df_stocks['isNew']==0)]print(f'股票池数量：{len(df_stocks)}')print(f'字段列表：{df_stocks.columns.tolist()}')

这段代码定义了统一的数据加载函数，所有接口都通过相同的方式读取，不需要处理不同的API协议和数据格式。

多接口联动的核心价值在于数据整合。下面这段代码是我用来构建单只股票完整画像的脚本：

defbuild_stock_profile(code):"""构建单只股票的完整画像"""profile={'代码':code}# 基础信息stock_info=df_stocks[df_stocks['dm']==code]iflen(stock_info)>0:profile['名称']=stock_info['mc'].iloc[0]profile['交易所']=stock_info['jys'].iloc[0]profile['是否创业板']=bool(stock_info['isCyb'].iloc[0])profile['是否科创板']=bool(stock_info['isKcb'].iloc[0])# 实时行情rt=load_realtime(code)ifrt:profile['现价']=float(rt.get('p',0))profile['涨跌幅']=float(rt.get('pc',0))profile['成交量']=float(rt.get('v',0))/10000profile['成交额']=float(rt.get('cje',0))/10000profile['振幅']=float(rt.get('zf',0))profile['换手率']=float(rt.get('hs',0))profile['最新更新时间']=rt.get('updateTime')# L2行情指标l2=load_l2(code)ifl2:profile['主力净流入']=float(l2.get('zlJlr',0))/10000profile['主力净比']=float(l2.get('zlJlb',0))profile['散户净比']=float(l2.get('shJlb',0))profile['DDX']=float(l2.get('ddx',0))profile['DDY']=float(l2.get('ddy',0))profile['DDZ']=float(l2.get('ddz',0))profile['5分钟主力净额']=float(l2.get('f5MinZlJe',0))/10000profile['近三日主力净额']=float(l2.get('f3DayZlJe',0))/10000# 五档盘口level5=load_level5(code)iflevel5:profile['委比']=float(level5.get('wb',0))profile['委差']=float(level5.get('wc',0))# 买盘和卖盘总挂单buy_total=sum(float(level5.get(f'bid{i}Volume',0))foriinrange(1,6))sell_total=sum(float(level5.get(f'ask{i}Volume',0))foriinrange(1,6))profile['买盘总挂单']=buy_total profile['卖盘总挂单']=sell_total profile['买卖比']=buy_total/sell_totalifsell_total>0else0# 分时成交分析timedeal=load_timedeal(code)iftimedeal:df_td=pd.DataFrame(timedeal)df_td['分时时间']=pd.to_datetime(df_td['分时时间'],format='%H:%M:%S')df_td['成交量']=pd.to_numeric(df_td['成交量'],errors='coerce')df_td['成交价']=pd.to_numeric(df_td['成交价'],errors='coerce')iflen(df_td)>0:profile['日内最高价']=df_td['成交价'].max()profile['日内最低价']=df_td['成交价'].min()profile['日内均价']=df_td['成交价'].mean()profile['上涨分钟数']=(df_td['成交价'].pct_change()>0).sum()profile['下跌分钟数']=(df_td['成交价'].pct_change()<0).sum()profile['上涨占比']=profile['上涨分钟数']/len(df_td)# 逐笔交易分析tick=load_tick(code)iftick:df_tick=pd.DataFrame(tick)df_tick['成交量']=pd.to_numeric(df_tick['成交量'],errors='coerce')df_tick['成交价']=pd.to_numeric(df_tick['成交价'],errors='coerce')iflen(df_tick)>0:buy_mask=df_tick['交易方向']==1sell_mask=df_tick['交易方向']==2buy_amount=(df_tick.loc[buy_mask,'成交量']*df_tick.loc[buy_mask,'成交价']).sum()sell_amount=(df_tick.loc[sell_mask,'成交量']*df_tick.loc[sell_mask,'成交价']).sum()profile['逐笔成交笔数']=len(df_tick)profile['逐笔买入金额']=buy_amount/10000profile['逐笔卖出金额']=sell_amount/10000profile['逐笔净流入']=(buy_amount-sell_amount)/10000# 大单交易分析bigdeal=load_bigdeal(code)ifbigdeal:df_bd=pd.DataFrame(bigdeal)df_bd['成交量']=pd.to_numeric(df_bd['成交量'],errors='coerce')df_bd['成交价']=pd.to_numeric(df_bd['成交价'],errors='coerce')iflen(df_bd)>0:buy_mask=df_bd['交易方向']==1sell_mask=df_bd['交易方向']==2buy_amount=(df_bd.loc[buy_mask,'成交量']*df_bd.loc[buy_mask,'成交价']).sum()sell_amount=(df_bd.loc[sell_mask,'成交量']*df_bd.loc[sell_mask,'成交价']).sum()profile['大单交易笔数']=len(df_bd)profile['大单买入金额']=buy_amount/10000profile['大单卖出金额']=sell_amount/10000profile['大单净流入']=(buy_amount-sell_amount)/10000returnprofile# 构建单只股票的完整画像profile=build_stock_profile('600519')print('\n=== 股票完整画像 ===')fork,vinprofile.items():ifisinstance(v,float):print(f'{k}:{v:.4f}')else:print(f'{k}:{v}')

这段代码整合了6个接口的数据，构建了单只股票的完整画像。换成在线API，同样的操作至少要调用6个不同的接口，处理6种不同的数据格式，还要处理限流和超时。

多接口联动的最高境界是构建全市场扫描体系。下面这段代码是我实际在用的多维度股票筛选脚本：

defmulti_dimension_scan(top_n=10):"""多维度全市场扫描，筛选优质股票"""results=[]forcodeindf_stocks['dm']:profile=build_stock_profile(code)# 跳过数据不全的股票if'现价'notinprofileorprofile['现价']==0:continue# 多维度评分score=0# 基本面得分（排除退市风险）ifnotprofile.get('是否创业板')andnotprofile.get('是否科创板'):score+=10# 价格得分（避免过高或过低）if10<=profile['现价']<=200:score+=15# 涨跌幅得分（避免追高）if-3<=profile['涨跌幅']<=3:score+=10# 成交量得分（流动性充足）ifprofile.get('成交量',0)>100:score+=10# 主力资金得分ifprofile.get('主力净比',0)>5:score+=20elifprofile.get('主力净比',0)>0:score+=10# DDX得分ifprofile.get('DDX',0)>0.05:score+=15# 盘口得分ifprofile.get('买卖比',0)>1.2:score+=10# 日内趋势得分ifprofile.get('上涨占比',0)>0.5:score+=10# 大单资金得分ifprofile.get('大单净流入',0)>1000:score+=10profile['综合评分']=score results.append(profile)df_result=pd.DataFrame(results)# 筛选评分>60的股票candidates=df_result[df_result['综合评分']>60]candidates=candidates.sort_values('综合评分',ascending=False).head(top_n)print(f'\n==={datetime.now().strftime("%H:%M:%S")}多维度扫描结果 ===')print(f'符合条件股票数量：{len(candidates)}')print(candidates[['代码','名称','现价','涨跌幅','主力净比','DDX','买卖比','上涨占比','大单净流入','综合评分']])returncandidates# 运行多维度全市场扫描multi_dimension_scan(top_n=15)

这段代码通过多维度评分体系，从全市场中筛选出优质股票。我之前用在线API做同样的分析，因为数据延迟和不全，经常漏掉好股票；现在本地数据完整实时，筛选结果靠谱多了。

用本地数据做多接口联动，最大的体感差异是"统一性"和"效率"。以前用在线API，每个接口都有不同的协议和数据格式，需要写大量的适配代码；现在所有数据都按统一的目录结构和JSON格式存储，读取方式完全一致。而且本地读取没有并发上限，全市场5000多只股票的多维度扫描，几分钟就能跑完，换成在线API至少要等几小时。

还有个容易被忽视的点，数据的可追溯性。在线API的数据是一次性的，想查历史数据需要重新请求；本地数据在自己机器上，想查多久的历史数据都可以，而且数据格式和结构和当前数据完全一致，不需要重新适配。

数据名称汇总：