Rekey
4d66a86234
- AGENTS.md: AI Agent 规则，定义项目运行环境、架构约定、常用命令 - ENGINE.md: Python 引擎架构设计文档，涵盖策略管理、信号总线、回测引擎等模块设计
1338 lines
46 KiB
Markdown
1338 lines
46 KiB
Markdown
# 策略引擎开发文档
|
||
|
||
> 数字货币量化交易系统 — 策略引擎模块（Python 3.10+）设计与开发指南
|
||
|
||
---
|
||
|
||
## 目录
|
||
|
||
1. [概述与定位](#概述与定位)
|
||
2. [核心架构](#核心架构)
|
||
3. [策略基类设计](#策略基类设计)
|
||
4. [策略管理器](#策略管理器)
|
||
5. [信号分发器](#信号分发器)
|
||
6. [回测引擎](#回测引擎)
|
||
7. [参数优化器](#参数优化器)
|
||
8. [数据流与生命周期](#数据流与生命周期)
|
||
9. [与上下游模块的交互](#与上下游模块的交互)
|
||
10. [目录结构](#目录结构)
|
||
11. [开发步骤](#开发步骤)
|
||
12. [实现规范](#实现规范)
|
||
|
||
---
|
||
|
||
## 概述与定位
|
||
|
||
策略引擎是整个量化交易系统的**核心调度中心**，位于 Python 业务层，负责：
|
||
|
||
- **策略生命周期管理**：注册、热加载、启动、暂停、停止、卸载策略
|
||
- **实时信号生成**：消费 TS 数据模块推送的行情数据，驱动策略产生交易信号
|
||
- **回测验证**：使用历史数据模拟策略执行，评估收益、回撤、夏普比率等指标
|
||
- **参数搜索**：通过网格搜索或贝叶斯优化自动寻找最优策略参数
|
||
- **下游集成**：将交易信号通过信号分发器路由到交易执行模块和风控模块
|
||
|
||
### 在系统中的位置
|
||
|
||
```
|
||
┌──────────────────────────┐
|
||
│ 策略引擎 🐍 Python │
|
||
│ │
|
||
Redis Pub/Sub ──────► │ ┌────────────────────┐ │
|
||
行情数据 │ │ 信号分发器 │──► executor/
|
||
│ └────────────────────┘ │
|
||
TimescaleDB ◄─────── │ ┌────────────────────┐ │
|
||
历史数据（回测） │ │ 策略管理器 │ │
|
||
│ │ ┌──────┬──────┐ │ │
|
||
│ │ │策略A │策略B │ │ │
|
||
│ │ └──────┴──────┘ │ │
|
||
│ └────────────────────┘ │
|
||
│ ┌────────────────────┐ │
|
||
│ │ 回测引擎 │ │
|
||
│ └────────────────────┘ │
|
||
│ ┌────────────────────┐ │
|
||
│ │ 参数优化器 │ │
|
||
│ └────────────────────┘ │
|
||
└──────────────────────────┘
|
||
```
|
||
|
||
---
|
||
|
||
## 核心架构
|
||
|
||
### 模块依赖图
|
||
|
||
```
|
||
┌──────────────────┐
|
||
│ engine/common/ │ ← 配置、日志、数据模型、策略基类
|
||
└────────┬─────────┘
|
||
│
|
||
┌────────────────────┼────────────────────┐
|
||
│ │ │
|
||
┌────▼────┐ ┌──────▼──────┐ ┌─────▼──────┐
|
||
│ base.py │ │ manager.py │ │ consumers/ │
|
||
│策略基类 │◄───────│ 策略管理器 │ │ 行情消费 │
|
||
└────┬────┘ └──────┬──────┘ └────────────┘
|
||
│ │
|
||
│ ┌──────▼──────┐
|
||
┌────▼────┐ │ signals.py │ ┌─────────┐
|
||
│strategies│ │ 信号分发器 │─────►│executor/│
|
||
│ 策略实现 │ └─────────────┘ └─────────┘
|
||
└────┬────┘
|
||
│
|
||
┌────▼────┐ ┌──────────────┐
|
||
│backtest │ │ optimizer.py │
|
||
│回测引擎 │◄───────│ 参数优化器 │
|
||
└─────────┘ └──────────────┘
|
||
```
|
||
|
||
### 设计原则
|
||
|
||
| 原则 | 说明 |
|
||
|------|------|
|
||
| **插件化** | 策略以插件形式注册，新增策略无需修改引擎核心代码 |
|
||
| **异步优先** | 全部 I/O 操作使用 `asyncio`，保证高并发下的低延迟 |
|
||
| **类型安全** | 核心接口用 `ABC` + Pydantic 模型约束，运行时校验 |
|
||
| **无状态** | 引擎本身不持有策略状态，状态由策略实例自行管理 |
|
||
| **错误隔离** | 单个策略异常不影响其他策略运行，也不影响引擎主循环 |
|
||
| **可观测** | 每个生命周期事件埋点，输出结构化日志和 Prometheus 指标 |
|
||
|
||
---
|
||
|
||
## 策略基类设计
|
||
|
||
### 抽象基类
|
||
|
||
```python
|
||
# engine/common/base.py
|
||
from abc import ABC, abstractmethod
|
||
from typing import Optional
|
||
from pydantic import BaseModel
|
||
|
||
from .models import Ticker, Kline, OrderBook, Trade
|
||
|
||
|
||
class StrategyConfig(BaseModel):
|
||
"""策略配置（由具体策略子类化扩展）"""
|
||
name: str
|
||
symbol: str
|
||
exchange: str = "binance"
|
||
enabled: bool = True
|
||
max_position_pct: float = 0.1 # 最大仓位比例
|
||
stop_loss_pct: Optional[float] = None # 止损百分比
|
||
take_profit_pct: Optional[float] = None # 止盈百分比
|
||
|
||
|
||
class Signal(BaseModel):
|
||
"""交易信号"""
|
||
symbol: str
|
||
side: str # "BUY" / "SELL"
|
||
signal_type: str # "MARKET" / "LIMIT" / "CANCEL"
|
||
price: Optional[float] = None # 限价单价格
|
||
quantity: Optional[float] = None # 下单数量
|
||
confidence: float = 1.0 # 信号置信度 [0, 1]
|
||
reason: str = "" # 信号生成原因（便于审计）
|
||
timestamp: float # 信号时间戳
|
||
|
||
|
||
class BaseStrategy(ABC):
|
||
"""所有策略的抽象基类"""
|
||
|
||
# 策略类型标识，子类必须覆盖
|
||
strategy_type: str = "base"
|
||
|
||
def __init__(self, config: StrategyConfig):
|
||
self.config = config
|
||
self.is_running = False
|
||
self.pnl = 0.0
|
||
self.trade_count = 0
|
||
|
||
# ── 生命周期钩子 ──
|
||
|
||
async def on_start(self) -> None:
|
||
"""策略启动时调用（加载初始状态、预热等）"""
|
||
self.is_running = True
|
||
|
||
async def on_stop(self) -> None:
|
||
"""策略停止时调用（清理、平仓等）"""
|
||
self.is_running = False
|
||
|
||
async def on_pause(self) -> None:
|
||
"""策略暂停时调用"""
|
||
pass
|
||
|
||
async def on_resume(self) -> None:
|
||
"""策略恢复时调用"""
|
||
pass
|
||
|
||
# ── 行情事件处理器（子类按需实现）──
|
||
|
||
@abstractmethod
|
||
async def on_kline(self, kline: Kline) -> Optional[Signal]:
|
||
"""处理 K 线数据，返回交易信号"""
|
||
...
|
||
|
||
async def on_ticker(self, ticker: Ticker) -> Optional[Signal]:
|
||
"""处理 Ticker 数据（可选实现）"""
|
||
return None
|
||
|
||
async def on_trade(self, trade: Trade) -> Optional[Signal]:
|
||
"""处理逐笔成交数据（可选实现）"""
|
||
return None
|
||
|
||
async def on_orderbook(self, orderbook: OrderBook) -> Optional[Signal]:
|
||
"""处理深度数据（可选实现）"""
|
||
return None
|
||
|
||
# ── 交易反馈钩子 ──
|
||
|
||
async def on_order_filled(self, order_result) -> None:
|
||
"""订单成交回调"""
|
||
pass
|
||
|
||
async def on_order_cancelled(self, order_id: str) -> None:
|
||
"""订单撤销回调"""
|
||
pass
|
||
|
||
# ── 辅助方法 ──
|
||
|
||
async def get_klines(
|
||
self, interval: str, limit: int = 100
|
||
) -> list[Kline]:
|
||
"""从 TimescaleDB 获取历史 K 线（回测和预热用）"""
|
||
...
|
||
|
||
def log(self, level: str, message: str, **kwargs) -> None:
|
||
"""结构化日志"""
|
||
...
|
||
```
|
||
|
||
### 策略状态机
|
||
|
||
```
|
||
register
|
||
│
|
||
▼
|
||
┌────────┐ start ┌─────────┐
|
||
│ LOADED │ ──────────► │ RUNNING │
|
||
└────────┘ └────┬─────┘
|
||
▲ │
|
||
│ ┌────────┴────────┐
|
||
│ ▼ ▼
|
||
│ ┌─────────┐ ┌──────────┐
|
||
└─────────│ PAUSED │ │ STOPPING │
|
||
unload └────┬─────┘ └────┬─────┘
|
||
│ resume │
|
||
└───► RUNNING │
|
||
▼
|
||
┌──────────┐
|
||
│ STOPPED │
|
||
└────┬─────┘
|
||
│ unload
|
||
▼
|
||
(removed)
|
||
```
|
||
|
||
### 策略生命周期事件流
|
||
|
||
```
|
||
on_start()
|
||
│
|
||
▼
|
||
┌──────────────────────────────────────────────────────────┐
|
||
│ 主循环（事件驱动） │
|
||
│ │
|
||
│ on_kline(kline) ──► Signal? ──► signal_bus.emit(signal) │
|
||
│ on_ticker(ticker) ─► Signal? ──► signal_bus.emit(signal) │
|
||
│ on_orderbook(ob) ──► Signal? ──► signal_bus.emit(signal) │
|
||
│ │
|
||
│ on_order_filled(result) ←── executor 回调 │
|
||
│ on_order_cancelled(id) ←── executor 回调 │
|
||
└──────────────────────────────────────────────────────────┘
|
||
│
|
||
▼
|
||
on_stop()
|
||
```
|
||
|
||
---
|
||
|
||
## 策略管理器
|
||
|
||
策略管理器是引擎的入口组件，负责所有策略实例的注册、发现、加载和生命周期调度。
|
||
|
||
### 接口设计
|
||
|
||
```python
|
||
# engine/manager.py
|
||
import importlib
|
||
import pkgutil
|
||
from pathlib import Path
|
||
from typing import Type, Optional
|
||
from .common.logger import logger
|
||
|
||
from .common import BaseStrategy, StrategyConfig, Signal
|
||
from .signals import SignalBus
|
||
|
||
|
||
class StrategyManager:
|
||
"""
|
||
策略管理器：策略注册、热加载、生命周期调度
|
||
|
||
用法:
|
||
manager = StrategyManager(signal_bus=signal_bus)
|
||
manager.discover_strategies("strategies")
|
||
await manager.start_all()
|
||
"""
|
||
|
||
def __init__(self, signal_bus: SignalBus):
|
||
self._strategies: dict[str, BaseStrategy] = {}
|
||
self._strategy_classes: dict[str, Type[BaseStrategy]] = {}
|
||
self.signal_bus = signal_bus
|
||
|
||
# ── 策略发现与注册 ──
|
||
|
||
def discover_strategies(self, package_path: str = "strategies") -> list[str]:
|
||
"""
|
||
自动发现策略包下的所有策略类。
|
||
|
||
约定：
|
||
- 策略文件放在 `strategies/` 目录下
|
||
- 策略类必须继承 `BaseStrategy` 且在其模块中定义
|
||
|
||
返回发现的策略类型名列表。
|
||
"""
|
||
...
|
||
|
||
def register(self, strategy_cls: Type[BaseStrategy]) -> None:
|
||
"""手动注册策略类"""
|
||
...
|
||
|
||
# ── 策略实例管理 ──
|
||
|
||
def create(
|
||
self, strategy_type: str, config: StrategyConfig
|
||
) -> BaseStrategy:
|
||
"""
|
||
根据策略类型创建实例。
|
||
|
||
Raises:
|
||
ValueError: 策略类型未注册
|
||
"""
|
||
...
|
||
|
||
async def start(self, name: str) -> None:
|
||
"""启动指定策略"""
|
||
...
|
||
|
||
async def stop(self, name: str) -> None:
|
||
"""停止指定策略"""
|
||
...
|
||
|
||
async def pause(self, name: str) -> None:
|
||
"""暂停指定策略（保留状态，不接收行情）"""
|
||
...
|
||
|
||
async def resume(self, name: str) -> None:
|
||
"""恢复暂停的策略"""
|
||
...
|
||
|
||
async def restart(self, name: str, new_config: Optional[StrategyConfig] = None) -> BaseStrategy:
|
||
"""重启策略（可选地更新配置）"""
|
||
...
|
||
|
||
def get(self, name: str) -> Optional[BaseStrategy]:
|
||
"""获取策略实例"""
|
||
...
|
||
|
||
def list_strategies(self) -> list[dict]:
|
||
"""列出所有策略的状态"""
|
||
...
|
||
|
||
# ── 批量操作 ──
|
||
|
||
async def start_all(self) -> None:
|
||
"""启动所有 enabled 策略"""
|
||
...
|
||
|
||
async def stop_all(self) -> None:
|
||
"""停止所有策略"""
|
||
...
|
||
|
||
# ── 行情入口（由主循环调用）──
|
||
|
||
async def feed_kline(self, kline) -> None:
|
||
"""将 K 线数据分发到对应 symbol 的策略"""
|
||
...
|
||
|
||
async def feed_ticker(self, ticker) -> None:
|
||
"""将 Ticker 分发到对应 symbol 的策略"""
|
||
...
|
||
|
||
async def feed_orderbook(self, orderbook) -> None:
|
||
"""将 OrderBook 分发到对应 symbol 的策略"""
|
||
...
|
||
|
||
# ── 交易反馈入口 ──
|
||
|
||
async def notify_order_filled(self, strategy_name: str, order_result) -> None:
|
||
"""通知策略订单已成交"""
|
||
...
|
||
|
||
# ── 统计与监控 ──
|
||
|
||
def stats(self) -> dict:
|
||
"""返回引擎统计信息"""
|
||
...
|
||
```
|
||
|
||
### 策略热加载机制
|
||
|
||
```
|
||
策略文件变更
|
||
│
|
||
▼
|
||
┌──────────────────┐
|
||
│ 文件监听器 │
|
||
│ (watchdog) │
|
||
└────────┬─────────┘
|
||
│
|
||
┌──────────▼──────────┐
|
||
│ 重新 importlib.reload│
|
||
└──────────┬──────────┘
|
||
│
|
||
┌──────────▼──────────┐
|
||
│ 更新 _strategy_classes│
|
||
└──────────┬──────────┘
|
||
│
|
||
┌──────────▼──────────┐
|
||
│ 重新创建策略实例 │
|
||
│ (stop → create → start)│
|
||
└─────────────────────┘
|
||
```
|
||
|
||
---
|
||
|
||
## 信号分发器
|
||
|
||
信号分发器是策略到交易执行模块之间的**解耦层**，提供事件总线能力。
|
||
|
||
### 设计目标
|
||
|
||
- **解耦**：策略不直接调用交易 API，只 emit 信号
|
||
- **过滤**：同一 symbol 的冲突信号自动去重/合并
|
||
- **优先级**：支持信号优先级排序（高置信度优先）
|
||
- **限流**：防止策略过度交易，内置冷却时间
|
||
- **审计**：所有信号记录到数据库，可追溯
|
||
|
||
### 接口设计
|
||
|
||
```python
|
||
# engine/signals.py
|
||
import asyncio
|
||
from collections import defaultdict
|
||
from typing import Callable, Awaitable
|
||
|
||
from .common import Signal
|
||
|
||
|
||
SignalHandler = Callable[[Signal], Awaitable[None]]
|
||
|
||
|
||
class SignalBus:
|
||
"""信号事件总线"""
|
||
|
||
def __init__(self):
|
||
self._handlers: dict[str, list[SignalHandler]] = defaultdict(list)
|
||
self._signal_history: list[Signal] = []
|
||
self._cooldowns: dict[str, float] = {} # symbol -> 下次允许信号时间
|
||
|
||
def subscribe(self, event_type: str, handler: SignalHandler) -> None:
|
||
"""
|
||
订阅信号类型。
|
||
|
||
event_type: "order" / "alert" / "log" 或自定义
|
||
handler: async callable，接收 Signal 参数
|
||
"""
|
||
...
|
||
|
||
async def emit(self, signal: Signal) -> None:
|
||
"""
|
||
发布信号。会经过以下处理链：
|
||
1. 冷却检查
|
||
2. 冲突检测
|
||
3. 优先级排序
|
||
4. 分发给所有订阅者
|
||
5. 记录到审计日志
|
||
"""
|
||
...
|
||
|
||
def set_cooldown(self, symbol: str, seconds: float) -> None:
|
||
"""设置 symbol 的下单冷却时间"""
|
||
...
|
||
|
||
def recent_signals(self, symbol: str, n: int = 10) -> list[Signal]:
|
||
"""获取最近的信号历史"""
|
||
...
|
||
|
||
|
||
class SignalFilter:
|
||
"""信号过滤器基类（责任链模式）"""
|
||
|
||
async def filter(self, signal: Signal) -> Signal | None:
|
||
"""返回过滤后的信号，返回 None 表示丢弃"""
|
||
return signal
|
||
|
||
|
||
class CooldownFilter(SignalFilter):
|
||
"""冷却时间过滤器"""
|
||
...
|
||
|
||
|
||
class DuplicateFilter(SignalFilter):
|
||
"""重复信号过滤器（同一 symbol 短期内相同方向合并）"""
|
||
...
|
||
|
||
|
||
class ConfidenceFilter(SignalFilter):
|
||
"""置信度过滤器（低于阈值的信号丢弃）"""
|
||
...
|
||
```
|
||
|
||
### 信号流向
|
||
|
||
```
|
||
策略A.on_kline() ──► Signal ──┐
|
||
策略B.on_ticker() ─► Signal ──┤
|
||
策略C.on_kline() ──► Signal ──┤
|
||
▼
|
||
┌────────────────┐
|
||
│ SignalBus │
|
||
│ │
|
||
│ ┌──────────┐ │
|
||
│ │ 冷却检查 │ │
|
||
│ └────┬─────┘ │
|
||
│ ┌────▼─────┐ │
|
||
│ │ 去重合并 │ │
|
||
│ └────┬─────┘ │
|
||
│ ┌────▼─────┐ │
|
||
│ │ 优先级排序│ │
|
||
│ └────┬─────┘ │
|
||
└───────┼────────┘
|
||
│
|
||
┌─────────────┼─────────────┐
|
||
▼ ▼ ▼
|
||
┌──────────┐ ┌──────────┐ ┌──────────┐
|
||
│ executor │ │ risk/ │ │ monitor/ │
|
||
│ 交易执行 │ │ 风控 │ │ 日志 │
|
||
└──────────┘ └──────────┘ └──────────┘
|
||
```
|
||
|
||
---
|
||
|
||
## 回测引擎
|
||
|
||
回测引擎使用历史 K 线数据模拟策略执行，评估策略在历史行情中的表现。
|
||
|
||
### 设计目标
|
||
|
||
- **事件驱动**：模拟真实行情推送，按时间顺序逐根 K 线喂给策略
|
||
- **精确模拟**：考虑手续费、滑点、最小下单量等实际交易约束
|
||
- **多维度指标**：总收益率、年化收益、夏普比率、最大回撤、胜率、盈亏比
|
||
- **可视化输出**：资金曲线图、回撤曲线、交易点标注
|
||
|
||
### 架构
|
||
|
||
```
|
||
┌──────────────────────────┐
|
||
│ 回测引擎 │
|
||
│ │
|
||
历史K线数据 ─────►│ ┌────────────────────┐ │
|
||
(TimescaleDB) │ │ BacktestEngine │ │
|
||
│ │ │ │
|
||
策略配置 ────────►│ │ - 加载历史K线 │ │
|
||
│ │ - 逐根推送给策略 │ │
|
||
策略类 ──────────►│ │ - 模拟订单成交 │ │
|
||
│ │ - 计算性能指标 │ │
|
||
│ │ - 生成回测报告 │ │
|
||
│ └────────────────────┘ │
|
||
└──────────┬───────────────┘
|
||
│
|
||
▼
|
||
┌──────────────────────────┐
|
||
│ 回测结果 │
|
||
│ - equity_curve │
|
||
│ - trades │
|
||
│ - metrics │
|
||
│ - report (HTML/JSON) │
|
||
└──────────────────────────┘
|
||
```
|
||
|
||
### 接口设计
|
||
|
||
```python
|
||
# engine/backtest.py
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime
|
||
from typing import Optional, Type
|
||
from pydantic import BaseModel
|
||
|
||
from .common import BaseStrategy, StrategyConfig
|
||
|
||
|
||
@dataclass
|
||
class BacktestConfig:
|
||
"""回测配置"""
|
||
symbol: str
|
||
exchange: str = "binance"
|
||
interval: str = "1h"
|
||
start_time: datetime
|
||
end_time: datetime
|
||
|
||
# 交易成本
|
||
commission_pct: float = 0.001 # 手续费率 (0.1%)
|
||
slippage_pct: float = 0.0005 # 滑点 (0.05%)
|
||
min_order_qty: float = 0.001 # 最小下单量
|
||
|
||
# 资金
|
||
initial_capital: float = 10_000.0 # 初始资金
|
||
|
||
# 数据
|
||
warmup_bars: int = 100 # 预热 K 线条数
|
||
|
||
|
||
@dataclass
|
||
class BacktestTrade:
|
||
"""单笔回测交易记录"""
|
||
timestamp: datetime
|
||
symbol: str
|
||
side: str
|
||
price: float
|
||
quantity: float
|
||
commission: float
|
||
slippage: float
|
||
pnl: Optional[float] = None
|
||
reason: str = ""
|
||
|
||
|
||
@dataclass
|
||
class BacktestMetrics:
|
||
"""回测性能指标"""
|
||
total_return_pct: float # 总收益率
|
||
annual_return_pct: float # 年化收益率
|
||
sharpe_ratio: float # 夏普比率
|
||
sortino_ratio: float # 索提诺比率
|
||
max_drawdown_pct: float # 最大回撤
|
||
max_drawdown_duration: int # 最大回撤持续天数
|
||
win_rate: float # 胜率
|
||
profit_factor: float # 盈亏比
|
||
total_trades: int # 总交易次数
|
||
avg_holding_hours: float # 平均持仓时间（小时）
|
||
calmar_ratio: float # 卡尔玛比率
|
||
|
||
|
||
@dataclass
|
||
class BacktestResult:
|
||
"""完整回测结果"""
|
||
config: BacktestConfig
|
||
strategy_config: StrategyConfig
|
||
metrics: BacktestMetrics
|
||
trades: list[BacktestTrade]
|
||
equity_curve: list[dict] # [{timestamp, equity, drawdown}, ...]
|
||
|
||
|
||
class BacktestEngine:
|
||
"""回测引擎"""
|
||
|
||
def __init__(self, config: BacktestConfig):
|
||
self.config = config
|
||
self._klines: list = []
|
||
self._trades: list[BacktestTrade] = []
|
||
self._equity: list[dict] = []
|
||
|
||
async def load_data(self) -> None:
|
||
"""从 TimescaleDB 加载历史 K 线数据"""
|
||
...
|
||
|
||
async def run(
|
||
self,
|
||
strategy_cls: Type[BaseStrategy],
|
||
strategy_config: StrategyConfig,
|
||
) -> BacktestResult:
|
||
"""
|
||
执行回测。
|
||
|
||
流程：
|
||
1. 预热：先喂 warmup_bars 根 K 线但不产生交易
|
||
2. 回测循环：逐根 K 线推送给策略
|
||
3. 收到 Signal 时模拟订单成交
|
||
4. 每根 K 线更新资金曲线
|
||
5. 计算最终指标
|
||
"""
|
||
...
|
||
|
||
async def run_batch(
|
||
self,
|
||
strategy_cls: Type[BaseStrategy],
|
||
configs: list[StrategyConfig],
|
||
max_concurrency: int = 4,
|
||
) -> list[BacktestResult]:
|
||
"""批量回测（并行执行，用于参数扫描）"""
|
||
...
|
||
|
||
@staticmethod
|
||
def compute_metrics(
|
||
equity_curve: list[dict],
|
||
trades: list[BacktestTrade],
|
||
config: BacktestConfig,
|
||
) -> BacktestMetrics:
|
||
"""从资金曲线和交易记录计算性能指标"""
|
||
...
|
||
|
||
def generate_report(self, result: BacktestResult, format: str = "json") -> str:
|
||
"""生成回测报告（json/html/markdown）"""
|
||
...
|
||
|
||
def plot(self, result: BacktestResult) -> None:
|
||
"""绘制资金曲线和回撤图（matplotlib）"""
|
||
...
|
||
```
|
||
|
||
### 回测执行流程
|
||
|
||
```
|
||
开始
|
||
│
|
||
▼
|
||
┌─────────────────────┐
|
||
│ 加载历史 K 线数据 │
|
||
│ (TimescaleDB) │
|
||
└──────────┬──────────┘
|
||
│
|
||
▼
|
||
┌─────────────────────┐
|
||
│ 预热阶段 │
|
||
│ 喂入 warmup_bars │
|
||
│ 策略初始化指标 │
|
||
└──────────┬──────────┘
|
||
│
|
||
▼
|
||
┌─────────────────────┐
|
||
│ 回测主循环 │
|
||
│ │
|
||
│ for kline in data: │
|
||
│ signal = strat. │
|
||
│ on_kline(kline)│
|
||
│ if signal: │
|
||
│ simulate_trade │
|
||
│ update_equity() │
|
||
│ │
|
||
└──────────┬──────────┘
|
||
│
|
||
▼
|
||
┌─────────────────────┐
|
||
│ 计算性能指标 │
|
||
│ - 收益率 │
|
||
│ - 夏普比率 │
|
||
│ - 最大回撤 │
|
||
│ - 胜率等 │
|
||
└──────────┬──────────┘
|
||
│
|
||
▼
|
||
┌─────────────────────┐
|
||
│ 生成回测报告 │
|
||
│ - JSON 结果 │
|
||
│ - 资金曲线图 │
|
||
│ - HTML 报告 │
|
||
└─────────────────────┘
|
||
```
|
||
|
||
### 交易模拟逻辑
|
||
|
||
```
|
||
收到 Signal
|
||
│
|
||
▼
|
||
┌──────────────┐
|
||
│ 输入价格 = │
|
||
│ kline.close │
|
||
│ + slippage │
|
||
└──────┬───────┘
|
||
│
|
||
▼
|
||
┌──────────────┐
|
||
│ 检查余额 │──── 不足 ──► 跳过，记录日志
|
||
│ 检查最小下单量 │
|
||
└──────┬───────┘
|
||
│ 通过
|
||
▼
|
||
┌──────────────┐
|
||
│ 执行成交 │
|
||
│ - 更新持仓 │
|
||
│ - 扣除手续费 │
|
||
│ - 记录交易 │
|
||
│ - 计算盈亏 │
|
||
└──────────────┘
|
||
```
|
||
|
||
---
|
||
|
||
## 参数优化器
|
||
|
||
参数优化器基于回测引擎，自动搜索最优策略参数组合。
|
||
|
||
### 支持的优化策略
|
||
|
||
| 方法 | 适用场景 | 特点 |
|
||
|------|---------|------|
|
||
| **网格搜索** | 参数空间小（2-3 维） | 穷举所有组合，结果完备但慢 |
|
||
| **随机搜索** | 参数空间中等 | 比网格搜索更高效 |
|
||
| **贝叶斯优化** | 参数空间大 | 利用先验知识，收敛快 |
|
||
| **遗传算法** | 复杂非线性参数空间 | 全局搜索能力强 |
|
||
|
||
### 接口设计
|
||
|
||
```python
|
||
# engine/optimizer.py
|
||
from dataclasses import dataclass, field
|
||
from typing import Callable, Any
|
||
|
||
from backtest import BacktestEngine, BacktestConfig, BacktestResult
|
||
from common import BaseStrategy, StrategyConfig
|
||
|
||
|
||
@dataclass
|
||
class ParamSpec:
|
||
"""参数规格定义"""
|
||
name: str
|
||
type: str # "int" / "float" / "choice"
|
||
low: Optional[float] = None # 数值参数下界
|
||
high: Optional[float] = None # 数值参数上界
|
||
step: Optional[float] = None # 步长（网格搜索用）
|
||
choices: Optional[list[Any]] = None # 离散选择
|
||
log_scale: bool = False # 是否对数尺度搜索
|
||
|
||
|
||
@dataclass
|
||
class OptimizationConfig:
|
||
"""优化器配置"""
|
||
method: str = "bayesian" # "grid" / "random" / "bayesian" / "genetic"
|
||
objective: str = "sharpe_ratio" # 优化目标指标
|
||
max_iterations: int = 100 # 最大迭代次数
|
||
n_jobs: int = 4 # 并行任务数
|
||
early_stopping_rounds: int = 10 # 早停轮数
|
||
cv_folds: int = 3 # 交叉验证折数
|
||
|
||
|
||
@dataclass
|
||
class OptimizationResult:
|
||
"""单次优化结果"""
|
||
params: dict[str, Any] # 最优参数
|
||
score: float # 目标函数值
|
||
all_trials: list[dict] # 所有试验记录
|
||
best_result: BacktestResult # 最优参数的回测结果
|
||
|
||
|
||
class ParamOptimizer:
|
||
"""参数优化器"""
|
||
|
||
def __init__(
|
||
self,
|
||
backtest_config: BacktestConfig,
|
||
strategy_cls: Type[BaseStrategy],
|
||
base_config: StrategyConfig,
|
||
param_specs: list[ParamSpec],
|
||
opt_config: OptimizationConfig = OptimizationConfig(),
|
||
):
|
||
...
|
||
|
||
async def optimize(self) -> OptimizationResult:
|
||
"""
|
||
执行参数优化。
|
||
|
||
返回最优参数组合及对应的回测结果。
|
||
"""
|
||
...
|
||
|
||
def suggest_params(self, n: int = 5) -> list[dict]:
|
||
"""获取 top-n 参数组合建议"""
|
||
...
|
||
|
||
def plot_importance(self) -> None:
|
||
"""绘制参数重要性图"""
|
||
...
|
||
|
||
def plot_parallel_coordinates(self) -> None:
|
||
"""绘制参数平行坐标图"""
|
||
...
|
||
```
|
||
|
||
### 优化流程
|
||
|
||
```
|
||
┌────────────────┐
|
||
│ 定义参数空间 │
|
||
│ ParamSpec[] │
|
||
└───────┬────────┘
|
||
│
|
||
▼
|
||
┌────────────────┐
|
||
│ 选择优化算法 │
|
||
│ (网格/贝叶斯等) │
|
||
└───────┬────────┘
|
||
│
|
||
┌────────▼────────┐
|
||
│ 优化主循环 │
|
||
│ │
|
||
│ while n < max: │
|
||
│ params = │
|
||
│ suggest() │
|
||
│ result = │
|
||
│ backtest() │
|
||
│ score = │
|
||
│ metric() │
|
||
│ update() │
|
||
│ check_early()│
|
||
│ │
|
||
└────────┬────────┘
|
||
│
|
||
▼
|
||
┌────────────────┐
|
||
│ 返回最优结果 │
|
||
│ + 所有试验记录 │
|
||
└────────────────┘
|
||
```
|
||
|
||
### 目标函数
|
||
|
||
```
|
||
目标函数可选指标（最小化回撤时取负值）：
|
||
|
||
sharpe_ratio — 夏普比率（风险调整后收益，推荐）
|
||
sortino_ratio — 索提诺比率（只考虑下行波动）
|
||
total_return — 总收益率
|
||
calmar_ratio — 卡尔玛比率（收益/最大回撤）
|
||
profit_factor — 盈亏比
|
||
custom — 自定义组合指标
|
||
```
|
||
|
||
---
|
||
|
||
## 数据流与生命周期
|
||
|
||
### 实时交易数据流
|
||
|
||
```
|
||
Redis Pub/Sub engine/ executor/
|
||
───────────── ───────── ─────────
|
||
|
||
market:kline:* ──► RedisConsumer ──► Manager.feed_kline()
|
||
│
|
||
▼
|
||
路由到对应 symbol 策略
|
||
│
|
||
┌─────┴─────┐
|
||
▼ ▼
|
||
策略A.on_kline 策略B.on_kline
|
||
│ │
|
||
Signal? Signal?
|
||
│ │
|
||
└─────┬─────┘
|
||
▼
|
||
SignalBus.emit()
|
||
│
|
||
┌───────────┼───────────┐
|
||
▼ ▼ ▼
|
||
冷却检查 去重合并 优先级排序
|
||
│ │ │
|
||
└───────────┼───────────┘
|
||
▼
|
||
┌──────────┐
|
||
│ risk/ │ ← 风控检查
|
||
│ manager │
|
||
└────┬─────┘
|
||
│ passed
|
||
▼
|
||
┌──────────┐
|
||
│ executor │ ← 实际下单
|
||
└──────────┘
|
||
│
|
||
▼
|
||
订单状态回调
|
||
│
|
||
▼
|
||
Manager.notify_order_filled()
|
||
│
|
||
▼
|
||
策略.on_order_filled()
|
||
```
|
||
|
||
### 回测数据流
|
||
|
||
```
|
||
TimescaleDB ──► BacktestEngine.load_data()
|
||
│
|
||
▼
|
||
历史 K 线列表
|
||
│
|
||
▼
|
||
BacktestEngine.run()
|
||
│
|
||
▼
|
||
逐根推送 K 线 ──► 策略.on_kline()
|
||
│ │
|
||
│ Signal?
|
||
│ │
|
||
▼ ▼
|
||
更新资金曲线 simulate_trade()
|
||
│
|
||
▼
|
||
计算指标 ──► BacktestResult
|
||
```
|
||
|
||
### 引擎主循环
|
||
|
||
```python
|
||
# engine/main.py — 主程序入口示意
|
||
import asyncio
|
||
from config import settings
|
||
from common.storage import TimescaleDBReader
|
||
from manager import StrategyManager
|
||
from signals import SignalBus
|
||
|
||
|
||
async def main():
|
||
# 1. 初始化信号总线
|
||
signal_bus = SignalBus()
|
||
|
||
# 2. 初始化策略管理器
|
||
manager = StrategyManager(signal_bus)
|
||
manager.discover_strategies("strategies")
|
||
|
||
# 3. 创建并注册策略实例
|
||
for cfg in settings.strategies:
|
||
manager.create(cfg.type, cfg)
|
||
|
||
# 4. 启动所有策略
|
||
await manager.start_all()
|
||
|
||
# 5. 订阅 Redis 行情
|
||
consumer = RedisMarketConsumer(settings.redis_url)
|
||
await consumer.subscribe_kline(manager.feed_kline)
|
||
await consumer.subscribe_ticker(manager.feed_ticker)
|
||
|
||
# 6. 保持运行
|
||
await asyncio.Event().wait()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|
||
```
|
||
|
||
---
|
||
|
||
## 与上下游模块的交互
|
||
|
||
### 上游：TypeScript 数据模块
|
||
|
||
引擎通过 Redis Pub/Sub 消费 TS 侧推送的行情数据：
|
||
|
||
| Redis 频道 | 数据类型 | 对应策略方法 |
|
||
|-----------|---------|-------------|
|
||
| `market:ticker:{exchange}:{symbol}` | Ticker | `on_ticker()` |
|
||
| `market:kline:{exchange}:{symbol}:{interval}` | Kline | `on_kline()` |
|
||
| `market:trade:{exchange}:{symbol}` | Trade | `on_trade()` |
|
||
| `market:orderbook:{exchange}:{symbol}` | OrderBook | `on_orderbook()` |
|
||
|
||
数据格式使用 **MessagePack** 序列化，Python 侧用 `msgpack.unpackb()` 反序列化为 Pydantic 模型。
|
||
|
||
### 下游：交易执行模块
|
||
|
||
引擎通过 `SignalBus` 发布信号，executor 订阅：
|
||
|
||
```python
|
||
# executor 侧订阅
|
||
signal_bus.subscribe("order", executor.handle_signal)
|
||
```
|
||
|
||
### 下游：风控模块
|
||
|
||
风控模块也订阅信号总线，在 executor 执行前进行风控检查：
|
||
|
||
```python
|
||
# risk 侧订阅
|
||
signal_bus.subscribe("order", risk_manager.check_signal)
|
||
```
|
||
|
||
### 下游：监控模块
|
||
|
||
监控模块订阅所有信号用于记录和统计：
|
||
|
||
```python
|
||
# monitor 侧订阅
|
||
signal_bus.subscribe("*", monitor.record_signal)
|
||
```
|
||
|
||
---
|
||
|
||
## 目录结构
|
||
|
||
```
|
||
engine/
|
||
├── ENGINE.md # 本文档
|
||
├── __init__.py
|
||
├── env.yaml # 引擎环境配置
|
||
│
|
||
├── common/ # 通用模块（基础数据类型、策略基类、日志）
|
||
│ ├── __init__.py
|
||
│ ├── base.py # BaseStrategy 抽象基类 + Signal / StrategyConfig
|
||
│ ├── models.py # 数据模型：Ticker / Kline / Trade / OrderBook
|
||
│ └── logger.py # 结构化日志模块
|
||
│
|
||
├── manager.py # StrategyManager 策略管理器
|
||
├── signals.py # SignalBus 信号分发器 + SignalFilter 过滤器
|
||
├── backtest.py # BacktestEngine 回测引擎
|
||
├── optimizer.py # ParamOptimizer 参数优化器
|
||
├── main.py # 引擎主循环入口
|
||
│
|
||
├── indicators/ # 技术指标计算（基于 TA-Lib / pandas_ta）
|
||
│ ├── __init__.py
|
||
│ ├── trend.py # 趋势指标 (MA, EMA, MACD, ADX...)
|
||
│ ├── momentum.py # 动量指标 (RSI, Stochastic, CCI...)
|
||
│ ├── volatility.py # 波动率指标 (Bollinger, ATR, Keltner...)
|
||
│ ├── volume.py # 成交量指标 (OBV, VWAP, MFI...)
|
||
│ └── composite.py # 复合指标（自定义组合）
|
||
│
|
||
├── consumers/ # 行情消费器（Redis → 策略馈送）
|
||
│ ├── __init__.py
|
||
│ ├── redis.py # Redis Pub/Sub 消费者
|
||
│ └── nats.py # NATS 消费者（可选）
|
||
│
|
||
└── tests/ # 引擎测试
|
||
├── __init__.py
|
||
├── test_base.py
|
||
├── test_manager.py
|
||
├── test_signals.py
|
||
├── test_backtest.py
|
||
├── test_optimizer.py
|
||
└── fixtures/ # 测试夹具（模拟行情数据）
|
||
└── sample_klines.json
|
||
```
|
||
|
||
---
|
||
|
||
## 开发步骤
|
||
|
||
### 第一阶段：核心框架（第 1 周）
|
||
|
||
| 步骤 | 任务 | 产出 |
|
||
|------|------|------|
|
||
| 1.1 | 实现 `common/base.py` — `BaseStrategy`、`Signal`、`StrategyConfig` | 策略基类和类型定义 |
|
||
| 1.2 | 实现 `common/models.py` — `Ticker`、`Kline`、`Trade`、`OrderBook` Pydantic 模型 | 与 TS 对齐的数据模型 |
|
||
| 1.3 | 实现 `common/logger.py` — 结构化日志模块 | 统一日志输出 |
|
||
| 1.3 | 实现 `common/logger.py` — 结构化日志模块 | 统一日志输出 |
|
||
| 1.4 | 实现 `signals.py` — `SignalBus`、`CooldownFilter`、`DuplicateFilter` | 事件总线和过滤链 |
|
||
| 1.5 | 实现 `manager.py` — 策略注册、创建、启停 | 策略管理器 |
|
||
| 1.6 | 实现 `consumers/redis.py` — Redis Pub/Sub 消费者 | 实时行情接收 |
|
||
| 1.7 | 编写 `main.py` — 引擎主循环串联 | 端到端可运行 |
|
||
|
||
### 第二阶段：指标与策略示例（第 2 周）
|
||
|
||
| 步骤 | 任务 | 产出 |
|
||
|------|------|------|
|
||
| 2.1 | 实现 `indicators/` — MA、EMA、MACD、RSI、Bollinger | 技术指标库 |
|
||
| 2.2 | 编写示例策略 1：双均线交叉 | `strategies/ma_cross.py` |
|
||
| 2.3 | 编写示例策略 2：网格交易 | `strategies/grid_trading.py` |
|
||
| 2.4 | 集成测试：Redis 行情 → 策略 → 信号 | 端到端验证 |
|
||
|
||
### 第三阶段：回测（第 3 周）
|
||
|
||
| 步骤 | 任务 | 产出 |
|
||
|------|------|------|
|
||
| 3.1 | 实现 `backtest.py` — 回测引擎核心 | 事件驱动回测 |
|
||
| 3.2 | 实现交易模拟逻辑（手续费、滑点） | 精确回测 |
|
||
| 3.3 | 实现性能指标计算（夏普比率、最大回撤等） | 完整指标 |
|
||
| 3.4 | 实现回测报告生成（JSON/HTML） | 可视化报告 |
|
||
| 3.5 | 编写回测测试用例 | 回测验证 |
|
||
|
||
### 第四阶段：参数优化（第 4 周）
|
||
|
||
| 步骤 | 任务 | 产出 |
|
||
|------|------|------|
|
||
| 4.1 | 实现 `optimizer.py` — 网格搜索 | 穷举搜索 |
|
||
| 4.2 | 集成 Optuna 实现贝叶斯优化 | 智能搜索 |
|
||
| 4.3 | 实现交叉验证 | 防止过拟合 |
|
||
| 4.4 | 实现参数重要性分析 | 参数洞察 |
|
||
|
||
---
|
||
|
||
## 实现规范
|
||
|
||
### 异步编程规范
|
||
|
||
- 所有 I/O 操作使用 `async/await`
|
||
- 长时间计算任务使用 `asyncio.to_thread()` 放入线程池
|
||
- 避免在协程中阻塞（不用 `time.sleep`，用 `asyncio.sleep`）
|
||
- 策略的 `on_*` 方法必须是非阻塞的快速返回
|
||
|
||
### 错误处理
|
||
|
||
```python
|
||
# 策略错误隔离模式
|
||
async def feed_kline(self, kline) -> None:
|
||
for name, strategy in self._strategies.items():
|
||
if not strategy.is_running:
|
||
continue
|
||
if strategy.config.symbol != kline.symbol:
|
||
continue
|
||
try:
|
||
signal = await strategy.on_kline(kline)
|
||
if signal:
|
||
await self.signal_bus.emit(signal)
|
||
except Exception as e:
|
||
logger.error(f"Strategy {name} error: {e}", exc_info=True)
|
||
# 单策略异常不影响其他策略
|
||
```
|
||
|
||
### 日志规范
|
||
|
||
```python
|
||
from common.logger import logger
|
||
|
||
# 使用结构化日志
|
||
logger.info("strategy_signal", strategy="ma_cross", symbol="BTCUSDT",
|
||
side="BUY", confidence=0.85, timestamp=1234567890)
|
||
logger.warning("strategy_error", strategy="grid", error=str(e))
|
||
logger.info("backtest_complete", symbol="ETHUSDT", sharpe=1.45, trades=230)
|
||
```
|
||
|
||
### 类型规范
|
||
|
||
- 所有公开接口使用类型注解
|
||
- 数据模型使用 Pydantic `BaseModel`
|
||
- 策略配置继承 `StrategyConfig` 并扩展
|
||
- 信号必须通过 `Signal` 模型创建，禁止使用裸 dict
|
||
|
||
### 测试规范
|
||
|
||
- 核心逻辑（SignalBus、StrategyManager）单元测试覆盖率 > 90%
|
||
- 策略逻辑有独立单元测试
|
||
- 回测引擎用已知结果的静态数据集验证
|
||
- 使用 `pytest-asyncio` 支持异步测试
|
||
|
||
### 性能基准
|
||
|
||
| 指标 | 目标值 | 说明 |
|
||
|------|--------|------|
|
||
| 策略单次 `on_kline` 耗时 | < 1ms | 不含 I/O 的纯计算 |
|
||
| SignalBus 发送延迟 | < 0.1ms | emit 到 handler 开始执行 |
|
||
| 回测速度 | > 10000 bars/s | 10 万根 K 线 < 10 秒 |
|
||
| 100 策略并行 | CPU < 50% | 空策略情况下的资源占用 |
|
||
|
||
---
|
||
|
||
## 附录
|
||
|
||
### A. 示例策略实现（双均线交叉）
|
||
|
||
```python
|
||
# strategies/ma_cross.py
|
||
from typing import Optional
|
||
from common import BaseStrategy, StrategyConfig, Signal
|
||
|
||
|
||
class MACrossConfig(StrategyConfig):
|
||
fast_period: int = 7
|
||
slow_period: int = 25
|
||
|
||
|
||
class MACrossStrategy(BaseStrategy):
|
||
strategy_type = "ma_cross"
|
||
|
||
def __init__(self, config: MACrossConfig):
|
||
super().__init__(config)
|
||
self.config: MACrossConfig = config
|
||
self._prices: list[float] = []
|
||
self._last_signal: Optional[str] = None # "BUY" / "SELL"
|
||
|
||
def _sma(self, data: list[float], period: int) -> float:
|
||
if len(data) < period:
|
||
return 0.0
|
||
return sum(data[-period:]) / period
|
||
|
||
async def on_kline(self, kline) -> Optional[Signal]:
|
||
self._prices.append(kline.close)
|
||
|
||
fast = self._sma(self._prices, self.config.fast_period)
|
||
slow = self._sma(self._prices, self.config.slow_period)
|
||
|
||
if fast == 0 or slow == 0:
|
||
return None
|
||
|
||
if fast > slow and self._last_signal != "BUY":
|
||
self._last_signal = "BUY"
|
||
return Signal(
|
||
symbol=self.config.symbol,
|
||
side="BUY",
|
||
signal_type="MARKET",
|
||
confidence=0.7,
|
||
reason=f"Golden cross: MA{self.config.fast_period} > MA{self.config.slow_period}",
|
||
timestamp=kline.timestamp,
|
||
)
|
||
|
||
if fast < slow and self._last_signal != "SELL":
|
||
self._last_signal = "SELL"
|
||
return Signal(
|
||
symbol=self.config.symbol,
|
||
side="SELL",
|
||
signal_type="MARKET",
|
||
confidence=0.7,
|
||
reason=f"Death cross: MA{self.config.fast_period} < MA{self.config.slow_period}",
|
||
timestamp=kline.timestamp,
|
||
)
|
||
|
||
return None
|
||
```
|
||
|
||
### B. 引擎配置示例（engine/env.yaml）
|
||
|
||
```yaml
|
||
# engine/env.yaml
|
||
engine:
|
||
name: "trade-engine"
|
||
strategies_package: "strategies"
|
||
max_concurrent_strategies: 50
|
||
|
||
redis:
|
||
url: "redis://localhost:6379/0"
|
||
channels:
|
||
kline: "market:kline:*"
|
||
ticker: "market:ticker:*"
|
||
orderbook: "market:orderbook:*"
|
||
|
||
signals:
|
||
cooldown_seconds: 1.0 # 同 symbol 下单冷却时间
|
||
min_confidence: 0.3 # 最低信号置信度
|
||
max_signals_per_minute: 60 # 每分钟最大信号数
|
||
|
||
backtest:
|
||
default_commission: 0.001
|
||
default_slippage: 0.0005
|
||
default_capital: 10000.0
|
||
|
||
optimizer:
|
||
default_method: "bayesian"
|
||
default_objective: "sharpe_ratio"
|
||
max_iterations: 200
|
||
n_jobs: 4
|
||
early_stopping: 20
|
||
```
|
||
|
||
### C. 相关文档索引
|
||
|
||
| 文档 | 路径 |
|
||
|------|------|
|
||
| 系统总体架构 | `../README.md` |
|
||
| 数据模块开发指南 | `../data/ENGINE.md` (待创建) |
|
||
| 交易执行模块 | `../executor/ENGINE.md` (待创建) |
|
||
| 风控模块 | `../risk/ENGINE.md` (待创建) |
|
||
| 公共模块 | `./common/__init__.py` |
|