supermind logo
同花顺logo

分时数据捕捉尾盘异动：最后 30 分钟的量价信号最值得看

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

很多人盯盘盯了一整天，其实最该认真看的只有两个时段：开盘 30 分钟和收盘 30 分钟。

开盘 30 分钟反映隔夜消息的定价，这个没什么好分析的——价格已经跳完了。

但尾盘 30 分钟不一样。这段时间的成交行为往往比全天更有信息量：

  • 尾盘放量拉升：可能是主力在收盘前建仓，不想让第二天的开盘价太低。
  • 尾盘放量跳水：可能是机构在出货，或者有利空消息在盘后公布前已经传开。
  • 尾盘突然缩量走平：大概率是多空双方都在等消息，观望情绪浓厚。

问题是：你不可能同时盯着 20 只票的尾盘走势。

AlphaFeed 的 intraday 接口可以拉到任意一只票的分钟线数据，intraday_batch 可以批量拉多只票。拿到分钟线之后，用代码提取尾盘信号，比肉眼盯盘效率高得多。

1. 拉一只票的分钟线数据

from alphafeed import AlphaFeed
import pandas as pd

af = AlphaFeed()

# 当日 1 分钟线
df = af.klines.intraday("600519.SH", to_dataframe=True)
print(f"共 {len(df)} 根分钟线")
print(df[["trade_time", "open", "high", "low", "close", "volume"]].tail(10))

返回的 DataFrame 包含从 9:30 到 15:00 的每一根分钟线。trade_time 字段格式是 2026-07-14 14:51:00，可以直接做时间筛选。

2. 提取尾盘 30 分钟的数据

A 股下午交易时段是 13:00–15:00，尾盘 30 分钟就是 14:30–15:00：

from alphafeed import AlphaFeed
import pandas as pd

af = AlphaFeed()

df = af.klines.intraday("600519.SH", to_dataframe=True)
df["trade_time"] = pd.to_datetime(df["trade_time"])

# 提取尾盘 30 分钟
late_session = df[df["trade_time"].dt.time >= pd.Timestamp("14:30:00").time()].copy()

# 提取非尾盘部分（对比用）
early_session = df[df["trade_time"].dt.time < pd.Timestamp("14:30:00").time()].copy()

print(f"全天分钟线: {len(df)} 根")
print(f"尾盘 30 分钟: {len(late_session)} 根")
print(f"尾盘成交量占全天: {late_session['volume'].sum() / df['volume'].sum():.1%}")

正常情况下尾盘 30 分钟的成交量占全天的 15%–20%。如果超过 25%，就算尾盘放量了。

3. 尾盘信号一：尾盘放量拉升

尾盘价格上涨 + 成交量明显放大，可能意味着有资金在"抢筹"：

from alphafeed import AlphaFeed
import pandas as pd

af = AlphaFeed()

def detect_late_rally(symbol: str) -> dict | None:
    """检测尾盘放量拉升"""
    df = af.klines.intraday(symbol, to_dataframe=True)
    if len(df) < 30:
        return None

    df["trade_time"] = pd.to_datetime(df["trade_time"])
    late = df[df["trade_time"].dt.time >= pd.Timestamp("14:30:00").time()]
    early = df[df["trade_time"].dt.time < pd.Timestamp("14:30:00").time()]

    if len(late) == 0 or len(early) == 0:
        return None

    # 尾盘涨幅 = 尾盘收盘价 vs 14:30 的价格
    late_open = late["open"].iloc[0]
    late_close = late["close"].iloc[-1]
    late_return = (late_close - late_open) / late_open

    # 尾盘成交量占比
    late_vol_pct = late["volume"].sum() / df["volume"].sum() if df["volume"].sum() > 0 else 0

    # 尾盘分钟均量 vs 早盘分钟均量
    late_avg_vol = late["volume"].mean()
    early_avg_vol = early["volume"].mean()
    vol_ratio = late_avg_vol / early_avg_vol if early_avg_vol > 0 else 0

    # 判断条件：尾盘涨 > 0.5% + 尾盘量比 > 1.5 倍
    if late_return > 0.005 and vol_ratio > 1.5:
        return {
            "symbol": symbol,
            "尾盘涨幅": late_return,
            "尾盘量占比": late_vol_pct,
            "尾盘量比": vol_ratio,
            "信号": "尾盘放量拉升 📈",
        }

    return None

