### 📌 摘要

> 我以前做量化选股时，最开始把“扫描股票”和“计算策略”写在了一起，结果股票池一扩大，代码很快变得难以维护。本文以我实际做过的量化扫描任务为例，使用 Python 获取全市场行情，并通过 Pandas 构建一套可扩展的多条件股票扫描流程，同时讨论实时行情、历史 K 线和复权数据在策略中的不同用途。

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

我曾经维护过一个盘中股票扫描脚本。

最开始的目标很简单：

> 每隔一段时间，把当天涨幅比较大、成交活跃的股票找出来。

第一版程序只有一个核心循环：

```text
股票 A → 获取行情 → 判断
股票 B → 获取行情 → 判断
股票 C → 获取行情 → 判断
……
```

当股票只有几百只的时候，我觉得这种代码没有什么问题。

后来策略开始扩展，我又增加了几个条件：

- 涨跌幅；
- 振幅；
- 换手率；
- 成交量；
- 最新价格；
- 历史价格趋势。

问题从这时候开始出现。

我的代码逐渐变成了：

```text
获取股票
    ↓
计算指标
    ↓
判断条件 A
    ↓
判断条件 B
    ↓
判断条件 C
    ↓
再次请求历史数据
    ↓
判断条件 D
```

每增加一个策略条件，我都需要修改数据请求部分。

有一次我甚至发现，同一只股票在不同模块里出现了不同的数据时间点。

这件事让我重新思考：

> 一个真正可维护的股票行情扫描器，应该先把市场数据变成统一的数据集，再让策略对这个数据集做筛选。

于是我把扫描器改成了“数据层 + 因子层 + 筛选层”的结构。

## 二、问题根因分析

我后来把问题归结成三个核心矛盾。

### 1. 行情字段和策略字段不是一个概念

行情接口告诉我的是：

- 最新价；
- 昨收；
- 成交量；
- 涨跌幅；
- 振幅；
- 换手率。

而策略真正关心的是：

- 是否突破；
- 是否放量；
- 是否满足涨幅条件；
- 是否处于某个价格区间。

所以我不应该让策略直接依赖原始行情字段。

更好的方式是：

```text
原始行情
    ↓
标准化
    ↓
策略因子
    ↓
扫描条件
```

### 2. 全市场扫描和候选股票深度分析应该分开

如果我有 5500 只股票，第一阶段只需要判断：

> 哪些股票值得进一步研究？

那么我不需要对 5500 只股票全部拉取几十个月历史数据。

我的做法后来变成：

```text
5500只股票
    ↓
实时行情初筛
    ↓
100只候选
    ↓
历史K线分析
    ↓
20只重点观察
```

这种两阶段扫描方式比一开始就对整个股票池进行重计算更合理。

### 3. “涨得多”并不等于“值得进入策略池”

这是我在做扫描器过程中体会比较深的一点。

例如一只股票上涨 8%，如果成交量、换手率等条件完全不符合策略，那么它可能只是一个噪声样本。

所以扫描器不能只做单一条件排序。

它应该支持组合条件。

## 三、解决方案探索与选型

### 方案一：自己维护多个数据源

我以前确实这么做过。

一个数据源负责股票列表，一个数据源负责实时行情，另一个数据源负责历史 K 线。

这种架构初期很灵活，但最大的问题是字段标准化。

同一个“涨跌幅”，不同来源可能有不同字段名称和数据格式。

每多接一个数据源，我都需要增加一层适配代码。

### 方案二：自己计算全部行情字段

这个方案也有问题。

比如我可以通过最新价和昨收计算涨跌幅：

```text
(最新价 - 昨收) / 昨收
```

但如果数据接口已经直接提供标准化的涨跌幅，我没有必要重复计算。

我的原则后来变成：

> 数据源已经提供的标准行情字段优先使用；策略专属指标才由本地计算。

### 方案三：统一行情入口 + 本地策略计算

后来我开始使用这种结构。

我使用 QuantDash 获取标的池行情，再把数据交给 Pandas。

这样数据源负责：

> “市场现在是什么样？”

而我的代码负责：

> “按照我的策略，哪些股票值得关注？”

这两个问题分开以后，代码明显更容易维护。

## 四、具体实现：代码实战

### 1. 构建基础行情加载器

第一步，我先建立一个独立函数。

```python
import os
import sys

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()

    try:
        df = qd.quotes.get(
            universes=["CN_Stock"],
            to_dataframe=True,
        )
    except Exception as exc:
        raise RuntimeError(
            f"获取全市场行情失败: {exc}"
        ) from exc

    if df is None or df.empty:
        raise RuntimeError("行情数据为空")

    return df


if __name__ == "__main__":
    try:
        market = load_market_quotes()
        print(f"行情记录数量: {len(market)}")
        print(market.head().to_string(index=False))
    except Exception as exc:
        print(f"程序执行失败: {exc}", file=sys.stderr)
```

