怎么实现 A 股全市场扫描选股，筛出值得关注的标的

选股是量化里最高频的需求。

大多数人的做法是打开行情软件，手动翻看自选股、刷涨幅榜，或者跟着论坛帖子追热点。这种方式在股票数量少的时候还行，一旦想覆盖全市场——沪深京超过 5000 只股票——就完全靠不住了。

程序化选股的思路很直接：把全市场数据拉下来，按你定义的规则过滤，剩下的就是候选列表。这件事听起来需要很大的数据工程量，但如果数据接口支持"全市场池查询"，整个流程可以压缩到十几行 Python。

这篇文章用 AlphaFeed 的 Universe 接口做一个完整的全市场扫描选股系统，覆盖几种常见的筛选逻辑。

1. 全市场行情快照：一行代码拿到所有 A 股

AlphaFeed 的 af.quotes.get(universes="CN_Stock") 可以一次性获取全部 A 股的实时行情快照：

from alphafeed import AlphaFeed

af = AlphaFeed()

all_cn = af.quotes.get(universes="CN_Stock", to_dataframe=True)
print(f"A 股标的总数: {len(all_cn)}")
print(all_cn[["symbol", "last_price", "prev_close", "volume", "amount"]].head(10))

返回的每一行包含当前价格、昨收、成交量、成交额等字段，相当于你自己做了一个全市场实时扫描器的数据底座。

如果你还想看 ETF，可以加一个池：

all_etf = af.quotes.get(universes="CN_ETF", to_dataframe=True)
print(f"ETF 标的总数: {len(all_etf)}")

注意：Universe 查询需要 Starter 及以上的订阅权限。免费版可以先用 symbols 参数指定具体股票来练手。

2. 第一个筛选器：涨幅 + 成交额

最基础的选股条件：今天涨了，而且成交额达标（排除冷门票和异常波动）。

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

df = af.quotes.get(universes="CN_Stock", to_dataframe=True)

df["change_pct"] = (df["last_price"] - df["prev_close"]) / df["prev_close"]

selected = df[
    (df["change_pct"] > 0.02)
    & (df["change_pct"] < 0.098)
    & (df["amount"] > 5e8)
].copy()

selected = selected.sort_values("change_pct", ascending=False)

print(f"符合条件的标的: {len(selected)} 只")
print(selected[["symbol", "last_price", "change_pct", "amount"]].head(20))

这里 change_pct < 0.098 是为了排除涨停板——涨停意味着你大概率买不进去，放在选股结果里没有实操意义。

amount > 5e8 是成交额超过 5 亿元，保证流动性。这个阈值可以根据你的资金量调整：

资金量级	建议成交额下限
10 万以下	1 亿
10–100 万	3 亿
100 万以上	5 亿+

3. 量价异动选股：放量突破

单纯看涨幅意义有限——涨 3% 但缩量可能是虚涨，涨 3% 且放量才可能是资金在推。

问题在于，判断"放量"需要历史成交量来做基准。我们可以用最近 20 天的日均成交量作为参考：

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

df_today = af.quotes.get(universes="CN_Stock", to_dataframe=True)
df_today["change_pct"] = (df_today["last_price"] - df_today["prev_close"]) / df_today["prev_close"]

candidates = df_today[
    (df_today["change_pct"] > 0.01)
    & (df_today["amount"] > 3e8)
].copy()

symbols = candidates["symbol"].tolist()

if len(symbols) > 200:
    symbols = symbols[:200]

klines = af.klines.batch(
    symbols,
    period="1d",
    count=20,
    adjust="forward",
    to_dataframe=True,
    show_progress=True,
)

vol_ratio_list = []
for sym, kdf in klines.items():
    if len(kdf) < 10:
        continue
    avg_vol = kdf["volume"].mean()
    today_vol = candidates.loc[candidates["symbol"] == sym, "volume"].values
    if len(today_vol) == 0 or avg_vol == 0:
        continue
    ratio = today_vol[0] / avg_vol
    vol_ratio_list.append({
        "symbol": sym,
        "vol_ratio": round(ratio, 2),
        "change_pct": candidates.loc[candidates["symbol"] == sym, "change_pct"].values[0],
        "amount": candidates.loc[candidates["symbol"] == sym, "amount"].values[0],
    })

vol_df = pd.DataFrame(vol_ratio_list)
vol_df = vol_df[vol_df["vol_ratio"] > 2.0].sort_values("vol_ratio", ascending=False)

print(f"放量突破标的: {len(vol_df)} 只")
print(vol_df.head(20))

