75.3 回测框架搭建

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

# 数据加载类
class MultiFactorData(bt.feeds.PandasData):
    params = (
        ('datetime', None),
        ('factor', 6),
        ('returns', 7),
    )

# 多因子策略类
class MultiFactorStrategy(bt.Strategy):
    params = (
        ('lookback', 20),
        ('top_n', 50),
        ('rebalance_freq', 20),
    )

    def __init__(self):
        self.factor_data = self.datas[0].factor
        self.returns = self.datas[0].returns
        self.rebalance_counter = 0

    def next(self):
        # 按频率调仓
        self.rebalance_counter += 1
        if self.rebalance_counter % self.params.rebalance_freq != 0:
            return

        # 获取因子值
        current_factor = self.factor_data[0]

        # 计算因子排名
        factor_ranks = {}
        for i, data in enumerate(self.datas):
            factor_ranks[i] = data.factor[0]

        # 选择因子值最高的股票
        sorted_stocks = sorted(factor_ranks.items(),
                              key=lambda x: x[1],
                              reverse=True)
        top_stocks = [i for i, _ in sorted_stocks[:self.params.top_n]]

        # 调仓逻辑
        for i, data in enumerate(self.datas):
            current_position = self.getposition(data).size

            if i in top_stocks and current_position == 0:
                # 买入
                target_value = self.broker.getvalue() / self.params.top_n
                price = data.close[0]
                size = target_value / price
                self.buy(data=data, size=size)

            elif i not in top_stocks and current_position > 0:
                # 卖出
                self.sell(data=data, size=current_position)

# 回测引擎
def run_backtest(factor_data, returns, start_date, end_date):
    """
    运行回测
    """
    cerebro = bt.Cerebro()

    # 添加策略
    cerebro.addstrategy(MultiFactorStrategy,
                       lookback=20,
                       top_n=50,
                       rebalance_freq=20)

    # 加载数据
    for ticker in factor_data.columns:
        data = pd.DataFrame({
            'datetime': factor_data.index,
            'close': 1.0,  # 基准价格
            'factor': factor_data[ticker],
            'returns': returns[ticker]
        })
        data_feed = MultiFactorData(dataname=data)
        cerebro.adddata(data_feed)

    # 设置初始资金
    cerebro.broker.setcash(10000000)

    # 设置手续费
    cerebro.broker.setcommission(commission=0.001)

    # 添加分析器
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')

    # 运行回测
    results = cerebro.run()
    strat = results[0]

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

    return {
        'final_value': cerebro.broker.getvalue(),
        'sharpe_ratio': sharpe.get('sharperatio'),
        'max_drawdown': drawdown.get('max', {}).get('drawdown'),
        'total_return': returns.get('rtot')
    }

import numpy as np

