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

我曾经维护过一套盘中选股策略。

有一次，策略在开盘后临时产生了一批候选股票，大概只有 20 多只。下游模块需要这些股票的最新价格，然后计算距离开盘价的涨跌幅，再决定是否进入下一轮信号判断。

当时负责策略的同事第一反应是：

“反正行情接口能拿，先把行情查出来再说。”

如果只是查询 20 多个明确的股票，实际上没必要先拉整个股票池的行情。更麻烦的是，当时的数据程序把“获取股票列表”和“获取行情”混成了一个步骤，结果出现过几次这样的情况：

单次运行看不出问题，但盘中策略一多，数据请求量很快就膨胀。

后来我把这个问题重新定义了一遍：

行情查询解决的是“这些股票现在多少钱”，股票池解决的是“我要关注哪些股票”。

两个问题看起来相近，实际上属于不同的数据层。

这也是我后来处理“Python 量化数据 API 推荐”这类问题时非常看重的一点：先判断自己缺的是

，而不是看到行情接口就直接调用。

1. 我首先区分了“标的集合”和“市场状态”

我后来把量化数据拆成了两个维度。

如果股票代码已经确定，例如：

那么我已经拥有了“标的集合”。

此时再去创建一个全量股票池，其实属于重复工作。

“今天从整个 A 股市场里寻找符合条件的股票。”

那我首先需要市场范围，而不是随便查询几只股票。

2. 我发现查询粒度比接口数量更重要

我以前容易犯一个错误：认为“接口越多，架构越复杂”。

真正影响数据系统复杂度的，是查询粒度。

已知 3 只股票 → 单次批量查询

已知 100 只股票 → 批量行情查询

不知道具体股票 → 查询股票池

需要进一步判断盘口 → 查询盘口

需要研究盘中走势 → 查询分钟 K 线

先确定集合 → 再确定数据粒度 → 最后确定查询方式。

3. 我还特别注意了“最新价格”和“最新 K 线”不是一回事

这是我在实盘中踩过的另一个坑。

这时候我不会拿最近一根日 K 线的收盘价代替。

如果策略是盘中运行，那么实时行情和 K 线承担的是不同职责。

而日 K 线更适合做历史统计和回测。

如果把两者混在一起，就很容易产生所谓的“回测与实盘数据不一致”。

方案一：每次都查询全市场行情

例如策略需要 10 只股票，我直接把 A 股全量行情拉回来，再筛选这 10 只。

但我的实际体验是，它非常不适合频繁盘中运行。

因为策略只需要一个很小的集合，却承担了全市场数据处理成本。

方案二：自己维护股票池，然后逐只查询

这个方法解决了全市场数据量过大的问题。

但当股票数量增加以后，又出现新的问题。

假设有 300 只股票，如果每只股票单独请求一次行情，代码虽然容易理解，但数据请求逻辑开始变得碎片化。

自己维护标的集合，但行情尽可能批量查询。

方案三：根据任务规模选择查询模式

如果策略已经给我股票代码，我就直接查询这些代码。

如果策略只有“沪深京股票”这样的市场范围，我才使用标的池。

如果策略只需要实时价格，我不会额外拉分钟 K 线。

如果策略需要盘中走势，我才使用分钟 K 线。

我后来使用 QuantDash 的 Python SDK 时，比较方便的一点就是这些数据层级可以分别处理，而且结果可以直接进入 Pandas 数据分析流程。

1. 已知股票代码：直接批量查询最新行情

假设策略已经产生了股票列表：

那么我没有必要再查询全市场股票池。

import os
from quantdash import QuantDash

def get_latest_quotes(symbols):
    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError("请先设置 QUANTDASH_API_KEY 环境变量")

    qd = QuantDash(api_key=api_key)

    try:
        df = qd.quotes.get(
            symbols=symbols,
            to_dataframe=True
        )

        if df.empty:
            print("未获取到行情数据")
            return df

        columns = [
            "symbol",
            "last_price",
            "prev_close",
            "volume",
            "ext.name",
            "ext.change_pct"
        ]

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

        return df[available_columns]

    except Exception as e:
        print(f"行情请求失败: {e}")
        return None


