用 Python 做日内分时分析：VWAP、量价分布与盘中节奏

大多数量化教程只讲日 K 线。原因很简单：日线数据容易拿、逻辑清晰、回测方便。但只看日线，你会丢掉一个重要维度——盘中发生了什么。

同样是涨 3%，有的票是早盘高开后一路横盘，有的是尾盘最后半小时拉起来的。这两种走势的含义完全不同，但在日 K 线上看起来可能一模一样。

分时数据（分钟线）可以帮你看到日线看不到的东西：盘中量价分布、主力交易时段、VWAP（成交量加权均价）等。AlphaFeed 的 af.klines.intraday() 接口提供当日的分钟级数据，这篇文章会用它做几个实用的日内分析。

1. 获取分钟级分时数据

from alphafeed import AlphaFeed

af = AlphaFeed()

df = af.klines.intraday(
    "600519.SH",
    period="1m",
    to_dataframe=True,
)

df = df.sort_values("trade_time").reset_index(drop=True)
print(f"数据量: {len(df)} 条")
print(df[["trade_time", "open", "high", "low", "close", "volume", "amount"]].head(10))

period 支持 1m、5m、15m、30m、60m，按需选择粒度。1m 是最细的——A 股一个交易日 240 分钟（9:30–11:30、13:00–15:00），所以一天最多 240 条。

也可以批量获取多只股票的分时数据：

symbols = ["600519.SH", "000001.SZ", "300750.SZ"]

intraday_data = af.klines.intraday_batch(
    symbols,
    period="5m",
    to_dataframe=True,
    show_progress=True,
)

for sym, idf in intraday_data.items():
    print(f"{sym}: {len(idf)} 条 5 分钟线")

2. 画出分时走势图

拿到数据后，第一件事是画出来看看：

import pandas as pd
import matplotlib.pyplot as plt
from alphafeed import AlphaFeed

af = AlphaFeed()

df = af.klines.intraday("600519.SH", period="1m", to_dataframe=True)
df = df.sort_values("trade_time").reset_index(drop=True)

fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 8), height_ratios=[3, 1], sharex=True)

ax1.plot(df.index, df["close"], color="#1f77b4", linewidth=1)
ax1.set_ylabel("价格")
ax1.set_title("600519.SH 分时走势")
ax1.grid(True, alpha=0.3)

ax2.bar(df.index, df["volume"], color="#2ca02c", alpha=0.6, width=0.8)
ax2.set_ylabel("成交量")
ax2.set_xlabel("分钟序号")

plt.tight_layout()
plt.savefig("intraday_chart.png", dpi=150)
plt.show()

这张图能直观地告诉你：今天的价格是怎么一步步走出来的，哪些时段成交量集中。

3. 计算 VWAP

VWAP（Volume Weighted Average Price，成交量加权均价）是日内交易中最重要的参考价格之一。机构交易员经常用 VWAP 来评估自己的执行效果——如果你买入的均价低于 VWAP，说明执行得不错。

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

df = af.klines.intraday("600519.SH", period="1m", to_dataframe=True)
df = df.sort_values("trade_time").reset_index(drop=True)

df["cum_amount"] = df["amount"].cumsum()
df["cum_volume"] = df["volume"].cumsum()
df["vwap"] = df["cum_amount"] / df["cum_volume"]

print(f"当日 VWAP: {df['vwap'].iloc[-1]:.2f}")
print(f"最新价:    {df['close'].iloc[-1]:.2f}")
print(f"偏离度:    {(df['close'].iloc[-1] / df['vwap'].iloc[-1] - 1) * 100:.2f}%")

print("\nVWAP 走势（每 30 分钟）:")
print(df.iloc[::30][["trade_time", "close", "vwap"]].to_string(index=False))

VWAP 的几个实际用途：

用途	说明
执行基准	买入均价 < VWAP = 买得便宜
支撑/阻力参考	价格在 VWAP 上方运行通常偏强
日内策略信号	价格上穿/下穿 VWAP 可作为方向信号
大单拆分目标	机构的 VWAP 算法单就是追踪这个价格

4. 成交量分布：钱在什么时候进场

A 股的成交量在一天内分布非常不均匀。一般规律是：开盘前 30 分钟和收盘前 30 分钟成交量最大，中间时段相对冷清。

我们可以量化地看看是不是这样：

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

df = af.klines.intraday("600519.SH", period="5m", to_dataframe=True)
df = df.sort_values("trade_time").reset_index(drop=True)

df["trade_time_str"] = df["trade_time"].astype(str)

total_vol = df["volume"].sum()
df["vol_pct"] = df["volume"] / total_vol * 100

print("=== 各时段成交量占比 ===")
for _, row in df.iterrows():
    bar = "█" * int(row["vol_pct"] * 2)
    print(f"{row['trade_time_str'][-8:]}  {row['vol_pct']:5.1f}%  {bar}")

如果你发现某只票今天的成交量分布和往常不一样——比如通常尾盘量最大，但今天早盘就放了巨量——这可能意味着有大资金在抢跑。

