企业官网建设流程全解析

API回测一场梦，本地数据才是真：WorkBuddy如何帮我构建高效策略回测体系

写在前面

做量化的人，都有过"回测跑一天"的经历。

写好一个策略，想看看历史表现，结果从API拉数据就花了半天。如果要跑多个策略、多只股票、多个时间段，时间成本更高。更头疼的是，API有频率限制，拉数据拉太快会被限流，只能慢慢等。

后来我开始用本地数据落盘方案，把历史数据存储在本地文件中，才发现回测原来可以这么快——WorkBuddy直接读本地文件，分钟级完成回测，效率提升了几十倍。

今天想分享的，就是WorkBuddy如何结合本地数据（ig50），构建一套高效的策略回测体系。

第一阶段：API回测的痛点

典型的回测流程

用API做回测，典型流程是这样的：

1. WorkBuddy调API，获取股票列表 2. WorkBuddy调API，获取第一只股票的历史K线 3. WorkBuddy运行策略，计算收益 4. WorkBuddy调API，获取第二只股票的历史K线 5. WorkBuddy运行策略，计算收益 6. ... 重复几百次 ...

每个API调用，都要经过网络请求、数据返回、数据解析，整个过程通常需要几百毫秒到几秒。如果要回测100只股票，光拉数据就要几分钟到十几分钟。

真实的时间消耗

我之前用API跑过一个均线策略的回测，回测50只股票、1年历史数据：

而且这还是顺利的情况，中间如果遇到API限流，时间会更长。

第二阶段：本地数据让回测分钟级完成

WorkBuddy读本地文件的速度

换成本地数据方案之后，回测的速度提升了一个量级：读本地的速度取决于你的CPU、内存和硬盘速度，如果再用多线程并行读取，回测速度还能进一步提升。

importjsonimportpandasaspdfrompathlibimportPathfromconcurrent.futuresimportThreadPoolExecutor DATA_DIR=Path('/data')defload_kline_data(args):"""WorkBuddy多线程读取单只股票历史K线"""code,start_date,end_date=args kline_path=DATA_DIR/'stock-a'/'历史K线'/codetry:withopen(kline_path,'r',encoding='utf-8')asf:data=json.load(f)except:returnNonedf=pd.DataFrame(data)df['日期']=pd.to_datetime(df['日期'])df=df[(df['日期']>=start_date)&(df['日期']<=end_date)]returncode,df# WorkBuddy多线程批量读取历史K线stock_codes=['000001','000002','600036','600519']args=[(code,'2024-01-01','2024-12-31')forcodeinstock_codes]withThreadPoolExecutor(max_workers=8)asexecutor:kline_results=list(executor.map(load_kline_data,args))# 过滤空数据kline_results=[rforrinkline_resultsifrisnotNone]

同样的回测任务，用本地数据：

回测时间从11分钟缩短到2分钟，效率提升近6倍。在CPU、内存、硬盘足够的情况下，多线程读取数据的速度上限取决于你的硬件配置。

WorkBuddy可以批量读取数据

更关键的是，用本地数据之后，WorkBuddy可以批量读取多只股票的数据：

defbatch_backtest(stock_codes,strategy_func,start_date,end_date):"""WorkBuddy批量回测"""results=[]# WorkBuddy批量读取数据forcodeinstock_codes:# 读取历史K线（本地，毫秒级）kline_path=DATA_DIR/'stock-a'/'历史K线'/codetry:withopen(kline_path,'r',encoding='utf-8')asf:data=json.load(f)except:continuedf=pd.DataFrame(data)df['日期']=pd.to_datetime(df['日期'])df=df[(df['日期']>=start_date)&(df['日期']<=end_date)]ifdf.empty:continue# WorkBuddy运行策略signals=strategy_func(df)# WorkBuddy计算收益profit=calculate_profit(signals,df)win_rate=calculate_win_rate(signals,df)max_drawdown=calculate_max_drawdown(signals,df)results.append({'code':code,'profit':profit,'win_rate':win_rate,'max_drawdown':max_drawdown})returnpd.DataFrame(results)# WorkBuddy回测50只股票results=batch_backtest(stock_codes,ma_strategy,'2024-01-01','2024-12-31')

