在上一篇中，学习了布林震荡策略的核心逻辑、参数优化方法，并掌握了基础的代码框架。
本篇来学习实战环节，通过Python量化工具，完整实现布林震荡策略与布林突破策略的双策略系统，对比两种策略在不同市场环境下的表现，并通过参数优化构建更稳健的交易体系。
一、双策略系统的核心逻辑
同时运行布林震荡策略和布林突破策略，通过市场状态识别模块自动切换主策略，同时保留两种策略的交易信号，实现“趋势行情抓大行情，震荡行情赚小波动”的组合效果。
1、策略切换规则
- 趋势行情识别：当ADX指标>25时，判定为趋势行情，优先执行布林突破策略
- 震荡行情识别：当ADX指标<20时，判定为震荡行情，优先执行布林震荡策略
- 过渡行情：当20≤ADX≤25时，两种策略同时运行，通过信号过滤器筛选最优交易
2、双策略信号融合
表格
二、完整Python代码实现
以下是基于Backtrader框架的双策略系统完整代码，包含策略切换、信号过滤、绩效分析等完整模块：
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
以下是基于Backtrader框架的布林双策略系统完整可运行代码，包含策略切换、信号过滤、绩效分析等完整模块：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)
代码核心特性说明：
- 策略自动切换：通过ADX指标自动识别市场状态，趋势行情执行布林突破策略，震荡行情执行布林震荡策略，过渡行情双策略并行
- 双重风险控制：两个策略均设置固定比例止损，同时结合仓位成本动态管理止损位
- 完整绩效分析：自动输出夏普率、最大回撤、总交易次数、盈利交易次数等关键指标
- 灵活参数配置：可通过
params参数快速调整布林带周期、止损比例等核心参数
使用说明：
- 数据替换：将代码中的模拟数据替换为实际行情数据（CSV格式，需包含
open,high,low,close,volume列） - 参数优化：可通过
cerebro.optstrategy()对两个子策略的参数分别进行网格搜索，寻找最优组合。
