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

选股最怕什么？漏掉。

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

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

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

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

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

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

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

```python
# 找"缩量后放量"的票
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% 但没涨停的票——如果同时放量了，说明有资金在大力买入但还没到极限。

```python
# 大涨 + 放量但未涨停
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. 异动规则四：连续下跌后企稳（可能的底部信号）

```python
# 找连跌后企稳的票
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. 异动规则五：突破近期新高

```python
# 突破 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. 完整的每日扫描脚本

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

```python
# 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 自动运行：

```bash
# 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 只"的问题，剩下的判断还是你来做。

---

参考文献：
- Python SDK 快速开始：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart


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

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

编程语言

Python实现A股全市场量价异动扫描，5分钟筛选5000+股票。通过放量、缩量后放量、大涨未涨停、连跌企稳、突破新高5大规则，快速发现市场暗线。AlphaFeed接口实现高效批量数据获取，解决人工盯盘遗漏问题，为投资决策提供数据支持。

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

每天收盘后 5 分钟，用 Python 扫出A股全市场量价异动-腾讯云开发者社区-腾讯云

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

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

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