背景¶
backtrader 已经比较完善了，我想要借鉴量化投资框架中其他项目的优势，继续改进优化 backtrader。
任务¶
阅读研究分析 backtrader 这个项目的源代码，了解这个项目。
阅读研究分析/Users/yunjinqi/Documents/量化交易框架/signal_trading
借鉴这个新项目的优点和功能，给 backtrader 优化改进提供新的建议
写需规文档和设计文档放到这个文档的最下面，方便后续借鉴
signal_trading 项目简介¶
signal_trading 是一个交易信号生成和管理框架，具有以下核心特点：
信号生成: 交易信号生成
信号管理: 信号管理系统
信号聚合: 多信号聚合
信号评估: 信号有效性评估
信号回测: 信号回测验证
信号订阅: 信号订阅分发
重点借鉴方向¶
信号系统: 信号系统设计
信号定义: 信号定义标准
信号聚合: 多信号聚合
信号评估: 信号评估方法
信号分发: 信号分发机制
信号存储: 信号持久化
signal_trading 项目分析与 backtrader 优化设计文档¶
一、signal_trading 项目分析¶
1.1 项目概述¶
signal_trading 是一个基于 backtrader 构建的加密货币交易信号生成和管理框架，具有以下特点：
| 特性 | 描述 |
|——|——|
| 架构模式| 基于 backtrader 的策略模式实现 |
|CLI 工具| 使用 Click 框架提供命令行接口 |
|数据源| 支持 CCXT 加密货币交易所数据 |
|存储| SQLAlchemy ORM 用于交易记录持久化 |
|日志| 使用 loguru 进行结构化日志记录 |
|参数优化 | 支持多参数组合并行回测 |
1.2 项目结构分析¶
signal_trading/
├── signal_trading.py # 主入口，启动横幅显示
├── cli/ # 命令行接口模块
│ ├── cli.py # Click 命令组路由
│ ├── back_strategy.py # 回测命令实现
│ ├── live_trading.py # 实盘交易命令
│ ├── candles.py # K 线数据获取
│ └── utils.py # 工具函数和结果处理
├── strategy/ # 交易策略实现
│ ├── SMA.py # 简单移动平均策略
│ ├── EMA.py # 指数移动平均策略
│ ├── EMACross.py # EMA 交叉策略
│ ├── Busy.py # 均值回归策略
│ ├── Oscillation.py # RSI+布林带震荡策略
│ └── AscendWave.py # 趋势跟踪策略
├── broker/ # 经纪商实现
│ ├── CCXTBroker.py # CCXT 交易所适配器
│ ├── CCXTData.py # CCXT 数据源
│ ├── CCXTStore.py # 存储实现
│ └── OKXData.py # OKX 交易所专用
├── analyzer/ # 分析器
│ └── PositionReturn.py # 持仓收益分析
└── model/ # 数据模型
└── TradeModel.py # SQLAlchemy 交易模型
```bash
### 1.3 核心设计模式分析
#### 1.3.1 状态机模式（Busy 策略）
```python
class Busy(bt.Strategy):
def __init__(self):
self._open_order = None # 跟踪未完成订单
self.op = bt.Order.Buy # 当前期望操作状态
self.buy_price = None # 记录买入价格
def next(self):
if self._open_order:
return # 有未完成订单，不生成新信号
if self.op == bt.Order.Buy:
# 生成买入信号
if self.data.close[0] <= self.short_ma[0] *(1 - self.params.below):
self._open_order = True
self.buy(...)
elif self.op == bt.Order.Sell:
# 生成卖出信号
if self.data.close[0] >= self.buy_price*(1 + self.params.net_profit):
self._open_order = True
self.sell(...)
def notify_order(self, order):
if order.status == bt.Order.Completed:
self._open_order = False
self.op = bt.Order.Sell if order.isbuy() else bt.Order.Buy
```bash
- *设计优点**：
- 清晰的状态转换逻辑
- 防止在未完成订单时生成新信号
- 状态切换由订单完成事件驱动
#### 1.3.2 多指标确认机制（Oscillation 策略）
```python
class Oscillation(bt.Strategy):
def handle_oscillating_market(self):
# 多指标组合确认
recent_rsi_list = np.array([self.rsi[-i] for i in range(1, 5)])
is_rsi_downward = np.all(np.diff(recent_rsi_list) < 0)
close_values = np.array([self.data.close[-i] for i in range(3)])
close_trend = np.all(np.diff(close_values) >= 0)
# 买入信号需要多个条件同时满足
if (self.rsi[0] < self.p.rsi_buy_signal and
is_rsi_downward and
current_close > self.data.close[-1]):
# 生成买入信号
self._sumit_buy_order(...)
```bash
- *设计优点**：
- 多指标相互验证，提高信号可靠性
- 趋势判断使用 numpy 数组操作提高效率
- 可配置的信号阈值参数
#### 1.3.3 风险管理框架
```python
class Oscillation(bt.Strategy):
def risk_management(self):
if self._op == bt.Order.Sell:
current_price = self.data.close[0]
# 动态止损检查
if current_price < (self._buy_price *(1 - self.p.stop_loss)):
self._sumit_sell_order(current_price, size, bt.Order.Limit)
def next(self):
# 先检查风险管理
self.risk_management()
# 再处理交易信号
self.handle_oscillating_market()
```bash
- *设计优点**：
- 风险检查优先于信号生成
- 止损基于入场价格动态计算
- 独立的风险管理方法，便于测试和维护
#### 1.3.4 参数化 CLI 工具
```python
@click.group()
@click.option('--cash', default=10000, help='初始投入资金')
@click.option('--f', '--file', 'filepath', required=True, help="数据文件 csv")
@click.option('--opt', default=False, is_flag=True, help="参数寻优")
def back_strategy(ctx, cash, filepath, opt):
"""策略回测"""
ctx.obj = {'cash': cash, 'filepath': filepath, 'opt': opt}
@back_strategy.command()
@click.option('--short_period', type=COMMA_SEPARATED_LIST_INT, required=True)
@click.option('--long_period', type=COMMA_SEPARATED_LIST_INT, required=True)
@click.option('--below', type=COMMA_SEPARATED_LIST, required=True)
def sma_busy(ctx, short_period, long_period, below):
# 支持多参数组合
cerebro.optstrategy(Busy,
short_period=short_period,
long_period=long_period,
below=below)
results = cerebro.run()
```bash
- *设计优点**：
- 命令行接口清晰，参数可配置
- 支持逗号分隔的多参数值进行优化
- 结果自动导出为 Excel 格式
#### 1.3.5 交易记录持久化
```python
class TradeRecord(Base):
__tablename__ = 'TradeRecord'
trade_id = Column(String(200), nullable=False)
symbol = Column(String(200), nullable=False)
timestamp = Column(String(20), nullable=False)
exec_type = Column(String(20), nullable=False)
price = Column(String(200), default='0')
status = Column(String(20), nullable=False)
side = Column(String(20), nullable=False)
executed_price = Column(String(200), default='0')
executed_size = Column(String(200), default='0')
fee = Column(JSON) # 支持 JSON 格式存储复杂费用结构
```bash
- *设计优点**：
- 完整的交易记录结构
- JSON 字段支持复杂的费用结构
- 时区感知的时间戳处理
### 1.4 信号系统特点总结
#### 1.4.1 信号生成特点
| 特点 | 实现 |
|------|------|
| **条件触发**| 基于技术指标阈值触发信号 |
|**多指标确认**| 组合多个指标进行信号验证 |
|**趋势判断**| 使用 numpy 数组操作判断价格趋势 |
|**状态管理**| 状态机模式管理买入/卖出状态 |
|**订单跟踪**| 跟踪未完成订单防止重复信号 |
#### 1.4.2 信号管理特点
| 特点 | 实现 |
|------|------|
|**订单状态跟踪**| `_open_order`标志跟踪订单状态 |
|**操作状态管理**| `op`变量管理期望操作方向 |
|**入场价格记录**| `buy_price`记录用于止损止盈计算 |
|**执行结果统计** | 胜率、盈亏比等统计指标 |
#### 1.4.3 信号评估特点
```python
def generate_combinations_report(self):
"""生成策略执行报告"""
avg_profit = self.TotalProfit / self.WinningTrades if self.WinningTrades else 0
avg_loss = self.TotalLoss / self.LosingTrades if self.LosingTrades else 0
winning_rate = (self.WinningTrades /
(self.WinningTrades + self.LosingTrades) * 100) if self.WinningTrades else 0
profit_loss_ratio = (avg_profit / avg_loss) if avg_loss != 0 else 0
return {
"手续费": self.commission,
"胜率": f"{winning_rate:.2f}%",
"获胜": self.WinningTrades,
"失败": self.LosingTrades,
"盈亏比": f"{profit_loss_ratio:.2f}",
"利润(平均)": avg_profit,
"亏损(平均)": avg_loss,
"总利润": self.TotalProfit,
"总亏损": self.TotalLoss,
"净利润": self.TotalProfit - self.TotalLoss - self.commission,
}
```bash
### 1.5 与 backtrader 的对比分析
| 特性 | backtrader 现状 | signal_trading 实现 |
|------|---------------|-------------------|
| **信号系统**| 内置于 Strategy 中 | 独立的信号生成/管理模块 |
|**状态管理**| 由用户自行实现 | 标准化的状态机模式 |
|**风险管理**| 通过订单管理 | 独立的风险管理方法 |
|**CLI 工具**| 无 | 完整的 Click CLI |
|**参数优化**| `optstrategy` | CLI 参数+Excel 导出 |
|**交易记录**| 内存中 | SQLAlchemy 持久化 |
|**日志系统**| 标准 logging | loguru 结构化日志 |
|**统计报告**| Analyzer 分析器 | 策略内置统计 |
### 1.6 可借鉴的核心优势
1.**标准化状态管理**：统一的状态机模式处理买入/卖出转换
1. **独立风险管理**：将风险检查与信号生成分离
2. **多指标确认机制**：提高信号可靠性的组合指标方法
3. **CLI 工具链**：完整的命令行工作流支持
4. **交易记录持久化**：完整的交易历史存储方案
5. **结构化日志**：loguru 提供更好的日志体验
6. **参数优化工作流**：从回测到结果导出的完整流程
- --
## 二、需求规格说明
### 2.1 功能性需求
#### FR1: 信号系统基础架构
- *描述**：为 backtrader 构建一个标准化的交易信号系统，提供信号生成、状态管理和生命周期管理的统一接口。
- *需求详情**：
- FR1.1: 定义`Signal`基类，包含信号类型、强度、触发条件等属性
- FR1.2: 实现`SignalGenerator`抽象基类，定义信号生成接口
- FR1.3: 实现`SignalManager`类，管理信号状态和生命周期
- FR1.4: 支持信号优先级和信号过滤机制
- FR1.5: 提供信号历史记录功能
- *验收标准**：
- 信号系统与现有 Strategy 类兼容
- 信号状态转换正确无误
- 支持多种信号类型（买入、卖出、平仓、持仓调整）
#### FR2: 信号状态管理
- *描述**：实现标准化的信号状态机，管理从生成到执行完成的完整生命周期。
- *需求详情**：
- FR2.1: 定义信号状态枚举（PENDING, SUBMITTED, FILLED, CANCELLED, REJECTED）
- FR2.2: 实现状态转换逻辑和验证
- FR2.3: 支持信号超时自动取消
- FR2.4: 提供信号状态查询接口
- FR2.5: 支持信号依赖关系（前置信号完成才触发后续信号）
- *验收标准**：
- 状态转换遵循预定义规则
- 异常状态能够正确处理
- 状态查询响应时间<1ms
#### FR3: 多指标信号确认
- *描述**：支持多个技术指标组合确认，提高信号可靠性。
- *需求详情**：
- FR3.1: 实现`SignalConfirmator`基类
- FR3.2: 提供常用确认策略（AND、OR、加权投票、机器学习）
- FR3.3: 支持自定义确认函数
- FR3.4: 提供指标权重配置
- FR3.5: 支持动态确认阈值调整
- *验收标准**：
- 支持至少 3 种确认策略
- 确认延迟<100ms
- 支持最多 10 个指标组合确认
#### FR4: 风险管理系统
- *描述**：将风险管理从信号生成中分离，提供独立的风险检查和决策模块。
- *需求详情**：
- FR4.1: 实现`RiskManager`基类
- FR4.2: 提供常用风险策略（止损、止盈、仓位控制、最大回撤）
- FR4.3: 支持动态止损（移动止损、ATR 止损）
- FR4.4: 支持组合级别风险控制
- FR4.5: 提供风险事件日志和统计
- *验收标准**：
- 风险检查在信号生成前执行
- 支持至少 5 种风险策略
- 风险事件记录完整
#### FR5: 信号评估系统
- *描述**：提供信号有效性评估和统计功能，支持策略优化。
- *需求详情**：
- FR5.1: 实现信号执行率统计
- FR5.2: 计算信号胜率、盈亏比
- FR5.3: 提供信号分布分析（按时间段、市场状态）
- FR5.4: 支持信号归因分析
- FR5.5: 提供信号质量评分
- *验收标准**：
- 提供至少 10 种评估指标
- 统计结果准确率>99.9%
- 支持历史信号回溯分析
#### FR6: 信号持久化
- *描述**：提供信号存储和查询功能，支持信号历史分析和审计。
- *需求详情**：
- FR6.1: 实现信号数据模型
- FR6.2: 支持多种存储后端（SQLite、PostgreSQL、MongoDB）
- FR6.3: 提供信号查询接口
- FR6.4: 支持信号导出（CSV、Excel、JSON）
- FR6.5: 提供信号数据清理和归档功能
- *验收标准**：
- 支持至少 3 种存储后端
- 信号写入延迟<10ms
- 支持百万级信号查询
#### FR7: CLI 工具链
- *描述**：提供完整的命令行工具，支持策略开发、回测、优化和部署。
- *需求详情**：
- FR7.1: 实现策略回测命令
- FR7.2: 实现参数优化命令
- FR7.3: 实现信号分析命令
- FR7.4: 实现策略部署命令
- FR7.5: 提供配置文件支持（TOML/YAML）
- *验收标准**：
- 命令行操作符合 Unix 惯例
- 命令执行有清晰提示
- 错误信息友好可读
#### FR8: 结构化日志
- *描述**：集成结构化日志系统，提供更好的调试和监控能力。
- *需求详情**：
- FR8.1: 集成 loguru 或类似日志库
- FR8.2: 支持日志级别动态调整
- FR8.3: 提供结构化日志输出（JSON 格式）
- FR8.4: 支持日志轮转和归档
- FR8.5: 集成信号事件日志
- *验收标准**：
- 日志不影响策略性能（<5%开销）
- 支持日志查询和过滤
- 日志格式统一规范
### 2.2 非功能性需求
#### NFR1: 性能要求
- 信号生成延迟 < 1ms
- 信号管理开销 < 2%
- 支持每秒 1000+信号处理
- 内存占用增长线性可控
#### NFR2: 可扩展性
- 支持自定义信号类型
- 支持插件式风险策略
- 支持自定义存储后端
- 支持自定义日志格式
#### NFR3: 兼容性
- 与现有 backtrader API 完全兼容
- 支持 Python 3.8+
- 支持 Windows/Linux/macOS
#### NFR4: 可靠性
- 信号状态持久化，支持断线恢复
- 异常处理完善，不导致主流程中断
- 提供信号校验机制
- --
## 三、系统设计
### 3.1 架构设计
#### 3.1.1 整体架构
```bash
┌─────────────────────────────────────────────────────────────────┐
│ Application Layer │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ CLI Tools │ │ Web API │ │ Notebooks │ │
│ └───────┬──────┘ └──────┬───────┘ └──────┬───────┘ │
└──────────┼─────────────────┼──────────────────┼─────────────────┘
│ │ │
┌──────────┼─────────────────┼──────────────────┼─────────────────┐
│ │ Signal Trading Framework Layer │
│ ┌───────▼────────┐ ┌─────▼──────┐ ┌─────▼──────┐ │
│ │ SignalEngine │ │RiskManager │ │ SignalStore │ │
│ └───────┬────────┘ └─────┬──────┘ └─────┬──────┘ │
│ ┌───────▼────────┐ ┌─────▼──────┐ ┌─────▼──────┐ │
│ │SignalGenerator │ │SignalEvaluator│ SignalReporter│ │
│ └────────────────┘ └────────────┘ └────────────┘ │
└──────────┼─────────────────┼──────────────────┼─────────────────┘
│ │ │
┌──────────┼─────────────────┼──────────────────┼─────────────────┐
│ │ Backtrader Core Layer │
│ ┌───────▼────────┐ ┌─────▼──────┐ ┌─────▼──────┐ │
│ │ Cerebro │ │ Strategy │ │ Analyzer │ │
│ └────────────────┘ └────────────┘ └────────────┘ │
└─────────────────────────────────────────────────────────────────┘
```bash
#### 3.1.2 模块组织
```bash
backtrader/
├── signals/ # 信号系统模块
│ ├── __init__.py
│ ├── base.py # 信号基类和枚举
│ ├── generator.py # 信号生成器
│ ├── manager.py # 信号管理器
│ ├── confirmator.py # 信号确认器
│ ├── evaluator.py # 信号评估器
│ └── registry.py # 信号注册表
├── risk/ # 风险管理模块
│ ├── __init__.py
│ ├── base.py # 风险管理基类
│ ├── managers.py # 风险管理器实现
│ ├── limits.py # 风险限制器
│ └── events.py # 风险事件
├── storage/ # 存储模块
│ ├── __init__.py
│ ├── base.py # 存储基类
│ ├── signal_store.py # 信号存储
│ ├── models.py # 数据模型
│ └── backends.py # 存储后端
├── cli/ # CLI 工具模块
│ ├── __init__.py
│ ├── main.py # 主命令入口
│ ├── backtest.py # 回测命令
│ ├── optimize.py # 优化命令
│ ├── analyze.py # 分析命令
│ └── utils.py # 工具函数
└── utils/
├── logger.py # 日志配置
└── formatters.py # 格式化工具
```bash
### 3.2 详细设计
#### 3.2.1 信号基类设计
```python
# backtrader/signals/base.py
from enum import Enum, auto
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, Dict, Any, List
import uuid
class SignalType(Enum):
"""信号类型枚举"""
BUY = auto()
SELL = auto()
SHORT = auto()
COVER = auto()
HOLD = auto()
ADJUST = auto()
class SignalStatus(Enum):
"""信号状态枚举"""
PENDING = auto() # 待处理
CONFIRMING = auto() # 确认中
CONFIRMED = auto() # 已确认
SUBMITTED = auto() # 已提交
PARTIAL_FILLED = auto() # 部分成交
FILLED = auto() # 已成交
CANCELLED = auto() # 已取消
REJECTED = auto() # 已拒绝
EXPIRED = auto() # 已过期
class SignalPriority(Enum):
"""信号优先级"""
LOW = 1
NORMAL = 2
HIGH = 3
EMERGENCY = 4
@dataclass
class Signal:
"""信号数据类"""
# 基本信息
signal_id: str = field(default_factory=lambda: str(uuid.uuid4()))
signal_type: SignalType = SignalType.HOLD
status: SignalStatus = SignalStatus.PENDING
# 目标信息
data_name: str = '' # 数据源名称
target_symbol: str = '' # 目标标的
# 价格和数量
price: Optional[float] = None
size: Optional[float] = None
order_type: str = 'market' # market, limit, stop, stop_limit
# 元数据
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
expires_at: Optional[datetime] = None
priority: SignalPriority = SignalPriority.NORMAL
# 信号来源
generator: str = '' # 生成器名称
strategy_id: str = '' # 策略 ID
# 上下文信息
context: Dict[str, Any] = field(default_factory=dict)
# 确认信息
confirmations: List[str] = field(default_factory=list)
rejection_reason: Optional[str] = None
# 关联信息
parent_signal_id: Optional[str] = None
child_signal_ids: List[str] = field(default_factory=list)
def __post_init__(self):
"""初始化后处理"""
if self.signal_type == SignalType.HOLD:
self.status = SignalStatus.FILLED
def is_active(self) -> bool:
"""检查信号是否处于活跃状态"""
return self.status in (
SignalStatus.PENDING,
SignalStatus.CONFIRMING,
SignalStatus.CONFIRMED,
SignalStatus.SUBMITTED,
SignalStatus.PARTIAL_FILLED
)
def is_completed(self) -> bool:
"""检查信号是否已完成"""
return self.status in (
SignalStatus.FILLED,
SignalStatus.CANCELLED,
SignalStatus.REJECTED,
SignalStatus.EXPIRED
)
def is_expired(self) -> bool:
"""检查信号是否过期"""
if self.expires_at is None:
return False
return datetime.now() > self.expires_at
def add_confirmation(self, source: str):
"""添加确认来源"""
if source not in self.confirmations:
self.confirmations.append(source)
self.updated_at = datetime.now()
def to_dict(self) -> Dict[str, Any]:
"""转换为字典"""
return {
'signal_id': self.signal_id,
'signal_type': self.signal_type.name,
'status': self.status.name,
'data_name': self.data_name,
'target_symbol': self.target_symbol,
'price': self.price,
'size': self.size,
'order_type': self.order_type,
'created_at': self.created_at.isoformat(),
'updated_at': self.updated_at.isoformat(),
'expires_at': self.expires_at.isoformat() if self.expires_at else None,
'priority': self.priority.name,
'generator': self.generator,
'strategy_id': self.strategy_id,
'context': self.context,
'confirmations': self.confirmations,
'rejection_reason': self.rejection_reason,
'parent_signal_id': self.parent_signal_id,
'child_signal_ids': self.child_signal_ids,
}
class SignalException(Exception):
"""信号异常基类"""
pass
class SignalValidationError(SignalException):
"""信号验证异常"""
pass
class SignalStateError(SignalException):
"""信号状态异常"""
pass
class SignalExpiredError(SignalException):
"""信号过期异常"""
pass
```bash
#### 3.2.2 信号管理器设计
```python
# backtrader/signals/manager.py
from typing import Dict, List, Optional, Callable, Any
from threading import Lock
from collections import defaultdict
from datetime import datetime, timedelta
from .base import Signal, SignalStatus, SignalType, SignalStateError
from ..utils.logger import get_logger
logger = get_logger(__name__)
class SignalManager:
"""信号管理器 - 管理信号生命周期和状态"""
def __init__(self):
# 信号存储
self._signals: Dict[str, Signal] = {}
self._lock = Lock()
# 索引
self._by_status: Dict[SignalStatus, List[str]] = defaultdict(list)
self._by_type: Dict[SignalType, List[str]] = defaultdict(list)
self._by_strategy: Dict[str, List[str]] = defaultdict(list)
self._by_data: Dict[str, List[str]] = defaultdict(list)
# 状态机转换规则
self._state_transitions = {
SignalStatus.PENDING: [
SignalStatus.CONFIRMING,
SignalStatus.CONFIRMED,
SignalStatus.CANCELLED,
SignalStatus.REJECTED,
SignalStatus.EXPIRED,
],
SignalStatus.CONFIRMING: [
SignalStatus.CONFIRMED,
SignalStatus.PENDING, # 确认失败，回退
SignalStatus.CANCELLED,
],
SignalStatus.CONFIRMED: [
SignalStatus.SUBMITTED,
SignalStatus.CANCELLED,
],
SignalStatus.SUBMITTED: [
SignalStatus.PARTIAL_FILLED,
SignalStatus.FILLED,
SignalStatus.CANCELLED,
SignalStatus.REJECTED,
],
SignalStatus.PARTIAL_FILLED: [
SignalStatus.FILLED,
SignalStatus.CANCELLED,
],
# FILLED, CANCELLED, REJECTED, EXPIRED 是终态
}
# 回调函数
self._callbacks: Dict[SignalStatus, List[Callable]] = defaultdict(list)
def add_signal(self, signal: Signal) -> bool:
"""添加新信号"""
with self._lock:
# 验证信号
if not self._validate_signal(signal):
return False
# 检查重复
if signal.signal_id in self._signals:
logger.warning(f"Signal {signal.signal_id} already exists")
return False
# 存储信号
self._signals[signal.signal_id] = signal
self._update_indexes(signal, True)
logger.info(f"Added signal: {signal.signal_id} "
f"type={signal.signal_type.name} "
f"strategy={signal.strategy_id}")
# 触发回调
self._trigger_callbacks(signal.status, signal)
return True
def update_signal_status(
self,
signal_id: str,
new_status: SignalStatus,
reason: Optional[str] = None
) -> bool:
"""更新信号状态"""
with self._lock:
signal = self._signals.get(signal_id)
if not signal:
logger.warning(f"Signal {signal_id} not found")
return False
# 验证状态转换
if new_status not in self._state_transitions.get(signal.status, []):
raise SignalStateError(
f"Invalid state transition: {signal.status.name} -> {new_status.name}"
)
# 更新索引
self._update_indexes(signal, False)
old_status = signal.status
signal.status = new_status
signal.updated_at = datetime.now()
if reason:
if new_status == SignalStatus.REJECTED:
signal.rejection_reason = reason
signal.context['status_change_reason'] = reason
self._update_indexes(signal, True)
logger.info(f"Signal {signal_id} status: {old_status.name} -> {new_status.name}")
# 触发回调
self._trigger_callbacks(new_status, signal)
return True
def get_signal(self, signal_id: str) -> Optional[Signal]:
"""获取信号"""
return self._signals.get(signal_id)
def get_signals_by_status(self, status: SignalStatus) -> List[Signal]:
"""按状态获取信号"""
signal_ids = self._by_status.get(status, [])
return [self._signals[sid] for sid in signal_ids if sid in self._signals]
def get_signals_by_type(self, signal_type: SignalType) -> List[Signal]:
"""按类型获取信号"""
signal_ids = self._by_type.get(signal_type, [])
return [self._signals[sid] for sid in signal_ids if sid in self._signals]
def get_signals_by_strategy(self, strategy_id: str) -> List[Signal]:
"""按策略获取信号"""
signal_ids = self._by_strategy.get(strategy_id, [])
return [self._signals[sid] for sid in signal_ids if sid in self._signals]
def get_active_signals(self) -> List[Signal]:
"""获取所有活跃信号"""
active = []
for status in (SignalStatus.PENDING, SignalStatus.CONFIRMING,
SignalStatus.CONFIRMED, SignalStatus.SUBMITTED,
SignalStatus.PARTIAL_FILLED):
active.extend(self.get_signals_by_status(status))
return active
def get_pending_signals(self) -> List[Signal]:
"""获取待处理信号"""
return self.get_signals_by_status(SignalStatus.PENDING)
def cancel_signal(self, signal_id: str, reason: Optional[str] = None) -> bool:
"""取消信号"""
return self.update_signal_status(signal_id, SignalStatus.CANCELLED, reason)
def expire_old_signals(self, timeout_seconds: Optional[int] = None) -> int:
"""过期旧信号"""
now = datetime.now()
expired_count = 0
for signal in list(self._signals.values()):
if signal.is_expired():
self.update_signal_status(signal.signal_id, SignalStatus.EXPIRED)
expired_count += 1
elif timeout_seconds:
age = (now - signal.created_at).total_seconds()
if age > timeout_seconds and signal.is_active():
self.update_signal_status(signal.signal_id, SignalStatus.EXPIRED)
expired_count += 1
return expired_count
def register_callback(
self,
status: SignalStatus,
callback: Callable[[Signal], Any]
):
"""注册状态变化回调"""
self._callbacks[status].append(callback)
def get_statistics(self) -> Dict[str, Any]:
"""获取信号统计信息"""
stats = {
'total': len(self._signals),
'by_status': {},
'by_type': {},
'active': len(self.get_active_signals()),
}
for status in SignalStatus:
count = len(self._by_status.get(status, []))
stats['by_status'][status.name] = count
for signal_type in SignalType:
count = len(self._by_type.get(signal_type, []))
stats['by_type'][signal_type.name] = count
return stats
def _validate_signal(self, signal: Signal) -> bool:
"""验证信号"""
try:
# 基本字段验证
if not signal.data_name:
logger.error("Signal validation failed: missing data_name")
return False
if signal.signal_type in (SignalType.BUY, SignalType.SELL,
SignalType.SHORT, SignalType.COVER):
if signal.price is None and signal.order_type == 'limit':
logger.error("Signal validation failed: limit order requires price")
return False
if signal.size is None or signal.size <= 0:
logger.error("Signal validation failed: invalid size")
return False
return True
except Exception as e:
logger.error(f"Signal validation error: {e}")
return False
def _update_indexes(self, signal: Signal, add: bool):
"""更新索引"""
action_func = (lambda lst, item: lst.append(item)) if add else (
lambda lst, item: lst.remove(item) if item in lst else None)
action_func(self._by_status[signal.status], signal.signal_id)
action_func(self._by_type[signal.signal_type], signal.signal_id)
if signal.strategy_id:
action_func(self._by_strategy[signal.strategy_id], signal.signal_id)
if signal.data_name:
action_func(self._by_data[signal.data_name], signal.signal_id)
def _trigger_callbacks(self, status: SignalStatus, signal: Signal):
"""触发状态回调"""
for callback in self._callbacks.get(status, []):
try:
callback(signal)
except Exception as e:
logger.error(f"Callback error for status {status.name}: {e}")
```bash
#### 3.2.3 信号生成器设计
```python
# backtrader/signals/generator.py
from abc import ABC, abstractmethod
from typing import Optional, List, Dict, Any
from dataclasses import dataclass
from .base import Signal, SignalType, SignalStatus, SignalPriority
from .manager import SignalManager
from .confirmator import SignalConfirmator
from ..utils.logger import get_logger
logger = get_logger(__name__)
@dataclass
class GeneratorConfig:
"""生成器配置"""
enable_confirmation: bool = True
confirmation_timeout: float = 5.0 # 秒
signal_timeout: float = 60.0 # 秒
priority: SignalPriority = SignalPriority.NORMAL
auto_submit: bool = True
class SignalGenerator(ABC):
"""信号生成器抽象基类"""
def __init__(
self,
strategy,
manager: Optional[SignalManager] = None,
config: Optional[GeneratorConfig] = None
):
self.strategy = strategy
self.manager = manager or SignalManager()
self.config = config or GeneratorConfig()
# 确认器列表
self.confirmators: List[SignalConfirmator] = []
# 统计信息
self.stats = {
'generated': 0,
'confirmed': 0,
'rejected': 0,
'submitted': 0,
}
def add_confirmator(self, confirmator: 'SignalConfirmator'):
"""添加确认器"""
self.confirmators.append(confirmator)
def generate_buy_signal(
self,
data_name: str,
price: Optional[float] = None,
size: Optional[float] = None,
order_type: str = 'market',
context: Optional[Dict[str, Any]] = None,
- *kwargs
) -> Optional[Signal]:
"""生成买入信号"""
return self._create_signal(
signal_type=SignalType.BUY,
data_name=data_name,
price=price,
size=size,
order_type=order_type,
context=context or {},
- *kwargs
)
def generate_sell_signal(
self,
data_name: str,
price: Optional[float] = None,
size: Optional[float] = None,
order_type: str = 'market',
context: Optional[Dict[str, Any]] = None,
- *kwargs
) -> Optional[Signal]:
"""生成卖出信号"""
return self._create_signal(
signal_type=SignalType.SELL,
data_name=data_name,
price=price,
size=size,
order_type=order_type,
context=context or {},
- *kwargs
)
def generate_short_signal(
self,
data_name: str,
price: Optional[float] = None,
size: Optional[float] = None,
order_type: str = 'market',
context: Optional[Dict[str, Any]] = None,
- *kwargs
) -> Optional[Signal]:
"""生成做空信号"""
return self._create_signal(
signal_type=SignalType.SHORT,
data_name=data_name,
price=price,
size=size,
order_type=order_type,
context=context or {},
- *kwargs
)
def generate_cover_signal(
self,
data_name: str,
price: Optional[float] = None,
size: Optional[float] = None,
order_type: str = 'market',
context: Optional[Dict[str, Any]] = None,
- *kwargs
) -> Optional[Signal]:
"""生成平空信号"""
return self._create_signal(
signal_type=SignalType.COVER,
data_name=data_name,
price=price,
size=size,
order_type=order_type,
context=context or {},
- *kwargs
)
def _create_signal(
self,
signal_type: SignalType,
data_name: str,
price: Optional[float],
size: Optional[float],
order_type: str,
context: Dict[str, Any],
- *kwargs
) -> Optional[Signal]:
"""创建信号"""
# 获取数据源
data = self._get_data(data_name)
if data is None:
logger.error(f"Data feed {data_name} not found")
return None
# 计算默认价格和数量
if price is None:
price = data.close[0] if order_type != 'market' else None
if size is None:
size = self._calculate_default_size(signal_type, data)
# 创建信号
signal = Signal(
signal_type=signal_type,
data_name=data_name,
target_symbol=getattr(data, '_name', data_name),
price=price,
size=size,
order_type=order_type,
priority=self.config.priority,
generator=self.__class__.__name__,
strategy_id=getattr(self.strategy, '_strategy_id', 'unknown'),
context=context,
- *kwargs
)
self.stats['generated'] += 1
# 添加到管理器
if not self.manager.add_signal(signal):
return None
# 处理确认
if self.config.enable_confirmation and self.confirmators:
self.manager.update_signal_status(
signal.signal_id,
SignalStatus.CONFIRMING
)
self._process_confirmation(signal)
else:
self.manager.update_signal_status(
signal.signal_id,
SignalStatus.CONFIRMED
)
# 自动提交
if signal.status == SignalStatus.CONFIRMED and self.config.auto_submit:
self._submit_signal(signal)
return signal
def _process_confirmation(self, signal: Signal):
"""处理信号确认"""
confirmed = True
for confirmator in self.confirmators:
result = confirmator.confirm(signal, self.strategy)
if result.confirmed:
signal.add_confirmation(confirmator.name)
logger.debug(f"Signal {signal.signal_id} confirmed by {confirmator.name}")
else:
confirmed = False
logger.warning(
f"Signal {signal.signal_id} rejected by {confirmator.name}: "
f"{result.reason}"
)
if confirmed and len(signal.confirmations) > 0:
self.manager.update_signal_status(signal.signal_id, SignalStatus.CONFIRMED)
self.stats['confirmed'] += 1
else:
self.manager.update_signal_status(
signal.signal_id,
SignalStatus.REJECTED,
reason=f"Failed confirmation: {len(signal.confirmations)} passed"
)
self.stats['rejected'] += 1
def _submit_signal(self, signal: Signal):
"""提交信号到策略执行"""
try:
order = None
data = self._get_data(signal.data_name)
if signal.signal_type == SignalType.BUY:
order = self.strategy.buy(
data=data,
size=signal.size,
price=signal.price,
exectype=self._get_order_type_code(signal.order_type)
)
elif signal.signal_type == SignalType.SELL:
order = self.strategy.sell(
data=data,
size=signal.size,
price=signal.price,
exectype=self._get_order_type_code(signal.order_type)
)
elif signal.signal_type == SignalType.SHORT:
order = self.strategy.sell(
data=data,
size=signal.size,
price=signal.price,
exectype=self._get_order_type_code(signal.order_type)
)
elif signal.signal_type == SignalType.COVER:
order = self.strategy.buy(
data=data,
size=signal.size,
price=signal.price,
exectype=self._get_order_type_code(signal.order_type)
)
if order:
self.manager.update_signal_status(signal.signal_id, SignalStatus.SUBMITTED)
signal.context['order_ref'] = order.ref
self.stats['submitted'] += 1
except Exception as e:
logger.error(f"Error submitting signal {signal.signal_id}: {e}")
self.manager.update_signal_status(
signal.signal_id,
SignalStatus.REJECTED,
reason=str(e)
)
def _get_data(self, data_name: str):
"""获取数据源"""
for data in self.strategy.datas:
if getattr(data, '_name', None) == data_name:
return data
return self.strategy.datas[0] if self.strategy.datas else None
def _calculate_default_size(self, signal_type: SignalType, data) -> float:
"""计算默认数量"""
cash = self.strategy.broker.getcash()
price = data.close[0]
if signal_type in (SignalType.SHORT, SignalType.SELL):
position = self.strategy.getposition(data)
if position.size > 0:
return position.size
return 0
if price > 0:
return cash / price *0.95 # 使用 95%的现金
return 0
def _get_order_type_code(self, order_type: str):
"""获取订单类型代码"""
import backtrader as bt
mapping = {
'market': bt.Order.Market,
'limit': bt.Order.Limit,
'stop': bt.Order.Stop,
'stop_limit': bt.Order.StopLimit,
}
return mapping.get(order_type, bt.Order.Market)
@abstractmethod
def next(self):
"""每根 K 线调用，生成信号"""
pass
```bash
#### 3.2.4 信号确认器设计
```python
# backtrader/signals/confirmator.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional, List, Callable, Any, Dict
from .base import Signal, SignalType
from ..utils.logger import get_logger
logger = get_logger(__name__)
@dataclass
class ConfirmationResult:
"""确认结果"""
confirmed: bool
reason: Optional[str] = None
score: float = 0.0 # 确认得分 0-1
class SignalConfirmator(ABC):
"""信号确认器抽象基类"""
def __init__(self, name: str, weight: float = 1.0):
self.name = name
self.weight = weight
@abstractmethod
def confirm(self, signal: Signal, strategy) -> ConfirmationResult:
"""确认信号"""
pass
class IndicatorConfirmator(SignalConfirmator):
"""指标确认器 - 基于技术指标确认"""
def __init__(
self,
name: str,
indicator_getter: Callable,
condition: Callable[[Any], bool],
weight: float = 1.0
):
super().__init__(name, weight)
self.indicator_getter = indicator_getter
self.condition = condition
def confirm(self, signal: Signal, strategy) -> ConfirmationResult:
"""基于指标条件确认"""
try:
indicator_value = self.indicator_getter(strategy)
passed = self.condition(indicator_value)
return ConfirmationResult(
confirmed=passed,
reason=f"{self.name}: {indicator_value} {'meets' if passed else 'fails'} condition",
score=1.0 if passed else 0.0
)
except Exception as e:
logger.error(f"Error in {self.name} confirmation: {e}")
return ConfirmationResult(
confirmed=False,
reason=f"{self.name} error: {e}",
score=0.0
)
class TrendConfirmator(SignalConfirmator):
"""趋势确认器 - 确认价格趋势"""
def __init__(self, period: int = 3, weight: float = 1.0):
super().__init__("TrendConfirmator", weight)
self.period = period
def confirm(self, signal: Signal, strategy) -> ConfirmationResult:
"""确认价格趋势"""
import numpy as np
data = strategy.datas[0] # 简化处理
close_values = [data.close[-i] for i in range(self.period)]
if signal.signal_type in (SignalType.BUY, SignalType.COVER):
# 买入信号需要上升趋势
is_upward = all(close_values[i] >= close_values[i+1]
for i in range(len(close_values)-1))
return ConfirmationResult(
confirmed=is_upward,
reason=f"Trend: {'upward' if is_upward else 'not upward'}",
score=1.0 if is_upward else 0.3
)
else:
# 卖出信号需要下降趋势
is_downward = all(close_values[i] <= close_values[i+1]
for i in range(len(close_values)-1))
return ConfirmationResult(
confirmed=is_downward,
reason=f"Trend: {'downward' if is_downward else 'not downward'}",
score=1.0 if is_downward else 0.3
)
class RSIDirectionConfirmator(SignalConfirmator):
"""RSI 方向确认器"""
def __init__(self, period: int = 4, weight: float = 1.0):
super().__init__("RSIDirectionConfirmator", weight)
self.period = period
def confirm(self, signal: Signal, strategy) -> ConfirmationResult:
"""确认 RSI 方向"""
import numpy as np
rsi = strategy.rsi # 假设策略有 rsi 指标
recent_rsi = np.array([rsi[-i] for i in range(1, self.period + 1)])
is_descending = np.all(np.diff(recent_rsi) < 0)
is_ascending = np.all(np.diff(recent_rsi) > 0)
if signal.signal_type in (SignalType.BUY, SignalType.COVER):
# 买入信号：RSI 下降表示超卖
return ConfirmationResult(
confirmed=is_descending,
reason=f"RSI direction: {'descending' if is_descending else 'not descending'}",
score=1.0 if is_descending else 0.5
)
else:
# 卖出信号：RSI 上升表示超买
return ConfirmationResult(
confirmed=is_ascending,
reason=f"RSI direction: {'ascending' if is_ascending else 'not ascending'}",
score=1.0 if is_ascending else 0.5
)
class VolumeConfirmator(SignalConfirmator):
"""成交量确认器"""
def __init__(self, period: int = 5, min_ratio: float = 1.2, weight: float = 1.0):
super().__init__("VolumeConfirmator", weight)
self.period = period
self.min_ratio = min_ratio
def confirm(self, signal: Signal, strategy) -> ConfirmationResult:
"""确认成交量支持"""
import numpy as np
data = strategy.datas[0]
current_volume = data.volume[0]
avg_volume = np.mean([data.volume[-i] for i in range(1, self.period + 1)])
volume_ratio = current_volume / avg_volume if avg_volume > 0 else 0
confirmed = volume_ratio >= self.min_ratio
return ConfirmationResult(
confirmed=confirmed,
reason=f"Volume ratio: {volume_ratio:.2f} (min: {self.min_ratio})",
score=min(volume_ratio / self.min_ratio, 1.0) if confirmed else volume_ratio / self.min_ratio
)
class CompositeConfirmator(SignalConfirmator):
"""组合确认器 - 支持多种确认策略"""
def __init__(
self,
confirmators: List[SignalConfirmator],
strategy: str = 'AND', # AND, OR, WEIGHTED, MAJORITY
min_score: float = 0.6,
weight: float = 1.0
):
super().__init__("CompositeConfirmator", weight)
self.confirmators = confirmators
self.strategy = strategy
self.min_score = min_score
def confirm(self, signal: Signal, strategy) -> ConfirmationResult:
"""组合确认"""
results = []
for confirmator in self.confirmators:
result = confirmator.confirm(signal, strategy)
results.append(result)
total_weight = sum(c.weight for c in self.confirmators)
reasons = []
if self.strategy == 'AND':
confirmed = all(r.confirmed for r in results)
score = sum(r.score for r in results) / len(results) if results else 0
reasons = [r.reason for r in results]
elif self.strategy == 'OR':
confirmed = any(r.confirmed for r in results)
score = max(r.score for r in results) if results else 0
reasons = [r.reason for r in results if r.confirmed]
elif self.strategy == 'WEIGHTED':
weighted_score = sum(
r.score*c.weight for r, c in zip(results, self.confirmators)
) / total_weight if total_weight > 0 else 0
confirmed = weighted_score >= self.min_score
score = weighted_score
reasons = [r.reason for r in results]
elif self.strategy == 'MAJORITY':
confirmed_count = sum(1 for r in results if r.confirmed)
confirmed = confirmed_count > len(results) / 2
score = confirmed_count / len(results) if results else 0
reasons = [r.reason for r in results]
else:
confirmed = False
score = 0
return ConfirmationResult(
confirmed=confirmed,
reason=f"Composite ({self.strategy}): " + "; ".join(reasons),
score=score
)
```bash
#### 3.2.5 风险管理器设计
```python
# backtrader/risk/managers.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional, List, Dict, Any, Callable
from enum import Enum, auto
from ..utils.logger import get_logger
from ..signals.base import Signal, SignalType
logger = get_logger(__name__)
class RiskAction(Enum):
"""风险动作"""
ALLOW = auto() # 允许信号
MODIFY = auto() # 修改信号
REJECT = auto() # 拒绝信号
CLOSE_POSITION = auto() # 平仓
@dataclass
class RiskCheckResult:
"""风险检查结果"""
action: RiskAction
reason: Optional[str] = None
modified_size: Optional[float] = None
modified_price: Optional[float] = None
risk_level: float = 0.0 # 0-1
class RiskManager(ABC):
"""风险管理器抽象基类"""
def __init__(self, name: str):
self.name = name
self.enabled = True
@abstractmethod
def check_signal(self, signal: Signal, strategy) -> RiskCheckResult:
"""检查信号"""
pass
class StopLossRiskManager(RiskManager):
"""止损风险管理器"""
def __init__(
self,
stop_loss_pct: float = 0.03,
trailing_stop: bool = False,
trailing_pct: float = 0.02
):
super().__init__("StopLossRiskManager")
self.stop_loss_pct = stop_loss_pct
self.trailing_stop = trailing_stop
self.trailing_pct = trailing_pct
self.entry_prices: Dict[str, float] = {}
self.peak_prices: Dict[str, float] = {}
def check_signal(self, signal: Signal, strategy) -> RiskCheckResult:
"""检查是否触发止损"""
if signal.signal_type in (SignalType.BUY, SignalType.SHORT):
# 记录入场价格
if signal.price:
self.entry_prices[signal.data_name] = signal.price
self.peak_prices[signal.data_name] = signal.price
return RiskCheckResult(action=RiskAction.ALLOW)
# 检查止损条件
data = self._get_data(signal, strategy)
if data is None:
return RiskCheckResult(action=RiskAction.ALLOW)
entry_price = self.entry_prices.get(signal.data_name)
if entry_price is None:
return RiskCheckResult(action=RiskAction.ALLOW)
current_price = data.close[0]
position = strategy.getposition(data)
if position.size > 0: # 多头持仓
loss_pct = (entry_price - current_price) / entry_price
# 更新移动止损
if self.trailing_stop and current_price > self.peak_prices.get(signal.data_name, entry_price):
self.peak_prices[signal.data_name] = current_price
stop_price = current_price*(1 - self.trailing_pct)
else:
stop_price = entry_price*(1 - self.stop_loss_pct)
if current_price <= stop_price:
return RiskCheckResult(
action=RiskAction.CLOSE_POSITION,
reason=f"Stop loss triggered: {current_price:.2f} <= {stop_price:.2f}",
risk_level=loss_pct
)
elif position.size < 0: # 空头持仓
loss_pct = (current_price - entry_price) / entry_price
if self.trailing_stop and current_price < self.peak_prices.get(signal.data_name, entry_price):
self.peak_prices[signal.data_name] = current_price
stop_price = current_price*(1 + self.trailing_pct)
else:
stop_price = entry_price*(1 + self.stop_loss_pct)
if current_price >= stop_price:
return RiskCheckResult(
action=RiskAction.CLOSE_POSITION,
reason=f"Stop loss triggered: {current_price:.2f} >= {stop_price:.2f}",
risk_level=loss_pct
)
return RiskCheckResult(action=RiskAction.ALLOW)
def _get_data(self, signal: Signal, strategy):
"""获取数据源"""
for data in strategy.datas:
if getattr(data, '_name', None) == signal.data_name:
return data
return strategy.datas[0] if strategy.datas else None
class TakeProfitRiskManager(RiskManager):
"""止盈风险管理器"""
def __init__(self, take_profit_pct: float = 0.05):
super().__init__("TakeProfitRiskManager")
self.take_profit_pct = take_profit_pct
self.entry_prices: Dict[str, float] = {}
def check_signal(self, signal: Signal, strategy) -> RiskCheckResult:
"""检查是否触发止盈"""
if signal.signal_type in (SignalType.BUY, SignalType.SHORT):
if signal.price:
self.entry_prices[signal.data_name] = signal.price
return RiskCheckResult(action=RiskAction.ALLOW)
data = self._get_data(signal, strategy)
if data is None:
return RiskCheckResult(action=RiskAction.ALLOW)
entry_price = self.entry_prices.get(signal.data_name)
if entry_price is None:
return RiskCheckResult(action=RiskAction.ALLOW)
current_price = data.close[0]
position = strategy.getposition(data)
if position.size > 0: # 多头持仓
profit_pct = (current_price - entry_price) / entry_price
if profit_pct >= self.take_profit_pct:
return RiskCheckResult(
action=RiskAction.CLOSE_POSITION,
reason=f"Take profit triggered: +{profit_pct:.2%}",
risk_level=profit_pct
)
elif position.size < 0: # 空头持仓
profit_pct = (entry_price - current_price) / entry_price
if profit_pct >= self.take_profit_pct:
return RiskCheckResult(
action=RiskAction.CLOSE_POSITION,
reason=f"Take profit triggered: +{profit_pct:.2%}",
risk_level=profit_pct
)
return RiskCheckResult(action=RiskAction.ALLOW)
def _get_data(self, signal: Signal, strategy):
"""获取数据源"""
for data in strategy.datas:
if getattr(data, '_name', None) == signal.data_name:
return data
return strategy.datas[0] if strategy.datas else None
class PositionSizeRiskManager(RiskManager):
"""仓位风险管理器"""
def __init__(
self,
max_position_pct: float = 0.95,
max_total_position_pct: float = 1.0
):
super().__init__("PositionSizeRiskManager")
self.max_position_pct = max_position_pct
self.max_total_position_pct = max_total_position_pct
def check_signal(self, signal: Signal, strategy) -> RiskCheckResult:
"""检查仓位大小"""
if signal.size is None:
return RiskCheckResult(action=RiskAction.ALLOW)
cash = strategy.broker.getcash()
value = strategy.broker.getvalue()
data = self._get_data(signal, strategy)
if data is None:
return RiskCheckResult(action=RiskAction.ALLOW)
current_price = data.close[0] if signal.price is None else signal.price
position_value = signal.size*current_price
# 单笔仓位检查
if position_value > cash*self.max_position_pct:
modified_size = (cash*self.max_position_pct) / current_price
return RiskCheckResult(
action=RiskAction.MODIFY,
reason=f"Position size exceeds {self.max_position_pct:.0%} of cash",
modified_size=modified_size,
risk_level=position_value / cash
)
# 总仓位检查
total_position = sum(
abs(strategy.getposition(d).size*d.close[0])
for d in strategy.datas
)
new_total_position = total_position + position_value
if new_total_position > value*self.max_total_position_pct:
max_allowed = (value* self.max_total_position_pct) - total_position
modified_size = max(max_allowed / current_price, 0)
return RiskCheckResult(
action=RiskAction.MODIFY if modified_size > 0 else RiskAction.REJECT,
reason=f"Total position exceeds {self.max_total_position_pct:.0%} of portfolio",
modified_size=modified_size if modified_size > 0 else None,
risk_level=new_total_position / value
)
return RiskCheckResult(action=RiskAction.ALLOW)
def _get_data(self, signal: Signal, strategy):
"""获取数据源"""
for data in strategy.datas:
if getattr(data, '_name', None) == signal.data_name:
return data
return strategy.datas[0] if strategy.datas else None
class MaxDrawdownRiskManager(RiskManager):
"""最大回撤风险管理器"""
def __init__(self, max_drawdown_pct: float = 0.15):
super().__init__("MaxDrawdownRiskManager")
self.max_drawdown_pct = max_drawdown_pct
self.peak_value = 0
def check_signal(self, signal: Signal, strategy) -> RiskCheckResult:
"""检查回撤"""
current_value = strategy.broker.getvalue()
# 更新峰值
if current_value > self.peak_value:
self.peak_value = current_value
if self.peak_value > 0:
drawdown_pct = (self.peak_value - current_value) / self.peak_value
if drawdown_pct >= self.max_drawdown_pct:
return RiskCheckResult(
action=RiskAction.REJECT,
reason=f"Max drawdown exceeded: {drawdown_pct:.2%} >= {self.max_drawdown_pct:.2%}",
risk_level=drawdown_pct
)
return RiskCheckResult(action=RiskAction.ALLOW)
class CompositeRiskManager:
"""组合风险管理器"""
def __init__(self, managers: List[RiskManager]):
self.managers = managers
self.risk_events: List[Dict[str, Any]] = []
def check_signal(self, signal: Signal, strategy) -> RiskCheckResult:
"""执行所有风险检查"""
highest_risk_action = RiskAction.ALLOW
final_reason = ""
modified_size = signal.size
max_risk_level = 0.0
for manager in self.managers:
if not manager.enabled:
continue
result = manager.check_signal(signal, strategy)
# 更新风险等级
if result.risk_level > max_risk_level:
max_risk_level = result.risk_level
# 确定动作优先级
action_priority = {
RiskAction.REJECT: 4,
RiskAction.CLOSE_POSITION: 3,
RiskAction.MODIFY: 2,
RiskAction.ALLOW: 1,
}
if action_priority[result.action] > action_priority[highest_risk_action]:
highest_risk_action = result.action
final_reason = result.reason
if result.modified_size is not None:
modified_size = result.modified_size
# 记录风险事件
if highest_risk_action != RiskAction.ALLOW:
self.risk_events.append({
'timestamp': strategy.datetime.datetime(0) if hasattr(strategy, 'datetime') else None,
'signal_id': signal.signal_id,
'action': highest_risk_action.name,
'reason': final_reason,
'risk_level': max_risk_level,
})
return RiskCheckResult(
action=highest_risk_action,
reason=final_reason,
modified_size=modified_size,
risk_level=max_risk_level
)
```bash
#### 3.2.6 CLI 工具设计
```python
# backtrader/cli/main.py
import click
import importlib
from pathlib import Path
from typing import Optional
import pandas as pd
import backtrader as bt
from ..utils.logger import setup_logger, get_logger
from ..signals.manager import SignalManager
from ..risk.managers import (
CompositeRiskManager,
StopLossRiskManager,
TakeProfitRiskManager,
PositionSizeRiskManager,
)
logger = get_logger(__name__)
class CommaSeparatedList(click.ParamType):
"""逗号分隔列表参数类型"""
name = "list"
def convert(self, value, param, ctx):
try:
if ',' in value:
return [x.strip() for x in value.split(',')]
return [value]
except ValueError:
self.fail(f"{value!r} is not a valid list", param, ctx)
class CommaSeparatedFloatList(click.ParamType):
"""逗号分隔浮点数列表参数类型"""
name = "float_list"
def convert(self, value, param, ctx):
try:
return [float(x.strip()) for x in value.split(',')]
except ValueError:
self.fail(f"{value!r} is not a valid float list", param, ctx)
class CommaSeparatedIntList(click.ParamType):
"""逗号分隔整数列表参数类型"""
name = "int_list"
def convert(self, value, param, ctx):
try:
return [int(x.strip()) for x in value.split(',')]
except ValueError:
self.fail(f"{value!r} is not a valid int list", param, ctx)
COMMA_LIST = CommaSeparatedList()
FLOAT_LIST = CommaSeparatedFloatList()
INT_LIST = CommaSeparatedIntList()
@click.group()
@click.option('--debug/--no-debug', default=False, help='Enable debug mode')
@click.option('--log-file', type=click.Path(), help='Log file path')
@click.pass_context
def cli(ctx, debug, log_file):
"""Backtrader Signal Trading CLI"""
ctx.ensure_object(dict)
ctx.obj['debug'] = debug
ctx.obj['log_file'] = log_file
# 设置日志
setup_logger(debug=debug, log_file=log_file)
@cli.command()
@click.option('--data', '-d', required=True, type=click.Path(exists=True),
help='Data file (CSV)')
@click.option('--strategy', '-s', required=True, help='Strategy class name')
@click.option('--module', '-m', default='backtrader.strategy',
help='Strategy module path')
@click.option('--cash', default=10000, type=float, help='Initial cash')
@click.option('--commission', default=0.001, type=float, help='Commission rate')
@click.option('--fromdate', type=str, help='Start date (YYYY-MM-DD)')
@click.option('--todate', type=str, help='End date (YYYY-MM-DD)')
@click.option('--output', '-o', type=click.Path(), help='Output file')
@click.option('--plot', is_flag=True, help='Plot results')
@click.option('--cerebro-kw', type=str, help='Cerebro kwargs (JSON)')
@click.pass_context
def run(ctx, data, strategy, module, cash, commission, fromdate, todate,
output, plot, cerebro_kw):
"""Run a single backtest"""
from ..utils.cerebro_factory import CerebroFactory
# 创建 Cerebro
cerebro = CerebroFactory.create_default(
cash=cash,
commission=commission
)
# 加载数据
df = pd.read_csv(data, index_col='datetime', parse_dates=True)
data_feed = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data_feed)
# 加载策略
strategy_class = _load_strategy_class(module, strategy)
cerebro.addstrategy(strategy_class)
# 运行
results = cerebro.run()
# 输出结果
_print_results(results, output)
# 绘图
if plot:
cerebro.plot()
@cli.command()
@click.option('--data', '-d', required=True, type=click.Path(exists=True),
help='Data file (CSV)')
@click.option('--strategy', '-s', required=True, help='Strategy class name')
@click.option('--module', '-m', default='backtrader.strategy',
help='Strategy module path')
@click.option('--cash', default=10000, type=float, help='Initial cash')
@click.option('--commission', default=0.001, type=float, help='Commission rate')
@click.option('--short-period', type=INT_LIST, help='Short period(s) for optimization')
@click.option('--long-period', type=INT_LIST, help='Long period(s) for optimization')
@click.option('--output', '-o', type=click.Path(), help='Output Excel file')
@click.option('--maxcpus', default=4, type=int, help='Maximum CPUs for optimization')
@click.pass_context
def optimize(ctx, data, strategy, module, cash, commission,
short_period, long_period, output, maxcpus):
"""Run strategy optimization"""
cerebro = bt.Cerebro(runonce=True, preload=True, optreturn=False, maxcpus=maxcpus)
# 加载数据
df = pd.read_csv(data, index_col='datetime', parse_dates=True)
data_feed = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data_feed)
cerebro.broker.setcash(cash)
cerebro.broker.setcommission(commission=commission)
# 加载策略进行优化
strategy_class = _load_strategy_class(module, strategy)
# 构建参数网格
opt_params = {}
if short_period:
opt_params['short_period'] = short_period
if long_period:
opt_params['long_period'] = long_period
cerebro.optstrategy(strategy_class, **opt_params)
# 运行优化
results = cerebro.run()
# 输出结果
_print_optimization_results(results, strategy, output)
@cli.command()
@click.option('--data', '-d', required=True, type=click.Path(exists=True),
help='Data file (CSV)')
@click.option('--config', '-c', type=click.Path(exists=True),
help='Configuration file (TOML/YAML)')
@click.option('--signal-stats', is_flag=True, help='Show signal statistics')
@click.option('--trade-stats', is_flag=True, help='Show trade statistics')
@click.pass_context
def analyze(ctx, data, config, signal_stats, trade_stats):
"""Analyze backtest results"""
# 加载配置
if config:
import toml
cfg = toml.load(config)
else:
cfg = {}
# 分析逻辑
click.echo("Analysis results...")
def _load_strategy_class(module: str, class_name: str):
"""动态加载策略类"""
mod = importlib.import_module(module)
return getattr(mod, class_name)
def _print_results(results, output: Optional[str] = None):
"""打印回测结果"""
strat = results[0][0]
click.echo("\n" + "=" *50)
click.echo("Backtest Results")
click.echo("="*50)
click.echo(f"Final Value: {strat.broker.getvalue():.2f}")
# 获取分析器结果
for analyzer in strat.analyzers:
result = analyzer.get_analysis()
click.echo(f"\n{analyzer.__class__.__name__}:")
click.echo(result)
if output:
_save_results(results, output)
def _print_optimization_results(results, strategy_name: str, output: Optional[str] = None):
"""打印优化结果"""
results_list = []
for result in results:
for strat in result:
params = {k: v for k, v in strat.params.__dict__.items()
if not k.startswith('_')}
final_value = strat.broker.getvalue()
results_list.append({
- *params,
'final_value': final_value,
})
df = pd.DataFrame(results_list)
df = df.sort_values('final_value', ascending=False)
click.echo("\n" + "=" *50)
click.echo(f"Optimization Results: {strategy_name}")
click.echo("="*50)
click.echo(df.to_string(index=False))
if output:
df.to_excel(output, index=False)
click.echo(f"\nResults saved to {output}")
def _save_results(results, output: str):
"""保存结果"""
import json
from pathlib import Path
output_path = Path(output)
output_path.parent.mkdir(parents=True, exist_ok=True)
# 准备数据
data = {
'final_value': results[0][0].broker.getvalue(),
'params:': results[0][0].params.__dict__,
}
# 保存
if output.endswith('.json'):
with open(output, 'w') as f:
json.dump(data, f, indent=2)
else:
with open(output, 'w') as f:
f.write(str(data))
if __name__ == '__main__':
cli()
```bash
#### 3.2.7 日志工具设计
```python
# backtrader/utils/logger.py
import sys
from loguru import logger
from typing import Optional
from pathlib import Path
class SignalLogger:
"""信号交易日志记录器"""
def __init__(self, log_file: Optional[str] = None, debug: bool = False):
self.log_file = log_file
self.debug = debug
self._setup_logger()
def _setup_logger(self):
"""配置日志"""
# 移除默认处理器
logger.remove()
# 控制台输出
log_level = "DEBUG" if self.debug else "INFO"
logger.add(
sys.stderr,
format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
"<level>{level: <8}</level> | "
"<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> | "
"<level>{message}</level>",
level=log_level,
colorize=True,
)
# 文件输出
if self.log_file:
log_path = Path(self.log_file)
log_path.parent.mkdir(parents=True, exist_ok=True)
# 普通日志
logger.add(
self.log_file,
format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} | {message}",
level="DEBUG",
rotation="100 MB",
retention="30 days",
compression="zip",
)
# 结构化 JSON 日志
json_log = self.log_file.replace('.log', '.json')
logger.add(
json_log,
format="{message}",
level="DEBUG",
serialize=True,
rotation="100 MB",
retention="30 days",
)
def get_logger(self, name: str):
"""获取命名日志记录器"""
return logger.bind(name=name)
# 全局日志配置
_logger_config: Optional[SignalLogger] = None
def setup_logger(log_file: Optional[str] = None, debug: bool = False):
"""设置全局日志"""
global _logger_config
_logger_config = SignalLogger(log_file=log_file, debug=debug)
def get_logger(name: str):
"""获取日志记录器"""
if _logger_config is None:
setup_logger()
return logger.bind(name=name)
# 信号事件专用日志
def log_signal_event(
event_type: str,
signal_id: str,
signal_type: str,
- *kwargs
):
"""记录信号事件"""
logger.bind(
event_type=event_type,
signal_id=signal_id,
signal_type=signal_type,
- *kwargs
).info(f"Signal event: {event_type} for signal {signal_id}")
def log_risk_event(
event_type: str,
signal_id: str,
action: str,
reason: str,
risk_level: float,
- *kwargs
):
"""记录风险事件"""
logger.bind(
event_type=event_type,
signal_id=signal_id,
action=action,
reason=reason,
risk_level=risk_level,
- *kwargs
).warning(f"Risk event: {action} for signal {signal_id} - {reason}")
```bash
### 3.3 使用示例
#### 3.3.1 基础信号策略
```python
# examples/signal_strategy_example.py
import backtrader as bt
from backtrader.signals import SignalGenerator
from backtrader.signals.confirmator import (
TrendConfirmator,
RSIDirectionConfirmator,
CompositeConfirmator
)
from backtrader.risk.managers import (
CompositeRiskManager,
StopLossRiskManager,
TakeProfitRiskManager,
PositionSizeRiskManager,
)
class MySignalStrategy(bt.Strategy):
"""使用信号系统的策略"""
params = (
('short_period', 50),
('long_period', 200),
('stop_loss', 0.03),
('take_profit', 0.05),
)
def __init__(self):
# 指标
self.short_ma = bt.indicators.EMA(self.data.close, period=self.params.short_period)
self.long_ma = bt.indicators.EMA(self.data.close, period=self.params.long_period)
self.rsi = bt.indicators.RSI(self.data.close, period=14)
# 信号生成器
self.signal_gen = SignalGenerator(strategy=self)
# 信号确认器
confirmator = CompositeConfirmator(
confirmators=[
TrendConfirmator(period=3),
RSIDirectionConfirmator(period=4),
],
strategy='AND'
)
self.signal_gen.add_confirmator(confirmator)
# 风险管理器
self.risk_manager = CompositeRiskManager([
StopLossRiskManager(stop_loss_pct=self.params.stop_loss),
TakeProfitRiskManager(take_profit_pct=self.params.take_profit),
PositionSizeRiskManager(max_position_pct=0.95),
])
# 状态
self._op = bt.Order.Buy
def next(self):
# 风险检查
if self._op == bt.Order.Sell:
# 创建平仓信号进行风险检查
from backtrader.signals.base import Signal, SignalType
check_signal = Signal(
signal_type=SignalType.SELL,
data_name=self.data._name,
size=self.getposition(self.data).size,
)
risk_result = self.risk_manager.check_signal(check_signal, self)
if risk_result.action.name == 'CLOSE_POSITION':
self.signal_gen.generate_sell_signal(self.data._name)
self._op = bt.Order.Buy
return
if len(self.data) < self.params.long_period:
return
# 生成交易信号
if self._op == bt.Order.Buy:
# 买入条件
if self.data.close[0] < self.short_ma[0] *0.98:
signal = self.signal_gen.generate_buy_signal(self.data._name)
# 检查风险
if signal:
risk_result = self.risk_manager.check_signal(signal, self)
if risk_result.action.name == 'REJECT':
self.signal_gen.manager.cancel_signal(
signal.signal_id,
reason=risk_result.reason
)
return
if risk_result.action.name == 'MODIFY':
signal.size = risk_result.modified_size
elif self._op == bt.Order.Sell:
# 卖出条件
if self.data.close[0] > self.short_ma[0]* 1.02:
self.signal_gen.generate_sell_signal(self.data._name)
self._op = bt.Order.Buy
def notify_order(self, order):
if order.status == bt.Order.Completed:
if order.isbuy():
self._op = bt.Order.Sell
else:
self._op = bt.Order.Buy
# 运行
if __name__ == '__main__':
cerebro = bt.Cerebro()
# 添加数据
data = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data)
# 添加策略
cerebro.addstrategy(MySignalStrategy)
# 运行
results = cerebro.run()
cerebro.plot()
```bash
#### 3.3.2 CLI 使用
```bash
# 基础回测
python -m backtrader.cli run \
- -data data.csv \
- -strategy MySignalStrategy \
- -module examples.signal_strategy_example \
- -cash 10000 \
- -plot
# 参数优化
python -m backtrader.cli optimize \
- -data data.csv \
- -strategy MySignalStrategy \
- -module examples.signal_strategy_example \
- -short-period 30,40,50,60 \
- -long-period 150,180,200,220 \
- -output optimization_results.xlsx
```bash
- --
## 四、实施计划
### 第一阶段：信号基础架构 (2 周)
1. 实现 Signal 基类和枚举
2. 实现 SignalManager
3. 编写单元测试
### 第二阶段：信号生成与确认 (2 周)
1. 实现 SignalGenerator
2. 实现各种 SignalConfirmator
3. 集成测试
### 第三阶段：风险管理系统 (2 周)
1. 实现各种 RiskManager
2. 实现 CompositeRiskManager
3. 风险事件记录
### 第四阶段：CLI 工具 (1 周)
1. 实现基础 CLI 命令
2. 参数优化功能
3. 结果导出功能
### 第五阶段：集成与文档 (1 周)
1. 与现有 backtrader 集成
2. 编写使用文档
3. 性能优化
- --
## 五、总结
本设计文档借鉴了 signal_trading 项目的以下核心优势：
1. **标准化状态管理**：通过 SignalManager 统一管理信号状态
2. **独立风险管理**：将风险检查从策略逻辑中分离
3. **多指标确认机制**：通过 Confirmator 提高信号可靠性
4. **CLI 工具链**：完整的命令行工作流
5. **结构化日志**：使用 loguru 提供更好的调试体验
6. **信号评估系统**：完整的信号统计和分析功能
这些改进将使 backtrader 成为一个更加专业、可靠的量化交易框架，特别适合生产环境的算法交易应用。