vol_ratio > 2.0 表示今日成交量是近 20 日均值的 2 倍以上。这个倍数越大，信号越强烈，但也要注意区分"放量上涨"和"放量出货"。

4. 动量选股：最近 N 天涨幅排名

另一种经典选股思路是动量因子——过去一段时间涨得好的股票，短期内可能继续涨（当然也可能反转，这就是为什么你需要回测）。

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

df_today = af.quotes.get(universes="CN_Stock", to_dataframe=True)

liquid = df_today[df_today["amount"] > 3e8].copy()
symbols = liquid["symbol"].tolist()

if len(symbols) > 500:
    symbols = symbols[:500]

klines = af.klines.batch(
    symbols,
    period="1d",
    count=20,
    adjust="forward",
    to_dataframe=True,
    show_progress=True,
)

momentum_list = []
for sym, kdf in klines.items():
    kdf = kdf.sort_values("trade_date").reset_index(drop=True)
    if len(kdf) < 15:
        continue
    ret_20d = kdf["close"].iloc[-1] / kdf["close"].iloc[0] - 1
    momentum_list.append({
        "symbol": sym,
        "name": kdf["name"].iloc[0] if "name" in kdf.columns else sym,
        "momentum_20d": round(ret_20d, 4),
        "close": kdf["close"].iloc[-1],
    })

mom_df = pd.DataFrame(momentum_list)
mom_df = mom_df.sort_values("momentum_20d", ascending=False)

print("=== 20 日动量 Top 20 ===")
print(mom_df.head(20).to_string(index=False))

print("\n=== 20 日动量 Bottom 10（反转候选？）===")
print(mom_df.tail(10).to_string(index=False))

动量选股经常出现在因子投资研究中。学术上一般用过去 12 个月收益（扣掉最近 1 个月）做截面排序。上面用 20 天只是一个短周期版本，适合快速筛选近期趋势股。

5. 技术面选股：均线多头排列

均线多头排列是很多技术面选股系统的基本条件之一：5 日均线 > 10 日均线 > 20 日均线 > 60 日均线，说明各周期趋势一致向上。

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

df_today = af.quotes.get(universes="CN_Stock", to_dataframe=True)
liquid = df_today[df_today["amount"] > 3e8].copy()
symbols = liquid["symbol"].tolist()[:500]

klines = af.klines.batch(
    symbols,
    period="1d",
    count=80,
    adjust="forward",
    to_dataframe=True,
    show_progress=True,
)

bullish_list = []
for sym, kdf in klines.items():
    kdf = kdf.sort_values("trade_date").reset_index(drop=True)
    if len(kdf) < 60:
        continue

    ma5 = kdf["close"].rolling(5).mean().iloc[-1]
    ma10 = kdf["close"].rolling(10).mean().iloc[-1]
    ma20 = kdf["close"].rolling(20).mean().iloc[-1]
    ma60 = kdf["close"].rolling(60).mean().iloc[-1]

    if ma5 > ma10 > ma20 > ma60:
        bullish_list.append({
            "symbol": sym,
            "name": kdf["name"].iloc[0] if "name" in kdf.columns else sym,
            "close": kdf["close"].iloc[-1],
            "ma5": round(ma5, 2),
            "ma10": round(ma10, 2),
            "ma20": round(ma20, 2),
            "ma60": round(ma60, 2),
        })

bull_df = pd.DataFrame(bullish_list)
print(f"均线多头排列标的: {len(bull_df)} 只")
print(bull_df.head(20).to_string(index=False))

这种筛选的作用不是直接告诉你"该买什么"，而是缩小研究范围。从 5000 只缩到 50 只，然后你再对这 50 只做进一步分析（看基本面、行业逻辑、估值等）。

6. 组合条件选股：把多个信号叠加

实战中很少用单一条件选股。更常见的做法是把几个条件叠起来——涨幅适中、放量、动量为正、均线方向一致：

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

df_today = af.quotes.get(universes="CN_Stock", to_dataframe=True)
df_today["change_pct"] = (df_today["last_price"] - df_today["prev_close"]) / df_today["prev_close"]

liquid = df_today[
    (df_today["amount"] > 5e8)
    & (df_today["change_pct"] > 0)
    & (df_today["change_pct"] < 0.098)
].copy()

symbols = liquid["symbol"].tolist()[:300]

klines = af.klines.batch(
    symbols,
    period="1d",
    count=80,
    adjust="forward",
    to_dataframe=True,
    show_progress=True,
)