对比多只股票的量分布

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

symbols = ["600519.SH", "000001.SZ", "300750.SZ"]

intraday_data = af.klines.intraday_batch(symbols, period="30m", to_dataframe=True)

for sym, idf in intraday_data.items():
    idf = idf.sort_values("trade_time").reset_index(drop=True)
    total = idf["volume"].sum()
    idf["vol_pct"] = idf["volume"] / total * 100

    first_30min = idf["vol_pct"].iloc[0] if len(idf) > 0 else 0
    last_30min = idf["vol_pct"].iloc[-1] if len(idf) > 0 else 0

    print(f"{sym}: 首30min={first_30min:.1f}%  尾30min={last_30min:.1f}%  "
          f"首尾合计={first_30min + last_30min:.1f}%")

5. 日内波动率分析

日线的波动率告诉你"这只票每天大概波动多少"。分钟线的波动率告诉你"这只票在一天之内什么时候波动最剧烈"。

import pandas as pd
import numpy as np
from alphafeed import AlphaFeed

af = AlphaFeed()

df = af.klines.intraday("600519.SH", period="5m", to_dataframe=True)
df = df.sort_values("trade_time").reset_index(drop=True)

df["ret"] = df["close"].pct_change()

df["minute_index"] = range(len(df))

df["rolling_vol"] = df["ret"].rolling(6).std() * np.sqrt(48)

print("=== 日内波动率变化 ===")
print(df[["trade_time", "close", "ret", "rolling_vol"]].dropna().to_string(index=False))

high_vol_periods = df[df["rolling_vol"] > df["rolling_vol"].quantile(0.8)].copy()
print(f"\n高波动时段 (Top 20%): {len(high_vol_periods)} 个 5 分钟 bar")
if not high_vol_periods.empty:
    print(high_vol_periods[["trade_time", "close", "rolling_vol"]].to_string(index=False))

这个分析的价值在于：如果你知道某只票通常在 10:00–10:30 波动率最高，那你的限价单策略就应该在这个时段更谨慎地设置挂单价格。

6. 日内动量与反转

日线上存在"动量效应"（涨了还会涨），日内也有类似的规律，但方向可能不同。很多研究发现，A 股日内存在"上午动量、下午反转"的倾向：

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

df = af.klines.intraday("600519.SH", period="1m", to_dataframe=True)
df = df.sort_values("trade_time").reset_index(drop=True)

df["trade_time_str"] = df["trade_time"].astype(str)

morning = df[df["trade_time_str"].str.contains(r"09:|10:|11:")].copy()
afternoon = df[df["trade_time_str"].str.contains(r"13:|14:")].copy()

if not morning.empty and not afternoon.empty:
    morning_ret = morning["close"].iloc[-1] / morning["open"].iloc[0] - 1
    afternoon_ret = afternoon["close"].iloc[-1] / afternoon["open"].iloc[0] - 1
    full_day_ret = df["close"].iloc[-1] / df["open"].iloc[0] - 1

    print(f"上午收益: {morning_ret:+.4f} ({morning_ret*100:+.2f}%)")
    print(f"下午收益: {afternoon_ret:+.4f} ({afternoon_ret*100:+.2f}%)")
    print(f"全天收益: {full_day_ret:+.4f} ({full_day_ret*100:+.2f}%)")

    if morning_ret > 0 and afternoon_ret < 0:
        print("→ 今日呈现上午涨、下午回落的模式")
    elif morning_ret < 0 and afternoon_ret > 0:
        print("→ 今日呈现上午跌、下午反弹的模式")
    elif morning_ret > 0 and afternoon_ret > 0:
        print("→ 今日上下午均上涨，趋势延续")
    else:
        print("→ 今日上下午均下跌")

要做更严谨的研究，你需要用历史分钟线（而不只是今天一天）来统计这种上下午的关系。可以每天收盘后用日 K 和分时数据积累样本。

7. VWAP 偏离度选股

把 VWAP 分析从单只票扩展到多只票，可以做一个简单的日内选股信号：当前价格低于 VWAP 且差距较大的票，可能存在日内反弹机会（当然也可能是真的在跌）。

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

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

intraday_data = af.klines.intraday_batch(
    watchlist,
    period="1m",
    to_dataframe=True,
    show_progress=True,
)

vwap_results = []
for sym, idf in intraday_data.items():
    idf = idf.sort_values("trade_time").reset_index(drop=True)
    if len(idf) < 10:
        continue

    cum_amount = idf["amount"].sum()
    cum_volume = idf["volume"].sum()
    vwap = cum_amount / cum_volume if cum_volume > 0 else 0

    last_price = idf["close"].iloc[-1]
    deviation = (last_price / vwap - 1) * 100 if vwap > 0 else 0

    vwap_results.append({
        "symbol": sym,
        "last_price": last_price,
        "vwap": round(vwap, 2),
        "deviation_pct": round(deviation, 2),
    })

