实时行情进 DataFrame 后，如何筛出量化候选股

原创

用户9138916

发布于 2026-09-23 06:45:00

2140

实时行情进入 Pandas DataFrame 以后，最容易写错的不是筛选条件本身，而是字段口径、空值、单位和排序。比如 change_pct 是 0.03 还是 3？volume 能不能直接跨市场比较？timestamp 是否需要先去重？这些问题如果不先处理，候选池看起来有结果，实际可能混入停牌、异常价、字段缺失或重复标的。

本文解决一个具体问题：拿到一批实时行情 DataFrame 后，如何只用已经返回的字段，如涨跌幅、成交量、成交额、昨收、最新价等，构建一个可复用的量化候选筛选函数。这里不讨论预测收益，也不把筛选结果当作投资建议；它只是回测、盘中监控或人工复核前的一层数据过滤。

示例数据源使用 QuantDash Python SDK 的实时行情接口。它在本文中只承担“把实时行情转换为 DataFrame”的角色；筛选逻辑完全基于 Pandas，可替换为其他返回同类字段的数据源。

环境与字段前提

建议在文章发布或项目落地时锁定关键依赖版本，避免未来 SDK 或字段格式变化导致示例含义不一致：

Python：3.10 或更高版本；
QuantDash SDK：quantdash==0.1.0；
Pandas：建议 2.x；
API Key：通过环境变量 QUANTDASH_API_KEY 提供，避免写进代码。

安装依赖可使用 pip install quantdash==0.1.0 pandas。如果在 Windows PowerShell 中设置环境变量，可使用 $env:QUANTDASH_API_KEY = 'YOUR_API_KEY'；macOS/Linux 可使用 export QUANTDASH_API_KEY='YOUR_API_KEY'。

QuantDash 实时行情接口按标的查询时的典型写法如下：

from quantdash import QuantDash

qd = QuantDash()

quotes = qd.quotes.get(
    symbols=['600519.SH', '000001.SZ', 'AAPL.US'],
    to_dataframe=True,
)

print(quotes.columns.tolist())
print(quotes.head())

需要注意两点。

第一，实时行情基础字段通常包括 symbol、last_price、prev_close、volume 等；OpenAPI 中还列出 region、open、high、low、amount、timestamp 等字段。扩展字段可能以 ext.name、ext.change_pct、ext.turnover_rate 这类展开列名出现。实际使用前应打印列名确认，不要假定所有市场、所有标的都具备完全相同的扩展字段。

第二，ext.change_pct 在官方口径中按小数表示，例如 0.01 表示 1%。筛选时如果写成 change_pct > 3，含义就会变成上涨超过 300%，这通常不是想要的条件。

不要一上来就筛选：先做 DataFrame 规范化

实时行情 DataFrame 的主要风险包括：重复标的、数值列被解析成字符串、昨收为 0、最新价缺失、字段名在不同数据源之间不一致。下面这个函数把这些问题集中处理。

import pandas as pd


def normalize_quote_df(df: pd.DataFrame) -> pd.DataFrame:
    '''
    对实时行情 DataFrame 做最小规范化：
    1. 检查必要字段；
    2. 去重；
    3. 将核心数值列转为 numeric；
    4. 统一生成 change_pct_calc 与 change_pct_used。
    '''
    if df is None or df.empty:
        return pd.DataFrame()

    result = df.copy()

    required = ['symbol', 'last_price', 'prev_close', 'volume']
    missing = [col for col in required if col not in result.columns]
    if missing:
        raise ValueError(f'实时行情缺少必要字段: {missing}')

    result = result.drop_duplicates(subset=['symbol'], keep='last')

    numeric_cols = [
        'last_price', 'prev_close', 'open', 'high', 'low',
        'volume', 'amount', 'ext.change_pct', 'ext.turnover_rate',
        'ext.amplitude',
    ]
    for col in numeric_cols:
        if col in result.columns:
            result[col] = pd.to_numeric(result[col], errors='coerce')

    valid_prev_close = result['prev_close'] > 0
    result['change_pct_calc'] = pd.NA
    result.loc[valid_prev_close, 'change_pct_calc'] = (
        result.loc[valid_prev_close, 'last_price']
        / result.loc[valid_prev_close, 'prev_close']
        - 1
    )
    result['change_pct_calc'] = pd.to_numeric(
        result['change_pct_calc'], errors='coerce'
    )

    if 'ext.change_pct' in result.columns:
        result['change_pct_used'] = result['ext.change_pct'].fillna(
            result['change_pct_calc']
        )
    else:
        result['change_pct_used'] = result['change_pct_calc']

    return result

这里的独立判断是：即使数据源已经返回了涨跌幅，也建议同时用 last_price / prev_close - 1 计算一个 change_pct_calc。它不是为了替代官方字段，而是用于发现明显异常，例如 ext.change_pct 与本地计算值差异过大、昨收缺失、停牌或复权口径混入实时行情等问题。

