backtest_engine.py

import backtrader as bt
import pandas as pd
from datetime import datetime
import json

class BacktestEngine:
    def __init__(self, config_file='config.json'):
        """
        初始化回测引擎

        参数:
        config_file: 配置文件路径
        """
        self.config = self.load_config(config_file)
        self.cerebro = bt.Cerebro()
        self.value_history = []  # 用于记录资产价值历史

    def load_config(self, config_file):
        """
        加载配置文件

        参数:
        config_file: 配置文件路径

        返回:
        dict: 配置参数字典
        """
        try:
            with open(config_file, 'r', encoding='utf-8') as f:
                config = json.load(f)
            return config
        except Exception as e:
            print(f"加载配置文件时出错: {e}")
            return {}

    def add_strategy(self, strategy_class, **kwargs):
        """
        添加策略到回测引擎

        参数:
        strategy_class: 策略类
        **kwargs: 策略参数
        """
        self.cerebro.addstrategy(strategy_class, **kwargs)

    def add_data(self, data_feed):
        """
        添加数据到回测引擎

        参数:
        data_feed: Backtrader数据feed
        """
        self.cerebro.adddata(data_feed)

    def set_initial_cash(self, cash):
        """
        设置初始资金

        参数:
        cash: 初始资金
        """
        self.cerebro.broker.setcash(cash)

    def add_analyzers(self):
        """
        添加分析器
        """
        # 添加收益率分析器
        self.cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
        # 添加夏普比率分析器
        self.cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
        # 添加最大回撤分析器
        self.cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')

    def run(self):
        """
        运行回测

        返回:
        list: 回测结果
        """
        # 添加观察器来记录资产价值
        self.cerebro.addobserver(bt.observers.Value)

        # 添加分析器
        self.add_analyzers()

        # 运行回测
        results = self.cerebro.run()

        return results

    def get_results(self, results):
        """
        获取回测结果

        参数:
        results: 回测结果

        返回:
        dict: 包含收益率、夏普比率、最大回撤等指标的字典
        """
        if not results:
            return {}

        strat = results[0]

        # 获取分析器结果
        returns = strat.analyzers.returns.get_analysis()
        sharpe = strat.analyzers.sharpe.get_analysis()
        drawdown = strat.analyzers.drawdown.get_analysis()

        # 获取资产价值历史（从观察器）
        value_history = []
        if hasattr(strat, '_value') and strat._value:
            value_history = list(strat._value)

        # 构建结果字典
        result_dict = {
            'total_return': returns.get('rtot', 0),
            'annual_return': returns.get('ravg', 0) * 252,  # 年化收益率
            'sharpe_ratio': sharpe.get('sharperatio', 0),
            'max_drawdown': drawdown.get('max.drawdown', 0),
            'value_history': value_history  # 资金曲线
        }

        return result_dict

    def add_benchmark(self, benchmark_data):
        """
        添加基准股票数据

        参数:
        benchmark_data: 基准股票数据
        """
        # 添加基准数据
        self.cerebro.adddata(benchmark_data, name='benchmark')