本文会从我实际遇到的行情扫描问题出发，分析逐股票请求的工程痛点，并通过 Python 实战展示如何获取 A 股全量行情、筛选股票以及进一步进入策略计算。

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

我做量化策略时，经常会遇到一个看起来非常简单的问题：

“今天所有A股里面，哪些股票满足我的条件？”

如果只看一两只股票，这件事非常简单。

但如果策略需要扫描整个 A 股市场，事情马上就不一样了。

我以前比较直接的做法是准备一个股票代码列表，然后循环请求：

for symbol in symbols:
    quote = get_quote(symbol)
    process(quote)


股票数量少的时候确实也没什么问题。

真正让我头疼的是股票池扩大以后，数据请求、异常处理、结果拼接、字段统一全部变成了额外工作。

最后我的策略代码里出现了大量与策略本身无关的内容：

for symbol in symbols:
    try:
        data = get_quote(symbol)

        if data is None:
            continue

        if data["change_pct"] > threshold:
            selected.append(data)

    except Exception as e:
        print(symbol, e)


我写的已经不是选股策略，而是在写一个简易行情数据基础设施。

而这恰恰不是策略研究应该承担的事情。

我后来把这个问题拆开看，发现主要有三个层次。

1. 股票代码管理本身就是一个问题

A股并不是简单地只有一个代码。

我在不同数据源之间切换时，经常需要处理类似：

如果自己维护股票列表，还要考虑股票池更新、交易所后缀以及新增标的。

“我要扫描全部股票”实际上首先是一个 Universe 管理问题。

2. 行情请求和策略逻辑混在一起

获取股票列表
↓
请求行情
↓
处理异常
↓
拼接DataFrame
↓
计算指标
↓
筛选股票


数据层
  ↓
标准化DataFrame
  ↓
策略层
  ↓
选股结果


策略只关心数据，不应该关心数据到底是怎么来的。

3. 真正需要的是“标的池查询”

我后来使用的思路发生了变化。

在提供的 SDK 中，行情接口支持通过标的池查询全量行情。

在确定这个方向之前，我其实考虑过几种方案。

对于一个研究型策略来说，这些代码基本都是“外围工程”。

方案二：免费数据源 + 自己封装

因为真正麻烦的不是第一次把数据取出来，而是后面的异常情况。

请求失败怎么办？
某个标的数据为空怎么办？
字段不一致怎么办？
股票池怎么更新？
回测代码和实盘代码如何保持一致？

这些问题最终都会回到自己的数据层。

方案三：把行情数据直接标准化成 DataFrame

我后来使用 QuantDash Python SDK，把数据获取和策略处理分开。

SDK 原生支持 Pandas，因此拿到行情以后可以直接进入筛选、计算和分析流程。

第一步：获取两只股票实时行情

我通常会先用少量股票验证数据链路。

import os

from quantdash import QuantDash


def main():
    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=["600519.SH", "000001.SZ"],
            to_dataframe=True
        )

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

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

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

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

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


if __name__ == "__main__":
    main()


这里有一个我比较在意的工程习惯：

第二步：从逐股票扫描升级成全量股票池扫描

真正让我觉得代码发生变化的是这里。

如果我要扫描 A 股股票池，核心代码可以直接写成：

import os

from quantdash import QuantDash


def main():
    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

        print(f"获取到 {len(df)} 条行情")
        print(df.head())

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


if __name__ == "__main__":
    main()


对我来说，这里最重要的并不是代码少了多少行。

股票池管理和行情获取被数据层接管了。

第三步：拿到行情后直接进入选股逻辑

例如，我需要先筛选当前涨跌幅字段存在的股票，可以直接在 DataFrame 上操作：

import os

from quantdash import QuantDash


def main():
    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

        required_columns = {"symbol", "ext.change_pct"}

        if not required_columns.issubset(df.columns):
            missing = required_columns - set(df.columns)
            print(f"缺少必要字段: {missing}")
            return

        result = df[df["ext.change_pct"] > 0.05].copy()

        print(f"筛选出 {len(result)} 只股票")
        print(
            result[
                ["symbol", "ext.change_pct"]
            ].sort_values(
                "ext.change_pct",
                ascending=False
            ).head(20).to_string(index=False)
        )

    except Exception as e:
        print(f"策略执行失败: {e}")


if __name__ == "__main__":
    main()


5000只A股实时行情怎么扫？我把量化行情扫描从15行代码压到了1行

编程语言

量化实盘全市场扫描的核心痛点在于行情数据获取的稳定性与效率。本文通过Python实战，展示如何利用QuantDash SDK将复杂的数据请求抽象为“标的池→DataFrame”模式，将全市场A股扫描简化为一行代码，实现从逐股循环到全量查询的升级，并快速进入策略筛选环节。

数据接口

数据处理

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

5000只A股实时行情怎么扫？我把量化行情扫描从15行代码压到了1行

5000只A股实时行情怎么扫？我把量化行情扫描从15行代码压到了1行

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