多接口联动实战本地数据引擎如何构建完整量化分析体系做量化投资的人都明白单一数据源远不足以支撑完整的策略分析。你需要实时行情看当前价格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_DIRPath(/ig50-data)# 统一的数据加载函数defload_json(path):读取JSON文件ifnotpath.exists():returnNonewithopen(path,r,encodingutf-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,encodingutf-8)asf:stock_listjson.load(f)df_stockspd.DataFrame(stock_list)df_stocksdf_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_infodf_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])# 实时行情rtload_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行情指标l2load_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# 五档盘口level5load_level5(code)iflevel5:profile[委比]float(level5.get(wb,0))profile[委差]float(level5.get(wc,0))# 买盘和卖盘总挂单buy_totalsum(float(level5.get(fbid{i}Volume,0))foriinrange(1,6))sell_totalsum(float(level5.get(fask{i}Volume,0))foriinrange(1,6))profile[买盘总挂单]buy_total profile[卖盘总挂单]sell_total profile[买卖比]buy_total/sell_totalifsell_total0else0# 分时成交分析timedealload_timedeal(code)iftimedeal:df_tdpd.DataFrame(timedeal)df_td[分时时间]pd.to_datetime(df_td[分时时间],format%H:%M:%S)df_td[成交量]pd.to_numeric(df_td[成交量],errorscoerce)df_td[成交价]pd.to_numeric(df_td[成交价],errorscoerce)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)# 逐笔交易分析tickload_tick(code)iftick:df_tickpd.DataFrame(tick)df_tick[成交量]pd.to_numeric(df_tick[成交量],errorscoerce)df_tick[成交价]pd.to_numeric(df_tick[成交价],errorscoerce)iflen(df_tick)0:buy_maskdf_tick[交易方向]1sell_maskdf_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# 大单交易分析bigdealload_bigdeal(code)ifbigdeal:df_bdpd.DataFrame(bigdeal)df_bd[成交量]pd.to_numeric(df_bd[成交量],errorscoerce)df_bd[成交价]pd.to_numeric(df_bd[成交价],errorscoerce)iflen(df_bd)0:buy_maskdf_bd[交易方向]1sell_maskdf_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# 构建单只股票的完整画像profilebuild_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_n10):多维度全市场扫描筛选优质股票results[]forcodeindf_stocks[dm]:profilebuild_stock_profile(code)# 跳过数据不全的股票if现价notinprofileorprofile[现价]0:continue# 多维度评分score0# 基本面得分排除退市风险ifnotprofile.get(是否创业板)andnotprofile.get(是否科创板):score10# 价格得分避免过高或过低if10profile[现价]200:score15# 涨跌幅得分避免追高if-3profile[涨跌幅]3:score10# 成交量得分流动性充足ifprofile.get(成交量,0)100:score10# 主力资金得分ifprofile.get(主力净比,0)5:score20elifprofile.get(主力净比,0)0:score10# DDX得分ifprofile.get(DDX,0)0.05:score15# 盘口得分ifprofile.get(买卖比,0)1.2:score10# 日内趋势得分ifprofile.get(上涨占比,0)0.5:score10# 大单资金得分ifprofile.get(大单净流入,0)1000:score10profile[综合评分]score results.append(profile)df_resultpd.DataFrame(results)# 筛选评分60的股票candidatesdf_result[df_result[综合评分]60]candidatescandidates.sort_values(综合评分,ascendingFalse).head(top_n)print(f\n{datetime.now().strftime(%H:%M:%S)}多维度扫描结果 )print(f符合条件股票数量{len(candidates)})print(candidates[[代码,名称,现价,涨跌幅,主力净比,DDX,买卖比,上涨占比,大单净流入,综合评分]])returncandidates# 运行多维度全市场扫描multi_dimension_scan(top_n15)这段代码通过多维度评分体系从全市场中筛选出优质股票。我之前用在线API做同样的分析因为数据延迟和不全经常漏掉好股票现在本地数据完整实时筛选结果靠谱多了。用本地数据做多接口联动最大的体感差异是统一性和效率。以前用在线API每个接口都有不同的协议和数据格式需要写大量的适配代码现在所有数据都按统一的目录结构和JSON格式存储读取方式完全一致。而且本地读取没有并发上限全市场5000多只股票的多维度扫描几分钟就能跑完换成在线API至少要等几小时。还有个容易被忽视的点数据的可追溯性。在线API的数据是一次性的想查历史数据需要重新请求本地数据在自己机器上想查多久的历史数据都可以而且数据格式和结构和当前数据完全一致不需要重新适配。数据名称汇总实时行情数据3秒落盘数据存放目录/time/real/{股票代码}L2行情指标3秒落盘数据存放目录/time/real/trace/l2sign/{股票代码}买卖五档盘口3秒落盘数据存放目录/time/real/trace/level5/{股票代码}当天分时成交数据存放目录/time/real/trace/timedeal/{股票代码}当天逐笔交易数据存放目录/time/real/trace/onebyone/{股票代码}当天大单交易数据存放目录/time/real/trace/bigdeal/{股票代码}资料参考ig50.com/index_realtime.html如有侵犯网站权利请联系我撤回。
