策略回测框架

回测是量化交易的核心环节，通过在历史数据上模拟策略运行，评估策略的潜在表现。本章将详细介绍Backtrader回测框架的使用方法。

回测基础概念

什么是回测？

回测（Backtesting）是将交易策略应用于历史数据，模拟策略在过去的表现。通过回测，我们可以：

• 验证策略逻辑是否正确
• 评估策略的收益和风险特征
• 优化策略参数
• 发现策略的潜在问题

但回测结果不能保证未来表现，因为：

• 历史不会简单重复
• 回测可能存在过拟合
• 实盘交易面临更多现实约束

回测的关键要素

数据质量：垃圾进垃圾出，高质量的历史数据是回测的基础。数据需要准确、完整、经过调整（分红拆股）。

交易成本：包括佣金、印花税、滑点等。忽略交易成本会高估策略收益，尤其是高频策略。

流动性约束：大额交易可能无法按预期价格成交，需要考虑市场深度和冲击成本。

幸存者偏差：如果只使用当前仍在交易的股票数据，会忽略已经退市的股票，导致回测结果偏乐观。

前视偏差：在交易决策时使用了当时不可得的信息，如用收盘价做日内交易决策。

Backtrader框架介绍

Backtrader是一个功能完整的Python回测框架，具有以下特点：

• 支持多种数据源
• 内置丰富的技术指标
• 灵活的策略定义方式
• 支持多策略、多资产回测
• 详细的交易记录和分析报告

安装Backtrader

pip install backtrader

Backtrader核心组件

Backtrader的核心组件包括：

Cerebro：回测引擎，负责协调各个组件运行

Strategy：策略类，定义交易逻辑

Data Feed：数据源，提供价格数据

Indicator：技术指标，用于策略分析

Sizer：仓位管理器，决定每次交易的股数

Analyzer：分析器，计算策略绩效指标

Observer：观察者，记录回测过程中的数据

Plot：绘图器，可视化回测结果

第一个回测策略

下面实现一个简单的均线交叉策略：

import backtrader as bt
import yfinance as yf
from datetime import datetime

# 定义策略类
class MAStrategy(bt.Strategy):
    params = (
        ('short_period', 5),   # 短期均线周期
        ('long_period', 20),   # 长期均线周期
    )
    
    def __init__(self):
        # 计算移动平均线
        self.ma_short = bt.indicators.SMA(self.data.close, period=self.params.short_period)
        self.ma_long = bt.indicators.SMA(self.data.close, period=self.params.long_period)
        
        # 记录订单状态
        self.order = None
        self.buy_price = None
        self.buy_comm = None
    
    def notify_order(self, order):
        """订单状态变化时调用"""
        if order.status in [order.Submitted, order.Accepted]:
            return
        
        if order.status in [order.Completed]:
            if order.isbuy():
                self.buy_price = order.executed.price
                self.buy_comm = order.executed.comm
                print(f'买入执行: 价格{order.executed.price:.2f}, '
                      f'成本{order.executed.value:.2f}, '
                      f'手续费{order.executed.comm:.2f}')
            else:
                print(f'卖出执行: 价格{order.executed.price:.2f}, '
                      f'成本{order.executed.value:.2f}, '
                      f'手续费{order.executed.comm:.2f}')
        
        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            print('订单取消/保证金不足/拒绝')
        
        self.order = None
    
    def notify_trade(self, trade):
        """交易完成时调用"""
        if not trade.isclosed:
            return
        print(f'交易盈亏: 毛利{trade.pnl:.2f}, 净利{trade.pnlcomm:.2f}')
    
    def next(self):
        """每个交易日调用一次，核心策略逻辑"""
        # 如果有未完成订单，不操作
        if self.order:
            return
        
        # 检查是否有持仓
        if not self.position:
            # 没有持仓，检查买入条件
            if self.ma_short[0] > self.ma_long[0] and self.ma_short[-1] <= self.ma_long[-1]:
                # 短期均线上穿长期均线，金叉买入
                print(f'买入信号: {self.data.datetime.date()}, 价格{self.data.close[0]:.2f}')
                self.order = self.buy()
        else:
            # 有持仓，检查卖出条件
            if self.ma_short[0] < self.ma_long[0] and self.ma_short[-1] >= self.ma_long[-1]:
                # 短期均线下穿长期均线，死叉卖出
                print(f'卖出信号: {self.data.datetime.date()}, 价格{self.data.close[0]:.2f}')
                self.order = self.sell()

# 创建回测引擎
cerebro = bt.Cerebro()

# 添加策略
cerebro.addstrategy(MAStrategy)

# 获取数据
data = yf.download('AAPL', start='2020-01-01', end='2023-12-31')

# 转换为Backtrader数据格式
data_feed = bt.feeds.PandasData(
    dataname=data,
    datetime=None,  # 使用DataFrame的索引作为日期
    open='Open',
    high='High',
    low='Low',
    close='Close',
    volume='Volume',
    openinterest=-1  # 没有持仓量数据
)

# 添加数据
cerebro.adddata(data_feed)

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

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

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

# 运行回测
print(f'初始资金: {cerebro.broker.getvalue():.2f}')
results = cerebro.run()
print(f'最终资金: {cerebro.broker.getvalue():.2f}')

# 获取分析结果
strat = results[0]
print(f'夏普比率: {strat.analyzers.sharpe.get_analysis()["sharperatio"]}')
print(f'最大回撤: {strat.analyzers.drawdown.get_analysis()["max"]["drawdown"]:.2f}%')
print(f'年化收益: {strat.analyzers.returns.get_analysis()["rnorm100"]:.2f}%')

# 绘制结果
cerebro.plot(style='candlestick')

策略参数优化

Backtrader支持参数优化，可以自动测试不同参数组合的表现：

# 参数优化
cerebro = bt.Cerebro()

# 添加策略，使用optstrategy进行参数优化
cerebro.optstrategy(
    MAStrategy,
    short_period=range(3, 10),   # 短期均线周期范围
    long_period=range(15, 30)    # 长期均线周期范围
)

# 添加数据
cerebro.adddata(data_feed)

# 设置初始资金和手续费
cerebro.broker.setcash(100000.0)
cerebro.broker.setcommission(commission=0.001)

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

# 运行优化（maxcpu=1表示单进程）
results = cerebro.run(maxcpu=1)

# 输出优化结果
for result in results:
    sharpe = result[0].analyzers.sharpe.get_analysis()['sharperatio']
    params = result[0].params
    print(f'参数: short={params.short_period}, long={params.long_period}, 夏普比率: {sharpe}')

自定义指标

Backtrader允许自定义技术指标：

class CustomRSI(bt.Indicator):
    """自定义RSI指标"""
    lines = ('rsi',)
    params = (('period', 14),)
    
    def __init__(self):
        self.addminperiod(self.params.period)
    
    def next(self):
        # 计算价格变化
        deltas = [self.data.close[-i] - self.data.close[-i-1] for i in range(self.params.period)]
        
        # 分离上涨和下跌
        gains = [d if d > 0 else 0 for d in deltas]
        losses = [-d if d < 0 else 0 for d in deltas]
        
        # 计算平均上涨和下跌
        avg_gain = sum(gains) / self.params.period
        avg_loss = sum(losses) / self.params.period
        
        # 计算RSI
        if avg_loss == 0:
            self.lines.rsi[0] = 100
        else:
            rs = avg_gain / avg_loss
            self.lines.rsi[0] = 100 - (100 / (1 + rs))

# 在策略中使用自定义指标
class RSIStrategy(bt.Strategy):
    params = (('rsi_period', 14), ('oversold', 30), ('overbought', 70))
    
    def __init__(self):
        self.rsi = CustomRSI(self.data, period=self.params.rsi_period)
    
    def next(self):
        if not self.position:
            if self.rsi[0] < self.params.oversold:
                self.buy()
        else:
            if self.rsi[0] > self.params.overbought:
                self.sell()

仓位管理

Backtrader提供了多种仓位管理方式：

class SizerStrategy(bt.Strategy):
    params = (('size_pct', 0.95),)  # 使用95%的可用资金
    
    def __init__(self):
        self.ma = bt.indicators.SMA(self.data.close, period=20)
    
    def next(self):
        if not self.position:
            if self.data.close[0] > self.ma[0]:
                # 方法1：固定股数
                # self.buy(size=100)
                
                # 方法2：按资金比例
                cash = self.broker.getcash()
                price = self.data.close[0]
                size = int((cash * self.params.size_pct) / price)
                self.buy(size=size)
                
                # 方法3：使用Sizer
                # self.buy()  # 需要先设置Sizer

# 设置Sizer
cerebro = bt.Cerebro()
cerebro.addstrategy(SizerStrategy)
cerebro.addsizer(bt.sizers.PercentSizer, percents=95)  # 使用95%资金

多资产回测

Backtrader支持同时回测多个资产：

class MultiAssetStrategy(bt.Strategy):
    params = (('period', 20),)
    
    def __init__(self):
        # 为每个数据创建指标
        self.mas = {}
        for data in self.datas:
            self.mas[data._name] = bt.indicators.SMA(data.close, period=self.params.period)
    
    def next(self):
        for data in self.datas:
            pos = self.getposition(data)
            if not pos:
                if data.close[0] > self.mas[data._name][0]:
                    self.buy(data=data)
            else:
                if data.close[0] < self.mas[data._name][0]:
                    self.sell(data=data)

# 创建回测引擎
cerebro = bt.Cerebro()
cerebro.addstrategy(MultiAssetStrategy)

# 添加多个数据
tickers = ['AAPL', 'MSFT', 'GOOGL']
for ticker in tickers:
    data = yf.download(ticker, start='2020-01-01', end='2023-12-31')
    data_feed = bt.feeds.PandasData(dataname=data, name=ticker)
    cerebro.adddata(data_feed)

