Python量化数据API批量K线为什么越处理越慢？我用Polars重做了数据处理层

一、问题场景：我在量化实盘中遇到的真实困境

我在量化项目里有一个很典型的任务：每天收盘之后，把股票池里的历史 K 线拉下来，然后计算收益率、均线、波动率、成交量变化等指标，最后交给策略回测模块。

刚开始的时候，我完全没有觉得这是一个问题。

毕竟 Python 里 Pandas 已经非常成熟，拿到数据之后直接 DataFrame 操作就可以了。

真正让我头疼，是股票池从几十只扩大到几百只、几千只之后。

我的代码最初大概是这样的思路：

当数据量比较小时，这套方式非常舒服。

但是当我开始处理较大的股票池后，我发现 CPU 使用率和内存占用开始明显增加。

import pandas as pd

frames = []

for df in dfs:
    frames.append(df)

all_data = pd.concat(frames, ignore_index=True)


问题并不是 Pandas “不能用”。

Pandas 非常适合做单表分析，但当数据处理开始变成批量数据工程时，单纯依赖 Pandas 就容易出现性能瓶颈。

这也是我后来决定把数据获取和数据计算拆开的原因。

数据接口负责把 K 线稳定拿回来，Pandas 负责兼容传统分析代码，而 Polars 负责更重的数据处理。

我后来使用 QuantDash 的批量 K 线接口获取数据，再把返回的 Pandas DataFrame 转换为 Polars DataFrame。

这个思路比直接寻找一个“完全替代 Pandas 的行情 SDK”更实际。

我后来把整个链路拆开之后，发现真正的问题主要有三个。

请求行情
↓
转换 DataFrame
↓
计算指标
↓
再次请求
↓
再次 concat
↓
再次计算

数据获取和数据处理互相阻塞。

行情接口返回速度不稳定的时候，计算层只能等待。

计算数据量变大之后，API 请求层又无法很好地复用。

数据源
↓
批量获取
↓
标准化
↓
Polars
↓
因子计算
↓
回测

2. 一个 DataFrame 一个 DataFrame 地拼接

这是我以前比较容易犯的错误。

假设有 1000 个标的，每个标的返回一个 DataFrame。

import pandas as pd

result = pd.DataFrame()

for df in frames:
    result = pd.concat([result, df], ignore_index=True)

这种写法对于小数据没什么感觉。

但是数据量一大，我就开始明显感受到内存复制和对象管理带来的开销。

拿一个
↓
拼一个
↓
再拿一个
↓
再拼一个

3. 数据源支持 Pandas，不代表计算层必须一直使用 Pandas

这是我后来比较重要的一个认知变化。

QuantDash Python SDK 原生返回 Pandas DataFrame，这对于 Python 量化开发非常方便。

但我没有必要因此让整个数据处理链都绑定在 Pandas 上。

QuantDash
    ↓
Pandas DataFrame
    ↓
Polars DataFrame
    ↓
计算


这样既保留了数据接口的便利性，又可以利用 Polars 做后续处理。

但到了更大规模之后，我发现问题仍然存在。

理论上完全可以自己维护 HTTP 请求，然后把不同市场的数据统一转换成 DataFrame。

但实际维护起来以后，我发现自己真正应该解决的是策略和数据处理问题，而不是长期维护各种行情接口。

特别是 A 股、美股、港股的数据格式、代码格式、交易时间等问题，很容易逐渐侵入业务代码。

方案三：数据获取和计算彻底解耦

我让行情 SDK 做它擅长的事情：

然后让 Polars 做它擅长的事情：

批量数据转换、过滤、分组和计算。

目前我的数据链路基本遵循这个原则。

四、具体实现：QuantDash批量K线 + Polars

我建议 API Key 放在环境变量里，而不是写死在代码中。

下面是我实际会采用的一种批量处理方式。

import os
import sys

import polars as pl
from quantdash import QuantDash


