📚 课程目录第4章数据处理器（DataHandler）

4. 数据处理器（DataHandler）

数据处理器（DataHandler）是事件驱动回测引擎的数据中枢，负责向策略模块提供统一、连续、可靠的市场数据。本章将围绕接口抽象、历史数据加载与滚动更新三大核心主题，构建一个可扩展、高性能的数据层。

4.1 数据处理器接口设计

接口设计的核心目标是解耦数据来源与策略逻辑。无论数据来自 CSV 文件、数据库还是实时 API，策略层都应通过同一套方法获取数据。我们定义抽象基类 DataHandler，所有具体实现必须继承并实现以下关键方法：

方法名	功能描述	返回值
`__init__(events, symbol_list, ...)`	初始化数据处理器，注册事件队列与交易标的列表	None
`update_bars()`	推进数据滚动，生成新的市场数据事件	None
`get_latest_bar(symbol)`	获取指定标的最新一根 K 线	Bar 对象或 dict
`get_latest_bars(symbol, N)`	获取指定标的最近 N 根 K 线	list
`get_latest_bar_datetime(symbol)`	获取最新 K 线的时间戳	datetime
`get_latest_bar_value(symbol, field)`	获取最新 K 线的指定字段（如 close、volume）	float / int
`continue_backtest()`	判断回测是否还有未处理的数据	bool

设计原则：所有方法均以 symbol 作为第一参数，确保多标的回测场景下数据访问的清晰性。接口只暴露数据获取方法，不暴露数据存储细节，实现“依赖倒置”。

4.2 历史数据加载：从 CSV 到内存

历史数据加载是回测的起点。以最常见的 CSV 文件为例，我们实现 CSVDataHandler，在初始化阶段完成以下步骤：

文件扫描：根据 symbol_list 在指定目录下查找对应的 CSV 文件。
数据校验：检查必需列（datetime, open, high, low, close, volume）是否存在。
排序与索引：按时间升序排列，并以 datetime 列为索引。
存储至字典：以 {symbol: DataFrame} 的形式存入内存，便于快速访问。

import pandas as pd
from abc import ABC, abstractmethod

class DataHandler(ABC):
    @abstractmethod
    def update_bars(self):
        raise NotImplementedError

    @abstractmethod
    def get_latest_bar(self, symbol):
        raise NotImplementedError

class CSVDataHandler(DataHandler):
    def __init__(self, events, csv_dir, symbol_list):
        self.events = events
        self.csv_dir = csv_dir
        self.symbol_list = symbol_list
        self.symbol_data = {}
        self.latest_symbol_data = {}
        self._load_csv_files()

    def _load_csv_files(self):
        for sym in self.symbol_list:
            file_path = f"{self.csv_dir}/{sym}.csv"
            df = pd.read_csv(file_path, parse_dates=['datetime'])
            df.sort_values('datetime', inplace=True)
            self.symbol_data[sym] = df.iterrows()
            self.latest_symbol_data[sym] = []

避坑提示：CSV 文件中的时间戳必须统一为 UTC 或同一时区，否则多标的数据对齐会出现偏移。建议在加载时统一转换为 pd.Timestamp 并指定时区。

4.3 数据滚动更新机制

滚动更新是事件驱动引擎的“心跳”。每次调用 update_bars()，数据处理器从每个标的的数据迭代器中取出下一根 K 线，封装为 BarEvent 并放入事件队列。核心逻辑如下：

def update_bars(self):
    for sym in self.symbol_list:
        try:
            index, row = next(self.symbol_data[sym])
            bar_event = BarEvent(
                symbol=sym,
                datetime=row['datetime'],
                open=row['open'],
                high=row['high'],
                low=row['low'],
                close=row['close'],
                volume=row['volume']
            )
            self.events.put(bar_event)
            self.latest_symbol_data[sym].append(bar_event)
            # 保持滚动窗口大小，例如只保留最近 100 根
            if len(self.latest_symbol_data[sym]) > 100:
                self.latest_symbol_data[sym].pop(0)
        except StopIteration:
            continue

滚动更新机制需要特别注意以下两点：

时间对齐：所有标的在同一轮 update_bars() 中推进到同一时间戳，避免策略在某一时刻看到 A 股的最新数据而 B 股还停留在上一分钟。
窗口管理：策略通常只需要最近 N 根 K 线计算指标，因此 latest_symbol_data 应维护一个固定长度的滚动窗口，防止内存无限增长。

4.4 数据处理器架构总览

下图展示了数据处理器在回测引擎中的位置及其内部数据流：

4.5 多标的时间对齐策略

当回测涉及多个标的时，不同标的的数据频率可能不一致（例如股票日线 vs 期货分钟线）。时间对齐策略决定了回测的准确性。常见策略对比如下：

策略名称	实现方式	适用场景	优缺点
同步推进	每次 update_bars() 只推进到所有标的中最早的下一个时间戳	标的频率相同或相近	逻辑简单，但低频标的会拖慢高频标的
独立推进	每个标的独立推进，策略按需查询	标的频率差异大	灵活性高，但策略需自行处理时间错位
主时钟推进	以主标的时间为基准，其他标的向前填充	主策略标的明确	实现复杂，但最贴近真实交易

  推荐实践：在回测引擎初期采用“同步推进”策略，代码简洁且不易出错。当需要回测跨频率策略（如日线选股 + 分钟线择时）时，再升级为“主时钟推进”模式。

4.6 性能优化与内存管理

回测时间跨度长、标的数量多时，数据处理器的性能会成为瓶颈。以下是三个关键优化点：

使用迭代器而非一次性加载：对于超长历史数据（如 10 年 tick 级数据），使用 pd.read_csv(chunksize=...) 或自定义迭代器，避免将全部数据读入内存。
滚动窗口剪裁：latest_symbol_data 只保留策略所需的最小窗口长度（如 100 根 K 线），超出部分及时丢弃。
事件对象复用：BarEvent 对象在生成后不再修改，可考虑使用对象池减少 GC 压力。

# 使用 chunksize 分块读取大文件示例
def _load_csv_files_chunked(self):
    for sym in self.symbol_list:
        file_path = f"{self.csv_dir}/{sym}.csv"
        # 只读取必要的列，减少内存占用
        dtypes = {'open': 'f4', 'high': 'f4', 'low': 'f4', 'close': 'f4', 'volume': 'i4'}
        reader = pd.read_csv(file_path, usecols=['datetime', 'open', 'high', 'low', 'close', 'volume'],
                             dtype=dtypes, parse_dates=['datetime'], chunksize=5000)
        # 将每个 chunk 存入临时列表，最后合并排序
        chunks = [chunk for chunk in reader]
        df = pd.concat(chunks, ignore_index=True)
        df.sort_values('datetime', inplace=True)
        self.symbol_data[sym] = df.iterrows()
        self.latest_symbol_data[sym] = []