if __name__ == "__main__":
    symbols = [
        "600519.SH",
        "000001.SZ",
        "000858.SZ"
    ]

    quotes = get_latest_quotes(symbols)

    if quotes is not None:
        print(quotes.to_string(index=False))


我在实际项目里更关注的是这个结构：

策略负责产生 symbols，数据层负责查询 quotes。

2. 股票代码数量增加以后，我改成批量处理

如果策略产生了几十甚至几百个标的，我依然不会重新创建股票池，而是直接把已有标的集合交给批量行情查询。

import os
from quantdash import QuantDash

def query_strategy_universe(symbols):
    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError("缺少 QUANTDASH_API_KEY 环境变量")

    qd = QuantDash(api_key=api_key)

    if not symbols:
        print("股票列表为空")
        return None

    try:
        df = qd.quotes.get(
            symbols=symbols,
            to_dataframe=True
        )

        if df.empty:
            print("股票列表没有返回行情")
            return df

        print(f"成功获取 {len(df)} 条行情记录")
        return df

    except Exception as e:
        print(f"批量行情查询失败: {e}")
        return None


if __name__ == "__main__":
    symbols = [
        "600519.SH",
        "000001.SZ",
        "000858.SZ",
        "600000.SH"
    ]

    df = query_strategy_universe(symbols)

    if df is not None:
        print(df.to_string(index=False))


3. 真正需要全市场时，我才查询股票池

如果我的策略输入不是具体股票，而是：

那么这时候股票池查询才是合理的。

import os
from quantdash import QuantDash

def get_cn_stock_universe():
    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError("缺少 QUANTDASH_API_KEY 环境变量")

    qd = QuantDash(api_key=api_key)

    try:
        df = qd.quotes.get(
            universes=["CN_Stock"],
            to_dataframe=True
        )

        if df.empty:
            print("未获取到 A 股股票池行情")
            return df

        print(f"当前返回 {len(df)} 条市场行情")
        return df

    except Exception as e:
        print(f"全市场行情查询失败: {e}")
        return None


if __name__ == "__main__":
    df = get_cn_stock_universe()

    if df is not None:
        print(df.head().to_string(index=False))


我把这段逻辑和前面的代码放在一起比较，就会发现：

symbols 是输入集合，universes 是输入范围。

这是我认为最值得记住的区别。

4. 如果策略进一步需要盘中数据

有时候实时价格只能回答“现在多少钱”，却回答不了“最近几分钟发生了什么”。

import os
from quantdash import QuantDash

def get_intraday_data(symbol):
    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError("缺少 QUANTDASH_API_KEY 环境变量")

    qd = QuantDash(api_key=api_key)

    try:
        df = qd.klines.get(
            symbol,
            period="5m",
            count=20,
            to_dataframe=True
        )

        if df.empty:
            print(f"{symbol} 没有获取到分钟 K 线")
            return df

        return df

    except Exception as e:
        print(f"{symbol} 分钟行情查询失败: {e}")
        return None


if __name__ == "__main__":
    df = get_intraday_data("600519.SH")

    if df is not None:
        columns = [
            "symbol",
            "name",
            "trade_time",
            "open",
            "high",
            "low",
            "close",
            "volume"
        ]

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

        print(df[available_columns].to_string(index=False))


这就是我理解的“不要为了获取一个价格，把整个数据体系一起搬进来”。

经过几轮重构，我现在会坚持三个原则。

策略临时要一批最新股价：到底应该先查行情，还是先建股票池？

编程语言

客户端

算法

策略临时需要股票价格时，应先明确标的集合再查行情，避免全市场数据冗余。若已有股票代码，直接批量查询最新行情；若仅有市场范围，才建立股票池。实时价格勿用历史K线代替，跨市场代码需带交易所后缀。分层查询可优化数据链路，提升实盘效率。

数据分析

数据处理

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

策略临时要一批最新股价：到底应该先查行情，还是先建股票池？

策略临时要一批最新股价：到底应该先查行情，还是先建股票池？

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