def load_batch_klines(symbols):
    """获取批量日K线，并转换为Polars DataFrame。"""

    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError(
            "未检测到 QUANTDASH_API_KEY，请先配置 API Key。"
        )

    qd = QuantDash(api_key=api_key)

    try:
        result = qd.klines.batch(
            symbols,
            period="1d",
            count=100,
            to_dataframe=True,
            show_progress=True,
        )
    except Exception as exc:
        raise RuntimeError(f"批量获取K线失败: {exc}") from exc

    frames = []

    for symbol, pandas_df in result.items():
        if pandas_df is None or pandas_df.empty:
            print(f"{symbol}: 没有返回数据")
            continue

        try:
            frame = pl.from_pandas(pandas_df)
        except Exception as exc:
            print(f"{symbol}: Pandas 转 Polars 失败: {exc}")
            continue

        frames.append(frame)

    if not frames:
        return pl.DataFrame()

    return pl.concat(
        frames,
        how="vertical_relaxed",
    )


def calculate_features(df):
    """使用Polars计算简单的量化特征。"""

    if df.is_empty():
        return df

    required_columns = {
        "symbol",
        "trade_date",
        "open",
        "high",
        "low",
        "close",
        "volume",
    }

    missing = required_columns - set(df.columns)

    if missing:
        raise ValueError(
            f"缺少必要字段: {sorted(missing)}"
        )

    result = (
        df.sort(["symbol", "trade_date"])
        .with_columns(
            (
                pl.col("close")
                .pct_change()
                .over("symbol")
            ).alias("daily_return")
        )
        .with_columns(
            (
                pl.col("close")
                .rolling_mean(window_size=20)
                .over("symbol")
            ).alias("ma20")
        )
        .with_columns(
            (
                pl.col("volume")
                .rolling_mean(window_size=20)
                .over("symbol")
            ).alias("volume_ma20")
        )
    )

    return result


def main():
    symbols = [
        "600519.SH",
        "000001.SZ",
        "000858.SZ",
    ]

    try:
        df = load_batch_klines(symbols)

        if df.is_empty():
            print("未获取到任何K线数据。")
            return

        print("原始数据:")
        print(df.head())

        result = calculate_features(df)

        print("\n计算后的数据:")
        print(
            result.select(
                [
                    "symbol",
                    "trade_date",
                    "close",
                    "daily_return",
                    "ma20",
                    "volume_ma20",
                ]
            ).tail(20)
        )

    except Exception as exc:
        print(f"程序执行失败: {exc}")
        sys.exit(1)


if __name__ == "__main__":
    main()

这里有一个我认为特别值得注意的地方：

我没有让 Polars 负责调用行情 API。

QuantDash 的批量接口本身已经可以一次请求多只标的，然后返回以标的代码为 key 的数据结果。

所以我只在接口层拿到 Pandas DataFrame，然后立即转换成 Polars。

这样做的好处是架构非常简单。

五、我后来怎么处理更大的数据量

当数据量继续增加后，我不会把所有字段全部保留下来。

那么后面就尽量只保留这些字段。

import polars as pl


def select_ohlcv_columns(df):
    columns = [
        "symbol",
        "trade_date",
        "open",
        "high",
        "low",
        "close",
        "volume",
    ]

    available = [
        column
        for column in columns
        if column in df.columns
    ]

    return df.select(available)

我还会把“数据获取”和“数据计算”做成两个函数。

以后如果我要把行情数据保存到 Parquet，或者换成本地缓存，就不需要修改因子计算代码。

编程语言

后端

数据库

利用QuantDash批量K线API与Polars构建高效数据处理层，解决量化回测中Pandas拼接慢、内存开销大的问题。实现数据获取与计算解耦，通过Polars向量化操作提升批量处理性能，并分享性能基准测试与避坑经验，为多市场量化系统提供可扩展架构参考。

数据清洗

数据接口

数据转换

数据处理

Python

Linux

2026上云采购 | AI焕新·智启新局

thpc

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

Python量化数据API批量K线为什么越处理越慢？我用Polars重做了数据处理层

Python量化数据API批量K线为什么越处理越慢？我用Polars重做了数据处理层

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