4. 数据获取与处理：期货行情数据获取（tick/bar）、数据清洗与对齐、数据存储与加载

做量化交易，数据就是你的弹药库。弹药不行，枪法再好也白搭。

这一章，咱们聊聊怎么搞到靠谱的期货行情数据，怎么把脏数据洗干净，怎么让不同周期的数据对齐，最后怎么存起来方便以后用。说白了，就是一套从「生数据」到「熟数据」的完整流水线。

4.1 期货行情数据获取：tick 与 bar

期货数据分两种：tick 和 bar。我刚开始做回测时，总觉得用日线就够了。后来发现，做日内策略没有 tick 数据，简直像盲人摸象。

4.1.1 Tick 数据

Tick 数据是交易所推送的最小粒度数据。每一笔成交、每一笔挂单变动，都会产生一个 tick。国内期货的 tick 频率大概是 0.5 秒一次，但行情剧烈时可能一秒钟来好几笔。

一个标准的 tick 包含以下字段：

4.1.2 Bar 数据

Bar 数据是对 tick 数据的聚合。常见的周期有 1 分钟、5 分钟、15 分钟、1 小时、日线等。每个 bar 包含四个核心价格：开盘价、最高价、最低价、收盘价，外加成交量和持仓量。

在 vn.py 里，获取 bar 数据很简单。你可以用 CtaTemplate 自带的 on_bar 函数，也可以自己从 tick 合成 bar。我个人更推荐后者——因为自己合成的 bar，你能控制细节。

# 从 tick 合成 1 分钟 bar 的示例
class BarGenerator:
    def __init__(self, on_bar):
        self.on_bar = on_bar
        self.bar = None
        self.last_tick = None

    def update_tick(self, tick):
        if not self.bar:
            self.bar = BarData(
                symbol=tick.symbol,
                exchange=tick.exchange,
                datetime=tick.datetime.replace(second=0, microsecond=0),
                open_price=tick.last_price,
                high_price=tick.last_price,
                low_price=tick.last_price,
                close_price=tick.last_price,
                volume=tick.volume,
                open_interest=tick.open_interest
            )
        else:
            self.bar.high_price = max(self.bar.high_price, tick.last_price)
            self.bar.low_price = min(self.bar.low_price, tick.last_price)
            self.bar.close_price = tick.last_price
            self.bar.volume += tick.volume - self.last_tick.volume
            self.bar.open_interest = tick.open_interest

        self.last_tick = tick

        # 检查是否跨分钟
        if tick.datetime.minute != self.bar.datetime.minute:
            self.on_bar(self.bar)
            self.bar = None

4.2 数据清洗与对齐

数据拿到手，别急着用。先洗一洗。你想想看，交易所的数据偶尔也会抽风——比如某根 tick 的价格突然跳到了 99999，或者某分钟 bar 的成交量变成了负数。这些脏数据不处理，回测结果就是垃圾。

4.2.1 常见的数据问题

• 缺失值： 某个时间段没有数据，比如夜盘休市、节假日、或者网络中断。
• 异常值： 价格或成交量明显偏离正常范围。比如螺纹钢突然出现 10000 元/吨的成交。
• 重复值： 同一时间戳出现多条相同的数据。
• 时间戳错乱： 数据的时间顺序不对，后面的 tick 时间戳比前面的还早。

4.2.2 清洗流程

我一般按这个顺序处理：

1. 去重： 按 datetime 和 last_price 去重，保留第一条。
2. 排序： 按 datetime 升序排列。
3. 过滤异常： 价格超出 [前收盘价 * 0.9, 前收盘价 * 1.1] 范围的，标记为异常。
4. 填充缺失： 对于 bar 数据，缺失的 bar 用前一根 bar 的收盘价填充。

def clean_tick_data(df):
    # 1. 去重
    df = df.drop_duplicates(subset=['datetime', 'last_price'])

    # 2. 排序
    df = df.sort_values('datetime')

    # 3. 过滤异常价格
    prev_close = df['last_price'].iloc[0]
    df = df[
        (df['last_price'] >= prev_close * 0.9) &
        (df['last_price'] <= prev_close * 1.1)
    ]

    # 4. 重置索引
    df = df.reset_index(drop=True)
    return df

4.2.3 数据对齐

做多品种策略时，不同合约的交易时间可能不一样。比如螺纹钢有夜盘，但有些品种没有。这时候就需要对齐。

对齐的核心思路是：以最长的交易时间段为基准，其他品种缺失的时间段用 NaN 或前值填充。

def align_bars(bar_dict):
    """
    bar_dict: {symbol: DataFrame}
    返回对齐后的 DataFrame
    """
    # 找到所有时间戳的并集
    all_timestamps = set()
    for df in bar_dict.values():
        all_timestamps.update(df['datetime'])

    all_timestamps = sorted(all_timestamps)

    # 对每个品种，按时间戳对齐
    aligned_data = {'datetime': all_timestamps}
    for symbol, df in bar_dict.items():
        df = df.set_index('datetime')
        df = df.reindex(all_timestamps, method='ffill')
        aligned_data[symbol] = df['close_price'].values

    return pd.DataFrame(aligned_data)

4.3 数据存储与加载

数据洗好了，得找个地方存起来。总不能每次回测都从交易所重新下载吧？那太慢了。

4.3.1 存储格式选择

我个人习惯用 HDF5。原因很简单：vn.py 原生支持，读写速度够用，而且一个文件就能存多个品种的数据。

4.3.2 使用 vn.py 的数据库接口

vn.py 提供了统一的数据库接口，你不需要关心底层是 HDF5 还是 SQLite。只需要配置一下就行。

# 在 vn.py 中配置 HDF5 数据库
from vnpy.trader.database import database_manager
from vnpy.trader.constant import Exchange, Interval

# 保存数据
database_manager.save_bar_data(
    bars,
    symbol='rb2410',
    exchange=Exchange.SHFE,
    interval=Interval.MINUTE
)

# 加载数据
bars = database_manager.load_bar_data(
    symbol='rb2410',
    exchange=Exchange.SHFE,
    interval=Interval.MINUTE,
    start=datetime(2024, 1, 1),
    end=datetime(2024, 5, 20)
)

4.3.3 数据加载的注意事项

• 缓存机制： 如果同一个数据在回测中会被多次使用，建议加载后缓存到内存里。我一般用 lru_cache 装饰器。
• 数据类型： 加载时注意把 datetime 列转成 pandas 的 datetime 类型，否则后面做时间切片时会报错。
• 内存管理： 如果加载的数据量太大（比如几年的 tick 数据），建议分块加载，或者用 Dask 等工具做延迟计算。

from functools import lru_cache

@lru_cache(maxsize=10)
def load_bars(symbol, interval, start, end):
    """带缓存的 bar 数据加载"""
    bars = database_manager.load_bar_data(
        symbol=symbol,
        exchange=Exchange.SHFE,
        interval=interval,
        start=start,
        end=end
    )
    return bars

4.4 本章知识体系

下面这张图，把数据获取与处理的整个流程串起来了。你可以把它当作一个操作手册，每次做数据准备时对照着来。

嗯，数据这块就聊到这儿。记住一句话：数据质量决定了回测的上限，策略模型只是去逼近这个上限。 把数据搞扎实了，后面的工作才能事半功倍。