第三章 VWAP数据准备：获取历史Tick数据与分钟级K线数据，数据清洗与预处理实战

做VWAP策略，数据是地基。地基不稳，楼盖得再漂亮也得塌。

我见过太多人，一上来就调参数、写策略，结果回测曲线漂亮得像假的一样——后来一查，数据里全是坑。嗯，今天我们就来聊聊，怎么把数据这关给过了。

3.1 数据源的选择：Tick vs 分钟K线

先搞清楚我们要什么。VWAP的核心是「成交量加权」，所以你需要两样东西：价格和成交量。

数据类型	特点	适用场景
Tick数据	每笔成交记录，最精细	高频VWAP、日内T+0
分钟K线	聚合后的OHLCV	标准VWAP、中低频策略

我个人习惯：做回测用分钟K线，做实盘用Tick数据。为什么？回测时Tick数据量太大，动不动几个G，跑一次要半天。实盘时精度不够，VWAP算出来会偏。

3.2 获取Tick数据实战

假设我们从某数据商拿到CSV格式的Tick数据。结构大概长这样：

# tick_data_sample.csv
timestamp,price,volume,bid,ask
2024-01-02 09:30:00.123,10.25,100,10.24,10.26
2024-01-02 09:30:00.456,10.26,200,10.25,10.27
2024-01-02 09:30:00.789,10.24,150,10.23,10.25
...

读进来很简单，但坑在后面。

import pandas as pd

df_tick = pd.read_csv('tick_data_sample.csv', parse_dates=['timestamp'])
df_tick.set_index('timestamp', inplace=True)
print(df_tick.head())

你看，就这么几行。但我在项目中遇到过一个问题：时间戳精度不一致。有的数据商给到毫秒，有的给到微秒，还有的干脆只给到秒。合并多个数据源时，时间轴对不上，VWAP算出来全是错的。

3.3 分钟K线数据构建

Tick数据太细，我们通常要聚合到分钟级。这里有个关键点：成交量要累加，价格要加权平均。

# 将Tick数据聚合为1分钟K线
df_minute = df_tick.resample('1T').agg({
    'price': 'ohlc',        # 生成开盘、最高、最低、收盘
    'volume': 'sum',        # 成交量累加
    'bid': 'last',          # 最后一笔买价
    'ask': 'last'           # 最后一笔卖价
})

# 展平多级列索引
df_minute.columns = ['_'.join(col).strip() for col in df_minute.columns.values]
df_minute.rename(columns={
    'price_open': 'open',
    'price_high': 'high',
    'price_low': 'low',
    'price_close': 'close',
    'volume_sum': 'volume'
}, inplace=True)

print(df_minute.head())

说白了，这就是把一堆零散的成交记录，压缩成一根根K线。你想想看，如果直接用Tick数据算VWAP，计算量会爆炸。聚合到分钟级，既保留了日内波动特征，又大幅降低了数据量。

3.4 数据清洗：那些年我们踩过的坑

数据清洗是VWAP准备中最枯燥、但最重要的一步。我总结了三个必做动作：

3.4.1 缺失值处理

分钟K线里经常出现空值——比如某分钟没有成交。怎么办？

• 成交量缺失：直接填0。VWAP计算时，成交量=0的分钟不贡献权重。
• 价格缺失：用前一根K线的收盘价填充（向前填充）。
• 连续缺失超过5分钟：建议标记为异常，不要强行填充。

# 缺失值处理
df_minute['volume'].fillna(0, inplace=True)
df_minute['close'].fillna(method='ffill', inplace=True)
df_minute['open'].fillna(method='ffill', inplace=True)

# 检查连续缺失
null_streaks = df_minute['close'].isna().astype(int).groupby(
    df_minute['close'].notna().cumsum()
).cumsum()
if null_streaks.max() > 5:
    print("警告：存在连续5分钟以上缺失数据，建议检查数据源")

3.4.2 异常值检测

价格突然跳空、成交量突然暴增——这些可能是数据错误，也可能是真实行情。怎么区分？

我习惯用Z-score方法：

from scipy import stats

# 对价格变化率做Z-score检测
df_minute['return'] = df_minute['close'].pct_change()
z_scores = stats.zscore(df_minute['return'].dropna())
outliers = abs(z_scores) > 3

print(f"检测到异常价格变动：{outliers.sum()} 条")

嗯，这里要注意：不要直接删除异常值。我曾经删了一堆「异常」数据，结果发现那是真实的闪崩行情。正确的做法是：标记出来，人工复核。

3.4.3 时间对齐

不同交易所、不同品种的交易时间不一样。A股是9:30-15:00，中间有午休。期货有夜盘。如果你把全天数据混在一起算VWAP，结果会偏。

# 过滤A股交易时段（9:30-11:30, 13:00-15:00）
trading_mask = (
    (df_minute.index.time >= pd.Timestamp('09:30').time()) &
    (df_minute.index.time <= pd.Timestamp('11:30').time())
) | (
    (df_minute.index.time >= pd.Timestamp('13:00').time()) &
    (df_minute.index.time <= pd.Timestamp('15:00').time())
)

df_minute_clean = df_minute[trading_mask].copy()

3.5 知识体系总览

下面这张图，是我做VWAP数据准备时的完整流程。你可以把它当成一个检查清单：

3.6 实战：完整的数据预处理Pipeline

最后，我把上面所有步骤打包成一个函数。你以后做VWAP，直接调这个就行：

def prepare_vwap_data(tick_file, freq='1T'):
    """
    从Tick数据到VWAP可用数据的完整Pipeline
    
    参数:
        tick_file: Tick数据CSV路径
        freq: 聚合频率，默认1分钟
    
    返回:
        df_clean: 清洗后的分钟K线DataFrame
    """
    # 1. 读取数据
    df = pd.read_csv(tick_file, parse_dates=['timestamp'])
    df.set_index('timestamp', inplace=True)
    
    # 2. 聚合到分钟K线
    df_min = df.resample(freq).agg({
        'price': 'ohlc',
        'volume': 'sum'
    })
    df_min.columns = ['open', 'high', 'low', 'close', 'volume']
    
    # 3. 缺失值处理
    df_min['volume'].fillna(0, inplace=True)
    df_min['close'].fillna(method='ffill', inplace=True)
    df_min['open'].fillna(method='ffill', inplace=True)
    
    # 4. 异常值标记（不删除）
    df_min['return'] = df_min['close'].pct_change()
    z_scores = stats.zscore(df_min['return'].dropna())
    df_min['is_outlier'] = False
    df_min.loc[df_min['return'].notna(), 'is_outlier'] = abs(z_scores) > 3
    
    # 5. 过滤交易时段
    morning = (df_min.index.time >= pd.Timestamp('09:30').time()) & \
              (df_min.index.time <= pd.Timestamp('11:30').time())
    afternoon = (df_min.index.time >= pd.Timestamp('13:00').time()) & \
                (df_min.index.time <= pd.Timestamp('15:00').time())
    df_clean = df_min[morning | afternoon].copy()
    
    return df_clean

# 使用示例
df_vwap_ready = prepare_vwap_data('tick_data_sample.csv')
print(f"数据准备完成，共 {len(df_vwap_ready)} 根K线")

数据准备好了，下一步就是真正的VWAP计算了。不过那是下一章的事。今天先把数据关过了，后面就顺了。

---