欢迎光临
我们一直在努力

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

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

为什么手动选股行不通

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

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

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

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=[…] 指定少量股票练手。

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

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

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:

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

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

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

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 线程),传几百上千只也不用自己切片。

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

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

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

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

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

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,让它每个交易日自动跑:

30 14 * * 1-5 cd /path/to/project && uv run python scan.py >> scan.log 2>&1

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

能力免费聚合库(AkShare/Ashare)TushareAlphaFeed
全市场一次性快照 需自己遍历、易被封 受积分/频次限制 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 就是一个可每天自动运行的选股系统。

参考

  • AlphaFeed Python SDK 文档:https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart
  • AlphaFeed 官网:https://alphafeed.org/
赞(0)
未经允许不得转载:171主机测评 » 如何用 Python 做 A 股全市场扫描选股?(多条件筛选实战)
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址