10 KiB
10 KiB
多策略回测引擎
一个基于Python的股票交易策略回测系统，支持参数优化、绩效分析和灵活的股票筛选功能。
功能特性
- 多策略支持：可以同时运行和比较多个交易策略
- 参数优化：支持网格搜索和随机搜索算法优化策略参数
- 实时进度显示：显示回测和参数优化的进度和耗时
- 绩效指标计算：包括总收益率、年化收益率、最大回撤、夏普比率等
- 可视化报告：自动生成收益曲线、回撤曲线等图表
- 灵活配置：通过配置文件调整回测参数和策略参数
- 个股筛选：支持根据股票类型（主板、创业板、科创板、北交所）和是否为ST股票进行筛选
项目结构
backtrader/
├── analysis/ # 绩效分析模块
├── backtest/ # 回测引擎
├── data/ # 数据获取和处理
├── strategy/ # 策略定义
├── utils/ # 工具函数
├── config.py # 配置文件
├── main.py # 主入口文件
├── parameter_optimizer.py # 参数优化器
└── requirements.txt # 依赖包
安装与依赖
环境要求
- Python 3.7+
安装依赖
pip install -r requirements.txt
配置说明
配置文件位于 config.py，主要包含以下部分：
数据配置
# 数据配置
START_DATE = '20230101' # 回测开始日期
END_DATE = '20251231' # 回测结束日期
STOCK_POOL = ['600252.SH', '002925.SZ', '600362.SH'] # 自定义股票池，用于快速回测
个股筛选配置
可以通过配置文件灵活设置股票筛选条件，支持全市场回测和特定类型股票的筛选：
# 个股筛选配置
STOCK_FILTER_CONFIG = {
'include_st': False, # 是否包含ST/*ST股票，默认False
'include_bse': False, # 是否包含北交所股票（代码以8开头），默认False
'include_sse_star': False, # 是否包含科创板股票（代码以688开头），默认False
'include_gem': True, # 是否包含创业板股票（代码以3开头），默认True
'include_main': True, # 是否包含主板股票（代码以0或6开头），默认True
'custom_stocks': STOCK_POOL # 自定义股票列表，优先级最高
}
筛选规则说明：
custom_stocks: 自定义股票列表，设置后将忽略其他筛选条件include_st: 控制是否包含ST或*ST股票include_bse: 控制是否包含北交所股票（股票代码以8开头）include_sse_star: 控制是否包含科创板股票（股票代码以688开头）include_gem: 控制是否包含创业板股票（股票代码以3开头）include_main: 控制是否包含主板股票（股票代码以0或6开头）
使用场景示例：
-
全市场回测（排除ST和北交所）：
STOCK_FILTER_CONFIG = { 'include_st': False, 'include_bse': False, 'include_sse_star': True, 'include_gem': True, 'include_main': True, 'custom_stocks': None # 不使用自定义股票池，使用全市场筛选 } -
仅主板和创业板（排除科创板、北交所和ST）：
STOCK_FILTER_CONFIG = { 'include_st': False, 'include_bse': False, 'include_sse_star': False, 'include_gem': True, 'include_main': True, 'custom_stocks': None }
策略配置
# 策略参数
MA_PARAMS = {
'short_period': 5,
'medium_period': 10,
'long_period': 20,
'extra_long_period': 60
}
YIYANG_CONDITIONS = {
'min_pct_change': 2.0,
'min_volume_ratio': 1.0,
'max_ma_gap_pct': 4.0,
'min_entity_ratio': 0.6,
'confirm_days': 2
}
# 策略列表
STRATEGY_LIST = ['YiYangStrategy'] # 要运行的策略
回测配置
# 回测配置
BACKTEST = {
'initial_capital': 100000, # 初始资金
'commission_rate': 0.0003, # 佣金费率
'stamp_tax_rate': 0.001, # 印花税
'slippage_rate': 0.001, # 滑点
'position_ratio': 0.2, # 仓位比例
'stop_loss': 0.08, # 止损比例
'take_profit': 0.20 # 止盈比例
}
参数优化配置
# 参数优化配置
RUN_PARAMETER_OPTIMIZATION = False # 是否运行参数优化
OPTIMIZATION_CONFIG = {
'strategy_name': 'YiYangStrategy', # 要优化的策略名称
'metric': 'sharpe_ratio' # 优化指标
}
使用方法
运行回测
直接运行主入口文件：
python main.py
查看结果
回测完成后，会生成以下结果：
- 控制台输出：回测结果摘要、最近交易记录
- 日志文件：位于
logs/目录 - 图表文件：位于
charts/目录 - 参数优化结果：如果启用了参数优化，会生成
optimization_results.csv文件
添加新策略
1. 创建策略类
在 strategy/ 目录下创建新的策略文件，例如 my_strategy.py，继承自 BaseStrategy 类：
# strategy/my_strategy.py
from strategy.base_strategy import BaseStrategy
import pandas as pd
class MyStrategy(BaseStrategy):
def __init__(self, config=None):
super().__init__()
self.config = config or {}
self.params = {
'param1': 10,
'param2': 0.5
}
def initialize(self, data):
# 初始化策略
pass
def generate_signals(self, data):
# 生成交易信号
signals = pd.DataFrame(index=data.index)
signals['signal'] = 0
signals['strength'] = 0.0
signals['reason'] = ''
# 实现信号生成逻辑
# signals.loc[..., 'signal'] = 1 # 买入
# signals.loc[..., 'signal'] = -1 # 卖出
return signals
def on_signal(self, date, signal, price):
# 处理交易信号
if signal == 1:
return {'action': 'buy', 'quantity': 100}
elif signal == -1:
return {'action': 'sell', 'quantity': 100}
else:
return {'action': 'hold', 'quantity': 0}
2. 注册策略
在 main.py 中的 strategy_class_map 字典中注册新策略：
# main.py 中的 strategy_class_map
def main():
# ...
# 策略类映射，用于动态加载策略
strategy_class_map = {
'YiYangStrategy': YiYangStrategy,
'MyStrategy': MyStrategy # 添加新策略
}
# ...
3. 在配置文件中启用策略
在 config.py 中的 STRATEGY_LIST 中添加新策略：
# config.py
STRATEGY_LIST = ['YiYangStrategy', 'MyStrategy'] # 添加新策略
策略参数配置
1. 策略内部默认参数
在策略类的 __init__ 方法中设置默认参数：
class MyStrategy(BaseStrategy):
def __init__(self, config=None):
super().__init__()
self.params = {
'param1': 10, # 默认参数1
'param2': 0.5 # 默认参数2
}
2. 通过配置文件配置参数
在 config.py 中添加策略参数配置：
# config.py
MY_STRATEGY_PARAMS = {
'param1': 15, # 覆盖默认参数1
'param2': 0.7 # 覆盖默认参数2
}
在策略初始化时读取配置：
class MyStrategy(BaseStrategy):
def __init__(self, config=None):
super().__init__()
self.params = {
'param1': 10,
'param2': 0.5
}
# 从配置中更新参数
if config and hasattr(config, 'MY_STRATEGY_PARAMS'):
self.params.update(config.MY_STRATEGY_PARAMS)
策略参数优化
1. 启用参数优化
在 config.py 中设置：
# config.py
RUN_PARAMETER_OPTIMIZATION = True # 启用参数优化
2. 定义参数搜索空间
在 main.py 的 run_parameter_optimization 函数中定义参数搜索空间：
def run_parameter_optimization(config, data_fetcher):
# ...
# 定义参数搜索空间
param_space = {
'param1': [5, 10, 15, 20], # 参数1的搜索范围
'param2': [0.3, 0.5, 0.7, 0.9] # 参数2的搜索范围
}
# ...
3. 运行参数优化
运行主程序，参数优化会自动执行：
python main.py
4. 查看优化结果
参数优化完成后，会输出最佳参数组合，并生成 optimization_results.csv 文件，包含所有参数组合的回测结果。
常用优化指标
total_return: 总收益率annual_return: 年化收益率sharpe_ratio: 夏普比率max_drawdown: 最大回撤win_rate: 胜率profit_factor: 盈亏比
可以在 config.py 的 OPTIMIZATION_CONFIG 中指定优化指标：
OPTIMIZATION_CONFIG = {
'strategy_name': 'MyStrategy',
'metric': 'sharpe_ratio' # 使用夏普比率作为优化目标
}
性能优化建议
- 减少参数空间：在进行参数优化时，减少每个参数的取值数量可以显著提高优化速度
- 缩短回测周期：对于初步优化，可以使用较短的回测周期
- 使用缓存数据：确保数据获取模块使用缓存功能，避免重复请求数据
扩展功能
添加新的数据源
在 data/ 目录下创建新的数据源类，继承自 DataFetcher 基类：
# data/my_data.py
from data.data_fetcher import DataFetcher
class MyDataFetcher(DataFetcher):
def get_daily_data(self, ts_code, start_date, end_date):
# 实现数据获取逻辑
pass
添加新的绩效指标
在 backtest/performance.py 中添加新的绩效指标计算函数：
def calculate_new_metric(trades_df, equity_curve):
# 实现新指标的计算逻辑
return metric_value
日志管理
日志文件位于 logs/ 目录下，包括：
backtrader.log: 回测引擎日志parameter_optimizer.log: 参数优化日志
可以在 utils/logger.py 中配置日志级别和格式。
故障排除
- 数据获取失败：检查网络连接和API token配置
- 策略错误：查看日志文件中的错误信息，检查策略逻辑
- 参数优化缓慢：减少参数搜索空间或缩短回测周期
- 内存不足：减少同时回测的策略数量或缩短回测周期
示例
运行单个策略
python main.py
启用参数优化
# 在config.py中设置RUN_PARAMETER_OPTIMIZATION = True
python main.py
贡献
欢迎提交issue和pull request，一起完善这个回测引擎。
许可证
MIT License