4. 数据模块（上）：历史行情数据结构设计、OHLCV数据模型、数据加载与清洗

做市商回测，说白了就是拿历史数据模拟交易。数据要是错了，后面全白搭。我见过不少团队，策略逻辑写得花里胡哨，结果数据源有个字段类型搞错了，回测结果直接偏了十万八千里。所以这一章，咱们先把数据模块的根基打牢。

4.1 历史行情数据结构设计

做市商系统跟普通交易系统不一样。普通交易可能只看日线，但做市商得盯 tick 级数据，甚至毫秒级的订单簿快照。为什么？因为做市商赚的是买卖价差和返佣，几毫秒的延迟就可能让一笔交易从盈利变成亏损。

我个人习惯把历史行情数据分成三层：

• 原始层（Raw）：交易所直接推送的原始数据，一股脑全存下来。格式可能是 CSV、Parquet，或者直接存数据库。这层数据不做任何加工，只做归档。
• 标准层（Standard）：把原始数据解析成统一的字段结构。比如把不同交易所的「时间戳」字段都转成统一的 UTC 时间，把「价格」字段都转成浮点数。这层数据是给回测引擎直接用的。
• 特征层（Feature）：基于标准层计算出来的衍生数据。比如 5 档买卖盘的加权平均价、过去 1 秒的成交笔数、买卖价差的波动率等等。这层数据通常由回测引擎在运行时动态计算，但也可以预计算好存下来加速回测。

嗯，这里要注意：很多新手喜欢把原始层和标准层混在一起，结果数据清洗出问题后，原始数据也被污染了，想重新清洗都没办法。我建议你从一开始就严格分层。

4.2 OHLCV 数据模型

OHLCV 是 Open（开盘价）、High（最高价）、Low（最低价）、Close（收盘价）、Volume（成交量）的缩写。这是最基础也是最核心的数据模型。做市商回测虽然需要更细粒度的数据，但 OHLCV 仍然是所有策略的起点。

我设计 OHLCV 模型时，一般会包含以下字段：

字段名	类型	说明
symbol	string	交易对，比如 "BTCUSDT"
timestamp	int64	时间戳，单位毫秒，UTC
open	float64	开盘价
high	float64	最高价
low	float64	最低价
close	float64	收盘价
volume	float64	成交量（以基础币种计）
quote_volume	float64	成交额（以计价币种计）
trades	int32	成交笔数

你可能会问：为什么还要存 quote_volume 和 trades？做市商策略经常需要分析「大单」和「小单」的分布，光看 volume 不够，得结合成交笔数和成交额才能判断市场参与者的行为。我在做某个币种的回测时，发现 volume 很大但 trades 很少，说明有人在拆单吃单，这对做市商的库存管理影响很大。

下面是一个简单的 Python 数据模型定义：

from dataclasses import dataclass
from typing import List

@dataclass
class OHLCV:
    symbol: str
    timestamp: int  # 毫秒级时间戳
    open: float
    high: float
    low: float
    close: float
    volume: float
    quote_volume: float
    trades: int

    def spread(self) -> float:
        """计算该周期的平均价差（简化版）"""
        return (self.high - self.low) / self.close

    def vwap(self) -> float:
        """计算该周期的成交量加权均价"""
        if self.volume == 0:
            return self.close
        return self.quote_volume / self.volume

4.3 数据加载与清洗

数据加载，说白了就是把文件读进内存。但这里有个坑：不同交易所的数据格式千奇百怪。有的用 CSV，有的用 JSON，有的用 MessagePack。我建议你统一封装一个加载器接口，屏蔽底层差异。

我曾经接手过一个项目，数据加载逻辑散落在各个策略文件里，有的用 pd.read_csv，有的用 json.load，还有的直接从数据库拉。结果换数据源的时候，改了一个月还没改完。后来我统一写了一个 DataLoader 类，所有策略都通过它加载数据，问题才解决。

清洗环节，我总结了三个必须做的步骤：

1. 时间戳对齐：不同交易所的时间戳精度不一样，有的到秒，有的到毫秒，有的到微秒。统一转成毫秒级整数，并且检查是否有重复或缺失的时间点。
2. 异常值过滤：价格不能为负，成交量不能为负，最高价不能低于最低价。这些看似基础，但数据源偶尔会抽风。我记得有一次，某交易所的 API 返回了最高价为 0 的数据，导致回测引擎直接除零崩溃。
3. 缺失值处理：如果某个时间点的数据缺失，是向前填充、向后填充，还是直接跳过？这取决于你的策略。做市商策略通常对连续性要求高，我一般用向前填充，然后用一个标志位标记该数据是填充的。

下面是一个清洗函数的示例：

import pandas as pd
import numpy as np

def clean_ohlcv(df: pd.DataFrame) -> pd.DataFrame:
    """
    清洗 OHLCV 数据
    """
    # 1. 时间戳对齐：确保是毫秒级整数
    if df['timestamp'].dtype == 'datetime64[ns]':
        df['timestamp'] = df['timestamp'].astype(np.int64) // 1_000_000

    # 2. 排序并去重
    df = df.sort_values('timestamp').drop_duplicates(subset='timestamp')

    # 3. 异常值过滤
    df = df[(df['high'] >= df['low']) & (df['close'] > 0) & (df['volume'] >= 0)]

    # 4. 缺失值处理：向前填充
    df = df.set_index('timestamp').asfreq('1min', method='ffill').reset_index()

    return df

4.4 知识体系结构图

下面这张图概括了本章的核心逻辑。你可以看到数据从原始层流向标准层，再流向特征层，每一层都有对应的处理步骤。

这张图把数据分层和清洗流程串起来了。你写代码的时候，脑子里要有这张图，知道当前处理的是哪一层，下一步要做什么。

---