QQE Signals Strategy
Implements the Quantitative Qualitative Estimation technique on RSI. The indicator builds dynamic upper and lower bands around a smoothed RSI line and tracks band crosses to signal trend changes. When RSI crosses above the trailing band a long signal is generated; crosses below trigger exits.
By adapting the bands to volatility, QQE seeks to smooth noise while remaining responsive. The strategy focuses on long trades and relies on the engine's trade reversals to close positions.
Details
- Entry Criteria:
- Long: RSI smoothed line crosses above the trailing band.
- Exit Criteria:
- RSI falls below the opposite band or an opposite signal appears.
- Indicators:
- RSI (period 14, smoothing 5)
- QQE bands derived from ATR of RSI with factor 4.238
- Stops: None by default; relies on opposite signals.
- Default Values:
RsiPeriod= 14RsiSmoothing= 5QqeFactor= 4.238Threshold= 10
- Filters:
- Trend-following
- Single timeframe
- Indicators: RSI, QQE
- Stops: None
- Complexity: Moderate
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>
/// QQE Signals Strategy.
/// Uses RSI with threshold crossover for trade signals.
/// Buys when RSI crosses above upper threshold.
/// Sells when RSI crosses below lower threshold.
/// Exits at the 50 midline crossover.
/// </summary>
public class QqeSignalsStrategy : Strategy
{
private readonly StrategyParam<DataType> _candleTypeParam;
private readonly StrategyParam<int> _rsiPeriod;
private readonly StrategyParam<decimal> _upperThreshold;
private readonly StrategyParam<decimal> _lowerThreshold;
private readonly StrategyParam<int> _cooldownBars;
private RelativeStrengthIndex _rsi;
private decimal _prevRsi;
private int _cooldownRemaining;
public QqeSignalsStrategy()
{
_candleTypeParam = Param(nameof(CandleType), TimeSpan.FromMinutes(30).TimeFrame())
.SetDisplay("Candle type", "Candle type for strategy calculation.", "General");
_rsiPeriod = Param(nameof(RsiPeriod), 14)
.SetGreaterThanZero()
.SetDisplay("RSI Length", "RSI period", "QQE");
_upperThreshold = Param(nameof(UpperThreshold), 60m)
.SetDisplay("Upper Threshold", "Bullish threshold", "QQE");
_lowerThreshold = Param(nameof(LowerThreshold), 40m)
.SetDisplay("Lower Threshold", "Bearish threshold", "QQE");
_cooldownBars = Param(nameof(CooldownBars), 10)
.SetDisplay("Cooldown Bars", "Bars to wait between trades", "Risk");
}
public DataType CandleType
{
get => _candleTypeParam.Value;
set => _candleTypeParam.Value = value;
}
public int RsiPeriod
{
get => _rsiPeriod.Value;
set => _rsiPeriod.Value = value;
}
public decimal UpperThreshold
{
get => _upperThreshold.Value;
set => _upperThreshold.Value = value;
}
public decimal LowerThreshold
{
get => _lowerThreshold.Value;
set => _lowerThreshold.Value = value;
}
public int CooldownBars
{
get => _cooldownBars.Value;
set => _cooldownBars.Value = value;
}
/// <inheritdoc />
public override IEnumerable<(Security sec, DataType dt)> GetWorkingSecurities()
=> [(Security, CandleType)];
/// <inheritdoc />
protected override void OnReseted()
{
base.OnReseted();
_rsi = null;
_prevRsi = 0;
_cooldownRemaining = 0;
}
/// <inheritdoc />
protected override void OnStarted2(DateTime time)
{
base.OnStarted2(time);
_rsi = new RelativeStrengthIndex { Length = RsiPeriod };
var subscription = SubscribeCandles(CandleType);
subscription
.Bind(_rsi, OnProcess)
.Start();
var area = CreateChartArea();
if (area != null)
{
DrawCandles(area, subscription);
DrawOwnTrades(area);
}
}
private void OnProcess(ICandleMessage candle, decimal rsiVal)
{
if (candle.State != CandleStates.Finished)
return;
if (!_rsi.IsFormed)
{
_prevRsi = rsiVal;
return;
}
if (!IsFormedAndOnlineAndAllowTrading())
{
_prevRsi = rsiVal;
return;
}
if (_cooldownRemaining > 0)
{
_cooldownRemaining--;
_prevRsi = rsiVal;
return;
}
if (_prevRsi == 0)
{
_prevRsi = rsiVal;
return;
}
// RSI crosses above upper threshold (bullish signal)
var crossUp = rsiVal > UpperThreshold && _prevRsi <= UpperThreshold;
// RSI crosses below lower threshold (bearish signal)
var crossDown = rsiVal < LowerThreshold && _prevRsi >= LowerThreshold;
if (crossUp && Position <= 0)
{
if (Position < 0)
BuyMarket(Math.Abs(Position));
BuyMarket(Volume);
_cooldownRemaining = CooldownBars;
}
else if (crossDown && Position >= 0)
{
if (Position > 0)
SellMarket(Math.Abs(Position));
SellMarket(Volume);
_cooldownRemaining = CooldownBars;
}
// Exit long: RSI drops below 50
else if (Position > 0 && rsiVal < 50 && _prevRsi >= 50)
{
SellMarket(Math.Abs(Position));
_cooldownRemaining = CooldownBars;
}
// Exit short: RSI rises above 50
else if (Position < 0 && rsiVal > 50 && _prevRsi <= 50)
{
BuyMarket(Math.Abs(Position));
_cooldownRemaining = CooldownBars;
}
_prevRsi = rsiVal;
}
}
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 RelativeStrengthIndex
from StockSharp.Algo.Strategies import Strategy
class qqe_signals_strategy(Strategy):
"""QQE Signals Strategy."""
def __init__(self):
super(qqe_signals_strategy, self).__init__()
self._candle_type = self.Param("CandleType", DataType.TimeFrame(TimeSpan.FromMinutes(30))) \
.SetDisplay("Candle type", "Candle type for strategy calculation.", "General")
self._rsi_period = self.Param("RsiPeriod", 14) \
.SetDisplay("RSI Length", "RSI period", "QQE")
self._upper_threshold = self.Param("UpperThreshold", 60.0) \
.SetDisplay("Upper Threshold", "Bullish threshold", "QQE")
self._lower_threshold = self.Param("LowerThreshold", 40.0) \
.SetDisplay("Lower Threshold", "Bearish threshold", "QQE")
self._cooldown_bars = self.Param("CooldownBars", 10) \
.SetDisplay("Cooldown Bars", "Bars to wait between trades", "Risk")
self._rsi = None
self._prev_rsi = 0.0
self._cooldown_remaining = 0
@property
def candle_type(self):
return self._candle_type.Value
def OnReseted(self):
super(qqe_signals_strategy, self).OnReseted()
self._rsi = None
self._prev_rsi = 0.0
self._cooldown_remaining = 0
def OnStarted2(self, time):
super(qqe_signals_strategy, self).OnStarted2(time)
self._rsi = RelativeStrengthIndex()
self._rsi.Length = int(self._rsi_period.Value)
subscription = self.SubscribeCandles(self.candle_type)
subscription.Bind(self._rsi, self._on_process).Start()
area = self.CreateChartArea()
if area is not None:
self.DrawCandles(area, subscription)
self.DrawOwnTrades(area)
def _on_process(self, candle, rsi_val):
if candle.State != CandleStates.Finished:
return
if not self._rsi.IsFormed:
self._prev_rsi = float(rsi_val)
return
if not self.IsFormedAndOnlineAndAllowTrading():
self._prev_rsi = float(rsi_val)
return
rsi = float(rsi_val)
if self._cooldown_remaining > 0:
self._cooldown_remaining -= 1
self._prev_rsi = rsi
return
if self._prev_rsi == 0.0:
self._prev_rsi = rsi
return
upper = float(self._upper_threshold.Value)
lower = float(self._lower_threshold.Value)
cooldown = int(self._cooldown_bars.Value)
cross_up = rsi > upper and self._prev_rsi <= upper
cross_down = rsi < lower and self._prev_rsi >= lower
if cross_up and self.Position <= 0:
if self.Position < 0:
self.BuyMarket(Math.Abs(self.Position))
self.BuyMarket(self.Volume)
self._cooldown_remaining = cooldown
elif cross_down 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 rsi < 50 and self._prev_rsi >= 50:
self.SellMarket(Math.Abs(self.Position))
self._cooldown_remaining = cooldown
elif self.Position < 0 and rsi > 50 and self._prev_rsi <= 50:
self.BuyMarket(Math.Abs(self.Position))
self._cooldown_remaining = cooldown
self._prev_rsi = rsi
def CreateClone(self):
return qqe_signals_strategy()