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

我曾经做过一个典型的日频选股策略。

它的第一步不是计算因子，而是：

我原本认为这个任务没有什么技术含量。

无非就是调用API，然后把数据塞进 Pandas。

但真正上线以后，我才发现数据获取本身就是整个策略链路里非常关键的一环。

有一天，我发现策略运行时间比平时长。

可是后面的因子计算明显推迟了。

但测完之后，我发现一个非常尴尬的问题：

我根本不知道自己应该测什么。

300只股票
↓
获取日线
↓
转换成DataFrame
↓
检查缺失
↓
统一字段
↓
计算因子


所以真正应该测的是整个数据链路。

1. 单只标的测试无法代表股票池任务

我以前测试接口时喜欢拿一只熟悉的股票。

现在回头看，这个结论明显不完整。

股票数量一上来以后，问题就变成了：

2. 数据量比股票数量更值得关注

假设我要300只股票，每只10根日线。

如果我要300只股票，每只500根日线，那么就是15万行。

因此我现在做 Benchmark 时，一定会把：

3. 回测数据的“速度”和“口径”是两件事

我曾经因为复权口径没有统一，花了比优化程序更长的时间排查结果差异。

所以我现在会在数据层明确记录复权方式。

速度快，但数据口径错了，对量化系统没有意义。

我最早比较倾向于自己搭数据系统。

特别是当策略研究本身才是我的核心工作时，我并不希望大量时间都耗在数据采集。

但随着时间推移，我越来越不喜欢这个方案。

页面结构一变，代码就需要修改。

对于量化数据这种需要稳定运行的任务，我更希望数据获取层是一个明确的接口，而不是依赖页面结构。

最后，我更倾向于把数据访问层标准化。

我后来使用 QuantDash Python SDK，安装方式比较直接：

SDK 支持 Pandas 返回，也提供批量 K 线接口。

对我而言，这意味着我可以把数据获取层和策略计算层分开。

实际生产环境里，我会从数据库或股票池文件读取。

为了让测试代码保持完整，我这里使用一个明确的列表构造方法：

def build_test_symbols():
    symbols = []

    for i in range(150):
        symbols.append(
            f"{i + 600000:06d}.SH"
        )

    for i in range(150):
        symbols.append(
            f"{i + 1:06d}.SZ"
        )

    return symbols[:300]


，不代表这些代码当前一定都是有效、上市或有数据的证券。

在真实项目中，我会使用实际标的池。

import os
import time

from quantdash import QuantDash


def build_test_symbols():
    symbols = []

    for i in range(150):
        symbols.append(
            f"{i + 600000:06d}.SH"
        )

    for i in range(150):
        symbols.append(
            f"{i + 1:06d}.SZ"
        )

    return symbols[:300]


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        raise RuntimeError(
            "请设置环境变量 QUANTDASH_API_KEY"
        )

    qd = QuantDash()

    symbols = build_test_symbols()

    start = time.perf_counter()

    try:
        result = qd.klines.batch(
            symbols,
            period="1d",
            count=10,
            to_dataframe=True,
            show_progress=True,
        )
    except Exception as exc:
        print(f"批量获取失败：{exc}")
        return

    elapsed = time.perf_counter() - start

    success_count = 0
    total_rows = 0
    empty_symbols = []

    for symbol in symbols:
        df = result.get(symbol)

        if df is None or df.empty:
            empty_symbols.append(symbol)
            continue

        success_count += 1
        total_rows += len(df)

    print("\n===== 测试结果 =====")
    print(f"股票数量：{len(symbols)}")
    print(f"成功数量：{success_count}")
    print(f"空数据数量：{len(empty_symbols)}")
    print(f"数据行数：{total_rows}")
    print(f"端到端耗时：{elapsed:.4f} 秒")

    if empty_symbols:
        print("\n需要检查的标的：")
        for symbol in empty_symbols[:20]:
            print(symbol)


if __name__ == "__main__":
    main()

这段代码就是我现在比较认可的最小 Benchmark。

它没有试图证明某个固定速度。

把真实任务跑一遍，然后记录结果。

真正要比较数据源时，我不会只运行一次。

import os
import statistics
import time

from quantdash import QuantDash


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        raise RuntimeError(
            "请设置 QUANTDASH_API_KEY"
        )

    qd = QuantDash()

    symbols = [
        "600519.SH",
        "000001.SZ",
        "000858.SZ",
    ]

    elapsed_times = []

    for index in range(5):
        start = time.perf_counter()

        try:
            result = qd.klines.batch(
                symbols,
                period="1d",
                count=10,
                to_dataframe=True,
                show_progress=False,
            )
        except Exception as exc:
            print(
                f"第 {index + 1} 次请求失败：{exc}"
            )
            continue

        elapsed = time.perf_counter() - start
        elapsed_times.append(elapsed)

        rows = sum(
            len(df)
            for df in result.values()
            if df is not None and not df.empty
        )

        print(
            f"第 {index + 1} 次："
            f"{elapsed:.4f} 秒，"
            f"{rows} 行"
        )

    if not elapsed_times:
        print("没有成功完成测试")
        return

    print("\n===== 统计 =====")
    print(
        f"平均耗时："
        f"{statistics.mean(elapsed_times):.4f} 秒"
    )
    print(
        f"最短耗时："
        f"{min(elapsed_times):.4f} 秒"
    )
    print(
        f"最长耗时："
        f"{max(elapsed_times):.4f} 秒"
    )


if __name__ == "__main__":
    main()


因为如果第一次是网络特别顺畅，第二次刚好出现抖动，单次测试很容易误导自己。

import datetime
import os
import time

from quantdash import QuantDash


def timestamp_ms(year, month, day):
    dt = datetime.datetime(
        year,
        month,
        day,
    )
    return int(dt.timestamp() * 1000)


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        raise RuntimeError(
            "请设置 QUANTDASH_API_KEY"
        )

    qd = QuantDash()

    symbols = [
        "600519.SH",
        "000001.SZ",
        "000858.SZ",
    ]

    start_time = timestamp_ms(
        2026, 5, 1
    )

    end_time = timestamp_ms(
        2026, 5, 31
    )

    started = time.perf_counter()

    try:
        result = qd.klines.batch(
            symbols,
            period="1d",
            start_time=start_time,
            end_time=end_time,
            to_dataframe=True,
        )
    except Exception as exc:
        print(f"请求失败：{exc}")
        return

    elapsed = time.perf_counter() - started

    print(
        f"时间区间请求耗时："
        f"{elapsed:.4f} 秒"
    )

    for symbol, df in result.items():
        if df is None or df.empty:
            print(f"{symbol}: 无数据")
            continue

        print(
            f"{symbol}: "
            f"{len(df)} 条日线"
        )


if __name__ == "__main__":
    main()


我通常会把不同数据量的结果分别记录，而不是混在一起。

1. 不要把“接口延迟”和“任务耗时”混为一谈

300只股票日线数据多久能拿完？我从一次回测数据事故开始重新设计API Benchmark

编程语言

数据库

职业发展

量化实盘数据测试中，300只股票日线获取耗时受股票池、K线数量、批量接口及复权方式影响，单次测试不可靠。需固定输入条件并重复运行，统计成功率和稳定性。量化系统应关注“拿得快”与“拿得对”，统一数据口径，建立可复现的Benchmark流程，而非依赖固定响应时间。

数据一致性

数据管道

数据接口

数据处理

服务器

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

300只股票日线数据多久能拿完？我从一次回测数据事故开始重新设计API Benchmark

300只股票日线数据多久能拿完？我从一次回测数据事故开始重新设计API Benchmark

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