4. Tick级数据与OHLCV数据

高频交易的第一手数据源是逐笔成交明细（Tick数据），而策略回测与信号计算通常依赖压缩后的OHLCV数据。本章深入讲解Tick数据的格式与存储方案、OHLCV的生成逻辑，以及数据压缩与对齐的核心技术。

4.1 Tick数据格式与存储

Tick数据记录每一笔成交的详细信息，是高频交易最细粒度的数据源。不同交易所的Tick格式略有差异，但核心字段一致。

4.2 OHLCV数据生成

OHLCV（Open, High, Low, Close, Volume）是将Tick数据按固定时间窗口聚合后的标准金融数据形态。生成过程本质上是时间序列的降采样与统计。

import pandas as pd
import numpy as np

def tick_to_ohlcv(tick_df, freq='1min'):
    """
    将Tick级DataFrame转换为OHLCV数据
    tick_df: 必须包含 'price', 'volume', 'timestamp' 列
    freq: 聚合频率，如 '1min', '5min', '1s'
    """
    # 确保时间戳为索引
    tick_df = tick_df.set_index('timestamp').sort_index()
    
    # 使用resample进行聚合
    ohlcv = tick_df['price'].resample(freq).ohlc()
    ohlcv['volume'] = tick_df['volume'].resample(freq).sum()
    
    # 处理空值：若窗口内无成交，前向填充价格，成交量置0
    ohlcv.fillna(method='ffill', inplace=True)
    ohlcv['volume'].fillna(0, inplace=True)
    
    return ohlcv

4.3 数据压缩与对齐

高频策略常涉及多标的、多频率的数据源，必须进行严格的时间对齐与数据压缩，避免未来函数与数据泄露。

4.3.1 数据压缩策略

4.3.2 多标的时间对齐

当策略同时使用股票A的1分钟K线和股票B的Tick数据时，必须将两者对齐到统一时间轴。常用方法为向前填充（Forward Fill）与线性插值。

def align_multiple_assets(tick_dict, freq='1min'):
    """
    将多个资产的Tick数据对齐到统一OHLCV时间轴
    tick_dict: dict {asset_name: tick_df}
    """
    aligned = {}
    master_index = None
    
    # 先生成主时间轴（取所有资产时间范围的并集）
    for asset, df in tick_dict.items():
        ohlcv = tick_to_ohlcv(df, freq)
        if master_index is None:
            master_index = ohlcv.index
        else:
            master_index = master_index.union(ohlcv.index)
    
    # 对齐每个资产
    for asset, df in tick_dict.items():
        ohlcv = tick_to_ohlcv(df, freq)
        # 重采样到主时间轴，前向填充
        aligned[asset] = ohlcv.reindex(master_index).fillna(method='ffill')
        # 成交量缺失置0
        aligned[asset]['volume'].fillna(0, inplace=True)
    
    return aligned

4.4 实战：Tick数据存储与OHLCV生成

以下代码演示完整的Tick数据存储（Parquet格式）与OHLCV生成流水线，可直接用于高频策略的数据准备阶段。

import pandas as pd
import os
from pathlib import Path

class TickDataPipeline:
    """Tick数据存储与OHLCV生成流水线"""
    
    def __init__(self, data_dir='./tick_data'):
        self.data_dir = Path(data_dir)
        self.data_dir.mkdir(parents=True, exist_ok=True)
    
    def save_tick_parquet(self, tick_df, symbol, date):
        """将Tick数据存储为Parquet，按日期分区"""
        file_path = self.data_dir / f"{symbol}_{date}.parquet"
        tick_df.to_parquet(file_path, compression='snappy')
        return file_path
    
    def load_tick_parquet(self, symbol, date):
        """加载指定日期的Tick数据"""
        file_path = self.data_dir / f"{symbol}_{date}.parquet"
        if file_path.exists():
            return pd.read_parquet(file_path)
        else:
            raise FileNotFoundError(f"数据文件不存在: {file_path}")
    
    def generate_daily_ohlcv(self, symbol, date, freq='1min'):
        """生成并存储日内的OHLCV数据"""
        tick_df = self.load_tick_parquet(symbol, date)
        ohlcv = tick_to_ohlcv(tick_df, freq)
        
        # 添加日期列便于后续查询
        ohlcv['date'] = date
        ohlcv['symbol'] = symbol
        
        # 存储OHLCV
        ohlcv_path = self.data_dir / f"{symbol}_{date}_ohlcv_{freq}.parquet"
        ohlcv.to_parquet(ohlcv_path, compression='snappy')
        return ohlcv

# 使用示例
pipeline = TickDataPipeline()
# 假设已有tick_data DataFrame
# pipeline.save_tick_parquet(tick_data, 'AAPL', '2024-01-15')
# ohlcv_1min = pipeline.generate_daily_ohlcv('AAPL', '2024-01-15', '1min')