可以增加一个轻量校验函数，避免错误字段进入候选筛选：

def validate_quote_df(df: pd.DataFrame) -> None:
    if df.empty:
        raise ValueError('实时行情为空，无法筛选候选池')

    base_required = ['symbol', 'last_price', 'prev_close', 'volume', 'change_pct_used']
    missing = [col for col in base_required if col not in df.columns]
    if missing:
        raise ValueError(f'规范化后仍缺少字段: {missing}')

    duplicated = df['symbol'].duplicated().sum()
    if duplicated:
        raise ValueError(f'symbol 存在重复记录: {duplicated}')

    bad_price = df['last_price'].isna() | df['prev_close'].isna() | (df['prev_close'] <= 0)
    if bad_price.all():
        raise ValueError('所有标的价格字段都不可用，请检查数据源或市场状态')

一个可复用的候选筛选规则

候选筛选不应该追求复杂，而应该追求口径稳定。下面的规则只使用实时行情中常见的已返回字段：

条件	目的	字段
最新价和昨收有效	排除停牌、异常价或缺字段	`last_price`、`prev_close`
涨跌幅位于指定区间	过滤过弱或过热标的	`change_pct_used`
成交量达到下限	排除流动性太弱标的	`volume`
成交额达到下限，若字段存在	更接近资金活跃度	`amount`
最新价不低于开盘价，若字段存在	保留日内相对强势	`last_price`、`open`
换手率位于区间，若字段存在	排除过冷或极端异常	`ext.turnover_rate`

对应代码如下：

def select_candidates(
    quotes: pd.DataFrame,
    min_change_pct: float = 0.01,
    max_change_pct: float = 0.07,
    min_volume: float = 100_000,
    min_amount: float | None = 50_000_000,
    min_turnover_rate: float | None = None,
    max_turnover_rate: float | None = None,
    top_n: int = 20,
) -> pd.DataFrame:
    '''
    用实时行情字段筛选候选标的。

    change_pct 参数使用小数口径：
    0.01 表示 1%，0.07 表示 7%。
    '''
    df = normalize_quote_df(quotes)
    validate_quote_df(df)

    mask = (
        df['last_price'].notna()
        & df['prev_close'].notna()
        & (df['prev_close'] > 0)
        & df['volume'].notna()
        & (df['volume'] >= min_volume)
        & df['change_pct_used'].notna()
        & (df['change_pct_used'] >= min_change_pct)
        & (df['change_pct_used'] <= max_change_pct)
    )

    if 'amount' in df.columns and min_amount is not None:
        mask &= df['amount'].notna() & (df['amount'] >= min_amount)

    if 'open' in df.columns:
        mask &= df['open'].notna() & (df['last_price'] >= df['open'])

    if 'ext.turnover_rate' in df.columns and min_turnover_rate is not None:
        mask &= df['ext.turnover_rate'].notna()
        mask &= df['ext.turnover_rate'] >= min_turnover_rate

    if 'ext.turnover_rate' in df.columns and max_turnover_rate is not None:
        mask &= df['ext.turnover_rate'].notna()
        mask &= df['ext.turnover_rate'] <= max_turnover_rate

    candidates = df.loc[mask].copy()
    if candidates.empty:
        return candidates

    candidates['volume_rank_pct'] = candidates['volume'].rank(pct=True)
    candidates['change_rank_pct'] = candidates['change_pct_used'].rank(pct=True)

    if 'amount' in candidates.columns:
        candidates['amount_rank_pct'] = candidates['amount'].rank(pct=True)
    else:
        candidates['amount_rank_pct'] = 0.0

    candidates['score'] = (
        0.45 * candidates['change_rank_pct']
        + 0.35 * candidates['volume_rank_pct']
        + 0.20 * candidates['amount_rank_pct']
    )

    display_cols = [
        'symbol', 'last_price', 'prev_close', 'change_pct_used',
        'volume', 'score',
    ]
    optional_cols = ['amount', 'open', 'high', 'low', 'timestamp', 'ext.name']
    display_cols.extend([col for col in optional_cols if col in candidates.columns])

    return (
        candidates.sort_values(
            by=['score', 'change_pct_used', 'volume'],
            ascending=[False, False, False],
        )
        .loc[:, display_cols]
        .head(top_n)
        .reset_index(drop=True)
    )

这段代码的关键点不是权重本身，而是把“过滤”和“排序”分开：

过滤条件负责排除明显不适合进入候选池的标的；
排序分数只负责在候选池内部给出优先级；
score 不是收益预测值，也不代表胜率；
权重应根据策略目标调整，例如偏短线监控可提高成交额权重，偏趋势观察可提高涨跌幅稳定区间的权重。

