ETF资金流监测系统实战用Python构建场内场外套利信号识别ETF是A股一个特殊的存在——它既是一篮子股票又能在二级市场像股票一样交易。我做量化这几年专门搭建了一套ETF资金流监测系统用Python把场内场外的资金流数据打通识别套利信号。这篇文章把核心实现分享出来。ETF数据主要分四块。第一块是ETF基金列表路径base/jjlbList基金列表返回所有ETF基金的代码、名称、类型、跟踪指数。第二块是ETF基金行情路径time/jjHQ/{基金代码}返回ETF的实时行情。第三块是ETF基金规模变动路径time/jjgm/{基金代码}返回ETF的份额变动历史。第四块是行业资金流向路径all/zjlx/zjhhy行业资金流向返回各行业的资金流向数据。我把这四块数据全部沉到本地。先说ETF基金列表怎么读。importjsonimportosimportpandasaspd data_dirD:/stock_datadefread_etf_list():file_pathos.path.join(data_dir,base,jjlbList)ifnotos.path.exists(file_path):returnNonewithopen(file_path,r,encodingutf-8)asf:datajson.load(f)dfpd.DataFrame(data)df.columns[dm,mc,jjlb,gmbm,gmrb,gmsj,gmjz,gmje]returndfdefread_etf_realtime(dm):file_pathos.path.join(data_dir,time,jjHQ,dm)ifnotos.path.exists(file_path):returnNonewithopen(file_path,r,encodingutf-8)asf:datajson.load(f)dfpd.DataFrame([data])df.columns[dm,mc,now,zf,jk,zs,zsz,cjsl,cjje,zt,dt,wb,wp]returndf字段简写dm基金代码mc基金名称jjlb基金类型gmbm规模编码gmrb规模日期gmje规模金额。实时行情字段now当前价zf涨跌幅jk今开zs昨收zsz总市值cjsl成交量cjje成交金额zt涨停dt跌停wb委买wp委卖。第一步是分析ETF场内场外溢价。ETF同时在场内二级市场和场外一级市场交易两个价格之间的差额就是套利空间。defanalyze_etf_premium(dm,lookback_days30):realtimeread_etf_realtime(dm)ifrealtimeisNone:returnNoneetf_listread_etf_list()ifetf_listisNone:returnNonetarget_etfetf_list[etf_list[dm]dm]iflen(target_etf)0:returnNoneetf_iopvtarget_etf.iloc[0][gmjz]etf_pricerealtime[now].iloc[0]premium(etf_price-etf_iopv)/etf_iopv*100print(ETF{}场内场外溢价分析.format(dm))print(场内价格{:.4f}.format(etf_price))print(场外净值{:.4f}.format(etf_iopv))print(溢价率{:.2f}%.format(premium))signalNoneifpremium2.0:signal场内高溢价可考虑卖出套利elifpremium-1.0:signal场内折价可考虑买入套利else:signal溢价正常无套利空间return{premium:premium,signal:signal}第二步是分析ETF份额变动。ETF份额变动反映了资金对这只ETF的态度是大资金进场还是离场最直观的指标。defread_etf_share_change(dm):file_pathos.path.join(data_dir,time,jjgm,dm)ifnotos.path.exists(file_path):returnNonewithopen(file_path,r,encodingutf-8)asf:datajson.load(f)dfpd.DataFrame(data)df.columns[rq,fbdw,fqjz,fcye,zfdw,zqjz,zcye,jzzzl]returndfdefanalyze_etf_share_change(dm,days30):dfread_etf_share_change(dm)ifdfisNoneorlen(df)0:returnNonedf[fbdw]pd.to_numeric(df[fbdw],errorscoerce)df[zfdw]pd.to_numeric(df[zfdw],errorscoerce)df[rq]pd.to_datetime(df[rq],errorscoerce)recentdf.head(days)total_inflowrecent[fbdw].sum()/1e8total_outflowrecent[zfdw].sum()/1e8nettotal_inflow-total_outflowprint(ETF{}近{}日份额变动.format(dm,days))print(总申购{:.2f}亿份.format(total_inflow))print(总赎回{:.2f}亿份.format(total_outflow))print(净申购{:.2f}亿份.format(net))signalNoneifnet1.0:signal大额净申购资金流入elifnet-1.0:signal大额净赎回资金流出else:signal份额变动正常return{net:net,signal:signal}第三步是行业ETF联动分析。行业ETF和行业指数高度联动可以用行业资金流向数据来预判行业ETF的走势。defread_industry_flow():file_pathos.path.join(data_dir,all,zjlx,zjhhy)ifnotos.path.exists(file_path):returnNonewithopen(file_path,r,encodingutf-8)asf:datajson.load(f)dfpd.DataFrame(data)df.columns[rq,hy,zjl,zlJlr,zlJlb,shJlb,ddy,ddx,ddz,ddf,f5MinZlJe]returndfdeffind_industry_etf_signals(date_str):dfread_industry_flow()ifdfisNone:returnNonedf[rq]pd.to_datetime(df[rq],errorscoerce)targetdf[df[rq].dt.strftime(%Y-%m-%d)date_str]iflen(target)0:returnNonetarget[zlJlr]pd.to_numeric(target[zlJlr],errorscoerce)strongtarget[target[zlJlr]5]weaktarget[target[zlJlr]-5]print(日期{} 行业资金流向.format(date_str))print(强势行业)foridx,rowinstrong.iterrows():print( {}净流入{:.2f}亿.format(row[hy],row[zlJlr]))print(弱势行业)foridx,rowinweak.iterrows():print( {}净流出{:.2f}亿.format(row[hy],abs(row[zlJlr])))return{strong:strong,weak:weak}第四步是构建完整的ETF套利信号系统。我把上面几个模块组合起来。defetf_arbitrage_scan(date_str):etf_listread_etf_list()ifetf_listisNone:returnNoneetf_listetf_list[etf_list[jjlb].str.contains(ETF,naFalse)]signals[]fordminetf_list[dm].tolist():premium_resultanalyze_etf_premium(dm)share_resultanalyze_etf_share_change(dm,days10)ifpremium_resultisNoneorshare_resultisNone:continuepremiumpremium_result[premium]net_shareshare_result[net]ifpremium2.0andnet_share0.5:signals.append({dm:dm,signal:高溢价净申购可能有大资金申购套利关注卖出机会,premium:premium,net_share:net_share})elifpremium-1.0andnet_share-0.5:signals.append({dm:dm,signal:折价净赎回可能有大资金赎回套利关注买入机会,premium:premium,net_share:net_share})returnpd.DataFrame(signals)这套ETF系统跑了一年最关键的收获是ETF的资金流向比价格更值得关注。当某只ETF出现大额净申购但价格没怎么涨的时候往往是聪明钱在悄悄建仓。反过来当某只ETF大额净赎回但价格还坚挺的时候往往是陷阱。接口说明ETF基金列表base/jjlbListETF实时行情time/jjHQ/{dm}ETF份额变动time/jjgm/{dm}行业资金流向all/zjlx/zjhhy资料参考ig50 ETF资金流监测系统实战：用Python构建场内场外套利信号识别
ETF资金流监测系统实战用Python构建场内场外套利信号识别ETF是A股一个特殊的存在——它既是一篮子股票又能在二级市场像股票一样交易。我做量化这几年专门搭建了一套ETF资金流监测系统用Python把场内场外的资金流数据打通识别套利信号。这篇文章把核心实现分享出来。ETF数据主要分四块。第一块是ETF基金列表路径base/jjlbList基金列表返回所有ETF基金的代码、名称、类型、跟踪指数。第二块是ETF基金行情路径time/jjHQ/{基金代码}返回ETF的实时行情。第三块是ETF基金规模变动路径time/jjgm/{基金代码}返回ETF的份额变动历史。第四块是行业资金流向路径all/zjlx/zjhhy行业资金流向返回各行业的资金流向数据。我把这四块数据全部沉到本地。先说ETF基金列表怎么读。importjsonimportosimportpandasaspd data_dirD:/stock_datadefread_etf_list():file_pathos.path.join(data_dir,base,jjlbList)ifnotos.path.exists(file_path):returnNonewithopen(file_path,r,encodingutf-8)asf:datajson.load(f)dfpd.DataFrame(data)df.columns[dm,mc,jjlb,gmbm,gmrb,gmsj,gmjz,gmje]returndfdefread_etf_realtime(dm):file_pathos.path.join(data_dir,time,jjHQ,dm)ifnotos.path.exists(file_path):returnNonewithopen(file_path,r,encodingutf-8)asf:datajson.load(f)dfpd.DataFrame([data])df.columns[dm,mc,now,zf,jk,zs,zsz,cjsl,cjje,zt,dt,wb,wp]returndf字段简写dm基金代码mc基金名称jjlb基金类型gmbm规模编码gmrb规模日期gmje规模金额。实时行情字段now当前价zf涨跌幅jk今开zs昨收zsz总市值cjsl成交量cjje成交金额zt涨停dt跌停wb委买wp委卖。第一步是分析ETF场内场外溢价。ETF同时在场内二级市场和场外一级市场交易两个价格之间的差额就是套利空间。defanalyze_etf_premium(dm,lookback_days30):realtimeread_etf_realtime(dm)ifrealtimeisNone:returnNoneetf_listread_etf_list()ifetf_listisNone:returnNonetarget_etfetf_list[etf_list[dm]dm]iflen(target_etf)0:returnNoneetf_iopvtarget_etf.iloc[0][gmjz]etf_pricerealtime[now].iloc[0]premium(etf_price-etf_iopv)/etf_iopv*100print(ETF{}场内场外溢价分析.format(dm))print(场内价格{:.4f}.format(etf_price))print(场外净值{:.4f}.format(etf_iopv))print(溢价率{:.2f}%.format(premium))signalNoneifpremium2.0:signal场内高溢价可考虑卖出套利elifpremium-1.0:signal场内折价可考虑买入套利else:signal溢价正常无套利空间return{premium:premium,signal:signal}第二步是分析ETF份额变动。ETF份额变动反映了资金对这只ETF的态度是大资金进场还是离场最直观的指标。defread_etf_share_change(dm):file_pathos.path.join(data_dir,time,jjgm,dm)ifnotos.path.exists(file_path):returnNonewithopen(file_path,r,encodingutf-8)asf:datajson.load(f)dfpd.DataFrame(data)df.columns[rq,fbdw,fqjz,fcye,zfdw,zqjz,zcye,jzzzl]returndfdefanalyze_etf_share_change(dm,days30):dfread_etf_share_change(dm)ifdfisNoneorlen(df)0:returnNonedf[fbdw]pd.to_numeric(df[fbdw],errorscoerce)df[zfdw]pd.to_numeric(df[zfdw],errorscoerce)df[rq]pd.to_datetime(df[rq],errorscoerce)recentdf.head(days)total_inflowrecent[fbdw].sum()/1e8total_outflowrecent[zfdw].sum()/1e8nettotal_inflow-total_outflowprint(ETF{}近{}日份额变动.format(dm,days))print(总申购{:.2f}亿份.format(total_inflow))print(总赎回{:.2f}亿份.format(total_outflow))print(净申购{:.2f}亿份.format(net))signalNoneifnet1.0:signal大额净申购资金流入elifnet-1.0:signal大额净赎回资金流出else:signal份额变动正常return{net:net,signal:signal}第三步是行业ETF联动分析。行业ETF和行业指数高度联动可以用行业资金流向数据来预判行业ETF的走势。defread_industry_flow():file_pathos.path.join(data_dir,all,zjlx,zjhhy)ifnotos.path.exists(file_path):returnNonewithopen(file_path,r,encodingutf-8)asf:datajson.load(f)dfpd.DataFrame(data)df.columns[rq,hy,zjl,zlJlr,zlJlb,shJlb,ddy,ddx,ddz,ddf,f5MinZlJe]returndfdeffind_industry_etf_signals(date_str):dfread_industry_flow()ifdfisNone:returnNonedf[rq]pd.to_datetime(df[rq],errorscoerce)targetdf[df[rq].dt.strftime(%Y-%m-%d)date_str]iflen(target)0:returnNonetarget[zlJlr]pd.to_numeric(target[zlJlr],errorscoerce)strongtarget[target[zlJlr]5]weaktarget[target[zlJlr]-5]print(日期{} 行业资金流向.format(date_str))print(强势行业)foridx,rowinstrong.iterrows():print( {}净流入{:.2f}亿.format(row[hy],row[zlJlr]))print(弱势行业)foridx,rowinweak.iterrows():print( {}净流出{:.2f}亿.format(row[hy],abs(row[zlJlr])))return{strong:strong,weak:weak}第四步是构建完整的ETF套利信号系统。我把上面几个模块组合起来。defetf_arbitrage_scan(date_str):etf_listread_etf_list()ifetf_listisNone:returnNoneetf_listetf_list[etf_list[jjlb].str.contains(ETF,naFalse)]signals[]fordminetf_list[dm].tolist():premium_resultanalyze_etf_premium(dm)share_resultanalyze_etf_share_change(dm,days10)ifpremium_resultisNoneorshare_resultisNone:continuepremiumpremium_result[premium]net_shareshare_result[net]ifpremium2.0andnet_share0.5:signals.append({dm:dm,signal:高溢价净申购可能有大资金申购套利关注卖出机会,premium:premium,net_share:net_share})elifpremium-1.0andnet_share-0.5:signals.append({dm:dm,signal:折价净赎回可能有大资金赎回套利关注买入机会,premium:premium,net_share:net_share})returnpd.DataFrame(signals)这套ETF系统跑了一年最关键的收获是ETF的资金流向比价格更值得关注。当某只ETF出现大额净申购但价格没怎么涨的时候往往是聪明钱在悄悄建仓。反过来当某只ETF大额净赎回但价格还坚挺的时候往往是陷阱。接口说明ETF基金列表base/jjlbListETF实时行情time/jjHQ/{dm}ETF份额变动time/jjgm/{dm}行业资金流向all/zjlx/zjhhy资料参考ig50