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 Bands breakout strategy with optional balance-based position sizing.
/// </summary>
public class BollTradeStrategy : Strategy
{
private readonly StrategyParam<decimal> _maxVolume;
private readonly StrategyParam<decimal> _takeProfit;
private readonly StrategyParam<decimal> _stopLoss;
private readonly StrategyParam<decimal> _bandOffset;
private readonly StrategyParam<int> _bollingerPeriod;
private readonly StrategyParam<decimal> _bollingerDeviation;
private readonly StrategyParam<decimal> _lots;
private readonly StrategyParam<bool> _lotIncrease;
private readonly StrategyParam<DataType> _candleType;
private decimal _lotBaseline;
private decimal _pipSize;
private BollingerBands _bollinger;
private decimal? _longStop;
private decimal? _longTarget;
private decimal? _shortStop;
private decimal? _shortTarget;
public decimal TakeProfit
{
get => _takeProfit.Value;
set => _takeProfit.Value = value;
}
public decimal StopLoss
{
get => _stopLoss.Value;
set => _stopLoss.Value = value;
}
public decimal BollingerDistance
{
get => _bandOffset.Value;
set => _bandOffset.Value = value;
}
public int BollingerPeriod
{
get => _bollingerPeriod.Value;
set => _bollingerPeriod.Value = value;
}
public decimal BollingerDeviation
{
get => _bollingerDeviation.Value;
set => _bollingerDeviation.Value = value;
}
public decimal Lots
{
get => _lots.Value;
set => _lots.Value = value;
}
public bool LotIncrease
{
get => _lotIncrease.Value;
set => _lotIncrease.Value = value;
}
public DataType CandleType
{
get => _candleType.Value;
set => _candleType.Value = value;
}
public decimal MaxVolume
{
get => _maxVolume.Value;
set => _maxVolume.Value = value;
}
public BollTradeStrategy()
{
_maxVolume = Param(nameof(MaxVolume), 500m)
.SetGreaterThanZero()
.SetDisplay("Max Volume", "Upper bound for scaled volume", "Money Management");
_takeProfit = Param(nameof(TakeProfit), 3m)
.SetNotNegative()
.SetDisplay("Take Profit (pips)", "Distance to take profit expressed in pip units.", "Orders")
;
_stopLoss = Param(nameof(StopLoss), 20m)
.SetNotNegative()
.SetDisplay("Stop Loss (pips)", "Distance to stop loss expressed in pip units.", "Orders")
;
_bandOffset = Param(nameof(BollingerDistance), 0m)
.SetNotNegative()
.SetDisplay("Band Offset", "Extra pip offset beyond Bollinger Bands.", "Signals")
;
_bollingerPeriod = Param(nameof(BollingerPeriod), 20)
.SetGreaterThanZero()
.SetDisplay("Bollinger Period", "Length of the Bollinger Bands.", "Signals")
;
_bollingerDeviation = Param(nameof(BollingerDeviation), 1m)
.SetGreaterThanZero()
.SetDisplay("Bollinger Deviation", "Width multiplier of the Bollinger Bands.", "Signals")
;
_lots = Param(nameof(Lots), 1m)
.SetGreaterThanZero()
.SetDisplay("Base Volume", "Default trade volume in lots.", "Money Management");
_lotIncrease = Param(nameof(LotIncrease), true)
.SetDisplay("Scale Volume", "Increase volume with balance growth.", "Money Management");
_candleType = Param(nameof(CandleType), TimeSpan.FromHours(1).TimeFrame())
.SetDisplay("Candle Type", "Primary timeframe for signals.", "General");
}
public override IEnumerable<(Security sec, DataType dt)> GetWorkingSecurities()
{
yield return (Security, CandleType);
}
/// <inheritdoc />
protected override void OnReseted()
{
base.OnReseted();
_lotBaseline = 0m;
_pipSize = 0m;
_bollinger = null!;
ResetStops();
}
/// <inheritdoc />
protected override void OnStarted2(DateTime time)
{
base.OnStarted2(time);
Volume = Lots;
_pipSize = CalculatePipSize();
_lotBaseline = 0m;
if (LotIncrease && Lots > 0m)
{
var balance = Portfolio?.CurrentValue ?? 0m;
if (balance > 0m)
_lotBaseline = balance / Lots;
}
var bollinger = new BollingerBands
{
Length = BollingerPeriod,
Width = BollingerDeviation
};
var subscription = SubscribeCandles(CandleType);
subscription
.BindEx(bollinger, ProcessCandle)
.Start();
_bollinger = bollinger;
}
private decimal CalculatePipSize()
{
var step = Security?.PriceStep ?? 1m;
if (step <= 0m)
step = 1m;
if (step < 0.01m)
step *= 10m;
return step;
}
private decimal CalculateVolume()
{
var baseVolume = Lots;
if (!LotIncrease || _lotBaseline <= 0m)
return baseVolume;
var balance = Portfolio?.CurrentValue ?? 0m;
if (balance <= 0m)
return baseVolume;
var scaled = baseVolume * (balance / _lotBaseline);
return Math.Min(scaled, MaxVolume);
}
private void ProcessCandle(ICandleMessage candle, IIndicatorValue value)
{
if (candle.State != CandleStates.Finished)
return;
if (value is not BollingerBandsValue bollingerValue)
return;
if (bollingerValue.MovingAverage is not decimal middleBand ||
bollingerValue.UpBand is not decimal upperBand ||
bollingerValue.LowBand is not decimal lowerBand)
return;
var offset = _pipSize * BollingerDistance;
var upperThreshold = upperBand + offset;
var lowerThreshold = lowerBand - offset;
var shouldBuy = candle.ClosePrice < lowerThreshold;
var shouldSell = candle.ClosePrice > upperThreshold;
if (Position == 0)
{
if (shouldBuy)
{
EnterLong(candle);
}
else if (shouldSell)
{
EnterShort(candle);
}
return;
}
if (Position > 0)
{
// Close long positions when stop loss or take profit levels are hit.
if ((_longStop.HasValue && candle.LowPrice <= _longStop.Value) ||
(_longTarget.HasValue && candle.HighPrice >= _longTarget.Value))
{
SellMarket();
ResetStops();
}
}
else if (Position < 0)
{
// Close short positions when stop loss or take profit levels are hit.
if ((_shortStop.HasValue && candle.HighPrice >= _shortStop.Value) ||
(_shortTarget.HasValue && candle.LowPrice <= _shortTarget.Value))
{
BuyMarket();
ResetStops();
}
}
}
private void EnterLong(ICandleMessage candle)
{
var volume = CalculateVolume();
if (volume <= 0m)
return;
BuyMarket(volume);
// Store exit targets for the newly opened long trade.
_longStop = StopLoss > 0m ? candle.ClosePrice - _pipSize * StopLoss : null;
_longTarget = TakeProfit > 0m ? candle.ClosePrice + _pipSize * TakeProfit : null;
_shortStop = null;
_shortTarget = null;
}
private void EnterShort(ICandleMessage candle)
{
var volume = CalculateVolume();
if (volume <= 0m)
return;
SellMarket(volume);
// Store exit targets for the newly opened short trade.
_shortStop = StopLoss > 0m ? candle.ClosePrice + _pipSize * StopLoss : null;
_shortTarget = TakeProfit > 0m ? candle.ClosePrice - _pipSize * TakeProfit : null;
_longStop = null;
_longTarget = null;
}
private void ResetStops()
{
// Clear cached exit levels after a position is closed.
_longStop = null;
_longTarget = null;
_shortStop = null;
_shortTarget = null;
}
}
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, Unit, UnitTypes
from StockSharp.Algo.Indicators import BollingerBands
from StockSharp.Algo.Strategies import Strategy
class boll_trade_strategy(Strategy):
def __init__(self):
super(boll_trade_strategy, self).__init__()
self._take_profit = self.Param("TakeProfit", 3.0)
self._stop_loss = self.Param("StopLoss", 20.0)
self._band_offset = self.Param("BollingerDistance", 0.0)
self._bollinger_period = self.Param("BollingerPeriod", 20)
self._bollinger_deviation = self.Param("BollingerDeviation", 1.0)
self._lots = self.Param("Lots", 1.0)
self._lot_increase = self.Param("LotIncrease", True)
self._max_volume = self.Param("MaxVolume", 500.0)
self._candle_type = self.Param("CandleType", DataType.TimeFrame(TimeSpan.FromHours(1)))
self._pip_size = 1.0
self._long_stop = None
self._long_target = None
self._short_stop = None
self._short_target = None
@property
def TakeProfit(self):
return self._take_profit.Value
@TakeProfit.setter
def TakeProfit(self, value):
self._take_profit.Value = value
@property
def StopLoss(self):
return self._stop_loss.Value
@StopLoss.setter
def StopLoss(self, value):
self._stop_loss.Value = value
@property
def BollingerDistance(self):
return self._band_offset.Value
@BollingerDistance.setter
def BollingerDistance(self, value):
self._band_offset.Value = value
@property
def BollingerPeriod(self):
return self._bollinger_period.Value
@BollingerPeriod.setter
def BollingerPeriod(self, value):
self._bollinger_period.Value = value
@property
def BollingerDeviation(self):
return self._bollinger_deviation.Value
@BollingerDeviation.setter
def BollingerDeviation(self, value):
self._bollinger_deviation.Value = value
@property
def Lots(self):
return self._lots.Value
@Lots.setter
def Lots(self, value):
self._lots.Value = value
@property
def LotIncrease(self):
return self._lot_increase.Value
@LotIncrease.setter
def LotIncrease(self, value):
self._lot_increase.Value = value
@property
def MaxVolume(self):
return self._max_volume.Value
@MaxVolume.setter
def MaxVolume(self, value):
self._max_volume.Value = value
@property
def CandleType(self):
return self._candle_type.Value
@CandleType.setter
def CandleType(self, value):
self._candle_type.Value = value
def OnStarted2(self, time):
super(boll_trade_strategy, self).OnStarted2(time)
self.Volume = self._lots.Value
self._pip_size = self._calculate_pip_size()
self._lot_baseline = 0.0
self._long_stop = None
self._long_target = None
self._short_stop = None
self._short_target = None
if self.LotIncrease and float(self.Lots) > 0.0:
portfolio = self.Portfolio
balance = float(portfolio.CurrentValue) if portfolio is not None and portfolio.CurrentValue is not None else 0.0
if balance > 0.0:
self._lot_baseline = balance / float(self.Lots)
bollinger = BollingerBands()
bollinger.Length = self.BollingerPeriod
bollinger.Width = self.BollingerDeviation
subscription = self.SubscribeCandles(self.CandleType)
subscription.BindEx(bollinger, self.ProcessCandle).Start()
def _calculate_pip_size(self):
step = float(self.Security.PriceStep) if self.Security is not None and self.Security.PriceStep is not None else 1.0
if step <= 0.0:
step = 1.0
if step < 0.01:
step *= 10.0
return step
def ProcessCandle(self, candle, value):
if candle.State != CandleStates.Finished:
return
upper_band = value.UpBand
lower_band = value.LowBand
if upper_band is None or lower_band is None:
return
upper = float(upper_band)
lower = float(lower_band)
close = float(candle.ClosePrice)
high = float(candle.HighPrice)
low = float(candle.LowPrice)
offset = self._pip_size * float(self.BollingerDistance)
upper_threshold = upper + offset
lower_threshold = lower - offset
should_buy = close < lower_threshold
should_sell = close > upper_threshold
if self.Position == 0:
if should_buy:
self._enter_long(close)
elif should_sell:
self._enter_short(close)
return
if self.Position > 0:
if (self._long_stop is not None and low <= self._long_stop) or \
(self._long_target is not None and high >= self._long_target):
self.SellMarket()
self._reset_stops()
elif self.Position < 0:
if (self._short_stop is not None and high >= self._short_stop) or \
(self._short_target is not None and low <= self._short_target):
self.BuyMarket()
self._reset_stops()
def _calculate_volume(self):
base_volume = float(self.Lots)
if not self.LotIncrease or self._lot_baseline <= 0.0:
return base_volume
portfolio = self.Portfolio
balance = float(portfolio.CurrentValue) if portfolio is not None and portfolio.CurrentValue is not None else 0.0
if balance <= 0.0:
return base_volume
scaled = base_volume * (balance / self._lot_baseline)
return min(scaled, float(self.MaxVolume))
def _enter_long(self, close):
volume = self._calculate_volume()
if volume <= 0.0:
return
self.BuyMarket(volume)
sl = float(self.StopLoss)
tp = float(self.TakeProfit)
self._long_stop = close - self._pip_size * sl if sl > 0.0 else None
self._long_target = close + self._pip_size * tp if tp > 0.0 else None
self._short_stop = None
self._short_target = None
def _enter_short(self, close):
volume = self._calculate_volume()
if volume <= 0.0:
return
self.SellMarket(volume)
sl = float(self.StopLoss)
tp = float(self.TakeProfit)
self._short_stop = close + self._pip_size * sl if sl > 0.0 else None
self._short_target = close - self._pip_size * tp if tp > 0.0 else None
self._long_stop = None
self._long_target = None
def _reset_stops(self):
self._long_stop = None
self._long_target = None
self._short_stop = None
self._short_target = None
def OnReseted(self):
super(boll_trade_strategy, self).OnReseted()
self._pip_size = 1.0
self._reset_stops()
def CreateClone(self):
return boll_trade_strategy()