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

我曾经做过一个比较典型的盘中选股流程。

每天开盘后，策略首先面对的不是“买哪只股票”，而是：

程序启动后，先获取市场行情，然后把所有股票放进 Pandas DataFrame，再根据成交量、价格、涨跌幅等字段进行过滤。

但是运行一段时间之后，我开始发现一个问题：

数据获取和策略筛选耦合得太紧。

策略本来只是想找出符合条件的股票，却不得不承担整个市场行情数据的读取和清洗。

确定股票池 → 获取候选行情 → 筛选 → 对候选标的做进一步验证

这样改完以后，我对“先查行情还是先建股票池”的理解发生了变化。

如果目标是全市场扫描，我当然需要先拿到市场集合。

如果策略已经有了 30 只候选股票，那么再查询股票池就没有必要。

所以真正的问题不是“哪个先”，而是：

1. 我把量化数据处理拆成了三个层级

后来我在项目中形成了一套比较固定的数据分层。

“这些股票现在是什么状态？”

“这些股票过去发生过什么？”

这三个问题分别对应不同的数据请求。

如果把三者混在一起，就容易产生重复查询。

2. 股票池不是简单的股票代码列表

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


后来实际做策略才发现，股票池往往还需要携带策略上下文。

因此我现在会把“股票池”理解成一个策略对象，而不是一个简单列表。

真正请求行情时，再把其中的 symbol 提取出来。

3. 全市场行情和候选行情是两种不同任务

这一点在工程实践里非常明显。

“找出所有 A 股里符合某条件的股票。”

那么全市场行情本身就是输入数据。

“这 20 只候选股票现在有没有继续满足条件？”

那么全市场行情已经没有必要。

这也是为什么我现在不会简单地说“永远先建股票池”或者“永远先查行情”。

网页上能看到价格，我把价格抓下来就可以。

但真正进入策略之后，问题马上出现。

页面结构可能变化，字段命名也可能变化，更重要的是，我还得自己处理：

最后我发现，自己维护一个稳定的数据访问层，成本比最开始想象的大得多。

一个数据源负责 A 股，一个负责美股，另一个负责港股。

真正麻烦的不是“能不能获取”，而是字段定义经常不同。

例如同样是收盘价，不同接口的数据字段名称、时间格式和复权方式可能不同。

于是策略层不得不写很多适配代码。

后来我更看重数据接口是否能让我在策略层保持统一的数据模型。

这样我的策略代码不需要因为市场变化而完全重写。

我在使用 QuantDash SDK 时，正好采用了这种代码格式，并且可以通过 Pandas DataFrame 接收结果，这让后续数据处理比较自然。

我先从最简单的策略对象开始。

from dataclasses import dataclass
from typing import List


@dataclass
class StrategyUniverse:
    name: str
    symbols: List[str]


def build_candidate_universe():
    return StrategyUniverse(
        name="candidate_pool",
        symbols=[
            "600519.SH",
            "000001.SZ",
            "000858.SZ",
            "600000.SH"
        ]
    )


if __name__ == "__main__":
    universe = build_candidate_universe()

    print(f"股票池名称: {universe.name}")
    print(f"股票数量: {len(universe.symbols)}")
    print("股票列表:")
    for symbol in universe.symbols:
        print(symbol)


我特意把股票池和行情查询分开。

这样做的好处是，后面无论行情接口换成什么形式，策略本身都不需要修改。

2. 第二步：拿股票池查询最新行情

import os
from dataclasses import dataclass
from typing import List

from quantdash import QuantDash


@dataclass
class StrategyUniverse:
    name: str
    symbols: List[str]


def fetch_quotes(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(
            symbols=universe.symbols,
            to_dataframe=True
        )

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

        return df

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


if __name__ == "__main__":
    universe = StrategyUniverse(
        name="candidate_pool",
        symbols=[
            "600519.SH",
            "000001.SZ",
            "000858.SZ"
        ]
    )

    quotes = fetch_quotes(universe)

    if quotes is not None:
        columns = [
            "symbol",
            "last_price",
            "prev_close",
            "volume",
            "ext.name",
            "ext.change_pct"
        ]

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

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


StrategyUniverse → quotes.get → DataFrame

策略和数据访问层之间只有一个明确的数据契约。

3. 第三步：全市场扫描时，再使用市场股票池

如果我的策略还没有候选股票，而是需要从 A 股全市场筛选，那么流程就不一样。

import os
from quantdash import QuantDash


def load_market_quotes():
    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("没有获取到市场行情")
            return df

        return df

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


def filter_candidates(df):
    if df is None or df.empty:
        return df

    if "ext.change_pct" not in df.columns:
        print("行情数据缺少涨跌幅字段")
        return df.iloc[0:0]

    candidates = df[df["ext.change_pct"] > 0].copy()

    return candidates


if __name__ == "__main__":
    market_df = load_market_quotes()

    if market_df is not None:
        candidates = filter_candidates(market_df)

        print(f"筛选后候选数量: {len(candidates)}")
        print(candidates.head().to_string(index=False))


这段代码只是演示数据工程流程，并不代表这个筛选条件本身具有交易意义。

我在实际项目里，会把筛选条件放在策略层，不会让数据访问层决定策略逻辑。

4. 第四步：候选股票再做历史验证

当候选股票产生以后，我通常会进一步查看历史 K 线。

import os
from quantdash import QuantDash


def load_history(symbol, count=20):
    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="1d",
            count=count,
            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__":
    symbol = "600519.SH"

    df = load_history(symbol)

    if df is not None:
        columns = [
            "symbol",
            "name",
            "trade_date",
            "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))


这里我一般会特别明确复权口径。

如果我要计算历史收益率，可以使用前复权比例方式：

import os
from quantdash import QuantDash


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

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

    qd = QuantDash(api_key=api_key)

    try:
        return qd.klines.get(
            symbol,
            period="1d",
            count=60,
            adjust="forward",
            to_dataframe=True
        )

    except Exception as e:
        print(f"复权历史数据查询失败: {e}")
        return None


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

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


复权方式必须成为数据契约的一部分，而不能让不同策略各自随便决定。

量化策略临时选股时，我为什么坚持先确定股票池再请求行情

算法

编程语言

后端

量化选股数据流程应分层设计：先定义股票池（Universe）确定候选标的，再获取行情（Quotes）验证状态，最后用历史K线（Klines）做深度验证。全市场扫描与候选验证需区分处理，避免数据耦合。统一标的代码格式（如.SH/.US/.HK）和复权口径，可提升策略回测与实盘一致性，降低维护成本。

数据清洗

数据接口

数据结构

数据处理

数据流

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

量化策略临时选股时，我为什么坚持先确定股票池再请求行情

量化策略临时选股时，我为什么坚持先确定股票池再请求行情

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