第45篇 Python量化实战:编写回测布林双策略系统

import backtrader as btimport pandas as pdfrom datetime import datetime# 布林突破策略（趋势跟踪）class BollingerBandsBreakoutStrategy(bt.Strategy):    params = (        ('period', 20),        ('devfactor', 2.0),        ('adx_period', 14),    )    def __init__(self):        self.boll = bt.indicators.BollingerBands(self.data.close,                                                   period=self.params.period,                                                   devfactor=self.params.devfactor)        self.adx = bt.indicators.ADX(self.data, period=self.params.adx_period)        self.stop_loss_pct = 0.05  # 5%止损    def next(self):        if len(self) < max(self.params.period, self.params.adx_period):            return        position = self.getposition().size        # 趋势行情判定（ADX>25）        if self.adx < 25:            return  # 非趋势行情，不执行策略        if not position:            # 做多信号：价格突破上轨            if self.data.close > self.boll.lines.top:                stop_price = self.data.close * (1 - self.stop_loss_pct)                self.buy(exectype=bt.Order.Market)                self.sell(exectype=bt.Order.Stop, price=stop_price)            # 做空信号：价格跌破下轨            elif self.data.close < self.boll.lines.bot:                stop_price = self.data.close * (1 + self.stop_loss_pct)                self.sell(exectype=bt.Order.Market)                self.buy(exectype=bt.Order.Stop, price=stop_price)        elif position > 0:            # 止盈：价格回落至中轨下方            if self.data.close < self.boll.lines.mid:                self.close()            # 止损触发            elif self.data.close < self.getposition().price * (1 - self.stop_loss_pct):                self.close()        elif position < 0:            # 止盈：价格反弹至中轨上方            if self.data.close > self.boll.lines.mid:                self.close()            # 止损触发            elif self.data.close > self.getposition().price * (1 + self.stop_loss_pct):                self.close()# 布林震荡策略（均值回归）class BollingerBandsReversionStrategy(bt.Strategy):    params = (        ('period', 20),        ('devfactor', 2.0),        ('rsi_period', 14),    )    def __init__(self):        self.boll = bt.indicators.BollingerBands(self.data.close,                                                   period=self.params.period,                                                   devfactor=self.params.devfactor)        self.rsi = bt.indicators.RSI(self.data.close, period=self.params.rsi_period)        self.stop_loss_pct = 0.03  # 3%止损    def next(self):        if len(self) < max(self.params.period, self.params.rsi_period):            return        position = self.getposition().size        # 震荡行情判定（ADX<20）        if self.adx > 20:            return  # 非震荡行情，不执行策略        if not position:            # 做多信号：价格触及下轨+RSI超卖            if (self.data.close <= self.boll.lines.bot and                 self.rsi < 30):                stop_price = self.data.close * (1 - self.stop_loss_pct)                self.buy(exectype=bt.Order.Market)                self.sell(exectype=bt.Order.Stop, price=stop_price)            # 做空信号：价格触及上轨+RSI超买            elif (self.data.close >= self.boll.lines.top and                   self.rsi > 70):                stop_price = self.data.close * (1 + self.stop_loss_pct)                self.sell(exectype=bt.Order.Market)                self.buy(exectype=bt.Order.Stop, price=stop_price)        elif position > 0:            # 止盈：价格反弹至中轨上方            if self.data.close >= self.boll.lines.mid:                self.close()            # 止损触发            elif self.data.close < self.getposition().price * (1 - self.stop_loss_pct):                self.close()        elif position < 0:            # 止盈：价格回落至中轨下方            if self.data.close <= self.boll.lines.mid:                self.close()            # 止损触发            elif self.data.close > self.getposition().price * (1 + self.stop_loss_pct):                self.close()# 双策略融合主策略class DualStrategySystem(bt.Strategy):    params = (        ('breakout_period', 20),        ('reversion_period', 20),    )    def __init__(self):        # 初始化两个子策略        self.breakout_strategy = BollingerBandsBreakoutStrategy(self.data)        self.reversion_strategy = BollingerBandsReversionStrategy(self.data)        # 初始化ADX指标用于行情判定        self.adx = bt.indicators.ADX(self.data, period=14)    def next(self):        # 根据ADX值切换主策略        if self.adx > 25:            # 趋势行情，执行突破策略            self.breakout_strategy.next()        elif self.adx < 20:            # 震荡行情，执行震荡策略            self.reversion_strategy.next()        else:            # 过渡行情，两个策略同时运行            self.breakout_strategy.next()            self.reversion_strategy.next()# 完整回测流程if __name__ == '__main__':    cerebro = bt.Cerebro()    # 1. 加载数据（示例：从CSV文件读取，需包含open,high,low,close,volume列）    # data = bt.feeds.PandasData(dataname=pd.read_csv('510300.csv', parse_dates=True, index_col=0))    # cerebro.adddata(data)    # 2. 模拟生成测试数据（用于演示）    data = bt.feeds.RandomWalkData()    cerebro.adddata(data)    # 3. 添加双策略系统    cerebro.addstrategy(DualStrategySystem, breakout_period=20, reversion_period=20)    # 4. 设置初始资金    cerebro.broker.setcash(100000.0)    # 5. 设置交易手续费    cerebro.broker.setcommission(commission=0.001)  # 万分之一手续费    # 6. 添加绩效分析器    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='trade_analyzer')    # 7. 运行回测    print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())    results = cerebro.run()    print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue())    # 8. 输出绩效指标    strat = results[0]    print(f'Sharpe Ratio: {strat.analyzers.sharpe.get_analysis()["sharperatio"]:.2f}')    print(f'Maximum Drawdown: {strat.analyzers.drawdown.get

