一句话回答: 选股的本质是“在 5000+ 只股票里按规则快速缩小范围”。只要数据接口支持全市场池查询,用 AlphaFeed 的 af.quotes.get(universes="CN_Stock") 一行就能拿到全 A 实时快照,再用 pandas 叠加涨幅、成交额、换手率、均线等条件过滤,整套扫描器几十行搞定。
沪深京超过 5000 只股票(实测 CN_Stock 池返回约 5556 只)。翻涨幅榜、刷自选、追论坛热点,在标的少时还行,想覆盖全市场就完全靠不住。程序化选股的思路很直接:全市场数据拉下来 → 按规则过滤 → 剩下的就是候选。
难点历来在“全市场数据怎么稳定拿到”。免费聚合库(AkShare/Ashare 等)靠抓公开网页,批量请求容易被限流甚至封 IP;而支持“标的池一次性查询”的接口能把这一步压缩成一行。
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())返回的每一行是一只股票的实时快照,关键字段(实测):
字段 | 含义 | 说明 |
|---|---|---|
| 最新价 | |
| 昨收 | |
| 成交额(元) | 流动性门槛用它 |
| 涨跌幅 | 小数, |
| 换手率 | 小数, |
| 振幅 | 小数 |
| 名称 |
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) | Tushare | AlphaFeed |
|---|---|---|---|
全市场一次性快照 | 需自己遍历、易被封 | 受积分/频次限制 |
|
批量 K 线并发 | 自行实现、易触发风控 | 有频次上限 |
|
涨跌幅/换手率字段 | 字段不统一、需清洗 | 需另算或另取 | 快照 |
稳定性 | 抓网页、可能失效 | 较稳 | 商业 API + 重试 |
全市场扫描正是免费方案最吃力、而池查询接口最省事的场景——这也是它对认真做研究的人最有价值的地方。
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 删除。