【数据管道】性能提升 10 倍！用 Polars + QuantDash 极速处理全市场多周期行情数据

做多因子选股或多标的数据清洗时很多人习惯用 Pandas。但在处理日内分钟线如 1 分钟、5 分钟 K 线或者全市场几千只股票的历史数据时Pandas 的单线程计算瓶颈就暴露出来了[2]。近两年由 Rust 编写的 Polars 库凭借其极速的多线程并行和 Lazy Evaluation惰性计算机制已经成为量化界的新宠[7][8]。本文将演示如何结合 QuantDash 的多周期行情 API 和 Polars搭建一个高效的技术指标以计算 VWAP 和 20日 动量为例批量计算管道[9][10]。1. 为什么不用 Pandas而推荐 Polars QuantDash零内存拷贝对接QuantDash 的 Python SDK 返回标准的 Pandas DataFrame[3][11]。我们可以使用 polars.from_pandas() 实现零内存拷贝的极速转换。并行计算优势如果我们需要计算几十只自选股的技术指标Polars 的多线程并发特性能够把你的 CPU 核心全部吃满处理速度相比 Pandas 往往有数倍甚至十倍的提升。2. 代码实现极速指标管道下面的例子展示了如何一次性下载多只股票的 K 线数据通过 Polars 组装成大宽表并利用其高效的 Expression 语法并行计算多项因子[6]。import polars as pl import pandas as pd from quantdash import QuantDash import time # 初始化 QuantDash qd QuantDash(api_keyyour_quantdash_api_key) def fetch_and_build_pipeline(symbols: list, start_date: str, end_date: str): start_time time.time() all_dfs [] # 1. 批量高效获取多市场历史 K 线数据 for sym in symbols: try: # 获取 Pandas DataFrame pdf qd.stock.get_kline(symbolsym, start_datestart_date, end_dateend_date) # 添加一列标识代码 pdf[symbol] sym all_dfs.append(pdf) except Exception as e: print(f获取 {sym} 数据失败: {e}) # 合并 DataFrame 并转化为 Polars DataFrame if not all_dfs: return None combined_pdf pd.concat(all_dfs, ignore_indexTrue) # 转换 df pl.from_pandas(combined_pdf) print(f数据拉取与转换耗时: {time.time() - start_time:.2f} 秒总行数: {df.height}) # 2. 使用 Polars 惰性管道进行高性能计算 # 我们要计算 # - 20 日收盘价简单移动均线 (MA_20) # - 20 日累计涨跌幅 (Momentum_20) # - 典型价格 (Typical Price) calc_pipeline ( df.lazy() .sort([symbol, date]) .group_by(symbol) .agg([ pl.col(date), pl.col(close), pl.col(close).rolling_mean(window_size20).alias(MA_20), ((pl.col(close) - pl.col(close).shift(20)) / pl.col(close).shift(20)).alias(Momentum_20), ((pl.col(high) pl.col(low) pl.col(close)) / 3.0).alias(Typical_Price) ]) .explode([date, close, MA_20, Momentum_20, Typical_Price]) # 还原展平数据 ) # 3. 触发计算 result_df calc_pipeline.collect() print(f指标计算完成总耗时: {time.time() - start_time:.2f} 秒) return result_df if __name__ __main__: # 选取 A股/美股/港股 的头部标的测试多市场管道 test_symbols [600519.SH, AAPL.US, 00700.HK, 000001.SZ] processed_data fetch_and_build_pipeline( symbolstest_symbols, start_date2024-01-01, end_date2025-12-31 ) if processed_data is not None: print(processed_data.head(10))3. 性能优化总结通过把 QuantDash 规范化的数据源塞进 Polars 的 LazyFrame我们避免了传统 Pandas 里面大量的 groupby-apply 慢速循环。Polars 会自动在底层开启多线程对于包含几十万行记录的分钟线数据集这种架构能让你在本地笔记本上也能跑出“工作站级”的因子回测速度。相关链接QuantDash 官方QuantDashPython SDK 快速开始快速开始 - QuantDash