做市策略回测框架设计原则

做市策略的回测，说白了就是拿历史数据模拟交易，看看你的策略到底能不能赚钱。我做了这么多年量化，见过太多人一上来就写回测代码，结果改个参数要改半天，换个品种直接崩掉。嗯，这里面的坑，我基本都踩过。

今天我们就聊聊回测框架的设计原则。我个人习惯是，先把框架搭稳了，再往里填策略。你想想看，地基没打好，楼盖得再高也是危房。

一、模块化设计

模块化，说白了就是把一个复杂系统拆成一个个独立的小零件。每个零件只管自己的事，互不干扰。

为什么要模块化？

• 好维护：改一个模块，不影响其他模块。我曾经因为耦合太紧，改了个风控逻辑，结果把订单管理模块搞崩了，排查了两天...
• 好复用：写好的模块，换个策略照样用。比如数据加载模块，做市策略用，趋势策略也能用。
• 好测试：每个模块单独测，出问题定位快。

常见的模块划分：

模块名称	职责	典型接口
数据模块	加载、清洗、对齐行情数据	load_data(symbol, start, end)
策略模块	根据行情生成买卖信号	on_tick(tick) -> signal
风控模块	检查订单是否合规	check_order(order) -> bool
订单管理模块	执行订单、管理持仓	execute_order(signal) -> trade
绩效模块	计算收益、夏普、最大回撤等	calculate_pnl(trades) -> metrics

二、事件驱动架构

做市策略和普通策略不一样。普通策略可能每分钟算一次信号，但做市策略是逐笔驱动的——每来一个tick，你都得马上反应。

事件驱动是什么？

简单说，就是「发生什么事，就触发什么逻辑」。比如：

• 来了一个新tick → 更新订单簿 → 检查是否需要调整报价
• 成交了一笔 → 更新持仓 → 检查风险指标
• 到了收盘时间 → 计算当日绩效 → 生成报告

我刚开始做回测时，用的是循环轮询的方式——每秒钟查一次有没有新数据。结果呢？性能差不说，还容易漏掉关键事件。后来改成事件驱动，整个框架清爽多了。

事件驱动的核心组件：

• 事件队列：存放所有待处理的事件，按时间排序
• 事件处理器：根据事件类型，调用对应的处理函数
• 事件源：产生事件的地方，比如数据加载器、成交回报

# 一个简化的事件驱动回测引擎
class EventEngine:
    def __init__(self):
        self.queue = []
        self.handlers = {}
    
    def register_handler(self, event_type, handler):
        self.handlers[event_type] = handler
    
    def push_event(self, event):
        self.queue.append(event)
        self.queue.sort(key=lambda e: e.timestamp)  # 按时间排序
    
    def run(self):
        while self.queue:
            event = self.queue.pop(0)
            handler = self.handlers.get(event.type)
            if handler:
                handler(event)

三、数据与策略分离

这个原则，说白了就是「数据是数据，策略是策略，别混在一起」。

为什么必须分离？

• 数据格式会变：今天用CSV，明天可能用数据库，后天可能用API。策略不应该关心数据从哪来。
• 策略逻辑会变：同一个数据，你可能想试不同的策略。如果耦合在一起，每次换策略都得改数据代码。
• 方便回测和实盘切换：回测用历史数据，实盘用实时数据，策略代码完全不用改。

怎么实现分离？

我习惯用「数据适配器」模式。数据模块提供一个统一的接口，不管底层数据源是什么，上层策略看到的都是一样的格式。

# 数据适配器示例
class DataAdapter:
    def get_tick(self, symbol, timestamp):
        """返回统一的Tick格式"""
        pass
    
    def get_orderbook(self, symbol, timestamp):
        """返回统一的订单簿格式"""
        pass

# CSV数据源
class CSVDataAdapter(DataAdapter):
    def get_tick(self, symbol, timestamp):
        # 从CSV文件读取
        pass

# 数据库数据源
class DBDataAdapter(DataAdapter):
    def get_tick(self, symbol, timestamp):
        # 从数据库读取
        pass

四、性能考量

做市策略的回测，数据量通常很大。一天可能有几百万个tick，一年就是几亿条。性能不行，回测一次等半天，谁受得了？

性能瓶颈在哪？

• I/O瓶颈：从磁盘读数据太慢
• 计算瓶颈：每个tick都要算一遍信号，计算量很大
• 内存瓶颈：数据量太大，内存装不下

优化思路：

优化方向	具体做法	效果
数据加载	使用二进制格式（如Parquet）代替CSV	加载速度提升10倍以上
计算加速	使用NumPy向量化计算，避免逐行循环	计算速度提升100倍
内存管理	分块加载数据，处理完一块释放一块	内存占用降低80%
并行处理	多品种回测时，用多进程并行	时间缩短到1/N

知识体系总览

下面这张图，把四个设计原则的关系画出来了。你仔细看看，其实它们是环环相扣的。

这四个原则，说白了就是一套方法论。模块化让你改得动，事件驱动让你跑得准，数据与策略分离让你换得爽，性能考量让你等得起。缺一个，你的回测框架就会出问题。

我个人习惯是，先按模块化把架子搭好，然后实现事件驱动引擎，再把数据接口抽象出来，最后做性能优化。这个顺序，我试过很多次，是最顺的。

---

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