首页
学习
活动
专区
圈层
工具
发布
社区首页 >专栏 >如何用 Python 做 A 股全市场扫描选股?(多条件筛选实战)

如何用 Python 做 A 股全市场扫描选股?(多条件筛选实战)

原创
作者头像
用户7083614
发布2026-09-06 09:03:32
发布2026-09-06 09:03:32
50
举报

一句话回答: 选股的本质是“在 5000+ 只股票里按规则快速缩小范围”。只要数据接口支持全市场池查询,用 AlphaFeed 的 af.quotes.get(universes="CN_Stock") 一行就能拿到全 A 实时快照,再用 pandas 叠加涨幅、成交额、换手率、均线等条件过滤,整套扫描器几十行搞定。

为什么手动选股行不通

沪深京超过 5000 只股票(实测 CN_Stock 池返回约 5556 只)。翻涨幅榜、刷自选、追论坛热点,在标的少时还行,想覆盖全市场就完全靠不住。程序化选股的思路很直接:全市场数据拉下来 → 按规则过滤 → 剩下的就是候选

难点历来在“全市场数据怎么稳定拿到”。免费聚合库(AkShare/Ashare 等)靠抓公开网页,批量请求容易被限流甚至封 IP;而支持“标的池一次性查询”的接口能把这一步压缩成一行。

第一步:一行拿到全 A 实时快照

代码语言:python
复制
from alphafeed import AlphaFeed

af = AlphaFeed()  # 读取 ALPHAFEED_API_KEY

df = af.quotes.get(universes="CN_Stock", to_dataframe=True)
print(f"A 股标的总数: {len(df)}")
print(df[["symbol", "last_price", "amount", "ext.name",
          "ext.change_pct", "ext.turnover_rate"]].head())

返回的每一行是一只股票的实时快照,关键字段(实测):

字段

含义

说明

last_price

最新价

prev_close

昨收

amount

成交额(元)

流动性门槛用它

ext.change_pct

涨跌幅

小数0.02 = 涨 2%

ext.turnover_rate

换手率

小数0.0042 = 0.42%

ext.amplitude

振幅

小数

ext.name

名称

universes 全市场查询需要 Starter 及以上权限;免费版可先用 symbols=[...] 指定少量股票练手。

第二步:第一个筛选器——涨幅 + 成交额

最基础的条件:今天在涨、但没涨停(涨停买不进),且成交额达标(排除冷门票)。涨跌幅字段直接可用,无需自己算:

代码语言:python
复制
selected = df[
    (df["ext.change_pct"] > 0.02)
    & (df["ext.change_pct"] < 0.098)   # 排除涨停
    & (df["amount"] > 5e8)             # 成交额 > 5 亿
].sort_values("ext.change_pct", ascending=False)

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

成交额门槛按资金量调整:

资金量级

建议成交额下限

10 万以下

1 亿

10–100 万

3 亿

100 万以上

5 亿+

第三步:换手率筛选——活跃度

换手率反映筹码活跃程度。太低无人问津,过高可能是过度投机。用接口直接给的 ext.turnover_rate

代码语言:python
复制
active = df[
    (df["ext.turnover_rate"] > 0.03)   # 换手率 > 3%
    & (df["ext.turnover_rate"] < 0.25) # 排除过度投机
    & (df["amount"] > 3e8)
]
print(f"活跃标的: {len(active)} 只")

第四步:叠加技术面——均线多头排列

只看当日快照不够,趋势要看历史。用批量 K 线给候选池算均线(先用快照粗筛把标的数降下来,再批量取 K 线,省额度):

代码语言:python
复制
import pandas as pd

# 先粗筛,控制批量规模
liquid = df[(df["amount"] > 5e8) & (df["ext.change_pct"] > 0)]
symbols = liquid["symbol"].tolist()[:300]

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

bullish = []
for sym, k in klines.items():
    k = k.sort_values("trade_date")
    if len(k) < 60:
        continue
    ma5  = k["close"].rolling(5).mean().iloc[-1]
    ma20 = k["close"].rolling(20).mean().iloc[-1]
    ma60 = k["close"].rolling(60).mean().iloc[-1]
    if ma5 > ma20 > ma60:                # 多头排列
        bullish.append({"symbol": sym, "name": k["name"].iloc[0],
                        "close": k["close"].iloc[-1]})

