折腾了半年在线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_DIR=Path('/ig50-data')# 读取股票列表withopen(DATA_DIR/'base'/'gplist','r',encoding='utf-8')asf:stock_list=json.load(f)df_stocks=pd.DataFrame(stock_list)# 字段：dm(代码)、mc(名称)、jys(交易所)、isCyb(创业板)、isKcb(科创板)、isSt、isNew# 过滤掉ST股和新股，只看主板和创业板df_pool=df_stocks[(df_stocks['isSt']==0)&(df_stocks['isNew']==0)]print(f'候选股票数量：{len(df_pool)}')# 批量读取实时行情（纯本地文件读取，无网络请求、无并发限制）defload_realtime(code):path=DATA_DIR/'time'/'real'/codeifnotpath.exists():returnNonewithopen(path,'r',encoding='utf-8')asf:returnjson.load(f)# 全市场扫描，几千只股票本地读取几秒就跑完records=[]forcodeindf_pool['dm']:rt=load_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_realtime=pd.DataFrame(records)# 按成交额排序，看今天资金扎堆在哪print(df_realtime.sort_values('成交额',ascending=False).head(20))

这种读法最爽的地方是，全市场5000多只股票扫一遍，因为是本地文件IO，几秒钟就跑完了。换成在线API，这个量级的请求光限频就得等半天，更别说中途超时重试的折腾。

做回测的时候，K线数据的读取同样直接。下面这段是把某只股票的日线前复权数据读出来，算均线和金叉信号的写法：

# 读取K线数据（日前复权级别 Day_qfq）code='000001'kline_path=DATA_DIR/'time'/'real'/'time'/code/'Day_qfq'withopen(kline_path,'r',encoding='utf-8')asf:kline=json.load(f)df_k=pd.DataFrame(kline)# 字段：d(时间)、o(开)、h(高)、l(低)、c(收)、v(成交量)、e(成交额)、zd(涨跌幅)、hs(换手率)df_k['d']=pd.to_datetime(df_k['d'])df_k=df_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上穿MA60）golden_cross=df_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'}),type='candle',style='charles',volume=True,title=f'{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?maodian=rt-trace-title，如有侵犯网站权利，请联系我撤回。