vwap_df = pd.DataFrame(vwap_results).sort_values("deviation_pct")

print("=== VWAP 偏离度排行 ===")
print("正值 = 当前价 > VWAP（偏贵），负值 = 当前价 < VWAP（偏便宜）")
print(vwap_df.to_string(index=False))

VWAP 偏离度不是一个独立的交易信号——你不能仅仅因为"低于 VWAP"就买入。但它可以作为一个辅助指标：在你已经看好某只票的前提下，等价格回到 VWAP 附近或以下再买入，执行效果通常更好。

8. 盘中量价异动检测

结合分时数据，可以做一个简单的"放量异动"检测器：如果某个 5 分钟 bar 的成交量是当日平均的 3 倍以上，标记为异动。

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

def detect_volume_spike(symbol: str, threshold: float = 3.0) -> list:
    df = af.klines.intraday(symbol, period="5m", to_dataframe=True)
    df = df.sort_values("trade_time").reset_index(drop=True)

    if len(df) < 5:
        return []

    avg_vol = df["volume"].mean()
    spikes = []

    for _, row in df.iterrows():
        if avg_vol > 0 and row["volume"] > avg_vol * threshold:
            ret = (row["close"] - row["open"]) / row["open"] * 100 if row["open"] > 0 else 0
            spikes.append({
                "time": str(row["trade_time"]),
                "volume": row["volume"],
                "vol_ratio": round(row["volume"] / avg_vol, 1),
                "bar_return": f"{ret:+.2f}%",
                "close": row["close"],
            })

    return spikes

symbols = ["600519.SH", "000001.SZ", "300750.SZ", "002594.SZ"]

for sym in symbols:
    spikes = detect_volume_spike(sym, threshold=3.0)
    if spikes:
        print(f"\n[{sym}] 检测到 {len(spikes)} 次放量异动:")
        for s in spikes:
            print(f"  {s['time']} | 量比={s['vol_ratio']}x | 涨跌={s['bar_return']} | 价格={s['close']}")
    else:
        print(f"\n[{sym}] 今日无明显放量异动")

这种检测可以帮你捕捉"突然来了一笔大单"的时刻。结合盘口数据（第 10 篇），你可以进一步判断这笔量是买入还是卖出。

9. 历史分钟线的积累与分析

af.klines.intraday() 返回的是当天的分时数据。如果你想积累历史分钟线，可以每天定时存档：

import json
from datetime import datetime
from alphafeed import AlphaFeed

af = AlphaFeed()

def save_daily_intraday(symbols: list, period: str = "5m"):
    today = datetime.now().strftime("%Y%m%d")

    intraday_data = af.klines.intraday_batch(
        symbols, period=period, to_dataframe=True, show_progress=True,
    )

    for sym, idf in intraday_data.items():
        filename = f"intraday_{sym.replace('.', '_')}_{today}.csv"
        idf.to_csv(filename, index=False)
        print(f"已保存 {filename} ({len(idf)} 条)")

save_daily_intraday(["600519.SH", "000001.SZ", "300750.SZ"])

积累两三周的数据后，你就可以做更有意义的统计分析：平均量价分布、日内波动率季节性、VWAP 偏离后的价格回归概率等。

如果你需要更长时间的历史分钟线，也可以用 af.klines.get() 配合分钟级 period 来获取：

df_5m = af.klines.get(
    "600519.SH",
    period="5m",
    count=1000,
    adjust="forward",
    to_dataframe=True,
)
print(f"历史 5 分钟线: {len(df_5m)} 条")
print(f"起始: {df_5m['trade_time'].min()}")
print(f"截止: {df_5m['trade_time'].max()}")

10. 分时数据的局限性

局限	说明
数据量大	1 分钟线一天 240 条，100 只票一年就是 500 多万条，存储和计算都更重
噪声更多	分钟级别的价格波动大部分是随机噪声，信号提取更难
回测复杂	需要处理开盘集合竞价、午休、收盘等特殊时段
交易成本放大	日内策略交易频率高，手续费和滑点的影响比日线策略大得多
不适合新手入门	建议先在日线层面建立完整的研究流程，再下沉到分钟级

分时数据不是"更好的数据"，而是"另一个维度的数据"。它适合回答日线无法回答的问题：盘中什么时候最活跃？大单在什么价位进场？VWAP 偏离多少是正常范围？

结语

日 K 线是量化的基础视角，分时数据是进阶视角。

用 AlphaFeed 的 af.klines.intraday() 拿到分钟级数据后，你可以计算 VWAP、分析盘中量价分布、检测放量异动、研究日内动量与反转规律。这些分析不会直接给你一个"必赚"的信号，但它们会帮你理解：一根日 K 线的背后，盘中到底发生了什么。

对于有一定基础的量化研究者来说，分时数据是从"日线策略"迈向"更精细执行"和"更深入市场微观结构理解"的关键一步。

相关链接：

• AlphaFeed 官网：https://alphafeed.org/
• Python SDK 快速开始：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart