Стратегия Squeeze Pro Overlays
Стратегия Squeeze Pro Overlays отслеживает сжатие волатильности, когда полосы Боллинджера полностью находятся внутри нескольких каналов Кельтнера. После выхода из сжатия направление сделки определяется наклоном линейной регрессии закрытий.
Подробности
- Условия входа:
- Сжатие завершилось (полосы Боллинджера выходят за самый широкий канал Кельтнера).
- Лонг: наклон импульса > 0.
- Шорт: наклон импульса < 0.
- Направление: обе стороны.
- Условия выхода:
- Противоположный сигнал.
- Стопы: отсутствуют.
- Параметры по умолчанию:
SqueezeLength= 20
- Фильтры:
- Категория: Волатильностный пробой
- Направление: Обе
- Индикаторы: Полосы Боллинджера, каналы Кельтнера, линейная регрессия
- Стопы: Нет
- Сложность: Низкая
- Таймфрейм: Любой
- Сезонность: Нет
- Нейросети: Нет
- Дивергенция: Нет
- Уровень риска: Средний
namespace StockSharp.Samples.Strategies;
using System;
using System.Collections.Generic;
using Ecng.Common;
using StockSharp.Algo.Indicators;
using StockSharp.Algo.Strategies;
using StockSharp.BusinessEntities;
using StockSharp.Messages;
/// <summary>
/// Squeeze Pro Overlays Strategy.
/// Detects when BB is inside KC (squeeze), then trades on breakout direction.
/// Uses momentum (LinearRegSlope) to determine direction.
/// </summary>
public class SqueezeProOverlaysStrategy : Strategy
{
private readonly StrategyParam<DataType> _candleType;
private readonly StrategyParam<int> _squeezeLength;
private readonly StrategyParam<int> _cooldownBars;
private BollingerBands _bb;
private KeltnerChannels _kc;
private LinearRegSlope _slope;
private bool _wasSqueezed;
private int _cooldownRemaining;
public DataType CandleType
{
get => _candleType.Value;
set => _candleType.Value = value;
}
public int SqueezeLength
{
get => _squeezeLength.Value;
set => _squeezeLength.Value = value;
}
public int CooldownBars
{
get => _cooldownBars.Value;
set => _cooldownBars.Value = value;
}
public SqueezeProOverlaysStrategy()
{
_candleType = Param(nameof(CandleType), TimeSpan.FromMinutes(30).TimeFrame())
.SetDisplay("Candle Type", "Type of candles to use", "General");
_squeezeLength = Param(nameof(SqueezeLength), 20)
.SetGreaterThanZero()
.SetDisplay("Squeeze Length", "Calculation length", "Squeeze");
_cooldownBars = Param(nameof(CooldownBars), 10)
.SetDisplay("Cooldown Bars", "Bars to wait between trades", "Risk");
}
/// <inheritdoc />
public override IEnumerable<(Security sec, DataType dt)> GetWorkingSecurities()
=> [(Security, CandleType)];
/// <inheritdoc />
protected override void OnReseted()
{
base.OnReseted();
_bb = null;
_kc = null;
_slope = null;
_wasSqueezed = false;
_cooldownRemaining = 0;
}
/// <inheritdoc />
protected override void OnStarted2(DateTime time)
{
base.OnStarted2(time);
_bb = new BollingerBands { Length = SqueezeLength, Width = 2m };
_kc = new KeltnerChannels { Length = SqueezeLength, Multiplier = 1.5m };
_slope = new LinearRegSlope { Length = SqueezeLength };
var subscription = SubscribeCandles(CandleType);
subscription
.BindEx(_bb, _kc, _slope, OnProcess)
.Start();
var area = CreateChartArea();
if (area != null)
{
DrawCandles(area, subscription);
DrawIndicator(area, _bb);
DrawOwnTrades(area);
}
}
private void OnProcess(ICandleMessage candle, IIndicatorValue bbValue, IIndicatorValue kcValue, IIndicatorValue slopeValue)
{
if (candle.State != CandleStates.Finished)
return;
if (!_bb.IsFormed || !_kc.IsFormed || !_slope.IsFormed)
return;
if (bbValue.IsEmpty || kcValue.IsEmpty || slopeValue.IsEmpty)
return;
var bb = (BollingerBandsValue)bbValue;
var kc = (KeltnerChannelsValue)kcValue;
if (bb.UpBand is not decimal bbUpper || bb.LowBand is not decimal bbLower)
return;
if (kc.Upper is not decimal kcUpper || kc.Lower is not decimal kcLower)
return;
if (!IsFormedAndOnlineAndAllowTrading())
return;
if (_cooldownRemaining > 0)
{
_cooldownRemaining--;
return;
}
var slopeVal = slopeValue.ToDecimal();
var squeezed = bbUpper < kcUpper && bbLower > kcLower;
// Squeeze release: was squeezed, now not
if (_wasSqueezed && !squeezed)
{
if (slopeVal > 0 && Position <= 0)
{
if (Position < 0)
BuyMarket(Math.Abs(Position));
BuyMarket(Volume);
_cooldownRemaining = CooldownBars;
}
else if (slopeVal < 0 && Position >= 0)
{
if (Position > 0)
SellMarket(Math.Abs(Position));
SellMarket(Volume);
_cooldownRemaining = CooldownBars;
}
}
// Exit: slope reverses
else if (Position > 0 && slopeVal < 0)
{
SellMarket(Math.Abs(Position));
_cooldownRemaining = CooldownBars;
}
else if (Position < 0 && slopeVal > 0)
{
BuyMarket(Math.Abs(Position));
_cooldownRemaining = CooldownBars;
}
_wasSqueezed = squeezed;
}
}
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, CandleStates
from StockSharp.Algo.Indicators import BollingerBands, KeltnerChannels, LinearRegSlope, IndicatorHelper
from StockSharp.Algo.Strategies import Strategy
class squeeze_pro_overlays_strategy(Strategy):
"""Squeeze Pro Overlays Strategy."""
def __init__(self):
super(squeeze_pro_overlays_strategy, self).__init__()
self._candle_type = self.Param("CandleType", DataType.TimeFrame(TimeSpan.FromMinutes(30))) \
.SetDisplay("Candle Type", "Type of candles to use", "General")
self._squeeze_length = self.Param("SqueezeLength", 20) \
.SetDisplay("Squeeze Length", "Calculation length", "Squeeze")
self._cooldown_bars = self.Param("CooldownBars", 10) \
.SetDisplay("Cooldown Bars", "Bars to wait between trades", "Risk")
self._bb = None
self._kc = None
self._slope = None
self._was_squeezed = False
self._cooldown_remaining = 0
@property
def candle_type(self):
return self._candle_type.Value
def OnReseted(self):
super(squeeze_pro_overlays_strategy, self).OnReseted()
self._bb = None
self._kc = None
self._slope = None
self._was_squeezed = False
self._cooldown_remaining = 0
def OnStarted2(self, time):
super(squeeze_pro_overlays_strategy, self).OnStarted2(time)
sq_len = int(self._squeeze_length.Value)
self._bb = BollingerBands()
self._bb.Length = sq_len
self._bb.Width = 2.0
self._kc = KeltnerChannels()
self._kc.Length = sq_len
self._kc.Multiplier = 1.5
self._slope = LinearRegSlope()
self._slope.Length = sq_len
subscription = self.SubscribeCandles(self.candle_type)
subscription.BindEx(self._bb, self._kc, self._slope, self._on_process).Start()
area = self.CreateChartArea()
if area is not None:
self.DrawCandles(area, subscription)
self.DrawIndicator(area, self._bb)
self.DrawOwnTrades(area)
def _on_process(self, candle, bb_value, kc_value, slope_value):
if candle.State != CandleStates.Finished:
return
if not self._bb.IsFormed or not self._kc.IsFormed or not self._slope.IsFormed:
return
if bb_value.IsEmpty or kc_value.IsEmpty or slope_value.IsEmpty:
return
if bb_value.UpBand is None or bb_value.LowBand is None:
return
if kc_value.Upper is None or kc_value.Lower is None:
return
bb_upper = float(bb_value.UpBand)
bb_lower = float(bb_value.LowBand)
kc_upper = float(kc_value.Upper)
kc_lower = float(kc_value.Lower)
if not self.IsFormedAndOnlineAndAllowTrading():
return
if self._cooldown_remaining > 0:
self._cooldown_remaining -= 1
return
slope_val = float(IndicatorHelper.ToDecimal(slope_value))
squeezed = bb_upper < kc_upper and bb_lower > kc_lower
cooldown = int(self._cooldown_bars.Value)
if self._was_squeezed and not squeezed:
if slope_val > 0 and self.Position <= 0:
if self.Position < 0:
self.BuyMarket(Math.Abs(self.Position))
self.BuyMarket(self.Volume)
self._cooldown_remaining = cooldown
elif slope_val < 0 and self.Position >= 0:
if self.Position > 0:
self.SellMarket(Math.Abs(self.Position))
self.SellMarket(self.Volume)
self._cooldown_remaining = cooldown
elif self.Position > 0 and slope_val < 0:
self.SellMarket(Math.Abs(self.Position))
self._cooldown_remaining = cooldown
elif self.Position < 0 and slope_val > 0:
self.BuyMarket(Math.Abs(self.Position))
self._cooldown_remaining = cooldown
self._was_squeezed = squeezed
def CreateClone(self):
return squeeze_pro_overlays_strategy()