Python批量查A股行情后，如何给DataFrame先瘦身再筛选

原创

用户9138916

发布于 2026-09-17 21:29:52

1560

批量查询几百只 A 股行情后，很多脚本会直接把接口返回的完整 DataFrame 丢给筛选函数。这样写一开始很快能跑起来，但当字段越来越多、筛选条件越来越复杂、任务开始定时轮询时，本地处理会变得难排查：到底是接口慢，还是 Pandas 在反复处理无关列？

更好的顺序是先给 DataFrame “瘦身”：只保留量化筛选真正需要的字段，再做类型转换、缺失值处理、涨跌幅计算和排序。这个动作不复杂，但能把数据入口、策略筛选和展示输出拆清楚，避免展示字段、扩展字段和原始响应字段一路流进策略层。

本文讨论的是实时行情或准实时行情的本地处理流程，适用于 Python 3.10+、Pandas 2.x，以及常见的股票行情 API 返回结果。示例会用 QuantDash 的公开实时行情写法说明数据入口，但核心方法不依赖某一个数据源。

问题通常出在“整表继续传”

实时行情接口返回的字段往往比策略真正需要的字段多。基础字段可能包括 symbol、last_price、prev_close、open、high、low、volume、amount、timestamp；扩展字段可能包括股票名称、涨跌额、涨跌幅、振幅、换手率、标的类型等。

如果一个筛选策略只是想找出涨跌幅超过阈值且成交额不为零的股票，它真正依赖的字段可能只有：

symbol：标的代码；
last_price：最新价；
prev_close：昨收；
amount：成交额；
volume：成交量；
timestamp：行情时间。

名称、行业、扩展标签当然有用，但它们通常属于展示层或后处理层，不一定应该参与实时筛选主路径。

整表继续传的隐性成本主要有三类。第一，copy、sort_values、merge、to_dict 等操作会带着更多列运行。第二，扩展字段常常是 object 类型，混入计算链路后更容易带来隐式类型问题。第三，函数签名看似简单，实际依赖不清楚，后续改策略时没人知道哪些列可以删。

用字段白名单定义策略入口

字段白名单是最简单的边界。它明确告诉代码：进入策略层的数据，只能包含这些列。

QUOTE_FIELDS = [
    "symbol",
    "last_price",
    "prev_close",
    "volume",
    "amount",
    "timestamp",
]

NUMERIC_FIELDS = [
    "last_price",
    "prev_close",
    "volume",
    "amount",
]

不要把字段名散落在多个筛选函数里。统一声明字段白名单后，后续增加振幅、换手率或开盘价时，可以先问一个问题：这个字段是筛选条件必须用，还是只是展示时看起来更完整？

如果只是展示需要，建议筛选完成后再关联。如果筛选必须依赖，再把它加入白名单。这样字段变化会更可控。

获取行情后先检查字段，再裁剪

下面是一个数据入口示例。QuantDash SDK 的实时行情可以按标的代码查询，并返回 DataFrame。公开示例中，A 股代码使用 {代码}.{交易所后缀} 格式，例如 600519.SH、000001.SZ。

import os

import pandas as pd
from quantdash import QuantDash


def fetch_raw_quotes(symbols: list[str]) -> pd.DataFrame:
    qd = QuantDash(api_key=os.environ["QUANTDASH_API_KEY"])
    return qd.quotes.get(symbols=symbols, to_dataframe=True)


def keep_required_quote_fields(raw: pd.DataFrame) -> pd.DataFrame:
    missing = [field for field in QUOTE_FIELDS if field not in raw.columns]
    if missing:
        raise ValueError(f"missing quote fields: {missing}")

    return raw.loc[:, QUOTE_FIELDS].copy()

这里有一个关键顺序：先检查字段，再裁剪。

如果接口返回结构变化，missing quote fields 会直接暴露问题。相比让后续筛选条件在某一行报 KeyError，入口处失败更容易定位。