import backtrader as btimport pandas as pdfrom datetime import datetime# 布林突破策略（趋势跟踪）class BollingerBandsBreakoutStrategy(bt.Strategy):    params = (        ('period', 20),          # 布林带均线周期        ('devfactor', 2.0),      # 标准差倍数        ('adx_period', 14),      # ADX指标周期        ('stop_loss_pct', 0.05)  # 5%止损比例    )    def __init__(self):        self.boll = bt.indicators.BollingerBands(self.data.close,                                                   period=self.params.period,                                                   devfactor=self.params.devfactor)        self.adx = bt.indicators.ADX(self.data, period=self.params.adx_period)        self.stop_loss_pct = self.params.stop_loss_pct    def next(self):        if len(self) < max(self.params.period, self.params.adx_period):            return        position = self.getposition().size        # 仅在趋势行情（ADX>25）中执行策略        if self.adx < 25:            return        if not position:            # 做多信号：价格突破上轨            if self.data.close > self.boll.lines.top:                stop_price = self.data.close * (1 - self.stop_loss_pct)                self.buy(exectype=bt.Order.Market)                self.sell(exectype=bt.Order.Stop, price=stop_price)            # 做空信号：价格跌破下轨            elif self.data.close < self.boll.lines.bot:                stop_price = self.data.close * (1 + self.stop_loss_pct)                self.sell(exectype=bt.Order.Market)                self.buy(exectype=bt.Order.Stop, price=stop_price)        elif position > 0:            # 止盈：价格回落至中轨下方            if self.data.close < self.boll.lines.mid:                self.close()            # 止损触发            elif self.data.close < self.getposition().price * (1 - self.stop_loss_pct):                self.close()        elif position < 0:            # 止盈：价格反弹至中轨上方            if self.data.close > self.boll.lines.mid:                self.close()            # 止损触发            elif self.data.close > self.getposition().price * (1 + self.stop_loss_pct):                self.close()# 布林震荡策略（均值回归）class BollingerBandsReversionStrategy(bt.Strategy):    params = (        ('period', 20),          # 布林带均线周期        ('devfactor', 2.0),      # 标准差倍数        ('rsi_period', 14),      # RSI指标周期        ('stop_loss_pct', 0.03)  # 3%止损比例    )    def __init__(self):        self.boll = bt.indicators.BollingerBands(self.data.close,                                                   period=self.params.period,                                                   devfactor=self.params.devfactor)        self.rsi = bt.indicators.RSI(self.data.close, period=self.params.rsi_period)        self.stop_loss_pct = self.params.stop_loss_pct    def next(self):        if len(self) < max(self.params.period, self.params.rsi_period):            return        position = self.getposition().size        # 仅在震荡行情（ADX<20）中执行策略        if self.adx > 20:            return        if not position:            # 做多信号：价格触及下轨+RSI超卖            if (self.data.close <= self.boll.lines.bot and                 self.rsi < 30):                stop_price = self.data.close * (1 - self.stop_loss_pct)                self.buy(exectype=bt.Order.Market)                self.sell(exectype=bt.Order.Stop, price=stop_price)            # 做空信号：价格触及上轨+RSI超买            elif (self.data.close >= self.boll.lines.top and                   self.rsi > 70):                stop_price = self.data.close * (1 + self.stop_loss_pct)                self.sell(exectype=bt.Order.Market)                self.buy(exectype=bt.Order.Stop, price=stop_price)        elif position > 0:            # 止盈：价格反弹至中轨上方            if self.data.close >= self.boll.lines.mid:                self.close()            # 止损触发            elif self.data.close < self.getposition().price * (1 - self.stop_loss_pct):                self.close()        elif position < 0:            # 止盈：价格回落至中轨下方            if self.data.close <= self.boll.lines.mid:                self.close()            # 止损触发            elif self.data.close > self.getposition().price * (1 + self.stop_loss_pct):                self.close()# 双策略融合主策略class DualStrategySystem(bt.Strategy):    params = (        ('breakout_period', 20),        ('reversion_period', 20),    )    def __init__(self):        # 初始化两个子策略        self.breakout_strategy = BollingerBandsBreakoutStrategy(self.data)        self.reversion_strategy = BollingerBandsReversionStrategy(self.data)        # 初始化ADX指标用于行情判定        self.adx = bt.indicators.ADX(self.data, period=14)    def next(self):        # 根据ADX值切换主策略        if self.adx > 25:            # 趋势行情，执行突破策略            self.breakout_strategy.next()        elif self.adx < 20:            # 震荡行情，执行震荡策略            self.reversion_strategy.next()        else:            # 过渡行情，两个策略同时运行            self.breakout_strategy.next()            self.reversion_strategy.next()# 完整回测流程if __name__ == '__main__':    cerebro = bt.Cerebro()    # 1. 加载数据（示例：从CSV文件读取，需包含open,high,low,close,volume列）    # data = bt.feeds.PandasData(dataname=pd.read_csv('510300.csv', parse_dates=True, index_col=0))    # cerebro.adddata(data)    # 2. 模拟生成测试数据（用于演示）    data = bt.feeds.RandomWalkData()    cerebro.adddata(data)    # 3. 添加双策略系统    cerebro.addstrategy(DualStrategySystem, breakout_period=20, reversion_period=20)    # 4. 设置初始资金    cerebro.broker.setcash(100000.0)    # 5. 设置交易手续费    cerebro.broker.setcommission(commission=0.001)  # 万分之一手续费    # 6. 添加绩效分析器    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='trade_analyzer')    # 7. 运行回测    print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())    results = cerebro.run()    print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue())    # 8. 输出绩效指标    strat = results[0]    print(f'Sharpe Ratio: {strat.analyzers.sharpe.get_analysis()["sharperatio"]:.2f}')    print(f'Maximum Drawdown: {strat.analyzers.drawdown.get_analysis()["max"]["drawdown"]:.2f}%')    print(f'Total Trades: {strat.analyzers.trade_analyzer.get_analysis()["total"]["total"]}')    print(f'Winning Trades: {strat.analyzers.trade_analyzer.get_analysis()["won"]["total"]}')    # 9. 绘制结果    cerebro.plot(iplot=False)