BB Squeeze 策略
该策略通过监控布林带的收缩与扩张来捕捉波动性突破。当上轨和下轨相对中轨变得非常靠近时,被视为进入“挤压”状态。若之后波动性增加并且价格在挤压结束后收于布林带之外,系统将按照突破方向开仓。
所有交易均使用市价单执行。当价格在挤压后收于上轨之上时开多仓;当价格收于下轨之下时开空仓。策略仅处理已完成的K线,避免在K线形成期间产生提前信号。
算法无需保存完整的历史数据,而是比较当前带宽与前一条的带宽,以确认真实的扩张后再入场。这防止在长时间低波动期间频繁入场但没有实际突破的情况。
默认参数使用20周期、倍数为2的布林带,挤压阈值设为0.05,即带宽需小于中轨的5%才认定为低波动。蜡烛周期和所有数值参数都可以配置,并且在StockSharp环境中支持优化。
using System;
using System.Collections.Generic;
using Ecng.Common;
using StockSharp.Algo.Indicators;
using StockSharp.Algo.Strategies;
using StockSharp.BusinessEntities;
using StockSharp.Messages;
namespace StockSharp.Samples.Strategies;
/// <summary>
/// Bollinger Bands breakout strategy.
/// Enters on breakout above/below bands, exits at middle band.
/// </summary>
public class BbSqueezeStrategy : Strategy
{
private readonly StrategyParam<int> _bollingerPeriod;
private readonly StrategyParam<DataType> _candleType;
private decimal _prevClose;
private decimal _prevUpper;
private decimal _prevLower;
private bool _hasPrev;
public int BollingerPeriod { get => _bollingerPeriod.Value; set => _bollingerPeriod.Value = value; }
public DataType CandleType { get => _candleType.Value; set => _candleType.Value = value; }
public BbSqueezeStrategy()
{
_bollingerPeriod = Param(nameof(BollingerPeriod), 20)
.SetGreaterThanZero()
.SetDisplay("Bollinger Period", "Period of Bollinger Bands", "Parameters");
_candleType = Param(nameof(CandleType), TimeSpan.FromHours(4).TimeFrame())
.SetDisplay("Candle Type", "Type of candles used", "General");
}
public override IEnumerable<(Security sec, DataType dt)> GetWorkingSecurities()
=> [(Security, CandleType)];
protected override void OnReseted()
{
base.OnReseted();
_prevClose = 0;
_prevUpper = 0;
_prevLower = 0;
_hasPrev = false;
}
protected override void OnStarted2(DateTime time)
{
base.OnStarted2(time);
var bb = new BollingerBands { Length = BollingerPeriod, Width = 2m };
SubscribeCandles(CandleType).BindEx(bb, ProcessCandle).Start();
}
private void ProcessCandle(ICandleMessage candle, IIndicatorValue value)
{
if (candle.State != CandleStates.Finished) return;
var bbVal = (BollingerBandsValue)value;
if (bbVal.UpBand is not decimal upper ||
bbVal.LowBand is not decimal lower ||
bbVal.MovingAverage is not decimal middle)
return;
var close = candle.ClosePrice;
if (!_hasPrev)
{
_prevClose = close;
_prevUpper = upper;
_prevLower = lower;
_hasPrev = true;
return;
}
// Cross above upper band => buy
if (_prevClose <= _prevUpper && close > upper && Position <= 0)
{
if (Position < 0) BuyMarket();
BuyMarket();
}
// Cross below lower band => sell
else if (_prevClose >= _prevLower && close < lower && Position >= 0)
{
if (Position > 0) SellMarket();
SellMarket();
}
// Exit long at middle
else if (Position > 0 && close < middle)
{
SellMarket();
}
// Exit short at middle
else if (Position < 0 && close > middle)
{
BuyMarket();
}
_prevClose = close;
_prevUpper = upper;
_prevLower = lower;
}
}
import clr
clr.AddReference("StockSharp.Messages")
clr.AddReference("StockSharp.Algo")
clr.AddReference("StockSharp.Algo.Indicators")
clr.AddReference("StockSharp.Algo.Strategies")
from System import TimeSpan
from StockSharp.Messages import DataType, CandleStates
from StockSharp.Algo.Indicators import BollingerBands
from StockSharp.Algo.Strategies import Strategy
class bb_squeeze_strategy(Strategy):
def __init__(self):
super(bb_squeeze_strategy, self).__init__()
self._bollinger_period = self.Param("BollingerPeriod", 20) \
.SetDisplay("Bollinger Period", "Period of Bollinger Bands", "Parameters")
self._candle_type = self.Param("CandleType", DataType.TimeFrame(TimeSpan.FromHours(4))) \
.SetDisplay("Candle Type", "Type of candles used", "General")
self._prev_close = 0.0
self._prev_upper = 0.0
self._prev_lower = 0.0
self._has_prev = False
@property
def bollinger_period(self):
return self._bollinger_period.Value
@property
def candle_type(self):
return self._candle_type.Value
def OnReseted(self):
super(bb_squeeze_strategy, self).OnReseted()
self._prev_close = 0.0
self._prev_upper = 0.0
self._prev_lower = 0.0
self._has_prev = False
def OnStarted2(self, time):
super(bb_squeeze_strategy, self).OnStarted2(time)
bb = BollingerBands()
bb.Length = self.bollinger_period
bb.Width = 2.0
subscription = self.SubscribeCandles(self.candle_type)
subscription.BindEx(bb, self.on_process).Start()
area = self.CreateChartArea()
if area is not None:
self.DrawCandles(area, subscription)
self.DrawIndicator(area, bb)
self.DrawOwnTrades(area)
def on_process(self, candle, value):
if candle.State != CandleStates.Finished:
return
upper = float(value.UpBand)
lower = float(value.LowBand)
middle = float(value.MovingAverage)
if upper == 0 or lower == 0:
return
close = float(candle.ClosePrice)
if not self._has_prev:
self._prev_close = close
self._prev_upper = upper
self._prev_lower = lower
self._has_prev = True
return
# Cross above upper band => buy
if self._prev_close <= self._prev_upper and close > upper and self.Position <= 0:
if self.Position < 0:
self.BuyMarket()
self.BuyMarket()
# Cross below lower band => sell
elif self._prev_close >= self._prev_lower and close < lower and self.Position >= 0:
if self.Position > 0:
self.SellMarket()
self.SellMarket()
# Exit long at middle
elif self.Position > 0 and close < middle:
self.SellMarket()
# Exit short at middle
elif self.Position < 0 and close > middle:
self.BuyMarket()
self._prev_close = close
self._prev_upper = upper
self._prev_lower = lower
def CreateClone(self):
return bb_squeeze_strategy()