# 测试
result = detect_late_rally("600519.SH")
if result:
    print(f"{result['symbol']}: {result['信号']}")
    print(f"  尾盘涨幅: {result['尾盘涨幅']:+.2%}")
    print(f"  尾盘成交量占比: {result['尾盘量占比']:.1%}")
    print(f"  尾盘量比(vs早盘): {result['尾盘量比']:.1f}x")
else:
    print("今天没有尾盘拉升信号")

4. 尾盘信号二：尾盘放量杀跌

反过来的信号——尾盘价格下跌 + 放量，通常是资金离场的信号：

def detect_late_dump(symbol: str) -> dict | None:
    """检测尾盘放量杀跌"""
    df = af.klines.intraday(symbol, to_dataframe=True)
    if len(df) < 30:
        return None

    df["trade_time"] = pd.to_datetime(df["trade_time"])
    late = df[df["trade_time"].dt.time >= pd.Timestamp("14:30:00").time()]
    early = df[df["trade_time"].dt.time < pd.Timestamp("14:30:00").time()]

    if len(late) == 0 or len(early) == 0:
        return None

    late_open = late["open"].iloc[0]
    late_close = late["close"].iloc[-1]
    late_return = (late_close - late_open) / late_open
    late_avg_vol = late["volume"].mean()
    early_avg_vol = early["volume"].mean()
    vol_ratio = late_avg_vol / early_avg_vol if early_avg_vol > 0 else 0

    if late_return < -0.005 and vol_ratio > 1.5:
        return {
            "symbol": symbol,
            "尾盘跌幅": late_return,
            "尾盘量比": vol_ratio,
            "信号": "尾盘放量杀跌 📉",
        }

    return None

5. 尾盘信号三：尾盘突然异动（方向不限）

有时候尾盘的关键不是涨跌，而是波动突然变大——某一分钟成交量是前面均值的 5 倍以上：

def detect_late_spike(symbol: str) -> dict | None:
    """检测尾盘成交量突刺"""
    df = af.klines.intraday(symbol, to_dataframe=True)
    if len(df) < 30:
        return None

    df["trade_time"] = pd.to_datetime(df["trade_time"])

    # 全天分钟均量（不含集合竞价）
    main_session = df[df["trade_time"].dt.time >= pd.Timestamp("09:35:00").time()]
    avg_minute_vol = main_session["volume"].mean()

    if avg_minute_vol == 0:
        return None

    # 找尾盘中成交量最大的那一分钟
    late = df[df["trade_time"].dt.time >= pd.Timestamp("14:30:00").time()]
    if len(late) == 0:
        return None

    max_vol_idx = late["volume"].idxmax()
    max_vol = late.loc[max_vol_idx, "volume"]
    max_vol_time = late.loc[max_vol_idx, "trade_time"]
    max_vol_price = late.loc[max_vol_idx, "close"]

    spike_ratio = max_vol / avg_minute_vol

    if spike_ratio > 5:
        return {
            "symbol": symbol,
            "异动时间": str(max_vol_time),
            "异动量比": spike_ratio,
            "异动价格": max_vol_price,
            "信号": f"尾盘量突刺 ⚡ ({spike_ratio:.0f}x)",
        }

    return None

6. 批量扫描多只票的尾盘信号

重头戏来了。用 intraday_batch 一次拉多只票的分钟线，批量检测尾盘异动：

from alphafeed import AlphaFeed
import pandas as pd

af = AlphaFeed()

watchlist = [
    "600519.SH", "000001.SZ", "300750.SZ", "002594.SZ", "601318.SH",
    "000858.SZ", "600036.SH", "000333.SZ", "601012.SH", "600276.SH",
    "600900.SH", "601398.SH", "600030.SH", "000651.SZ", "002415.SZ",
]