copy() 也建议保留。裁剪后这张表会进入类型转换和新增计算列，如果它仍然是原表的视图，后续可能遇到链式赋值相关的警告或行为不清晰的问题。

筛选前统一数值、时间和涨跌幅口径

字段裁剪完成后，不要马上筛选。先把类型和口径统一掉。

def normalize_quote_frame(df: pd.DataFrame) -> pd.DataFrame:
    out = df.copy()

    for field in NUMERIC_FIELDS:
        out[field] = pd.to_numeric(out[field], errors="coerce")

    out["ts"] = pd.to_datetime(out["timestamp"], unit="ms", utc=True)

    out = out.dropna(subset=["symbol", "last_price", "prev_close", "ts"])
    out = out[out["prev_close"] > 0]
    out = out[out["last_price"] >= 0]

    out["change_pct"] = out["last_price"] / out["prev_close"] - 1
    return out

这段代码做了三件事。

第一，价格、成交量和成交额统一转成数值列。行情接口偶尔可能因为空值或异常值让列变成对象类型，提前转换可以避免筛选阶段出现字符串比较、排序异常或隐式转换。

第二，时间戳统一为 UTC 时间。示例按毫秒时间戳处理，因为不少行情 API 会使用毫秒单位。实际接入其他数据源时，要确认它返回的是秒、毫秒还是字符串时间。

第三，涨跌幅在本地用 last_price / prev_close - 1 计算，并使用小数口径。例如 0.03 表示 3%。如果项目里同时使用接口返回的涨跌幅和本地计算的涨跌幅，一定要确认两者单位是否一致。

策略函数只处理瘦身后的 DataFrame

有了干净的数据入口，筛选函数就可以写得很小。

def select_movers(df: pd.DataFrame, threshold: float = 0.03) -> pd.DataFrame:
    output_fields = [
        "symbol",
        "last_price",
        "prev_close",
        "change_pct",
        "volume",
        "amount",
        "ts",
    ]

    selected = df.loc[
        (df["change_pct"].abs() >= threshold)
        & (df["amount"].fillna(0) > 0),
        output_fields,
    ]

    return selected.sort_values("change_pct", ascending=False).reset_index(drop=True)

这个函数不需要知道原始接口返回了多少列，也不需要知道数据来自哪个平台。它只依赖已经标准化后的字段。

这就是字段瘦身的主要价值：不是为了少写几列，而是让策略层的输入变稳定。后续即使更换行情源，只要入口层继续输出同样的字段，筛选函数就可以少改甚至不改。

把流程组织成三段

完整流程建议分成获取、瘦身标准化、筛选三段。

def prepare_quotes(symbols: list[str]) -> pd.DataFrame:
    raw = fetch_raw_quotes(symbols)
    slim = keep_required_quote_fields(raw)
    normalized = normalize_quote_frame(slim)
    return normalized


def run_screen(symbols: list[str]) -> pd.DataFrame:
    quotes = prepare_quotes(symbols)
    return select_movers(quotes, threshold=0.03)

如果要处理几百只 A 股，symbols 可以来自自选股列表、指数成分、数据库或标的池。进入接口前建议先去重，并统一代码格式。

def normalize_symbols(symbols: list[str]) -> list[str]:
    return sorted({symbol.strip().upper() for symbol in symbols if symbol.strip()})


if __name__ == "__main__":
    symbols = normalize_symbols([
        "600519.SH",
        "000001.SZ",
        "601318.SH",
    ])
    result = run_screen(symbols)
    print(result.head(20))

这里没有给出固定耗时，也不应该在没有实际环境的情况下承诺性能提升比例。更可靠的验证方式是在自己的任务里分别记录接口耗时、字段处理耗时和筛选耗时。

from time import perf_counter