从拉取行情到输出候选池

下面给出一个最小串联示例。由于不同账号权限、市场状态和标的范围会影响返回结果，这里不展示伪造的运行输出；验证时应重点检查返回列名、候选池是否为空、涨跌幅单位是否符合预期。

from quantdash import QuantDash


def fetch_realtime_quotes(symbols: list[str]) -> pd.DataFrame:
    qd = QuantDash()
    return qd.quotes.get(
        symbols=symbols,
        to_dataframe=True,
    )


if __name__ == '__main__':
    symbols = [
        '600519.SH',
        '000001.SZ',
        '00700.HK',
        'AAPL.US',
    ]

    raw_quotes = fetch_realtime_quotes(symbols)
    print(raw_quotes.columns.tolist())

    candidates = select_candidates(
        raw_quotes,
        min_change_pct=0.01,
        max_change_pct=0.07,
        min_volume=100_000,
        min_amount=50_000_000,
        top_n=10,
    )

    print(candidates)

如果希望按标的池查询，QuantDash 实时行情接口也支持 universes=['CN_Stock'] 这类写法。实际工程中应先确认账号权限、市场范围和返回规模，再决定是否全市场拉取；不要在未确认限流和批量规模前，把全市场刷新写进高频循环。

常见坑：候选筛选看起来正确，实际口径错了

1. 把 0.03 当成 0.03%

在小数口径下，0.03 表示 3%。因此常见条件应写成 change_pct_used >= 0.03，而不是 >= 3。如果来自不同数据源，应先统一为小数口径，再复用筛选函数。

2. 只看成交量，不看成交额

成交量的单位和股票价格有关。低价股和高价股的成交量不可直接比较。若返回字段中存在 amount，建议同时设置成交额下限；如果没有 amount，至少不要把成交量排序解释成资金活跃度排序。

3. 不检查重复 symbol

实时任务中常见的问题是同一个标的被多个分组重复拉取，合并后出现重复行。drop_duplicates(subset=['symbol'], keep='last') 可以作为最低限度的保护，但更好的做法是在任务入口就去重。

4. 混用 A 股、港股、美股后直接比较阈值

不同市场的交易制度、价格单位、成交习惯和交易时段不同。本文示例支持混合标的输入，但实际策略筛选时更建议按市场拆分阈值，例如分别维护 A 股、港股、美股的 min_volume、min_amount 和涨跌幅区间。

如何验证这套筛选逻辑

没有必要先做复杂回测，可以从三个层面验证：

字段验证：打印 raw_quotes.columns.tolist()，确认 symbol、last_price、prev_close、volume 存在，并确认是否有 amount、open、ext.change_pct。
单位验证：任选几只标的，手工计算 last_price / prev_close - 1，与 change_pct_used 比较，确认差异在可解释范围内。
边界验证：把 min_change_pct 调高或调低，观察候选数量是否随阈值变化。如果阈值变化但结果完全不变，通常说明字段口径或条件没有生效。

如果要放进定时任务，还应额外记录每次刷新时间、输入标的数量、返回行数、候选数量和异常原因。日志不需要很复杂，但必须能回答“为什么这次没有候选”这个问题。

小结

实时行情进入 DataFrame 后，候选筛选的核心不是写一个很炫的打分公式，而是把字段口径统一、把空值和异常值挡在外面，再用少量稳定条件构建可解释的候选池。

一个较稳妥的实现顺序是：先获取实时行情，再规范化 DataFrame，然后校验必要字段，最后用涨跌幅、成交量、成交额、开盘价和换手率等已返回字段做过滤与排序。这样得到的候选列表可以用于盘中看板、策略观察或后续研究，但不应被直接解释为买卖信号。

FAQ

Q1：没有 `ext.change_pct` 字段还能筛选吗？

可以。只要有 last_price 和 prev_close，就能用 last_price / prev_close - 1 计算涨跌幅。本文代码会优先使用已返回的 ext.change_pct，缺失时回退到本地计算值。

Q2：成交量阈值应该怎么设？

不要使用一个永久固定的阈值覆盖所有市场。更合理的做法是按市场、股票价格区间或策略容量拆分阈值。若有 amount 字段，成交额通常比单纯成交量更适合表达资金活跃度。

Q3：候选池为空是不是接口有问题？

不一定。可能是阈值太严格、当前非活跃交易时段、输入标的太少，或者 amount、open 等可选字段缺失导致条件被过滤。可以先降低 min_change_pct、关闭 min_amount，再检查字段和数据状态。

Q4：这套规则能直接用于实盘交易吗？

不能直接等同于交易策略。它只完成实时行情层面的候选过滤，没有包含风险控制、交易成本、滑点、仓位管理、成交约束和回测验证。进入实盘前，应在完整策略框架中重新验证。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python