用 Python 做 A 股全市场扫描选股：从 5000+ 只股票中筛出值得关注的标的（附 GitHub 源码）
·
选股是量化交易中最高频、最核心的需求之一。
大多数投资者的常规做法是打开行情软件，手动翻看自选股、刷涨幅榜，或者跟着论坛帖子追热点。这种方式在市场标的较少时还勉强行得通，一旦想要覆盖全市场——尤其是当前沪深京超过 5000 只股票时——纯人工复盘就完全无法应付了，极易遗漏真正具有异动特征的黑马。
程序化选股的思路则非常直接：把全市场行情快照一次性拉下来，按照你定义的量价、动量或技术面规则进行多维过滤，剩下的就是你的高价值候选列表。
这听起来似乎需要庞大的分布式数据清洗工程，但如果数据接口本身支持“全市场池一键查询”，整个流程在 Python 中可以压缩到十几行。本文将带你基于 QuantDash 的行情快照与批量 K 线接口，构建一套完整的全市场多因子选股系统，并实现信号的自动化本地存档。
1. 全市场行情快照：一行代码拿到所有 A 股
传统的开源库（如 AkShare、efinance）在抓取全市场快照时，往往需要多重循环或多线程拼接，耗时且易被封 IP。
在 QuantDash 中，你可以通过 qd.quotes.get(universes="CN_Stock") 一键获取沪深京全部 A 股的实时行情快照，直接返回标准的 Pandas DataFrame：
from quantdash import QuantDash
# 初始化客户端，SDK 会自动从环境变量读取 QUANTDASH_API_KEY
qd = QuantDash(api_key="demo_public_token") # 演示环境可使用公共测试 Token
all_cn = qd.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))输出的Pandas DataFrame结果：
A 股标的总数: 5530
symbol last_price prev_close volume amount
0 603328.SH 8.51 8.68 113650 9.751710e+07
1 603334.SH 31.15 32.10 10357 3.288180e+07
2 920363.BJ 9.43 9.84 13751 1.325170e+07
3 300421.SZ 15.16 15.45 54446 8.294060e+07
4 920821.BJ 17.43 18.33 35797 6.395950e+07
5 301022.SZ 19.68 20.45 9460 1.879810e+07
6 688285.SH 6.68 6.94 21402 1.443650e+07
7 300675.SZ 11.06 11.41 19630 2.187310e+07
8 300058.SZ 11.21 11.95 2045459 2.328081e+09
9 688793.SH 14.64 15.13 9788 1.440380e+07返回的每一行都包含当前最新价格、昨收、当日成交量、当日成交额等时序快照字段，为我们搭建实时全市场扫描器提供了极简的数据底座。
如果你也做 ETF 轮动或交易，只需更换一个池子即可：
all_etf = qd.quotes.get(universes="CN_ETF", to_dataframe=True)
print(f"ETF 标的总数: {len(all_etf)}")注意：全量 Universe 实时快照查询需要基础及以上的订阅权限。免费版在测试时可以先使用 symbols 参数指定具体股票列表进行局部演练。
2. 第一个筛选器：涨幅 + 成交额
最基础的选股逻辑：过滤掉不涨的票，同时排除掉流动性极差的冷门票，且避开无法买入的涨停板。
import pandas as pd
from quantdash import QuantDash
qd = QuantDash(api_key="demo_public_token")
df = qd.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) & # 今日涨幅大于 2%
(df["change_pct"] < 0.098) & # 排除已封涨停板的标的（大概率买不进去，无实操意义）
(df["amount"] > 5e8) # 成交额大于 5 亿元，保证足够的资金流动性
].copy()
selected = selected.sort_values("change_pct", ascending=False)
print(f"最基础筛选符合条件的标的: {len(selected)} 只")
print(selected[["symbol", "last_price", "change_pct", "amount"]].head(20))输出结果：
基础筛选符合条件的标的: 76 只
symbol last_price change_pct amount
393 002156.SZ 76.64 0.097680 1.981788e+10
162 300779.SZ 77.22 0.095475 1.848529e+09
1567 301345.SZ 256.90 0.088605 5.832200e+08
1652 603065.SH 14.70 0.084871 6.752779e+08
2515 301127.SZ 19.05 0.083618 8.622514e+08
504 688220.SH 111.77 0.081471 1.759517e+09
995 688596.SH 48.38 0.077266 1.896488e+09
2009 001258.SZ 13.03 0.075970 2.713109e+09
2172 000021.SZ 41.99 0.074188 8.593148e+09
92 301382.SZ 35.60 0.072612 8.470731e+08
5469 301357.SZ 92.58 0.072397 9.233488e+08
3069 300062.SZ 6.80 0.070866 8.053601e+08
5022 688049.SH 44.37 0.070446 7.136064e+08
2842 688484.SH 35.70 0.070144 5.735927e+08
597 688361.SH 357.42 0.069032 4.621535e+09
3708 300427.SZ 6.36 0.065327 5.519051e+08
4147 688072.SH 791.00 0.064603 6.999719e+09
20 688630.SH 408.80 0.064583 3.062885e+09
2888 688099.SH 98.00 0.064523 2.987887e+09
4248 000779.SZ 9.99 0.062766 1.121072e+09这里的成交额下限（amount）可以根据你的个人资金体量进行弹性调整：
| 资金量级 | 建议日成交额下限 |
| 10 万以下 | 1 亿 |
| 10 万 – 100 万 | 3 亿 |
| 100 万以上 | 5 亿+ |
3. 量价异动选股：放量突破
在短线交易中，无量上涨往往是“空中楼阁”式的虚涨。涨 3% 但缩量，随时面临反转；涨 3% 并且成交量显著放大，才说明有增量资金在强力建仓。
要准确判断“放量”，需要历史成交量来做参考。我们使用该股票最近 20 天的日均成交量（MA20 Volume）作为量能基准：
import pandas as pd
from quantdash import QuantDash
qd = QuantDash(api_key="demo_public_token")
df_today = qd.quotes.get(universes="CN_Stock", to_dataframe=True)
df_today["change_pct"] = (df_today["last_price"] - df_today["prev_close"]) / df_today["prev_close"]
# 粗筛：过滤出今天上涨且成交额不低于 3 亿的标的
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] # 截取前200只做示范，防止请求过多
# 使用 batch 接口高并发获取这些候选股的历史日 K 线（近20个交易日）
klines = qd.klines.batch(
symbols=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
# 1. 计算过去 20 日的均量
avg_vol = kdf["volume"].mean()
# 2. 拿到今日实时成交量
today_vol = candidates.loc[candidates["symbol"] == sym, "volume"].values
if len(today_vol) == 0 or avg_vol == 0:
continue
# 3. 计算量比（今日量 / 20日均量）
ratio = today_vol[0] / avg_vol
vol_ratio_list.append({
"symbol": sym,
"vol_ratio": round(ratio, 2),
"change_pct": round(candidates.loc[candidates["symbol"] == sym, "change_pct"].values[0] * 100, 2),
"amount": candidates.loc[candidates["symbol"] == sym, "amount"].values[0],
})
vol_df = pd.DataFrame(vol_ratio_list)
# 筛选量比大于 2.0（即今日成交量超历史均值2倍以上）的放量股
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))4. 动量选股：最近 N 天涨幅排名
动量因子（Momentum）是学术界与业界公认的最有效因子之一。其基本假设是：在过去一段时间涨势凌厉的股票，短期内大概率会维持其趋势。
下面我们编写一个 20 日动量选股脚本：
import pandas as pd
from quantdash import QuantDash
qd = QuantDash(api_key="demo_public_token")
df_today = qd.quotes.get(universes="CN_Stock", to_dataframe=True)
liquid = df_today[df_today["amount"] > 3e8].copy()
symbols = liquid["symbol"].tolist()[:300]
# 拉取候选股过去 20 天的历史日 K
klines = qd.klines.batch(
symbols=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
# 计算 20 日累计收益率
ret_20d = kdf["close"].iloc[-1] / kdf["close"].iloc[0] - 1
momentum_list.append({
"symbol": sym,
"momentum_20d": round(ret_20d * 100, 2),
"close": kdf["close"].iloc[-1],
})
mom_df = pd.DataFrame(momentum_list)
mom_df = mom_df.sort_values("momentum_20d", ascending=False)
print("=== 20 日动量 Top 10 ===")
print(mom_df.head(10).to_string(index=False))实时数据输出：
=== 20 日动量 Top 10 ===
symbol momentum_20d close
002396.SZ 88.87 32.92
001258.SZ 86.94 13.03
300534.SZ 54.42 11.18
601168.SH 32.70 36.24
603338.SH 32.25 57.00
002860.SZ 22.86 17.79
688072.SH 22.60 791.00
300062.SZ 21.86 6.80
603619.SH 20.81 22.70
300996.SZ 19.72 12.935. 技术面选股：均线多头排列
技术分析者极度青睐多头排列形态：MA5>MA10>MA20>MA60 。这说明从短周期到中长周期，资金的平均买入成本在不断垫高，趋势完全一致向上。
import pandas as pd
from quantdash import QuantDash
qd = QuantDash(api_key="demo_public_token")
df_today = qd.quotes.get(universes="CN_Stock", to_dataframe=True)
liquid = df_today[df_today["amount"] > 3e8].copy()
symbols = liquid["symbol"].tolist()[:500]
# 获取 60 日的历史 K 线用以计算中长期均线
klines = qd.klines.batch(
symbols=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),
"ma20": round(ma20, 2),
"ma60": round(ma60, 2),
})
bullish_df = pd.DataFrame(bullish_list)
print(f"均线多头排列个股: {len(bullish_df)} 只")
print(bullish_df.head(10))输出实时数据：
均线多头排列个股: 27 只
symbol name close ma5 ma20 ma60
0 000166.SZ 申万宏源 4.64 4.69 4.59 4.55
1 002860.SZ 星帅尔 17.79 16.88 16.19 14.38
2 300458.SZ 全志科技 40.99 39.28 38.63 36.46
3 688347.SH 华虹宏力 329.90 358.04 339.20 254.02
4 300534.SZ 陇神戎发 11.18 12.63 9.46 8.64
5 688702.SH 盛科通信-U 390.78 388.73 374.49 321.93
6 601377.SH 兴业证券 6.08 6.16 6.14 6.02
7 688578.SH 艾力斯 119.24 127.09 122.75 100.19
8 002832.SZ 比音勒芬 22.81 22.14 21.25 20.35
9 601336.SH 新华保险 63.12 64.73 63.59 61.706. 组合条件选股：把多个信号叠加
在实战交易中，我们很少依赖单一因子选股，而是将流动性、技术面、量价和动量共振因子串联起来：
-
流动性门槛：日成交额 > 5 亿。
-
非涨停限制：排除无法建仓的涨停个股。
-
均线多头：MA5>MA20>MA60 。
-
量能放量：今日量比 > 1.5。
-
中期动量：20 日动量累计涨幅 > 5%。
import pandas as pd
from quantdash import QuantDash
qd = QuantDash(api_key="demo_public_token")
df_today = qd.quotes.get(universes="CN_Stock", to_dataframe=True)
df_today["change_pct"] = (df_today["last_price"] - df_today["prev_close"]) / df_today["prev_close"]
# 基础过滤：日成交额超 5 亿，涨幅在 0%-9.8% 之间
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 = qd.klines.batch(
symbols=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] * 100, 2),
"vol_ratio": round(vol_ratio, 2),
"momentum_20d": round(ret_20d * 100, 2),
})
final = pd.DataFrame(results).sort_values("momentum_20d", ascending=False)
print(f"多因子共振选出: {len(final)} 只")
print(final.to_string(index=False))输出实时结果：
多因子共振选出: 5 只
symbol name close change_pct vol_ratio momentum_20d
603137.SH 恒尚节能 26.51 2.75 1.56 125.43
300779.SZ 惠城环保 77.22 9.55 1.94 54.72
000779.SZ 甘咨询 9.99 6.28 2.25 28.90
002860.SZ 星帅尔 17.79 5.20 1.57 22.86
301536.SZ 星宸科技 131.80 0.27 1.79 9.087. 把选股结果自动保存下来
选股结果不仅是当日看一眼，在量化工程中，我们必须把每天的筛选结果进行标准格式存档（比如 JSON 或 CSV）。只有做好了信号存档，后续才能轻松验证“上周五选出来的个股，本周实际收益表现如何”：
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. Cursor / DeepSeek 量化选股专用 Prompt
如果你想将自己的奇思妙想快速转化为代码，或者打算建立一个定时发送复盘报告的脚本，可以直接复制以下这段 Prompt 扔给 AI 助手：
你是一个资深的 Python 量化开发专家。我正在使用 QuantDash 数据接口构建一个【A股全市场每日自动选股与推送系统】。
请记住我的核心开发规范，并在此基础上帮我实现：
1. 一键拿全市场行情快照：
df = qd.quotes.get(universes="CN_Stock", to_dataframe=True)
2. 批量并发获取历史 K 线：
klines = qd.klines.batch(symbols, period="1d", count=60, adjust="forward", to_dataframe=True)
3. 选股规则升级：
请帮我设计一个“N字型二度放量启动”选股器。
- 规则 A (流动性)：当日成交额大于 4 亿元。
- 规则 B (前期强势)：过去 20 个交易日内，至少出现过一次涨幅大于 9.5% 的大阳线。
- 规则 C (中期调整)：大阳线出现后，股价经历 5-10 天的缩量回踩，收盘价不破大阳线当天的开盘价。
- 规则 D (今日启动)：今日股价重回 5 日均线上方，且今日成交量大于 5 日滚动均量的 1.5 倍（温和放量突破）。
4. 代码要求：
请用 Pandas 向量化与循环高效结合完成，输出包含 symbol, name, close, change_pct 字段的选股 DataFrame，并自动以 "scan_n_shape_[YYYYMMDD].csv" 格式存档到本地文件夹。总结：建立你的程序化复盘习惯
全市场选股不是寻找“稳赚不赔的秘籍”，而是用程序和数据逻辑替代人的主观情绪，快速缩小高期望值交易的注意力范围。
借助 QuantDash 统一的 Universe 接口与 Batch 并发技术，你可以彻底告别笨重繁杂的多线程网络爬虫开发，用最优雅、标准的 Pandas 代码在数秒内完成 5000+ 标的的过滤。通过把每日的选股快照自动保存为 JSON 或 CSV 文件，你正在建立属于自己的第一个自动化因子库，这也是你迈向中高级量化研发无可替代的工程第一步。
“三步走”落地指引
-
第一步：获取完整源码。请访问官方开源托管仓库获取本文全套选股系统、复盘报告生成及定时任务配置脚本：https://github.com/quantdash-net/QuantDash（请认准官方 quantdash-net 组织，欢迎 Star 支持）。
-
第二步：申请专属密钥。注册获取您的个人免费/生产级 API Key：https://quantdash.net/。
-
第三步：查阅开发细节。更多高频行情、多周期 K 线、美股与港股 universes 全集接口参数请参考官方文档：https://docs.quantdash.net/。
这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。
更多推荐
13
0- 0
已为社区贡献6条内容
所有评论(0)