这个函数故意只做一件事：

> 把全市场行情拿回来。

它不负责选股。

这是我后来比较重视的一条原则。

### 2. 标准化扫描字段

接下来我定义一个行情标准化函数。

```python
import pandas as pd


def normalize_quotes(df):
    required = [
        "symbol",
        "last_price",
        "prev_close",
        "volume",
        "ext.change_pct",
        "ext.amplitude",
        "ext.turnover_rate",
    ]

    missing = [
        column for column in required
        if column not in df.columns
    ]

    if missing:
        raise ValueError(
            f"行情缺少字段: {missing}"
        )

    result = df[required].copy()

    numeric_columns = [
        "last_price",
        "prev_close",
        "volume",
        "ext.change_pct",
        "ext.amplitude",
        "ext.turnover_rate",
    ]

    for column in numeric_columns:
        result[column] = pd.to_numeric(
            result[column],
            errors="coerce",
        )

    return result.dropna(
        subset=["last_price", "prev_close"]
    )
```

我特别喜欢这种写法，因为策略代码以后不需要关心数据类型。

### 3. 创建一个多条件扫描器

例如，我现在想找：

- 涨幅大于 3%；
- 振幅大于 5%；
- 换手率大于 2%。

那么可以直接写成：

```python
def scan_candidates(df):
    result = df.copy()

    condition = (
        (result["ext.change_pct"] >= 0.03)
        & (result["ext.amplitude"] >= 0.05)
        & (result["ext.turnover_rate"] >= 0.02)
    )

    candidates = result.loc[condition].copy()

    candidates = candidates.sort_values(
        by=[
            "ext.change_pct",
            "ext.turnover_rate",
            "volume",
        ],
        ascending=[False, False, False],
    )

    return candidates
```

这时候策略条件已经完全脱离了数据获取逻辑。

如果明天我想改成：

> 跌幅超过 3%，同时振幅超过 6%。

只需要改筛选函数。

### 4. 增加价格区间条件

实际做股票扫描时，我经常会增加一些最基本的价格约束。

例如：

```python
def scan_by_price(df):
    condition = (
        (df["last_price"] >= 5)
        & (df["last_price"] <= 100)
        & (df["ext.change_pct"] >= 0.03)
    )

    return df.loc[condition].copy()
```

我这里并不是说这个条件本身有什么普适意义。

真正重要的是扫描器已经变成了一个可以自由组合条件的框架。

### 5. 完整运行版本

最后，我会把这些函数组合成一个可以直接运行的程序。

```python
import os
import sys

import pandas as pd
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()

    try:
        df = qd.quotes.get(
            universes=["CN_Stock"],
            to_dataframe=True,
        )
    except Exception as exc:
        raise RuntimeError(
            f"行情接口请求失败: {exc}"
        ) from exc

    if df is None or df.empty:
        raise RuntimeError("没有获取到全市场行情")

    return df


def normalize_quotes(df):
    required = [
        "symbol",
        "last_price",
        "prev_close",
        "volume",
        "ext.change_pct",
        "ext.amplitude",
        "ext.turnover_rate",
    ]

    missing = [
        column
        for column in required
        if column not in df.columns
    ]

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

    result = df[required].copy()

    numeric_columns = [
        "last_price",
        "prev_close",
        "volume",
        "ext.change_pct",
        "ext.amplitude",
        "ext.turnover_rate",
    ]

    for column in numeric_columns:
        result[column] = pd.to_numeric(
            result[column],
            errors="coerce",
        )

    return result.dropna(
        subset=["last_price", "prev_close"]
    )


def scan_market(df):
    condition = (
        (df["last_price"] >= 5)
        & (df["last_price"] <= 100)
        & (df["ext.change_pct"] >= 0.03)
        & (df["ext.amplitude"] >= 0.05)
        & (df["ext.turnover_rate"] >= 0.02)
    )

    candidates = df.loc[condition].copy()

    return candidates.sort_values(
        by=[
            "ext.change_pct",
            "ext.turnover_rate",
        ],
        ascending=[False, False],
    )


def main():
    try:
        raw = load_market_quotes()
        market = normalize_quotes(raw)
        candidates = scan_market(market)

        print(f"全市场有效行情: {len(market)}")
        print(f"扫描结果: {len(candidates)}")

        if candidates.empty:
            print("本轮没有符合条件的股票")
            return

        print(
            candidates.head(50).to_string(
                index=False
            )
        )

    except Exception as exc:
        print(
            f"扫描程序失败: {exc}",
            file=sys.stderr,
        )
        sys.exit(1)


if __name__ == "__main__":
    main()
```

这个版本就是我理解中的“最小可用量化股票扫描器”。

