2. 回测框架设计原则：模块化设计、事件驱动架构、数据与策略分离

做回测框架，说白了就是在搭积木。我见过太多人一上来就写个几百行的脚本，把所有逻辑揉在一起。结果呢？换个策略就得重写一半代码，换个数据源更是要命。今天咱们聊聊三个核心原则，这些都是我在实战中踩过坑之后总结出来的。

2.1 模块化设计：拆开才能复用

模块化设计，就是把系统拆成独立的功能块。每个模块只干一件事，干好一件事。我个人习惯把回测框架拆成这几个核心模块：

• 数据模块：负责获取、清洗、存储行情数据
• 策略模块：只负责生成交易信号，不关心数据从哪来
• 风控模块：检查订单是否合规，比如仓位限制、价格偏离
• 执行模块：模拟撮合，生成成交记录
• 分析模块：计算收益、夏普比率、最大回撤等指标

为什么要这么拆？我在项目中遇到过一件事：有个同事把数据清洗逻辑直接写在了策略里。后来换了数据源，格式不一样，他得改策略代码。你说这多麻烦？模块化之后，你只需要换数据模块，策略完全不用动。

2.2 事件驱动架构：让系统自己动起来

事件驱动架构，听起来高大上，其实很简单。就是系统里发生了一件事，然后自动触发后续的处理流程。比如：

• 新来了一个tick数据 → 触发策略计算
• 策略生成了信号 → 触发风控检查
• 风控通过了 → 触发订单执行
• 订单成交了 → 触发持仓更新

你想想看，如果不用事件驱动，你得写一堆if-else来判断「现在该干啥」。代码会变得又臭又长。事件驱动的好处是，每个模块只关心自己感兴趣的事件，收到事件就干活，干完活就发布新事件。

为什么会这样？因为事件驱动让系统的扩展性变得极好。你想加个日志模块？简单，让它订阅所有事件就行。你想加个实时监控？同样，订阅订单事件和持仓事件。完全不用改现有代码。

2.3 数据与策略分离：别让数据绑架策略

这个原则，我愿称之为回测框架的「黄金法则」。数据是数据，策略是策略，两者必须彻底分开。什么意思呢？

• 策略不应该知道数据是从CSV来的，还是从数据库来的，还是从API实时拉的
• 策略不应该关心数据是1分钟K线还是5分钟K线，它只关心「给我一个价格序列」
• 数据模块不应该知道策略在用什么指标，它只管提供干净的数据

我曾经犯过一个错误：在策略里直接写了数据文件的路径。后来数据文件挪了个位置，整个策略跑不起来了。你说冤不冤？

数据与策略分离，还有一个好处：你可以轻松地做「数据回放」。比如用历史数据模拟实时行情，测试策略在「如果当时发生了这件事」会怎么反应。这个能力在实盘前验证策略非常有用。

2.4 三个原则如何协同工作

这三个原则不是孤立的，它们互相配合。模块化设计提供了结构，事件驱动架构提供了流程，数据与策略分离提供了灵活性。我画了一张图，帮你理解它们之间的关系：

从这张图你能看到，所有模块都围绕着事件总线工作。数据模块发布行情事件，策略模块订阅行情事件并发布订单事件，风控模块订阅订单事件... 每个模块都是独立的，替换任何一个都不会影响其他模块。

2.5 代码示例：一个极简的事件驱动框架

说了这么多理论，咱们来点实际的。下面是一个极简的事件驱动框架代码，你可以在自己的项目里直接用：

from enum import Enum
from dataclasses import dataclass
from typing import Callable, Dict, List
from collections import defaultdict

class EventType(Enum):
    TICK = "tick"          # 行情数据
    SIGNAL = "signal"      # 交易信号
    ORDER = "order"        # 订单
    FILL = "fill"          # 成交
    POSITION = "position"  # 持仓变化

@dataclass
class Event:
    event_type: EventType
    data: dict
    timestamp: float

class EventBus:
    def __init__(self):
        self._subscribers: Dict[EventType, List[Callable]] = defaultdict(list)
    
    def subscribe(self, event_type: EventType, handler: Callable):
        """订阅事件"""
        self._subscribers[event_type].append(handler)
    
    def publish(self, event: Event):
        """发布事件"""
        for handler in self._subscribers[event.event_type]:
            handler(event)

# 使用示例
bus = EventBus()

def on_tick(event: Event):
    print(f"收到行情: {event.data}")

def on_signal(event: Event):
    print(f"收到信号: {event.data}")

bus.subscribe(EventType.TICK, on_tick)
bus.subscribe(EventType.SIGNAL, on_signal)

# 模拟数据模块发布行情
bus.publish(Event(EventType.TICK, {"price": 50000, "volume": 1.5}, 1234567890.0))

2.6 三个原则的落地检查清单

最后，我整理了一个检查清单。你在设计回测框架时，可以对照着看看自己有没有做到位：

原则	检查项	常见反例
模块化设计	每个模块是否只有一个职责？ 模块间是否通过接口通信？ 替换一个模块是否需要改其他模块？	策略代码里直接读写数据库
事件驱动架构	是否使用事件总线？ 模块是否只通过事件交互？ 事件类型是否清晰定义？	用全局变量传递订单状态
数据与策略分离	策略是否不关心数据来源？ 数据格式变化是否不影响策略？ 是否通过配置而非硬编码指定数据源？	策略里写死了CSV文件路径

嗯，这三个原则说起来简单，做起来需要一点经验积累。但只要你坚持用这套思路去设计，你的回测框架会越来越健壮，改起来也越来越轻松。我在做市商系统的回测框架时，就是靠这三个原则撑起了整个系统，后来加新策略、换数据源都特别顺畅。

---

公众号:蓝海资料掘金营，微信deep3321