WorkBuddy可以在几分钟内完成50只股票的回测，这在用API的时候是不可想象的。

第三阶段：多策略并行回测

WorkBuddy可以同时跑多个策略

用API的时候，跑多个策略很麻烦——每个策略要单独拉数据，重复的API调用浪费大量时间。

用本地数据之后，WorkBuddy可以一次读取数据，运行多个策略：

defmulti_strategy_backtest(stock_codes,strategies,start_date,end_date):"""WorkBuddy多策略并行回测"""all_results=[]forcodeinstock_codes:# WorkBuddy读取数据（只读一次）kline_path=DATA_DIR/'stock-a'/'历史K线'/codetry:withopen(kline_path,'r',encoding='utf-8')asf:data=json.load(f)except:continuedf=pd.DataFrame(data)df['日期']=pd.to_datetime(df['日期'])df=df[(df['日期']>=start_date)&(df['日期']<=end_date)]ifdf.empty:continue# WorkBuddy运行多个策略forstrategy_name,strategy_funcinstrategies.items():signals=strategy_func(df)profit=calculate_profit(signals,df)win_rate=calculate_win_rate(signals,df)max_drawdown=calculate_max_drawdown(signals,df)all_results.append({'code':code,'strategy':strategy_name,'profit':profit,'win_rate':win_rate,'max_drawdown':max_drawdown})returnpd.DataFrame(all_results)# WorkBuddy同时跑3个策略strategies={'均线策略':ma_strategy,'布林带策略':bollinger_strategy,'动量策略':momentum_strategy}results=multi_strategy_backtest(stock_codes,strategies,'2024-01-01','2024-12-31')# WorkBuddy对比不同策略的表现comparison=results.groupby('strategy').agg({'profit':'mean','win_rate':'mean','max_drawdown':'mean'}).sort_values('profit',ascending=False)print(comparison)

WorkBuddy可以在几分钟内跑完多个策略、多只股票的回测，然后对比不同策略的表现：

策略 平均收益 平均胜率 平均最大回撤 布林带策略 15.2% 55.3% -8.2% 均线策略 12.5% 52.1% -10.5% 动量策略 8.3% 48.7% -15.2%

这让策略优化的效率大大提升。

第四阶段：分钟级回测

WorkBuddy可以回测分钟级策略

做日内交易的人，需要回测分钟级策略。但分钟级数据量很大——一只股票1年的分钟K线，就有几万条记录。用API拉这些数据，非常慢。

用本地数据之后，WorkBuddy可以快速读取分钟级数据：

defminute_level_backtest(code,strategy_func,start_date,end_date):"""WorkBuddy分钟级回测"""# WorkBuddy读取分钟级K线（本地，毫秒级）minute_path=DATA_DIR/'stock-a'/'历史K线（分钟级）'/codewithopen(minute_path,'r',encoding='utf-8')asf:data=json.load(f)df=pd.DataFrame(data)df['时间']=pd.to_datetime(df['时间'])df=df[(df['时间']>=start_date)&(df['时间']<=end_date)]# WorkBuddy运行分钟级策略signals=strategy_func(df)# WorkBuddy计算分钟级收益profit=calculate_minute_profit(signals,df)returnprofit# WorkBuddy回测1分钟策略profit=minute_level_backtest('000001',minute_strategy,'2024-01-01','2024-12-31')

分钟级回测的数据量是日线级的几十倍，用API几乎跑不动，但用本地数据，WorkBuddy可以轻松完成。

分钟级回测的价值

分钟级回测，解决了日内交易最大的痛点：策略验证难。

很多日内策略看起来逻辑完美，但实际跑起来效果很差。因为日内交易的细节很多——滑点、流动性、盘口变化，这些都要在分钟级数据上验证。

WorkBuddy可以做分钟级回测，让策略在上实盘之前充分验证，大大降低风险。

第五阶段：融合多数据源的回测

