3. 事件类型定义：市场事件、订单事件、信号事件、账户事件、风控事件

好，咱们今天来聊聊回测系统的「血液」——事件。

我个人习惯把事件比作系统的神经信号。没有事件，你的策略再牛，也动不了。说白了，回测引擎就是个事件驱动的机器，它不停地接收事件、处理事件、产生新事件。

我在做第一个回测框架时，就犯过一个低级错误：把所有事件混在一个大字典里传。结果呢？调试的时候，我根本分不清这个事件是市场推送的，还是策略自己产生的。那叫一个乱。

所以，事件类型必须清晰定义。这是基本功。

3.1 事件基类设计

先搞个基类。所有事件都继承它。这样后面扩展起来方便。

from enum import Enum, auto
from dataclasses import dataclass
from datetime import datetime

class EventType(Enum):
    """事件类型枚举"""
    TICK = auto()        # 市场Tick事件
    BAR = auto()         # 市场Bar事件
    ORDER = auto()       # 订单事件
    SIGNAL = auto()      # 信号事件
    ACCOUNT = auto()     # 账户事件
    RISK = auto()        # 风控事件

@dataclass
class Event:
    """事件基类"""
    type: EventType
    timestamp: datetime
    # 每个事件必须带时间戳，这是回测的命脉

嗯，这里要注意：dataclass 是 Python 3.7+ 的特性。用起来很爽，省去写一堆 __init__ 的麻烦。

3.2 市场事件：Tick 和 Bar

市场事件是回测的「燃料」。没有它，你的策略就是无米之炊。

Tick 事件

Tick 是逐笔成交数据。精度最高，但数据量也最大。我在处理期货高频数据时，一天就能产生上百万条 Tick。

@dataclass
class TickEvent(Event):
    """Tick事件 - 逐笔成交"""
    symbol: str           # 合约代码
    price: float          # 成交价
    volume: int           # 成交量
    bid_price: float      # 买一价
    ask_price: float      # 卖一价
    bid_volume: int       # 买一量
    ask_volume: int       # 卖一量
    
    def __post_init__(self):
        self.type = EventType.TICK

为什么要有买卖盘口？因为你的订单能不能成交，得看盘口深度。我曾经遇到过一个策略，只看成交价，结果回测时收益漂亮，实盘一跑就亏。原因就是没考虑盘口流动性。

Bar 事件

Bar 是聚合数据。1分钟、5分钟、日线，都属于 Bar。大部分策略用 Bar 就够了。

@dataclass
class BarEvent(Event):
    """Bar事件 - K线数据"""
    symbol: str
    open: float
    high: float
    low: float
    close: float
    volume: int
    # 可选字段
    vwap: float = None    # 成交量加权均价
    count: int = None     # 成交笔数
    
    def __post_init__(self):
        self.type = EventType.BAR
我的经验：Bar 事件里一定要带上 vwap。很多策略用收盘价计算信号，但实际成交价可能偏离收盘价。用 VWAP 做回测，结果更接近实盘。

3.3 订单事件

订单事件是策略和交易所之间的「对话」。你下单、撤单、成交，都会产生订单事件。

class OrderStatus(Enum):
    CREATED = auto()      # 已创建
    SUBMITTED = auto()    # 已提交
    PARTIAL = auto()      # 部分成交
    FILLED = auto()       # 全部成交
    CANCELED = auto()     # 已撤销
    REJECTED = auto()     # 已拒绝

class OrderSide(Enum):
    BUY = auto()
    SELL = auto()

@dataclass
class OrderEvent(Event):
    """订单事件"""
    order_id: str         # 订单ID
    symbol: str
    side: OrderSide
    price: float
    quantity: int
    status: OrderStatus
    filled_qty: int = 0   # 已成交数量
    avg_price: float = 0.0  # 成交均价
    
    def __post_init__(self):
        self.type = EventType.ORDER

你想想看，如果订单状态只有「成交」和「未成交」，那部分成交的情况怎么处理？我早期就吃过这个亏，回测时假设订单要么全成要么全撤，结果实盘遇到大单拆分，策略逻辑全乱了。

3.4 信号事件

信号事件是策略的「大脑」产生的。它告诉系统：该买还是该卖。

@dataclass
class SignalEvent(Event):
    """信号事件 - 策略决策"""
    symbol: str
    action: str           # 'BUY', 'SELL', 'HOLD'
    strength: float = 1.0 # 信号强度 0~1
    meta: dict = None     # 附加信息
    
    def __post_init__(self):
        self.type = EventType.SIGNAL
        if self.meta is None:
            self.meta = {}

这里我建议把 strength 加上。为什么？因为不是所有信号都值得执行。信号强度低于 0.5 的，可以过滤掉。这能减少无效交易，降低手续费损耗。

3.5 账户事件

账户事件记录资金变动。每次成交、分红、手续费扣除，都会触发账户事件。

@dataclass
class AccountEvent(Event):
    """账户事件 - 资金变动"""
    account_id: str
    balance: float        # 当前余额
    frozen: float         # 冻结资金
    available: float      # 可用资金
    pnl: float            # 浮动盈亏
    margin: float = 0.0   # 占用保证金
    
    def __post_init__(self):
        self.type = EventType.ACCOUNT
避坑指南：我曾经在账户事件里只记录了余额，没记录冻结资金。结果回测时发现资金明明够，但下单时总是被拒绝。后来才意识到，有些资金已经被其他订单冻结了。所以，frozen 和 available 必须分开记录。

3.6 风控事件

风控事件是系统的「刹车」。当策略行为异常时，风控模块会发出警告或强制干预。

class RiskLevel(Enum):
    WARNING = auto()      # 警告
    REJECT = auto()       # 拒绝
    FORCE_CLOSE = auto()  # 强制平仓

@dataclass
class RiskEvent(Event):
    """风控事件"""
    rule_name: str        # 触发的风控规则
    level: RiskLevel
    message: str          # 描述信息
    detail: dict = None   # 详细信息
    
    def __post_init__(self):
        self.type = EventType.RISK
        if self.detail is None:
            self.detail = {}

常见的风控规则有哪些？我列一下：

  • 仓位限制：单品种持仓不能超过总资金的 20%
  • 撤单频率：每分钟撤单不超过 10 次
  • 亏损限制：当日亏损超过 5% 停止交易
  • 价格偏离：下单价格偏离市价超过 1% 拒绝

3.7 事件关系图

下面这张图，展示了事件在回测系统中的流转关系。我画了很久，希望能帮你理清思路。

事件驱动回测系统 - 事件流转图 市场数据源 Tick事件 Bar事件 策略引擎 接收市场事件 产生信号事件 管理账户状态 信号事件 订单管理 接收信号事件 生成订单事件 触发风控检查 订单事件 风控模块 风控事件 反馈/干预 图例： 市场数据 策略引擎 订单管理 风控模块 信号事件 订单事件 风控事件 反馈/干预

3.8 事件处理的核心逻辑

事件处理的核心，就是一个循环。我把它叫做「事件泵」。

class EventEngine:
    """事件引擎 - 核心调度器"""
    
    def __init__(self):
        self.queue = []  # 事件队列
        self.handlers = {
            EventType.TICK: [],
            EventType.BAR: [],
            EventType.SIGNAL: [],
            EventType.ORDER: [],
            EventType.ACCOUNT: [],
            EventType.RISK: [],
        }
    
    def register_handler(self, event_type, handler):
        """注册事件处理器"""
        self.handlers[event_type].append(handler)
    
    def put(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)
            # 分发事件
            for handler in self.handlers[event.type]:
                handler(event)
核心要点：事件队列必须按时间戳排序。这是回测和实盘最大的区别。实盘中事件是实时到达的，但回测中所有事件都是历史数据，必须保证时序正确。

好了，事件类型定义就讲到这里。这些代码我都在实际项目中用过，你可以直接拿去用。记住一点：事件定义得越清晰，后面的调试就越轻松。

专注资料整理

← 上一章 📖 返回目录 下一章 →