supermind logo
同花顺logo

每天收盘后 5 分钟，用 Python 扫出A股全市场异动

用户头像sh_*2176oo
2026-07-14 发布

每天收盘后 5 分钟，用 Python 扫出A股全市场量价异动

选股最怕什么？漏掉。

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

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

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

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

from alphafeed import AlphaFeed

af = AlphaFeed()

# 全部 A 股当日行情
all_cn = af.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"]

universes="CN_Stock" 返回沪深京所有股票的当前价格、昨收、成交量、成交额。这是整个扫描系统的数据底座。

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

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

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

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

all_cn = af.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 = af.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. 异动规则二：缩量后突然放量（底部放量）

比单纯放量更有价值的信号是：一只票连续多天缩量（没人关注），然后某天突然放量。这可能意味着有新资金开始介入。

# 找"缩量后放量"的票
bottom_volume = []

for sym, kdf in klines.items():
    if kdf is None or len(kdf) < 15:
        continue

    kdf = kdf.sort_values("trade_date").reset_index(drop=True)

    # 最近 5 天之前的 10 天均量（"缩量期"基准）
    base_vol = kdf["volume"].iloc[-15:-5].mean()
    # 最近 5 天均量（"缩量期"）
    quiet_vol = kdf["volume"].iloc[-5:-1].mean()
    # 今天成交量
    today_vol = kdf["volume"].iloc[-1]

    if base_vol == 0 or quiet_vol == 0:
        continue

    # 条件：最近 5 天缩量（< 基准的 0.7 倍），今天放量（> 基准的 2 倍）
    if quiet_vol < base_vol * 0.7 and today_vol > base_vol * 2:
        bottom_volume.append({
            "symbol": sym,
            "缩量期均量": int(quiet_vol),
            "今日成交量": int(today_vol),
            "放量倍数": round(today_vol / quiet_vol, 1),
        })

bv_df = pd.DataFrame(bottom_volume).sort_values("放量倍数", ascending=False)
print(f"\n=== 缩量后突然放量: {len(bv_df)} 只 ===")
print(bv_df.head(15).to_string(index=False))

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

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

# 大涨 + 放量但未涨停
hot_not_limit = active[
    (active["change_pct"] > 0.05) &
    (active["change_pct"] < 0.095) &
    (active["vol_ratio"] > 1.5)
].sort_values("change_pct", ascending=False)

print(f"\n=== 大涨放量但未涨停（5%–9.5%，量比>1.5x）: {len(hot_not_limit)} 只 ===")
print(hot_not_limit[["symbol", "last_price", "change_pct", "vol_ratio"]]
      .head(15).to_string(index=False))

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

# 找连跌后企稳的票
reversal_candidates = []

for sym, kdf in klines.items():
    if kdf is None or len(kdf) < 10:
        continue

    kdf = kdf.sort_values("trade_date").reset_index(drop=True)
    kdf["ret"] = kdf["close"].pct_change()

    # 最近 5 天中有 4 天以上下跌
    recent_5 = kdf["ret"].tail(6).iloc[:-1]  # 倒数第 6 到倒数第 2 天
    down_days = (recent_5 < 0).sum()
    cum_drop = recent_5.sum()

    # 今天翻红
    today_ret = kdf["ret"].iloc[-1]

    if down_days >= 4 and cum_drop < -0.05 and today_ret > 0.01:
        reversal_candidates.append({
            "symbol": sym,
            "连跌天数": down_days,
            "累计跌幅": f"{cum_drop:.1%}",
            "今日涨幅": f"{today_ret:+.1%}",
        })

rv_df = pd.DataFrame(reversal_candidates).sort_values("连跌天数", ascending=False)
print(f"\n=== 连跌后翻红: {len(rv_df)} 只 ===")
print(rv_df.head(15).to_string(index=False))

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

# 突破 20 日新高
breakout = []

for sym, kdf in klines.items():
    if kdf is None or len(kdf) < 20:
        continue

    kdf = kdf.sort_values("trade_date").reset_index(drop=True)

    prev_high = kdf["high"].iloc[-20:-1].max()  # 之前 19 天最高价
    today_high = kdf["high"].iloc[-1]
    today_close = kdf["close"].iloc[-1]

    # 今天的最高价突破了之前 19 天的最高，且收盘价在高位（不是冲高回落）
    if today_high > prev_high and today_close > prev_high * 0.98:
        breakout.append({
            "symbol": sym,
            "前高": prev_high,
            "今日最高": today_high,
            "收盘价": today_close,
            "突破幅度": f"{(today_close / prev_high - 1):+.1%}",
        })

bo_df = pd.DataFrame(breakout).sort_values("突破幅度", ascending=False)
print(f"\n=== 突破 20 日新高: {len(bo_df)} 只 ===")
print(bo_df.head(15).to_string(index=False))

7. 完整的每日扫描脚本

把上面的规则整合成一个脚本，收盘后一键运行：

# daily_scan.py
"""每日收盘后量价异动扫描"""

import pandas as pd
from datetime import datetime
from alphafeed import AlphaFeed

af = AlphaFeed()