它没有复杂的机器学习模型，也没有复杂的策略框架。

但是它已经具备：

- 全市场行情读取；
- 字段检查；
- 数据类型转换；
- 多条件筛选；
- 排序；
- 异常处理。

### 6. 候选股票进入历史数据分析

当候选股票数量明显减少之后，我才会继续请求历史 K 线。

```python
import os

from quantdash import QuantDash


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

    qd = QuantDash()

    try:
        return qd.klines.batch(
            symbols,
            period="1d",
            count=120,
            adjust="forward",
            to_dataframe=True,
            show_progress=True,
        )
    except Exception as exc:
        raise RuntimeError(
            f"历史行情请求失败: {exc}"
        ) from exc


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

    try:
        histories = load_history(symbols)

        for symbol, df in histories.items():
            print(f"--- {symbol} ---")

            if df.empty:
                print("没有历史数据")
                continue

            print(
                df[
                    [
                        "trade_date",
                        "open",
                        "high",
                        "low",
                        "close",
                        "volume",
                    ]
                ].tail(10).to_string(index=False)
            )

    except Exception as exc:
        print(f"执行失败: {exc}")
```

这里我明确使用了前复权比例方式，是因为后续如果我要计算历史收益率，需要让价格序列在公司分红、送转等事件发生后保持适合收益率分析的口径。

## 五、实践效果与避坑指南

### 避坑一：不要把一个扫描器写成一个巨大的函数

我见过很多量化脚本最后变成：

```text
main()
    获取数据
    清洗数据
    计算指标
    筛选
    排序
    输出
    请求历史数据
    再筛选
```

这种代码前期写起来快，后期维护非常痛苦。

我的习惯是至少拆成：

- 数据获取；
- 数据标准化；
- 扫描条件；
- 历史数据分析。

### 避坑二：不要把一个指标当成完整策略

涨幅、振幅、换手率本身只是字段或基础特征。

真正的策略应该由多个条件共同构成。

所以我的扫描器更像一个“候选生成器”，而不是直接替代整个交易策略。

### 避坑三：回测和实盘一定要确认价格口径

如果历史数据使用了前复权，而实时行情是不复权价格，那么两边直接比较可能产生误判。

我以前就在这里踩过坑。

所以只要涉及历史收益计算，我都会明确指定 adjust 参数，并在策略文档里记录价格口径。

## 六、总结与思考

我现在看待全市场股票扫描器的方式，已经从“怎么快速找到几只股票”变成了“怎么构建一个稳定的数据筛选流水线”。

最重要的不是筛选条件本身，而是把数据访问、标准化、因子计算和策略条件拆开。

我在自己的研究中使用 QuantDash 时，会把它放在行情数据入口这一层。至于最终采用什么筛选条件，仍然由自己的研究代码决定。这样做可以避免数据工具和策略逻辑过度耦合。

如果继续扩展，我会把这个扫描器进一步拆成：

```text
行情层
  ↓
数据质量层
  ↓
基础因子层
  ↓
策略筛选层
  ↓
候选股票层
  ↓
历史验证
```

我认为这比单纯增加更多选股条件更有价值。

## Q&A

### Q1: 可以同时扫描 A 股、美股和港股吗？

A: 可以按照不同市场的标的池分别获取行情。标的代码使用统一的代码和交易所后缀格式，例如 A 股可以使用 .SH、.SZ，美股使用 .US，港股使用 .HK。实际策略中，我会根据不同市场分别处理交易时间和数据特征。

### Q2: 为什么不直接给所有股票计算 120 日均线？

A: 技术上可以，但工程上没有必要。

如果第一阶段只是为了找候选股票，我通常先用实时行情做初筛，再对较小的候选集合获取历史数据。这样能够减少不必要的数据处理。

我以前做量化选股时，最开始把“扫描股票”和“计算策略”写在了一起，结果股票池一扩大，代码很快变得难以维护。本文以我实际做过的量化扫描任务为例，使用 Python 获取全市场行情，并通过 Pandas 构建一套可扩展的多条件股票扫描流程，同时讨论实时行情、历史 K 线和复权数据在策略中的不同用途。

如何构建一个全市场股票行情扫描器：从实时行情到量化选股

人工智能

后端

本文分享Python量化选股实战：构建可扩展的全市场股票扫描器，用Pandas统一处理行情数据，分离数据层、因子层与筛选层，实现多条件组合筛选并优化历史K线分析流程，避免代码混乱。

机器学习

数据分析

数据清洗

数据接口

数据处理

代码解释

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

如何构建一个全市场股票行情扫描器：从实时行情到量化选股-腾讯云开发者社区-腾讯云

如何构建一个全市场股票行情扫描器：从实时行情到量化选股

如何构建一个全市场股票行情扫描器：从实时行情到量化选股

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