折腾了半年在线API才明白，股票数据还是本地落盘最省事｜Python量化回测实战

折腾了半年在线API才明白股票数据还是本地落盘最省事Python量化回测实战做量化的人大概都绕不开数据这道坎。模型再精巧喂进去的数据慢半拍或者字段缺斤少两回测结果就是另一个故事了。我之前在A股数据上踩过不少坑从免费的开源库到付费的在线接口都试过一轮最近半年总算摸索出一套用着顺手的本地落盘方案配合Python做回测分析效率比以前高了一截想跟同样在折腾数据的朋友聊聊这套思路。在线API用起来最直观但问题也最多。Tushare的积分门槛和频次限制大家都经历过批量拉几千只股票的K线跑着跑着就被打回。AkShare免费但数据源稳定性看天吃饭盘中高峰期经常超时。Wind和专业终端质量没得说可成本不是个人玩家能长期扛的。更头疼的是延迟在线接口单次请求几十到几百毫秒听着不多可当你要把全市场5000多只股票的实时行情拉一遍做信号扫描这点延迟会被放大成几十秒甚至几分钟的等待日内策略根本等不起。还有个容易被忽视的点你查什么数据、查多频繁请求日志都在别人服务器上策略思路多少会留下痕迹。后来我把目光转向了本地数据引擎这条路。ig50这类把数据直接落盘到本地的方案思路和在线API完全不同——数据引擎在后台持续跑把行情、K线、财务数据写成JSON文件存到你自己的磁盘上程序直接读本地文件不走网络。股票列表在base/gplist目录下读出来能拿到代码、名称、交易所、是否创业板科创板这些基础字段实时行情在time/real/{股票代码}路径下每10秒更新一次、3秒左右落盘价格、涨跌幅、成交量这些字段都有历史K线在time/real/time/{股票代码}/{分时级别}下从5分钟到年线十几个级别都支持日线还分不复权、前复权、后复权三种。下面这段代码是我平时做全市场扫描的起手式先读股票列表再批量读实时行情做初步筛选。因为是读本地文件全程没有一次网络请求importjsonimportpandasaspdfrompathlibimportPath# 数据存放目录默认 /ig50-data安装时可自定义DATA_DIRPath(/ig50-data)# 读取股票列表withopen(DATA_DIR/base/gplist,r,encodingutf-8)asf:stock_listjson.load(f)df_stockspd.DataFrame(stock_list)# 字段dm(代码)、mc(名称)、jys(交易所)、isCyb(创业板)、isKcb(科创板)、isSt、isNew# 过滤掉ST股和新股只看主板和创业板df_pooldf_stocks[(df_stocks[isSt]0)(df_stocks[isNew]0)]print(f候选股票数量{len(df_pool)})# 批量读取实时行情纯本地文件读取无网络请求、无并发限制defload_realtime(code):pathDATA_DIR/time/real/codeifnotpath.exists():returnNonewithopen(path,r,encodingutf-8)asf:returnjson.load(f)# 全市场扫描几千只股票本地读取几秒就跑完records[]forcodeindf_pool[dm]:rtload_realtime(code)ifrt:records.append({代码:rt.get(dm),名称:rt.get(mc),现价:rt.get(p),涨跌幅:rt.get(pc),成交量:rt.get(v),成交额:rt.get(cje),振幅:rt.get(zf),})df_realtimepd.DataFrame(records)# 按成交额排序看今天资金扎堆在哪print(df_realtime.sort_values(成交额,ascendingFalse).head(20))这种读法最爽的地方是全市场5000多只股票扫一遍因为是本地文件IO几秒钟就跑完了。换成在线API这个量级的请求光限频就得等半天更别说中途超时重试的折腾。做回测的时候K线数据的读取同样直接。下面这段是把某只股票的日线前复权数据读出来算均线和金叉信号的写法# 读取K线数据日前复权级别 Day_qfqcode000001kline_pathDATA_DIR/time/real/time/code/Day_qfqwithopen(kline_path,r,encodingutf-8)asf:klinejson.load(f)df_kpd.DataFrame(kline)# 字段d(时间)、o(开)、h(高)、l(低)、c(收)、v(成交量)、e(成交额)、zd(涨跌幅)、hs(换手率)df_k[d]pd.to_datetime(df_k[d])df_kdf_k.set_index(d).sort_index()# 计算均线和金叉信号df_k[MA20]df_k[c].rolling(20).mean()df_k[MA60]df_k[c].rolling(60).mean()df_k[signal](df_k[MA20]df_k[MA60]).astype(int)# 找出金叉日期MA20上穿MA60golden_crossdf_k[(df_k[signal]1)(df_k[signal].shift(1)0)]print(f最近一次金叉{golden_cross.index[-1].date()})# 想画K线图的话mplfinance 直接能用importmplfinanceasmpf mpf.plot(df_k.tail(120).rename(columns{o:Open,h:High,l:Low,c:Close,v:Volume}),typecandle,stylecharles,volumeTrue,titlef{code}日线)用本地文件做回测最大的体感差异是敢跑了。以前用在线API全市场回测一次要小心翼翼控制请求频率生怕被封现在本地读取没有并发上限想跑多少只就跑多少只万只股票的分钟级回测从分钟级降到秒级。数据还在自己机器上策略逻辑不会通过请求记录泄露出去对在意隐私的团队来说这点挺关键。数据引擎后台自动更新也不用自己维护采集脚本和清洗逻辑接口变更、字段缺失这些烂摊子都不用操心。如果你也在被在线接口的频次限制和延迟折磨或者想让回测跑得更快更稳本地落盘这条路值得试试。数据自动更新、本地零延迟读取、没有并发上限配合Python做数据分析和策略验证整个流程会比想象中顺畅不少。用过就回不去了大概就是这种感觉。数据名称沪深京A股-股票交易数据-股票行情-实时行情数据3秒落盘本地路径数据存放目录/time/real/{股票代码}描述根据《股票列表》得到的股票代码获取实时交易数据您可以理解为日线的最新数据。更新频率交易时间段每10秒。完成更新耗时约3秒。数据格式{}关键字[代码, 名称, 更新时间yyyy-MM-dd HH:mm:ss, 数据时间yyyy-MM-dd HH:mm:ss, 当前价格元, 昨收价元, 开盘价元, 最高价元, 最低价元, 均价元成交量加权平均价, 涨跌额元, 涨跌幅%, 振幅%, 涨停价元, 跌停价元, 成交量股, 现量最新一笔成交量手, 成交额元]资料参考ig50.com/index_realtime.html?maodianrt-trace-title如有侵犯网站权利请联系我撤回。