WorkBuddy可以融合多个数据源

高级一点的策略，会结合多个数据源——比如结合主力资金流向、北向资金、龙虎榜等数据。用API做这种回测，要调多个API，非常慢。

用本地数据之后，WorkBuddy可以融合多个数据源做回测：

defmulti_source_backtest(code,start_date,end_date):"""WorkBuddy融合多数据源回测"""# WorkBuddy读取历史K线kline_path=DATA_DIR/'stock-a'/'历史K线'/codewithopen(kline_path,'r',encoding='utf-8')asf:kline_data=json.load(f)df_kline=pd.DataFrame(kline_data)# WorkBuddy读取主力资金流向fund_path=DATA_DIR/'stock-a'/'主力资金流向'withopen(fund_path,'r',encoding='utf-8')asf:fund_data=json.load(f)df_fund=pd.DataFrame([itemforiteminfund_dataifitem['股票代码']==code])# WorkBuddy读取北向资金north_path=DATA_DIR/'stock-center'/'北向资金'withopen(north_path,'r',encoding='utf-8')asf:north_data=json.load(f)df_north=pd.DataFrame([itemforiteminnorth_dataifitem['股票代码']==code])# WorkBuddy融合数据df_merged=pd.merge(df_kline,df_fund,on='日期',how='left')df_merged=pd.merge(df_merged,df_north,on='日期',how='left')# WorkBuddy运行融合策略signals=fusion_strategy(df_merged)# WorkBuddy计算收益profit=calculate_profit(signals,df_merged)returnprofit# WorkBuddy融合多数据源回测profit=multi_source_backtest('000001','2024-01-01','2024-12-31')

WorkBuddy可以一次读取多个数据源，融合分析，这在用API的时候几乎做不到。

第六阶段：踩坑与优化

教训一：数据格式要统一

不同数据源的字段名称、日期格式可能不一样，WorkBuddy读取之后要做统一处理：

defstandardize_data(df):"""WorkBuddy统一数据格式"""# 统一日期格式df['日期']=pd.to_datetime(df['日期'])# 统一数值类型forcolindf.columns:if'涨幅'incolor'比例'incol:df[col]=pd.to_numeric(df[col],errors='coerce')returndf

教训二：数据质量要检查

本地数据虽然快，但也要检查数据质量——比如是否有缺失、异常值等：

defcheck_data_quality(df):"""WorkBuddy检查数据质量"""# 检查缺失值missing=df.isnull().sum()ifmissing.any():print(f"⚠️ 数据存在缺失：{missing[missing>0]}")# 检查异常值forcolin['收盘价','成交量']:ifcolindf.columns:z_score=(df[col]-df[col].mean())/df[col].std()outliers=df[abs(z_score)>3]iflen(outliers)>0:print(f"⚠️{col}存在异常值：{len(outliers)}个")

教训三：回测不是实盘

回测的目的是验证策略逻辑，不是追求回测收益。过度优化回测参数，容易导致"过拟合"——回测表现很好，实盘表现很差。

我的原则是：回测验证逻辑，实盘验证策略。

总结：从API回测到本地回测

用WorkBuddy结合本地数据做回测的这段时间，我的策略开发效率提升了一个量级：

之前：API回测

• 拉数据慢，回测一场要等半天
• 多策略回测要重复拉数据
• 分钟级回测几乎做不了
• 多数据源融合很麻烦

现在：本地回测

• 读本地文件，回测分钟级完成
• 一次读取，多策略运行
• 分钟级回测轻松完成
• 多数据源融合很简单

从"API回测一场梦"到"本地回测分钟完成"，策略开发效率提升了10倍以上。

给同路人的建议

如果你也想用WorkBuddy做回测，建议从这几步开始：

1. 确定回测粒度：日线级还是分钟级？
2. 确定数据源：单数据源还是多数据源融合？
3. 确定策略类型：趋势跟踪、均值回归、还是动量策略？

把这些问题想清楚，再构建你的WorkBuddy回测体系。回测是验证工具，你的策略逻辑才是核心。

gitee开源地址
github开源地址