每天收盘后5分钟，用 Python 扫出 A 股全市场量价异动-平芜编程栈

选股最怕什么？漏掉。

5000 多只 A 股，你不可能全看一遍。大部分人只盯着自选股或者涨幅榜，但真正有意义的异动——比如一只长期缩量横盘的票突然放量 3 倍、一只连跌 7 天的票尾盘突然拉起来——往往发生在你根本没关注的票上。

解决思路很直接：每天收盘后跑一遍全市场扫描，把当天有“量价异动”的票自动筛出来。

QuantDash的 Universe 接口可以一次拉到全部 A 股的实时行情快照，配合 batch 接口拉历史 K 线，5 分钟内就能扫完 5000+ 只票。

1. 第一步：一行代码拿到全市场行情快照

fromquantdashimportQuantDash qd=QuantDash()# 全部 A 股当日行情快照all_cn=qd.quotes.get(universes=["CN_Stock"],to_dataframe=True)print(f"A 股标的总数:{len(all_cn)}")# 计算涨跌幅all_cn["change_pct"]=(all_cn["last_price"]-all_cn["prev_close"])/all_cn["prev_close"]

2. 异动规则一：当日放量（成交量突增）

放量是最重要的异动信号之一。一只票如果平时日均成交 5000 万，今天突然做了 2 个亿，说明有大量资金在交易——无论是买还是卖，都值得关注。

但“放量”需要和历史均量对比才有意义。光看当日成交量的绝对值没用——茅台日均成交 30 亿算缩量，一只小盘股做 3 亿就算天量。

importpandasaspdfromquantdashimportQuantDash qd=QuantDash()all_cn=qd.quotes.get(universes=["CN_Stock"],to_dataframe=True)all_cn["change_pct"]=(all_cn["last_price"]-all_cn["prev_close"])/all_cn["prev_close"]# 过滤掉成交额太小的标的（低于 5000 万，流动性不够）active=all_cn[all_cn["amount"]>5e7].copy()print(f"活跃标的:{len(active)}只")# 拉取这些票最近 20 天的 K 线，计算历史均量symbols=active["symbol"].tolist()# batch 自动分块并发，几百只票几秒搞定klines=qd.klines.batch(symbols,period="1d",count=20,adjust="forward",to_dataframe=True,show_progress=True,)# 计算每只票的 20 日均量avg_volumes={}forsym,kdfinklines.items():ifkdfisnotNoneandlen(kdf)>=10:avg_volumes[sym]=kdf["volume"].mean()active["avg_vol_20d"]=active["symbol"].map(avg_volumes)active=active.dropna(subset=["avg_vol_20d"])active["vol_ratio"]=active["volume"]/active["avg_vol_20d"]# 筛选放量票：今日成交量 > 2.5 倍 20 日均量volume_surge=active[active["vol_ratio"]>2.5].sort_values("vol_ratio",ascending=False)print(f"\n=== 放量异动（量比 > 2.5x）:{len(volume_surge)}只 ===")print(volume_surge[["symbol","last_price","change_pct","vol_ratio","amount",]].head(20).to_string(index=False))

量比 2.5 倍以上的票一般不会太多（每天几十只到上百只），但里面往往藏着当天市场的“暗线”。

3. 异动规则二：缩量后突然放量（底部放量）

比单天放量更有价值的是：前段时间极度缩量（无人关注、筹码锁定），今天突然放量。这往往是主力试盘或启动的信号。

判断逻辑：

之前 14 天成交量低迷；
今天成交量是前 14 天均量的 2.5 倍以上。

# 缩量后突然放量bottom_vol=[]forsym,kdfinklines.items():ifkdfisnotNoneandlen(kdf)>=15:# 前 14 天均量 vs 第 15 天（今天）成交量past_14_vol=kdf["volume"].iloc[:-1].mean()today_vol=kdf["volume"].iloc[-1]ifpast_14_vol>0and(today_vol/past_14_vol)>2.5:bottom_vol.append({"symbol":sym,"today_vol":today_vol,"past_avg_vol":past_14_vol,"surge_ratio":today_vol/past_14_vol,})bv_df=pd.DataFrame(bottom_vol).sort_values("surge_ratio",ascending=False)print(f"\n=== 缩量后突然放量:{len(bv_df)}只 ===")print(bv_df.head(15).to_string(index=False))

4. 异动规则三：大涨但不是涨停（有后续空间）

涨停的票你大概率买不到（挂单排不上）。更有实际意义的是那些涨了 5%–9% 但没涨停的票——如果同时放量了，说明有资金在大力买入但还没封板，后续仍有关注空间。

# 筛选大涨未涨停且放量的股票active_surge=active[(active["change_pct"]>=0.05)&(active["change_pct"]<0.095)&(active["vol_ratio"]>=2.0)].sort_values("change_pct",ascending=False)print(f"\n=== 大涨未涨停且放量:{len(active_surge)}只 ===")print(active_surge[["symbol","last_price","change_pct","vol_ratio","amount"]].head(15).to_string(index=False))

