选股最怕什么？漏掉。

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

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

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

---

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

from quantdash import QuantDash

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 亿就算天量。

import pandas as pd
from quantdash import QuantDash

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 = {}
for sym, kdf in klines.items():
  if kdf is not None and len(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 = []
for sym, kdf in klines.items():
  if kdf is not None and len(kdf) >= 15:
    # 前 14 天均量 vs 第 15 天（今天）成交量
    past_14_vol = kdf["volume"].iloc[:-1].mean()
    today_vol = kdf["volume"].iloc[-1]
    if past_14_vol > 0 and (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 = []
for sym, kdf in klines.items():
  if kdf is not None and len(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()
    )

    if down_days >= 4 and today_rebound:
      reversal_candidates.append(sym)

print(f"\n=== 连跌后企稳/放量反弹: {len(reversal_candidates)} 只 ===")
print(reversal_candidates[:15])

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

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

# 突破 20 日新高
breakout = []
for sym, kdf in klines.items():
  if kdf is not None and len(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()

    if today_close > past_20_high and today_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 脚本，每天收盘后直接运行即可：

import pandas as pd
from quantdash import QuantDash


def scan_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 = {}
  for sym, kdf in klines.items():
    if kdf is not None and len(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 = [], []
  for sym, kdf in klines.items():
    if kdf is None or len(kdf) < 20:
      continue

    recent_5 = kdf.iloc[-6:-1]
    today = kdf.iloc[-1]

    # 连跌企稳
    down_days = sum(recent_5["close"] < recent_5["open"])
    if (
        down_days >= 4
        and (today["close"] > today["open"])
        and (today["volume"] > recent_5["volume"].mean())
    ):
      reversals.append(sym)

    # 突破 20 日新高
    past_20_high = kdf["high"].iloc[:-1].max()
    if today["close"] > past_20_high and today["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()

8. 总结

每日复盘的核心不是看“谁涨得最好”，而是寻找“哪些标的正在发生结构性改变”。

通过 QuantDash 的全市场快照（quotes）接口 + 高效的批量 K 线（klines.batch）能力，你可以在几分钟内快速完成 A 股 5000 多只股票的量价异动筛选，真正做到全市场覆盖、不漏掉潜在机会。

如需进一步了解与接入：

• 官网地址：https://quantdash.net/
• 技术文档：https://docs.quantdash.net/