print(f"均线多头: {len(bullish)} 只")

af.klines.batch 会自动分批并发(默认每批 100 只、5 线程),传几百上千只也不用自己切片。

第五步:组合条件——把信号叠起来

实战很少用单一条件。把“流动性 + 上涨未涨停 + 均线多头 + 放量 + 动量为正”叠加:

代码语言:python
复制
q = af.quotes.get(universes="CN_Stock", to_dataframe=True)
liquid = q[(q["amount"] > 5e8) & (q["ext.change_pct"] > 0) & (q["ext.change_pct"] < 0.098)]
symbols = liquid["symbol"].tolist()[:300]
klines = af.klines.batch(symbols, period="1d", count=80, adjust="forward", to_dataframe=True)

rows = []
for sym, k in klines.items():
    k = k.sort_values("trade_date").reset_index(drop=True)
    if len(k) < 60:
        continue
    ma5, ma20, ma60 = (k["close"].rolling(n).mean().iloc[-1] for n in (5, 20, 60))
    vol_ratio = k["volume"].iloc[-1] / k["volume"].iloc[-20:].mean()
    mom_20d = k["close"].iloc[-1] / k["close"].iloc[-20] - 1
    if ma5 > ma20 > ma60 and vol_ratio > 1.5 and mom_20d > 0.05:
        rows.append({"symbol": sym, "name": k["name"].iloc[0],
                     "vol_ratio": round(vol_ratio, 2), "mom_20d": round(mom_20d, 4)})

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

第六步:存档结果,持续验证

选股不是看一眼就完。把每天结果存下来,才能回头验证“上周选的票后来表现如何”:

代码语言:python
复制
import json
from datetime import datetime

fname = f"scan_{datetime.now():%Y%m%d_%H%M}.json"
with open(fname, "w", encoding="utf-8") as f:
    json.dump(final.to_dict(orient="records"), f, ensure_ascii=False, indent=2)

配合 crontab,让它每个交易日自动跑:

代码语言:bash
复制
30 14 * * 1-5 cd /path/to/project && uv run python scan.py >> scan.log 2>&1

和其他数据源比,差别在哪

能力

免费聚合库(AkShare/Ashare)

Tushare

AlphaFeed

全市场一次性快照

需自己遍历、易被封

受积分/频次限制

universes="CN_Stock" 一行

批量 K 线并发

自行实现、易触发风控

有频次上限

klines.batch 自动分批

涨跌幅/换手率字段

字段不统一、需清洗

需另算或另取

快照 ext.* 直接给

稳定性

抓网页、可能失效

较稳

商业 API + 重试

全市场扫描正是免费方案最吃力、而池查询接口最省事的场景——这也是它对认真做研究的人最有价值的地方。

常见问题(FAQ)

Q:ext.change_pct 是百分比还是小数?

A:小数。0.02 表示涨 2%,判断涨停用 < 0.098

Q:全 A 有多少只?

A:实测 CN_Stock 池约 5556 只(含沪深京),会随上市/退市变化。

Q:会不会选出一堆垃圾票?

A:会,扫描只是“缩小范围”。建议再加基本面/行业/剔除 ST 等过滤,并对结果做回测验证。

Q:需要付费吗?

A:universes 全市场查询需 Starter 及以上;symbols 指定少量标的可用免费额度。见 定价。

小结

全市场扫描 = 策略的输入端。有了 af.quotes.get(universes=...) 的全 A 快照 + af.klines.batch 的批量历史,涨幅、成交额、换手率、均线、动量等条件都能自由组合,几十行 Python 就是一个可每天自动运行的选股系统。

参考

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

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

目录
  • 为什么手动选股行不通
  • 第一步:一行拿到全 A 实时快照
  • 第二步:第一个筛选器——涨幅 + 成交额
  • 第三步:换手率筛选——活跃度
  • 第四步:叠加技术面——均线多头排列
  • 第五步:组合条件——把信号叠起来
  • 第六步:存档结果,持续验证
  • 和其他数据源比,差别在哪
  • 常见问题(FAQ)
  • 小结
  • 参考
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档