def daily_scan():
    today = datetime.now().strftime("%Y-%m-%d")
    print(f"{'='*60}")
    print(f"  A 股量价异动扫描  {today}")
    print(f"{'='*60}\n")

    # 全市场行情
    print("正在拉取全市场行情...")
    all_cn = af.quotes.get(universes="CN_Stock", to_dataframe=True)
    all_cn["change_pct"] = (all_cn["last_price"] - all_cn["prev_close"]) / all_cn["prev_close"]
    active = all_cn[all_cn["amount"] > 5e7].copy()
    print(f"活跃标的: {len(active)} 只\n")

    # 批量拉 K 线
    print("正在拉取历史 K 线（用于对比）...")
    symbols = active["symbol"].tolist()
    klines = af.klines.batch(
        symbols, period="1d", count=20,
        adjust="forward", to_dataframe=True, show_progress=True,
    )

    # 计算量比
    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"]

    # ---- 异动 1: 放量 ----
    surge = active[active["vol_ratio"] > 2.5].sort_values("vol_ratio", ascending=False)
    print(f"\n📊 放量异动（量比 > 2.5x）: {len(surge)} 只")
    if len(surge) > 0:
        print(surge[["symbol", "change_pct", "vol_ratio"]].head(10).to_string(index=False))

    # ---- 异动 2: 大涨未涨停 ----
    hot = active[
        (active["change_pct"] > 0.05) &
        (active["change_pct"] < 0.095) &
        (active["vol_ratio"] > 1.5)
    ].sort_values("change_pct", ascending=False)
    print(f"\n🔥 大涨放量未涨停: {len(hot)} 只")
    if len(hot) > 0:
        print(hot[["symbol", "change_pct", "vol_ratio"]].head(10).to_string(index=False))

    # ---- 异动 3: 突破 20 日新高 ----
    breakout_list = []
    for sym, kdf in klines.items():
        if kdf is None or len(kdf) < 20:
            continue
        kdf = kdf.sort_values("trade_date").reset_index(drop=True)
        prev_high = kdf["high"].iloc[-20:-1].max()
        today_close = kdf["close"].iloc[-1]
        if today_close > prev_high:
            breakout_list.append({"symbol": sym, "突破幅度": today_close / prev_high - 1})

    bo_df = pd.DataFrame(breakout_list).sort_values("突破幅度", ascending=False)
    print(f"\n📈 突破 20 日新高: {len(bo_df)} 只")
    if len(bo_df) > 0:
        print(bo_df.head(10).to_string(index=False))

    # ---- 异动 4: 连跌后翻红 ----
    reversal_list = []
    for sym, kdf in klines.items():
        if kdf is None or len(kdf) < 10:
            continue
        kdf = kdf.sort_values("trade_date").reset_index(drop=True)
        kdf["ret"] = kdf["close"].pct_change()
        recent = kdf["ret"].tail(6).iloc[:-1]
        if (recent < 0).sum() >= 4 and recent.sum() < -0.05 and kdf["ret"].iloc[-1] > 0.01:
            reversal_list.append({
                "symbol": sym,
                "连跌天数": (recent < 0).sum(),
                "今日涨幅": kdf["ret"].iloc[-1],
            })

    rv_df = pd.DataFrame(reversal_list).sort_values("今日涨幅", ascending=False)
    print(f"\n🔄 连跌后翻红: {len(rv_df)} 只")
    if len(rv_df) > 0:
        print(rv_df.head(10).to_string(index=False))

    # 汇总
    total = len(surge) + len(hot) + len(bo_df) + len(rv_df)
    print(f"\n{'='*60}")
    print(f"  共发现 {total} 条异动信号（存在重叠）")
    print(f"{'='*60}")


if __name__ == "__main__":
    daily_scan()

设成每天 15:05 自动运行：

# crontab -e
5 15 * * 1-5 cd /path/to/project && uv run python daily_scan.py >> scan.log 2>&1

8. 性能：为什么能在 5 分钟内跑完

整个扫描流程的耗时瓶颈在"拉数据"，而不是计算：

步骤 耗时 说明
全市场行情快照 ~2 秒 af.quotes.get(universes="CN_Stock") 一次返回
筛选活跃标的 < 0.1 秒 pandas 过滤
批量拉 K 线 ~30–60 秒 af.klines.batch() 对 2000+ 只票拉 20 天数据
计算所有异动规则 < 1 秒 纯 pandas 运算

总共约 1–2 分钟。如果你的股票池更小（比如只扫成交额 top 500），十几秒就能跑完。

关键在于 AlphaFeed 的两个接口设计：

  1. universes 全量查询——不需要你预先维护一个代码列表，一行拿到全市场
  2. batch 批量并发——2000 只票的 K 线，SDK 自动分成 20 个块、5 线程并发、失败自动重试

如果用爬虫接口一只一只拉，2000 只票 × 0.5 秒 sleep = 16 分钟起步。

9. 扫描结果怎么用

扫出来的异动票不是"买入推荐"。它们是"值得你花时间去看一眼的票"。

合理的使用方式：

  1. 每天扫描 → 筛出 20–50 只异动票
  2. 人工翻一遍 → 挑出 5–10 只有逻辑的（比如行业有利好、技术形态好）
  3. 加入观察池 → 持续跟踪几天
  4. 确认信号持续 → 决定是否买入

扫描解决的是"从 5000 只里缩小到 50 只"的问题，剩下的判断还是你来做。

评论