# 一次拉取所有票的分钟线
print(f"正在拉取 {len(watchlist)} 只票的分时数据...")
all_intraday = af.klines.intraday_batch(watchlist, to_dataframe=True)
print(f"拉取完成\n")

signals = []

for sym, df in all_intraday.items():
    if df is None or len(df) < 30:
        continue

    df["trade_time"] = pd.to_datetime(df["trade_time"])
    late = df[df["trade_time"].dt.time >= pd.Timestamp("14:30:00").time()]
    early = df[df["trade_time"].dt.time < pd.Timestamp("14:30:00").time()]

    if len(late) == 0 or len(early) == 0:
        continue

    late_open = late["open"].iloc[0]
    late_close = late["close"].iloc[-1]
    late_return = (late_close - late_open) / late_open
    late_avg_vol = late["volume"].mean()
    early_avg_vol = early["volume"].mean()
    vol_ratio = late_avg_vol / early_avg_vol if early_avg_vol > 0 else 1
    late_vol_pct = late["volume"].sum() / df["volume"].sum() if df["volume"].sum() > 0 else 0

    # 全天涨跌幅
    day_open = df["open"].iloc[0]
    day_close = df["close"].iloc[-1]
    day_return = (day_close - day_open) / day_open

    # 尾盘最大分钟量
    avg_min_vol = df["volume"].mean()
    max_late_vol = late["volume"].max()
    spike = max_late_vol / avg_min_vol if avg_min_vol > 0 else 0

    signal_type = ""
    if late_return > 0.005 and vol_ratio > 1.5:
        signal_type = "尾盘放量拉升 📈"
    elif late_return < -0.005 and vol_ratio > 1.5:
        signal_type = "尾盘放量杀跌 📉"
    elif spike > 5:
        signal_type = f"尾盘量突刺 ⚡"
    elif late_vol_pct > 0.25:
        signal_type = "尾盘成交集中 🔔"

    if signal_type:
        signals.append({
            "代码": sym,
            "信号": signal_type,
            "全天涨跌": f"{day_return:+.2%}",
            "尾盘涨跌": f"{late_return:+.2%}",
            "尾盘量比": f"{vol_ratio:.1f}x",
            "尾盘量占比": f"{late_vol_pct:.0%}",
        })

if signals:
    sdf = pd.DataFrame(signals)
    print(f"=== 尾盘异动信号: {len(sdf)} 只 ===\n")
    print(sdf.to_string(index=False))
else:
    print("今天没有明显的尾盘异动")

7. 尾盘量价分布图：VWAP 偏离度

一个更精细的指标：尾盘成交价相对于全天 VWAP（成交量加权平均价）的偏离程度。

如果尾盘的成交集中在 VWAP 上方，说明资金在"高位"扫货；如果在 VWAP 下方，说明在"低位"出货。

from alphafeed import AlphaFeed
import pandas as pd

af = AlphaFeed()

def late_session_vwap_analysis(symbol: str) -> dict:
    """分析尾盘成交价 vs 全天 VWAP 的关系"""
    df = af.klines.intraday(symbol, to_dataframe=True)
    df["trade_time"] = pd.to_datetime(df["trade_time"])

    # 计算全天 VWAP
    df["turnover"] = df["close"] * df["volume"]
    total_turnover = df["turnover"].sum()
    total_volume = df["volume"].sum()
    vwap = total_turnover / total_volume if total_volume > 0 else df["close"].mean()

    # 尾盘成交量加权平均价
    late = df[df["trade_time"].dt.time >= pd.Timestamp("14:30:00").time()]
    late_turnover = (late["close"] * late["volume"]).sum()
    late_volume = late["volume"].sum()
    late_vwap = late_turnover / late_volume if late_volume > 0 else late["close"].mean()

    deviation = (late_vwap - vwap) / vwap

    return {
        "全天VWAP": vwap,
        "尾盘VWAP": late_vwap,
        "偏离度": deviation,
        "解读": "尾盘在高位成交（偏多）" if deviation > 0.003
                else "尾盘在低位成交（偏空）" if deviation < -0.003
                else "尾盘成交价与VWAP基本一致",
    }

