4. 策略与回测解耦
任务目标
在量化回测框架中，策略逻辑与回测引擎（数据加载、订单撮合、绩效统计）不应相互耦合。本案例定义一个抽象策略接口，使策略仅负责接收市场数据并输出交易信号，回测引擎通过该接口驱动任意策略，从而实现策略的即插即用。
核心设计
策略接口只暴露两个方法：init() 用于策略初始化（如预计算指标），on_bar() 在每根K线到达时被引擎调用，返回目标持仓比例（正数做多、负数做空、0表示空仓）。回测引擎不关心策略内部如何计算信号，只依赖该接口获取仓位指令。
完整实现
from abc import ABC, abstractmethod
from typing import List, Dict, Any
# ---------- 策略抽象接口 ----------
class Strategy(ABC):
"""策略基类：所有策略必须实现 init 和 on_bar"""
@abstractmethod
def init(self) -> None:
"""策略初始化，如预计算指标"""
pass
@abstractmethod
def on_bar(self, bar: Dict[str, Any], history: List[Dict[str, Any]]) -> float:
"""
每根K线回调，返回目标仓位比例
bar: 当前K线 {'open','high','low','close','volume','timestamp'}
history: 截至上一根K线的历史数据（不含当前bar）
返回: 目标仓位，1.0=满仓做多，-1.0=满仓做空，0=空仓
"""
pass
# ---------- 两个具体策略（插拔示例） ----------
class DualMAStrategy(Strategy):
"""双均线策略：快线上穿慢线做多，下穿做空"""
def __init__(self, fast: int = 5, slow: int = 20):
self.fast = fast
self.slow = slow
def init(self) -> None:
pass # 无需预计算
def on_bar(self, bar, history) -> float:
closes = [h['close'] for h in history] + [bar['close']]
if len(closes) < self.slow + 1:
return 0.0
fast_ma = sum(closes[-self.fast:]) / self.fast
slow_ma = sum(closes[-self.slow:]) / self.slow
if fast_ma > slow_ma:
return 1.0
elif fast_ma < slow_ma:
return -1.0
return 0.0
class BuyHoldStrategy(Strategy):
"""买入持有策略：始终满仓做多"""
def init(self) -> None:
pass
def on_bar(self, bar, history) -> float:
return 1.0
# ---------- 回测引擎（只依赖 Strategy 接口） ----------
class BacktestEngine:
def __init__(self, strategy: Strategy, data: List[Dict[str, Any]]):
self.strategy = strategy
self.data = data
self.position = 0.0 # 当前实际仓位
def run(self) -> Dict[str, Any]:
self.strategy.init()
history: List[Dict[str, Any]] = []
trades = []
equity_curve = []
for bar in self.data:
target_pos = self.strategy.on_bar(bar, history)
# 记录调仓（仅当目标仓位变化时）
if target_pos != self.position:
trades.append({
'timestamp': bar['timestamp'],
'from': self.position,
'to': target_pos,
'price': bar['close']
})
self.position = target_pos
# 按收盘价计算净值（简化：无手续费/滑点）
equity_curve.append({
'timestamp': bar['timestamp'],
'equity': 1.0 + self.position * (bar['close'] / self.data[0]['close'] - 1.0)
})
history.append(bar)
return {'trades': trades, 'equity_curve': equity_curve}
# ---------- 最小可运行示例 ----------
if __name__ == '__main__':
# 构造10根K线模拟数据（上涨趋势）
raw_data = []
price = 100.0
for i in range(10):
price *= 1.02 # 每日涨2%
raw_data.append({
'timestamp': f'2024-01-{i+1:02d}',
'open': price * 0.99,
'high': price * 1.01,
'low': price * 0.98,
'close': price,
'volume': 1000 + i * 10
})
# 插拔策略1：双均线
engine1 = BacktestEngine(DualMAStrategy(fast=2, slow=4), raw_data)
result1 = engine1.run()
print(f"双均线策略调仓次数: {len(result1['trades'])}")
# 插拔策略2：买入持有
engine2 = BacktestEngine(BuyHoldStrategy(), raw_data)
result2 = engine2.run()
print(f"买入持有策略调仓次数: {len(result2['trades'])}")
print(f"最终净值: {result2['equity_curve'][-1]['equity']:.4f}")运行与验证
- 运行方式：将上述代码保存为
strategy_decoupling.py，执行python strategy_decoupling.py。 - 预期输出（由代码逻辑直接推出）：
- 双均线策略在10根上涨K线中，快线(2日)始终高于慢线(4日)，因此首次信号为做多(1.0)，后续不变，调仓次数为 1。
- 买入持有策略始终满仓，调仓次数为 0。
- 最终净值 = 1.0 + 1.0 × (100×1.02¹⁰ / 100 − 1.0) = 1.0 + (1.02¹⁰ − 1.0) ≈ 1.2190。
- 验证要点：引擎代码中不出现任何策略特有逻辑（如均线计算、持有规则），只调用
init()和on_bar()接口；新增策略只需继承Strategy并实现两个方法即可接入引擎，无需修改引擎代码。
解耦验证
将 DualMAStrategy 替换为 BuyHoldStrategy 时，BacktestEngine 类无需任何改动，证明策略与回测引擎已解耦。若需接入新策略（如 RSI 策略），只需新增一个继承 Strategy 的类并实现 init/on_bar，即可直接传入引擎运行。