如何异步运行Python回测函数runBacktest以提升执行效率?
Python量化回测提速:解决asyncio无效的问题
为啥asyncio没效果?
你的runBacktest属于CPU密集型任务——大数据集迭代计算,而asyncio的核心是处理IO等待场景(比如网络请求、文件读写阻塞),对纯CPU计算的任务,Python的GIL锁会让异步任务无法真正并行,反而增加调度开销,自然看不到提速效果。
靠谱的提速方案
1. 直接用多进程(CPU密集场景最优解)
用concurrent.futures.ProcessPoolExecutor或multiprocessing绕开GIL,真正利用多核CPU并行跑不同参数的回测,这是CPU密集任务的正确打开方式。
示例代码:
from concurrent.futures import ProcessPoolExecutor def runBacktest(lenSmooth1): # 你的原回测逻辑,比如大数据集计算、指标生成等 ... if __name__ == "__main__": # 要测试的lenSmooth1参数列表 params = [5, 10, 15, 20, 25] # 进程数建议设为CPU核心数,别超太多 with ProcessPoolExecutor(max_workers=4) as executor: # 并行执行所有参数的回测,拿到结果列表 results = list(executor.map(runBacktest, params))
- 注意:如果回测涉及共享数据,要么让每个进程独立加载(避免大文件进程间拷贝的话,可使用
multiprocessing.Manager或共享内存对象),要么提前预处理好数据再传给每个进程,别在进程内重复读大文件。
2. 异步+多进程混合(若回测包含IO操作)
如果runBacktest里有IO环节(比如读取历史数据文件、调用外部行情接口),可以把IO部分用asyncio异步处理,CPU计算部分丢给进程池,既节省IO等待时间,又能并行计算。
代码思路示例:
import asyncio from concurrent.futures import ProcessPoolExecutor async def async_run_backtest(lenSmooth1, pool): # 异步读数据,避免IO等待阻塞 data = await async_read_large_data_file() # 把CPU密集计算交给进程池 result = await asyncio.get_event_loop().run_in_executor(pool, calc_backtest, lenSmooth1, data) return result def calc_backtest(lenSmooth1, data): # 抽离出纯CPU计算的回测逻辑 ... async def main(): params = [5,10,15,20] with ProcessPoolExecutor(max_workers=4) as pool: tasks = [async_run_backtest(p, pool) for p in params] results = await asyncio.gather(*tasks) if __name__ == "__main__": asyncio.run(main())
3. 先优化回测本身的代码(基础且高效)
不管用哪种并行方案,先优化runBacktest内部逻辑,能直接大幅提速:
- 把Python原生循环替换成
numpy/pandas的向量化操作,比如用df.rolling(window=lenSmooth1).mean()替代手写的滑动窗口循环。 - 用
numba给核心计算函数加速,给循环密集的函数添加@numba.jit(nopython=True)装饰器,直接将Python代码编译成机器码,速度能提升数倍。 - 缓存重复计算的结果,比如不同参数共用的历史数据预处理,只计算一次,避免每次回测重复执行。
- 减少内存拷贝,比如用
df.values代替df.to_numpy()(场景合适时),或使用内存视图操作数据。
测试小技巧
- 用
cProfile运行一次runBacktest,定位耗时占比最高的代码段,优先优化瓶颈,避免盲目并行。 - 先用小数据集验证并行逻辑的正确性,再用大数据集测试速度,避免无效调试。
内容的提问来源于stack exchange,提问作者AG23
相关产品推荐
相关产品推荐

