Bollinger Band Squeeze Strategy
该策略监控布林带宽度以发现低波动时期,当带宽相对于近期平均收窄时,预示即将出现波动扩张。
测试表明年均收益约为 100%,该策略在外汇市场表现最佳。
识别到“Squeeze”后,策略等待价格突破带外:收盘价上破上轨则做多,下破下轨则做空。若价格回到带内或触及止损,则平仓。
此方法适合交易波动性突破的投资者,通过带宽过滤,可避免震荡期的虚假信号。
细节
- 入场条件:
- 多头: 带宽收窄后,收盘价上破上轨
- 空头: 带宽收窄后,收盘价下破下轨
- 多/空: 双向
- 离场条件:
- 多头: 价格跌回带内
- 空头: 价格升回带内
- 止损: 通常为2倍ATR
- 默认值:
BollingerPeriod= 20BollingerMultiplier= 2.0mLookbackPeriod= 20CandleType= TimeSpan.FromMinutes(5)
- 过滤器:
- 类别: Breakout
- 方向: 双向
- 指标: Bollinger Bands
- 止损: 是
- 复杂度: 中等
- 时间框架: 日内
- 季节性: 否
- 神经网络: 否
- 背离: 否
- 风险等级: 中等
using System;
using System.Linq;
using System.Collections.Generic;
using Ecng.Common;
using Ecng.Collections;
using Ecng.Serialization;
using StockSharp.Algo.Indicators;
using StockSharp.Algo.Strategies;
using StockSharp.BusinessEntities;
using StockSharp.Messages;
namespace StockSharp.Samples.Strategies;
/// <summary>
/// Bollinger Band Squeeze strategy.
/// Trades when volatility decreases (bands squeeze) followed by a breakout.
/// </summary>
public class BollingerBandSqueezeStrategy : Strategy
{
private readonly StrategyParam<int> _bollingerPeriodParam;
private readonly StrategyParam<decimal> _bollingerMultiplierParam;
private readonly StrategyParam<int> _lookbackPeriodParam;
private readonly StrategyParam<DataType> _candleTypeParam;
private BollingerBands _bollinger;
private AverageTrueRange _atr;
private decimal _prevBollingerWidth;
private decimal _avgBollingerWidth;
private decimal _bollingerWidthSum;
private readonly Queue<decimal> _bollingerWidths = [];
/// <summary>
/// Bollinger Bands period.
/// </summary>
public int BollingerPeriod
{
get => _bollingerPeriodParam.Value;
set => _bollingerPeriodParam.Value = value;
}
/// <summary>
/// Bollinger Bands multiplier.
/// </summary>
public decimal BollingerMultiplier
{
get => _bollingerMultiplierParam.Value;
set => _bollingerMultiplierParam.Value = value;
}
/// <summary>
/// Period for averaging Bollinger width.
/// </summary>
public int LookbackPeriod
{
get => _lookbackPeriodParam.Value;
set => _lookbackPeriodParam.Value = value;
}
/// <summary>
/// Candle type for strategy.
/// </summary>
public DataType CandleType
{
get => _candleTypeParam.Value;
set => _candleTypeParam.Value = value;
}
/// <summary>
/// Constructor.
/// </summary>
public BollingerBandSqueezeStrategy()
{
_bollingerPeriodParam = Param(nameof(BollingerPeriod), 20)
.SetGreaterThanZero()
.SetDisplay("Bollinger Period", "Period for Bollinger Bands calculation", "Parameters")
.SetOptimize(10, 30, 5);
_bollingerMultiplierParam = Param(nameof(BollingerMultiplier), 2.0m)
.SetRange(0.1m, decimal.MaxValue)
.SetDisplay("Bollinger Multiplier", "Standard deviation multiplier for Bollinger Bands", "Parameters")
.SetOptimize(1.5m, 3.0m, 0.5m);
_lookbackPeriodParam = Param(nameof(LookbackPeriod), 20)
.SetGreaterThanZero()
.SetDisplay("Lookback Period", "Period for averaging Bollinger width", "Parameters")
.SetOptimize(10, 30, 5);
_candleTypeParam = Param(nameof(CandleType), TimeSpan.FromMinutes(5).TimeFrame())
.SetDisplay("Candle Type", "Candle type for strategy", "Common");
}
/// <inheritdoc />
public override IEnumerable<(Security sec, DataType dt)> GetWorkingSecurities()
{
return [(Security, CandleType)];
}
/// <inheritdoc />
protected override void OnReseted()
{
base.OnReseted();
_bollinger = null;
_atr = null;
_prevBollingerWidth = 0;
_avgBollingerWidth = 0;
_bollingerWidthSum = 0;
_bollingerWidths.Clear();
}
/// <inheritdoc />
protected override void OnStarted2(DateTime time)
{
base.OnStarted2(time);
// Initialize indicator
_bollinger = new BollingerBands
{
Length = BollingerPeriod,
Width = BollingerMultiplier
};
_atr = new AverageTrueRange { Length = BollingerPeriod };
// Create subscription and bind indicator
var subscription = SubscribeCandles(CandleType);
subscription
.BindEx(_bollinger, ProcessCandle)
.Start();
// Setup chart visualization if available
var area = CreateChartArea();
if (area != null)
{
DrawCandles(area, subscription);
DrawIndicator(area, _bollinger);
DrawOwnTrades(area);
}
// Enable position protection
StartProtection(
takeProfit: new Unit(0, UnitTypes.Absolute), // No take profit
stopLoss: new Unit(2, UnitTypes.Absolute) // Stop loss at 2*ATR
);
}
private void ProcessCandle(ICandleMessage candle, IIndicatorValue bollingerValue)
{
if (candle.State != CandleStates.Finished)
return;
if (!IsFormedAndOnlineAndAllowTrading())
return;
var bollingerTyped = (BollingerBandsValue)bollingerValue;
if (bollingerTyped.UpBand is not decimal upperBand)
return;
if (bollingerTyped.LowBand is not decimal lowerBand)
return;
// Calculate Bollinger width (upper - lower)
var bollingerWidth = upperBand - lowerBand;
// Track average Bollinger width over lookback period
_bollingerWidths.Enqueue(bollingerWidth);
_bollingerWidthSum += bollingerWidth;
if (_bollingerWidths.Count > LookbackPeriod)
{
var oldValue = _bollingerWidths.Dequeue();
_bollingerWidthSum -= oldValue;
}
if (_bollingerWidths.Count == LookbackPeriod)
{
_avgBollingerWidth = _bollingerWidthSum / LookbackPeriod;
// Detect Bollinger Band squeeze (narrowing bands)
bool isSqueeze = bollingerWidth < _avgBollingerWidth;
// Breakout after squeeze
if (isSqueeze)
{
// Upside breakout
if (candle.ClosePrice > upperBand && Position <= 0)
{
BuyMarket(Volume + Math.Abs(Position));
}
// Downside breakout
else if (candle.ClosePrice < lowerBand && Position >= 0)
{
SellMarket(Volume + Math.Abs(Position));
}
}
}
_prevBollingerWidth = bollingerWidth;
}
}
import clr
clr.AddReference("StockSharp.Messages")
clr.AddReference("StockSharp.Algo")
clr.AddReference("StockSharp.Algo.Indicators")
clr.AddReference("StockSharp.Algo.Strategies")
from System import TimeSpan, Math
from StockSharp.Messages import DataType, UnitTypes, Unit, CandleStates
from StockSharp.Algo.Indicators import BollingerBands, AverageTrueRange
from StockSharp.Algo.Strategies import Strategy
class bollinger_band_squeeze_strategy(Strategy):
def __init__(self):
super(bollinger_band_squeeze_strategy, self).__init__()
self._bollinger_period = self.Param("BollingerPeriod", 20) \
.SetDisplay("Bollinger Period", "Period for Bollinger Bands calculation", "Parameters")
self._bollinger_multiplier = self.Param("BollingerMultiplier", 2.0) \
.SetDisplay("Bollinger Multiplier", "Standard deviation multiplier for Bollinger Bands", "Parameters")
self._lookback_period = self.Param("LookbackPeriod", 20) \
.SetDisplay("Lookback Period", "Period for averaging Bollinger width", "Parameters")
self._candle_type = self.Param("CandleType", DataType.TimeFrame(TimeSpan.FromMinutes(5))) \
.SetDisplay("Candle Type", "Candle type for strategy", "Common")
self._prev_bollinger_width = 0.0
self._avg_bollinger_width = 0.0
self._bollinger_width_sum = 0.0
self._bollinger_widths = []
@property
def bollinger_period(self):
return self._bollinger_period.Value
@property
def bollinger_multiplier(self):
return self._bollinger_multiplier.Value
@property
def lookback_period(self):
return self._lookback_period.Value
@property
def candle_type(self):
return self._candle_type.Value
def OnReseted(self):
super(bollinger_band_squeeze_strategy, self).OnReseted()
self._prev_bollinger_width = 0.0
self._avg_bollinger_width = 0.0
self._bollinger_width_sum = 0.0
self._bollinger_widths = []
def OnStarted2(self, time):
super(bollinger_band_squeeze_strategy, self).OnStarted2(time)
bollinger = BollingerBands()
bollinger.Length = self.bollinger_period
bollinger.Width = self.bollinger_multiplier
subscription = self.SubscribeCandles(self.candle_type)
subscription.BindEx(bollinger, self.OnProcess).Start()
area = self.CreateChartArea()
if area is not None:
self.DrawCandles(area, subscription)
self.DrawIndicator(area, bollinger)
self.DrawOwnTrades(area)
self.StartProtection(
takeProfit=Unit(0, UnitTypes.Absolute),
stopLoss=Unit(2, UnitTypes.Absolute)
)
def OnProcess(self, candle, bollinger_value):
if candle.State != CandleStates.Finished:
return
if bollinger_value.UpBand is None:
return
upper_band = float(bollinger_value.UpBand)
if bollinger_value.LowBand is None:
return
lower_band = float(bollinger_value.LowBand)
bollinger_width = upper_band - lower_band
self._bollinger_widths.append(bollinger_width)
self._bollinger_width_sum += bollinger_width
if len(self._bollinger_widths) > self.lookback_period:
old_value = self._bollinger_widths.pop(0)
self._bollinger_width_sum -= old_value
if len(self._bollinger_widths) == self.lookback_period:
self._avg_bollinger_width = self._bollinger_width_sum / self.lookback_period
is_squeeze = bollinger_width < self._avg_bollinger_width
if is_squeeze:
if candle.ClosePrice > upper_band and self.Position <= 0:
self.BuyMarket()
elif candle.ClosePrice < lower_band and self.Position >= 0:
self.SellMarket()
self._prev_bollinger_width = bollinger_width
def CreateClone(self):
return bollinger_band_squeeze_strategy()