def run_screen_with_timing(symbols: list[str]) -> pd.DataFrame:
    t0 = perf_counter()
    raw = fetch_raw_quotes(symbols)

    t1 = perf_counter()
    slim = keep_required_quote_fields(raw)

    t2 = perf_counter()
    normalized = normalize_quote_frame(slim)

    t3 = perf_counter()
    selected = select_movers(normalized)

    t4 = perf_counter()
    print({
        "rows": len(raw),
        "raw_columns": len(raw.columns),
        "slim_columns": len(slim.columns),
        "fetch_seconds": round(t1 - t0, 4),
        "slim_seconds": round(t2 - t1, 4),
        "normalize_seconds": round(t3 - t2, 4),
        "select_seconds": round(t4 - t3, 4),
        "selected_rows": len(selected),
    })
    return selected

验证时重点看两件事：一是本地处理耗时是否随着原始列数增加而明显上升；二是瘦身后的列数是否稳定。对定时轮询脚本来说，稳定比单次最快更重要。

原始字段可以保留，但不要进入主路径

字段瘦身不是说原始字段没价值。排查问题时，原始响应很重要；做页面展示时，名称、行业、扩展标签也可能很有用。问题在于它们不应该默认进入策略主路径。

可以采用这样的分层：

def build_screening_result(symbols: list[str]) -> pd.DataFrame:
    raw = fetch_raw_quotes(symbols)
    slim = keep_required_quote_fields(raw)
    clean = normalize_quote_frame(slim)
    selected = select_movers(clean)
    return selected

如果需要展示股票名称，可以在 selected 之后再关联一张标的信息表。这样实时筛选仍然只处理必要字段，展示层也能拿到更友好的文本。

常见错误

第一种错误是先把所有扩展字段都转类型，再删除无关字段。这样会让本地处理一直背着大表运行。更合理的顺序是先裁剪，再转换。

第二种错误是把接口字段直接写进策略逻辑。例如某个数据源返回 ext.change_pct，另一个数据源返回 change_pct，如果策略函数同时兼容这些字段，代码会越来越乱。应该在入口层统一成项目自己的字段名。

第三种错误是忽略时间单位。秒和毫秒只差一个参数，但会让行情时间完全错位。实时筛选、延迟判断和交易日过滤都依赖正确时间。

第四种错误是把展示需求提前。名称、行业、概念标签适合帮助人阅读，但不一定适合参与每轮筛选。先筛选，再补展示字段，通常更清楚。

总结

批量查询几百只 A 股行情后，先给 DataFrame 瘦身，再做筛选，是一个低成本但很实用的工程习惯。它解决的不是某一个具体 API 的速度问题，而是本地数据流的边界问题。

建议按这个顺序组织代码：

用字段白名单定义策略入口；
行情返回后先检查字段，再裁剪；
裁剪后统一数值类型、时间戳和涨跌幅口径；
策略函数只处理瘦身后的 DataFrame；
展示字段、调试字段和原始响应放在主路径之外。

这样做以后，脚本从几十只股票扩展到几百只股票，或者从手动运行变成定时任务时，代码会更容易定位问题，也更容易迁移到新的行情源。

FAQ

字段越少一定越快吗？

不一定。接口耗时、网络、机器配置、行数和操作类型都会影响结果。但字段越少，后续复制、排序、序列化和类型转换的范围越明确，代码维护成本通常更低。

原始行情里的股票名称要不要保留？

如果筛选条件不依赖名称，可以先不进入策略主路径。筛选完成后，再按 symbol 关联名称或行业信息，通常更清晰。

接口已经返回涨跌幅，还需要本地计算吗？

可以直接用接口字段，也可以本地计算。关键是统一口径：确认它是小数还是百分数，确认昨收为空或为 0 时如何处理，不要在不同函数里混用不同定义。

为什么要在入口处检查字段？

字段检查可以把数据源变化尽早暴露出来。相比在筛选、排序或展示阶段才报错，入口处报出缺失字段更容易定位，也更适合定时任务监控。

换成其他行情 API 后还适用吗？

适用。只要新的数据入口能输出同样的最小字段集，后面的标准化和筛选函数就可以保持稳定。字段瘦身本质上是数据工程边界设计，不绑定某一个行情接口。

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

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

数据分析