5. 异动规则四：连续下跌后企稳（可能的底部信号）

市场调整时，连跌几天的票一旦出现止跌信号，很可能是反弹起点。

判断逻辑：

过去 5 天中有 4 天以上在下跌；
今天收阳线且成交量放大（高于前 5 天均量）。

# 找连跌后企稳的票reversal_candidates=[]forsym,kdfinklines.items():ifkdfisnotNoneandlen(kdf)>=6:recent_5=kdf.iloc[-6:-1]# 前 5 天today=kdf.iloc[-1]# 今天# 统计前 5 天中有几天下跌down_days=sum(recent_5["close"]<recent_5["open"])# 今天收阳且放量today_rebound=(today["close"]>today["open"])and(today["volume"]>recent_5["volume"].mean())ifdown_days>=4andtoday_rebound:reversal_candidates.append(sym)print(f"\n=== 连跌后企稳/放量反弹:{len(reversal_candidates)}只 ===")print(reversal_candidates[:15])

6. 异动规则五：突破近期新高

突破前高是经典的顺势策略逻辑。如果价格创了近 20 日新高，同时伴随成交量放大，说明突破是有效的（而非假突破）。

# 突破 20 日新高breakout=[]forsym,kdfinklines.items():ifkdfisnotNoneandlen(kdf)>=20:past_20_high=kdf["high"].iloc[:-1].max()today_close=kdf["close"].iloc[-1]today_vol=kdf["volume"].iloc[-1]avg_vol=kdf["volume"].mean()iftoday_close>past_20_highandtoday_vol>avg_vol*1.5:breakout.append({"symbol":sym,"close":today_close,"past_20_high":past_20_high,"vol_ratio":today_vol/avg_vol,})bo_df=pd.DataFrame(breakout).sort_values("vol_ratio",ascending=False)print(f"\n=== 突破 20 日新高（放量突破）:{len(bo_df)}只 ===")print(bo_df.head(15).to_string(index=False))

7. 完整扫描脚本（整合 5 种异动）

把上述逻辑整合为一个完整的 Python 脚本，每天收盘后直接运行即可：

importpandasaspdfromquantdashimportQuantDashdefscan_anomalies():qd=QuantDash()print("Step 1: 获取全市场行情快照...")all_cn=qd.quotes.get(universes=["CN_Stock"],to_dataframe=True)all_cn["change_pct"]=(all_cn["last_price"]-all_cn["prev_close"])/all_cn["prev_close"]# 过滤成交额 > 5000 万的活跃标的active=all_cn[all_cn["amount"]>5e7].copy()symbols=active["symbol"].tolist()print(f"活跃标的数量:{len(symbols)}")print("Step 2: 批量拉取近 20 日 K 线...")klines=qd.klines.batch(symbols,period="1d",count=20,adjust="forward",to_dataframe=True,show_progress=True,)# 计算 20 日均量avg_vols={}forsym,kdfinklines.items():ifkdfisnotNoneandlen(kdf)>=10:avg_vols[sym]=kdf["volume"].mean()active["avg_vol_20d"]=active["symbol"].map(avg_vols)active=active.dropna(subset=["avg_vol_20d"])active["vol_ratio"]=active["volume"]/active["avg_vol_20d"]# 1. 当日放量 > 2.5 倍surge=active[active["vol_ratio"]>2.5]# 2. 大涨（5%~9.5%）且放量 > 2.0 倍big_rise=active[(active["change_pct"]>=0.05)&(active["change_pct"]<0.095)&(active["vol_ratio"]>=2.0)]# 3. 连跌后企稳 & 4. 突破 20 日新高reversals,breakouts=[],[]forsym,kdfinklines.items():ifkdfisNoneorlen(kdf)<20:continuerecent_5=kdf.iloc[-6:-1]today=kdf.iloc[-1]# 连跌企稳down_days=sum(recent_5["close"]<recent_5["open"])if(down_days>=4and(today["close"]>today["open"])and(today["volume"]>recent_5["volume"].mean())):reversals.append(sym)# 突破 20 日新高past_20_high=kdf["high"].iloc[:-1].max()iftoday["close"]>past_20_highandtoday["volume"]>kdf["volume"].mean()*1.5:breakouts.append(sym)print("\n"+"="*50)print(f"【扫描完成】")print(f"1. 放量异动（>2.5x）:{len(surge)}只")print(f"2. 大涨未涨停（5%-9%, >2x量）:{len(big_rise)}只")print(f"3. 连跌企稳反弹:{len(reversals)}只")print(f"4. 突破 20 日新高:{len(breakouts)}只")print("="*50)if__name__=="__main__":scan_anomalies()