results = []
for sym, kdf in klines.items():
    kdf = kdf.sort_values("trade_date").reset_index(drop=True)
    if len(kdf) < 60:
        continue

    ma5 = kdf["close"].rolling(5).mean().iloc[-1]
    ma20 = kdf["close"].rolling(20).mean().iloc[-1]
    ma60 = kdf["close"].rolling(60).mean().iloc[-1]
    is_bullish = ma5 > ma20 > ma60

    avg_vol = kdf["volume"].iloc[-20:].mean()
    today_row = liquid[liquid["symbol"] == sym]
    if today_row.empty or avg_vol == 0:
        continue
    vol_ratio = today_row["volume"].values[0] / avg_vol

    ret_20d = kdf["close"].iloc[-1] / kdf["close"].iloc[-20] - 1

    if is_bullish and vol_ratio > 1.5 and ret_20d > 0.05:
        results.append({
            "symbol": sym,
            "name": kdf["name"].iloc[0] if "name" in kdf.columns else sym,
            "close": kdf["close"].iloc[-1],
            "change_pct": round(today_row["change_pct"].values[0], 4),
            "vol_ratio": round(vol_ratio, 2),
            "momentum_20d": round(ret_20d, 4),
        })

final = pd.DataFrame(results).sort_values("momentum_20d", ascending=False)
print(f"组合条件选出: {len(final)} 只")
print(final.to_string(index=False))

这段代码的逻辑是：

1. 流动性门槛：成交额 > 5 亿
2. 今日上涨但未涨停：排除无法买入和无涨幅的标的
3. 均线多头：MA5 > MA20 > MA60
4. 放量：量比 > 1.5
5. 20 日动量为正：涨幅 > 5%

你可以根据自己的交易风格调整任何一个参数。

7. 把选股结果保存下来

选股不是看一眼就完了。你需要把每天的结果存下来，才能回头验证"上周选出来的票，后来表现怎么样"。

import json
from datetime import datetime

output = {
    "scan_time": datetime.now().isoformat(),
    "filter_desc": "bullish_ma + vol_ratio>1.5 + momentum_20d>5%",
    "count": len(final),
    "symbols": final.to_dict(orient="records"),
}

filename = f"scan_{datetime.now().strftime('%Y%m%d_%H%M')}.json"
with open(filename, "w", encoding="utf-8") as f:
    json.dump(output, f, ensure_ascii=False, indent=2)

print(f"已保存到 {filename}")

长期积累下来，你会有一个选股信号和后续表现的对照数据库。这是做因子研究和信号评价的基础素材。

8. 定时运行：让选股每天自动跑

如果你不想每天手动执行，可以把脚本做成定时任务。

Linux/macOS 上用 crontab：

# 每个交易日 14:30 运行一次
30 14 * * 1-5 cd /path/to/project && uv run python scan_stocks.py >> scan.log 2>&1

或者用 Python 的 schedule 库在一个长驻进程里循环：

import schedule
import time

def daily_scan():
    # 把上面的选股逻辑封装成函数
    print("开始全市场扫描...")
    run_composite_scan()
    print("扫描完成")

schedule.every().day.at("14:30").do(daily_scan)

while True:
    schedule.run_pending()
    time.sleep(60)

9. 从扫描到策略：下一步怎么走

全市场选股本身不是策略，它是策略的输入端。拿到候选列表之后，你还需要回答几个问题：

问题	可能的方向
选出来的票后续表现如何？	对历史数据做回测：每天选出 Top 10，等权持有 5 天，看收益
哪些筛选条件贡献最大？	因子重要性分析：单独测试每个条件的 IC 值
会不会选出太多垃圾票？	加入基本面过滤（市值、行业、ST 剔除等）
参数是不是过拟合？	用滚动窗口做样本外检验
实盘能不能执行？	考虑滑点、涨停无法买入、停牌复牌等情况

全市场扫描给了你一个程序化的起点。有了这个起点，后面不管你是做日线轮动、周线动量、还是事件驱动，都有一个可复用的数据获取和筛选框架。

结语

选股的本质是"在大量标的中快速缩小注意力范围"。

传统方式是靠人肉翻看、靠消息面、靠别人的推荐。程序化方式是定义规则、全市场扫描、存档结果、持续验证。

AlphaFeed 的 Universe 查询让"全市场快照"这件事变得很轻——一个 API 调用就能拿到所有 A 股的实时数据。在此基础上，筛选逻辑、组合条件、定时运行、信号存档，全部用 Python 搭起来。

不要期望一个选股条件能"选出必涨的票"。但一个持续运行的选股系统，可以帮你系统性地观察市场、积累数据、迭代方法。

相关链接：

• AlphaFeed 官网：https://alphafeed.org/
• Python SDK 快速开始：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart
• 全市场行情文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart