策略引擎：策略生命周期与信号生成
策略引擎是整个交易系统的「大脑」。我做了这么多年量化，见过太多系统崩在策略管理上——不是信号发不出来，就是生命周期乱了套。今天咱们就把这块彻底讲透。
策略生命周期管理
一个策略从加载到销毁，经历四个核心状态：初始化、运行、暂停、停止。我习惯用状态机来管理，清晰且不容易出错。
核心状态定义：
- 初始化（INIT）：加载参数、建立数据连接、校验配置
- 运行（RUNNING）：正常接收行情、生成信号、执行交易
- 暂停（PAUSED）：停止信号生成，但保留上下文和连接
- 停止（STOPPED）：释放资源、保存状态、断开连接
为什么要单独搞个「暂停」？我在项目中遇到过，某次交易所升级接口，需要临时停掉部分策略。如果直接停止，重启时得重新加载几十G的历史数据。暂停就优雅多了——保留内存状态，恢复时秒级启动。
class StrategyState:
INIT = 0
RUNNING = 1
PAUSED = 2
STOPPED = 3
class StrategyEngine:
def __init__(self):
self.state = StrategyState.INIT
self.strategies = {}
def start(self, strategy_id):
if self.state != StrategyState.INIT:
raise RuntimeError("引擎已启动")
self.state = StrategyState.RUNNING
# 启动事件循环
self._event_loop.start()
def pause(self, strategy_id):
if self.state != StrategyState.RUNNING:
return # 非运行状态忽略
self.state = StrategyState.PAUSED
# 暂停信号生成，但保留数据流
self._event_loop.pause()
def stop(self, strategy_id):
self.state = StrategyState.STOPPED
self._event_loop.stop()
self._cleanup()注意：状态切换必须做合法性校验。我曾经见过一个系统，从「停止」直接切到「暂停」，结果资源没释放干净，内存泄漏跑了三天才崩。
策略基类与接口设计
接口设计我坚持一个原则：最小暴露原则。子类只需要关心三个方法：初始化、处理数据、生成信号。其他生命周期逻辑由基类封装。
from abc import ABC, abstractmethod
class BaseStrategy(ABC):
def __init__(self, strategy_id, config):
self.id = strategy_id
self.config = config
self.position = 0
self.signals = []
@abstractmethod
def on_init(self):
"""策略初始化：加载模型、建立指标计算器"""
pass
@abstractmethod
def on_tick(self, tick_data):
"""处理每笔行情"""
pass
@abstractmethod
def on_bar(self, bar_data):
"""处理K线数据"""
pass
def generate_signal(self, direction, price, volume):
"""生成信号（基类提供默认实现）"""
signal = {
'strategy_id': self.id,
'direction': direction,
'price': price,
'volume': volume,
'timestamp': time.time()
}
self.signals.append(signal)
return signal我个人习惯把generate_signal放在基类。为什么？因为信号格式必须统一，否则事件总线解析起来会疯掉。你想想看，十个策略各自定义信号结构，下游的订单管理器得写多少适配代码？
避坑指南：我曾经在基类里加了个validate_config()方法，所有策略启动前自动校验参数。那次有个同事把止损比例配成了负数，系统直接拒绝启动，避免了爆仓风险。建议你也加上。
策略信号生成与事件驱动
信号生成不是简单的「金叉买死叉卖」。真正的生产环境里，信号要经过三层过滤：
- 原始信号：策略逻辑算出来的买卖点
- 风控过滤：检查仓位、资金、频率限制
- 事件封装：包装成标准事件，推送到事件总线
事件驱动架构是策略引擎的基石。说白了，就是「有行情来了，谁关心谁处理」。我习惯用发布-订阅模式：
class EventBus:
def __init__(self):
self._handlers = {}
def subscribe(self, event_type, handler):
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
def publish(self, event):
event_type = event['type']
for handler in self._handlers.get(event_type, []):
handler(event)
# 策略引擎内部使用
class StrategyEngine:
def __init__(self, event_bus):
self.event_bus = event_bus
self.event_bus.subscribe('TICK', self._on_tick)
self.event_bus.subscribe('BAR', self._on_bar)
def _on_tick(self, event):
for strategy in self.active_strategies:
signal = strategy.on_tick(event['data'])
if signal:
# 风控过滤
if self._risk_check(signal):
self.event_bus.publish({
'type': 'SIGNAL',
'data': signal
})关键设计点：
- 事件类型用字符串常量，别用魔法数字
- 每个事件带时间戳，方便回溯
- 异步处理时注意线程安全，我习惯用队列解耦
信号生成这块，我踩过最大的坑是「信号抖动」。比如均线金叉后马上死叉，策略在几秒内反复开平仓。后来我在基类里加了信号去抖逻辑：
class BaseStrategy(ABC):
def __init__(self, strategy_id, config):
self._last_signal_time = 0
self._min_interval = config.get('min_signal_interval', 1.0) # 秒
def _can_generate_signal(self):
now = time.time()
if now - self._last_signal_time < self._min_interval:
return False
self._last_signal_time = now
return True嗯，这里要注意：去抖时间不能设太长，否则会错过真正的行情。我一般根据策略交易频率来调，高频策略设0.5秒，中低频设3-5秒。
整体架构图
下面这张图展示了策略引擎的核心流程，从行情输入到信号输出，每个环节都清晰可见：
从图里能看出来，策略引擎的核心就是「事件驱动 + 状态管理」。行情进来，经过事件总线分发给所有活跃策略，每个策略独立计算、生成信号，最后统一输出。暂停的策略不参与计算，但保留上下文——这就是我前面说的优雅降级。
个人经验：实际部署时，我会给每个策略分配独立的上下文对象，避免全局变量污染。曾经有个同事把止损价存在类变量里，两个策略共用同一个止损，结果一个亏了另一个也跟着平仓，亏了十几万。教训深刻啊。
好了，策略引擎的核心就这些。记住三个关键词：状态机、基类接口、事件驱动。把这套架子搭好，后面加策略就跟插积木一样简单。