cerebro.broker.setcash(100000.0)
cerebro.run()

回测报告分析

def print_analysis(results):
    """打印详细的回测分析报告"""
    strat = results[0]
    
    # 收益分析
    returns = strat.analyzers.returns.get_analysis()
    print('=' * 50)
    print('收益分析')
    print('=' * 50)
    print(f"总收益率: {returns.get('rtot', 0) * 100:.2f}%")
    print(f"年化收益率: {returns.get('rnorm100', 0):.2f}%")
    
    # 风险分析
    drawdown = strat.analyzers.drawdown.get_analysis()
    print('\n' + '=' * 50)
    print('风险分析')
    print('=' * 50)
    print(f"最大回撤: {drawdown.get('max', {}).get('drawdown', 0):.2f}%")
    print(f"最大回撤持续时间: {drawdown.get('max', {}).get('len', 0)} 天")
    
    # 夏普比率
    sharpe = strat.analyzers.sharpe.get_analysis()
    print('\n' + '=' * 50)
    print('风险调整收益')
    print('=' * 50)
    print(f"夏普比率: {sharpe.get('sharperatio', 'N/A')}")
    
    # 交易分析
    trades = strat.analyzers.trades.get_analysis()
    print('\n' + '=' * 50)
    print('交易分析')
    print('=' * 50)
    print(f"总交易次数: {trades.get('total', {}).get('total', 0)}")
    print(f"盈利交易: {trades.get('won', {}).get('total', 0)}")
    print(f"亏损交易: {trades.get('lost', {}).get('total', 0)}")
    if trades.get('total', {}).get('total', 0) > 0:
        win_rate = trades.get('won', {}).get('total', 0) / trades['total']['total'] * 100
        print(f"胜率: {win_rate:.2f}%")
        avg_win = trades.get('won', {}).get('pnl', {}).get('average', 0)
        avg_loss = trades.get('lost', {}).get('pnl', {}).get('average', 0)
        print(f"平均盈利: {avg_win:.2f}")
        print(f"平均亏损: {avg_loss:.2f}")
        if avg_loss != 0:
            print(f"盈亏比: {abs(avg_win / avg_loss):.2f}")

# 使用示例
cerebro = bt.Cerebro()
cerebro.addstrategy(MAStrategy)
cerebro.adddata(data_feed)
cerebro.broker.setcash(100000.0)
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')
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')

results = cerebro.run()
print_analysis(results)

回测常见陷阱

前视偏差

使用未来信息做决策是最常见的回测错误：

# 错误示例：使用当日收盘价做日内决策
def next(self):
    if self.data.close[0] > self.ma[0]:  # close[0]是当日收盘价
        self.buy()  # 但这个决策应该在开盘时做出

# 正确示例：使用前一日数据
def next(self):
    if self.data.close[-1] > self.ma[-1]:  # 使用昨日收盘价
        self.buy()  # 今日开盘买入

过拟合

过度优化参数会导致策略只对历史数据有效：

# 避免过拟合的方法：
# 1. 样本外测试：将数据分为训练集和测试集
# 2. 参数稳健性：测试参数小幅变化是否导致性能大幅下降
# 3. 交叉验证：使用滚动窗口回测
# 4. 简化策略：减少参数数量，避免过度复杂

# 样本外测试示例
train_data = data[:'2022-12-31']  # 训练集
test_data = data['2023-01-01':]   # 测试集

# 在训练集上优化参数
cerebro_train = bt.Cerebro()
cerebro_train.optstrategy(MAStrategy, short_period=range(3, 10), long_period=range(15, 30))
cerebro_train.adddata(bt.feeds.PandasData(dataname=train_data))
train_results = cerebro_train.run()

# 选择最优参数
best_params = max(train_results, key=lambda x: x[0].analyzers.sharpe.get_analysis()['sharperatio'])
best_short = best_params[0].params.short_period
best_long = best_params[0].params.long_period

# 在测试集上验证
cerebro_test = bt.Cerebro()
cerebro_test.addstrategy(MAStrategy, short_period=best_short, long_period=best_long)
cerebro_test.adddata(bt.feeds.PandasData(dataname=test_data))
test_results = cerebro_test.run()

忽略交易成本

# 设置真实的交易成本
cerebro.broker.setcommission(commission=0.0003)  # 佣金万三

# 设置滑点
cerebro.broker.set_slippage_perc(perc=0.0001)  # 滑点万分之一

# 或者使用固定滑点
cerebro.broker.set_slippage_fixed(spread=0.01)  # 固定滑点0.01元

小结

本章详细介绍了Backtrader回测框架的使用方法，包括策略定义、参数优化、多资产回测等内容。回测是量化交易的重要环节，但回测结果不能保证未来表现。在回测时需要注意避免前视偏差、过拟合等常见陷阱，并合理设置交易成本。下一章将学习风险管理的基础知识。