result = late_session_vwap_analysis("600519.SH")
print(f"全天 VWAP: {result['全天VWAP']:.2f}")
print(f"尾盘 VWAP: {result['尾盘VWAP']:.2f}")
print(f"偏离度: {result['偏离度']:+.3%}")
print(f"解读: {result['解读']}")

8. 尾盘信号的实际用法

尾盘信号不应该直接作为买卖依据，但可以作为"次日开盘前"的参考：

尾盘信号 可能含义 次日操作参考
放量拉升 有资金抢筹 如果次日高开，观察是否持续；低开可能是"诱多"
放量杀跌 有资金出逃 次日大概率低开，观望为主
量突刺 有大单成交 关注是买还是卖，看价格方向判断
成交集中 当天博弈激烈 次日波动可能加大
VWAP 偏高 尾盘买盘力量强 正面信号，但需配合日线趋势

9. 每日尾盘扫描脚本

把所有逻辑打包成一个每天 15:05 运行的脚本：

# late_scan.py
"""每日尾盘异动扫描"""

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

af = AlphaFeed()

WATCHLIST = [
    "600519.SH", "000001.SZ", "300750.SZ", "002594.SZ", "601318.SH",
    "000858.SZ", "600036.SH", "000333.SZ", "601012.SH", "600276.SH",
]

def scan():
    today = datetime.now().strftime("%Y-%m-%d")
    print(f"=== 尾盘异动扫描 {today} ===\n")

    # 批量拉分时数据
    all_data = af.klines.intraday_batch(WATCHLIST, to_dataframe=True)

    # 同时拉标的名称
    insts = af.instruments.batch(WATCHLIST)
    name_map = {i["symbol"]: i["name"] for i in insts}

    for sym, df in all_data.items():
        if df is None or len(df) < 30:
            continue

        df["trade_time"] = pd.to_datetime(df["trade_time"])
        late = df[df["trade_time"].dt.time >= pd.Timestamp("14:30:00").time()]
        early = df[df["trade_time"].dt.time < pd.Timestamp("14:30:00").time()]

        if len(late) == 0 or len(early) == 0:
            continue

        late_open = late["open"].iloc[0]
        late_close = late["close"].iloc[-1]
        late_ret = (late_close - late_open) / late_open
        late_avg = late["volume"].mean()
        early_avg = early["volume"].mean()
        vol_ratio = late_avg / early_avg if early_avg > 0 else 1

        name = name_map.get(sym, sym)

        flags = []
        if late_ret > 0.005 and vol_ratio > 1.5:
            flags.append("📈 放量拉升")
        if late_ret < -0.005 and vol_ratio > 1.5:
            flags.append("📉 放量杀跌")
        if late["volume"].max() > df["volume"].mean() * 5:
            flags.append("⚡ 量突刺")

        if flags:
            print(f"  {name}({sym}): {' '.join(flags)}")
            print(f"    尾盘涨跌 {late_ret:+.2%}  量比 {vol_ratio:.1f}x")

    print(f"\n扫描完成")

if __name__ == "__main__":
    scan()

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

10. AlphaFeed 分时接口的优势

做尾盘分析对数据接口有两个硬性要求：

  1. 分钟线数据要完整——从 9:30 到 15:00 每一分钟都有，不能漏。
  2. 批量拉取要快——你可能要同时分析 10–20 只票的分时数据。

AlphaFeed 的 intraday 和 intraday_batch 接口满足这两点：

  • 返回完整的 240 根分钟线（1 分钟周期）或 48 根（5 分钟周期）
  • intraday_batch 一次传多个标的，内部并发，15 只票的分时数据几秒搞定
  • 返回标准 DataFrame，列名一致（trade_time, open, high, low, close, volume），直接用 pandas 分析

如果用爬虫方案拉分时数据，通常需要逐只票请求、手动解析 HTML 或 JSON、处理反爬限制。一只票还好，15 只票就变成了一个工程问题。AlphaFeed 把这个工程问题缩减成了一行代码。

评论