class PerformanceAnalyzer:
    """
    绩效分析器
    """

    def __init__(self, portfolio_returns, benchmark_returns):
        self.portfolio_returns = portfolio_returns
        self.benchmark_returns = benchmark_returns

    def calculate_metrics(self):
        """计算绩效指标"""
        metrics = {}

        # 年化收益率
        metrics['annual_return'] = self._calculate_annual_return()

        # 夏普比率
        metrics['sharpe_ratio'] = self._calculate_sharpe_ratio()

        # 最大回撤
        metrics['max_drawdown'] = self._calculate_max_drawdown()

        # 胜率
        metrics['win_rate'] = self._calculate_win_rate()

        # 盈亏比
        metrics['win_loss_ratio'] = self._calculate_win_loss_ratio()

        # 信息比率
        metrics['information_ratio'] = self._calculate_information_ratio()

        # Alpha与Beta
        alpha, beta = self._calculate_alpha_beta()
        metrics['alpha'] = alpha
        metrics['beta'] = beta

        return metrics

    def _calculate_annual_return(self):
        """计算年化收益率"""
        total_return = (1 + self.portfolio_returns).prod() - 1
        years = len(self.portfolio_returns) / 252
        annual_return = (1 + total_return) ** (1/years) - 1
        return annual_return

    def _calculate_sharpe_ratio(self, risk_free_rate=0.03):
        """计算夏普比率"""
        excess_returns = self.portfolio_returns - risk_free_rate/252
        return np.sqrt(252) * excess_returns.mean() / excess_returns.std()

    def _calculate_max_drawdown(self):
        """计算最大回撤"""
        cum_returns = (1 + self.portfolio_returns).cumprod()
        rolling_max = cum_returns.expanding().max()
        drawdown = (cum_returns - rolling_max) / rolling_max
        return drawdown.min()

    def _calculate_win_rate(self):
        """计算胜率"""
        winning_days = (self.portfolio_returns > 0).sum()
        total_days = len(self.portfolio_returns)
        return winning_days / total_days

    def _calculate_win_loss_ratio(self):
        """计算盈亏比"""
        winning_returns = self.portfolio_returns[self.portfolio_returns > 0]
        losing_returns = self.portfolio_returns[self.portfolio_returns < 0]
        return abs(winning_returns.mean() / losing_returns.mean())

    def _calculate_information_ratio(self):
        """计算信息比率"""
        active_returns = self.portfolio_returns - self.benchmark_returns
        return np.sqrt(252) * active_returns.mean() / active_returns.std()

    def _calculate_alpha_beta(self):
        """计算Alpha与Beta"""
        covariance = np.cov(self.portfolio_returns, self.benchmark_returns)[0,1]
        benchmark_variance = self.benchmark_returns.var()
        beta = covariance / benchmark_variance

        alpha = (self.portfolio_returns.mean() -
                self.benchmark_returns.mean() * beta) * 252

        return alpha, beta

import matplotlib.pyplot as plt
import seaborn as sns

class BacktestVisualizer:
    """
    回测可视化
    """

    @staticmethod
    def plot_equity_curve(portfolio_returns, benchmark_returns):
        """绘制净值曲线"""
        fig, ax = plt.subplots(figsize=(12, 6))

        portfolio_cum = (1 + portfolio_returns).cumprod()
        benchmark_cum = (1 + benchmark_returns).cumprod()

        ax.plot(portfolio_cum.index, portfolio_cum, label='策略净值', linewidth=2)
        ax.plot(benchmark_cum.index, benchmark_cum, label='基准净值', linewidth=2, alpha=0.7)

        ax.set_xlabel('日期')
        ax.set_ylabel('净值')
        ax.set_title('策略净值曲线')
        ax.legend()
        ax.grid(True, alpha=0.3)

        return fig

    @staticmethod
    def plot_drawdown(portfolio_returns):
        """绘制回撤图"""
        fig, ax = plt.subplots(figsize=(12, 6))

        cum_returns = (1 + portfolio_returns).cumprod()
        rolling_max = cum_returns.expanding().max()
        drawdown = (cum_returns - rolling_max) / rolling_max

        ax.fill_between(drawdown.index, drawdown, 0, alpha=0.3, color='red')
        ax.plot(drawdown.index, drawdown, color='red', linewidth=2)

        ax.set_xlabel('日期')
        ax.set_ylabel('回撤')
        ax.set_title('策略回撤曲线')
        ax.grid(True, alpha=0.3)

        return fig

    @staticmethod
    def plot_monthly_returns(portfolio_returns):
        """绘制月度收益热图"""
        monthly_returns = portfolio_returns.resample('M').apply(
            lambda x: (1 + x).prod() - 1
        )

        monthly_returns_df = pd.DataFrame({
            'year': monthly_returns.index.year,
            'month': monthly_returns.index.month,
            'return': monthly_returns.values
        }).pivot('year', 'month', 'return')

        fig, ax = plt.subplots(figsize=(12, 8))
        sns.heatmap(monthly_returns_df, cmap='RdYlGn', center=0,
                   fmt='.2%', annot=True, ax=ax)

        ax.set_title('月度收益热图')
        ax.set_xlabel('月份')
        ax.set_ylabel('年份')

        return fig