Estratégia de Reversão no Keltner Channel
Canais baseados em volatilidade podem destacar movimentos sobreextendidos. Este método opera contra o preço quando ele ultrapassa o Keltner Channel, antecipando um retorno em direção à linha média. Utiliza uma média móvel exponencial e o ATR para dimensionar a largura do canal.
Os testes indicam um retorno anual médio de aproximadamente 106%. Funciona melhor no mercado de ações.
A cada vela concluída, a estratégia verifica se o fechamento está além da banda superior ou inferior e se a direção da vela está de acordo. Velas de alta fechando abaixo da banda inferior disparam entradas compradas, enquanto velas de baixa acima da banda superior geram vendas. As posições são encerradas quando o preço cruza a banda média ou quando o stop baseado em ATR é atingido.
Ao operar na direção oposta dos extremos de curto prazo, o sistema busca movimentos rápidos de reversão à média dentro de uma faixa mais ampla.
Detalhes
- Critérios de entrada: Fechamento fora do Keltner Channel na direção da vela.
- Comprado/Vendido: Ambos.
- Critérios de saída: Preço cruzando a banda média ou stop-loss.
- Stops: Sim, baseados em ATR.
- Valores padrão:
EmaPeriod= 20AtrPeriod= 14AtrMultiplier= 2.0StopLossAtrMultiplier= 2.0CandleType= 5 minute
- Filtros:
- Categoria: Reversão à média
- Direção: Ambos
- Indicadores: Keltner Channel
- Stops: Sim
- Complexidade: Básico
- Período: Intradiário
- Sazonalidade: Não
- Redes neurais: Não
- Divergência: Não
- Nível de risco: Médio
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>
/// Fades a directional outside close against separately configured EMA/ATR bands.
/// Exits at the EMA or native actual-fill protection with frozen signal ATR distance.
/// </summary>
public class KeltnerChannelReversalStrategy : Strategy
{
private readonly StrategyParam<int> _emaPeriod;
private readonly StrategyParam<int> _atrPeriod;
private readonly StrategyParam<decimal> _atrMultiplier;
private readonly StrategyParam<decimal> _stopLossAtrMultiplier;
private readonly StrategyParam<DataType> _candleType;
private Order _pendingOrder;
private Unit _stopDistance;
private bool _protectionStarted;
public int EmaPeriod { get => _emaPeriod.Value; set => _emaPeriod.Value = value; }
public int AtrPeriod { get => _atrPeriod.Value; set => _atrPeriod.Value = value; }
public decimal AtrMultiplier { get => _atrMultiplier.Value; set => _atrMultiplier.Value = value; }
public decimal StopLossAtrMultiplier { get => _stopLossAtrMultiplier.Value; set => _stopLossAtrMultiplier.Value = value; }
public DataType CandleType { get => _candleType.Value; set => _candleType.Value = value; }
public KeltnerChannelReversalStrategy()
{
_emaPeriod = Param(nameof(EmaPeriod), 20).SetGreaterThanZero()
.SetDisplay("EMA Period", "Close EMA length for middle band", "Indicators");
_atrPeriod = Param(nameof(AtrPeriod), 14).SetGreaterThanZero()
.SetDisplay("ATR Period", "Wilder ATR length for channel width and stop", "Indicators");
_atrMultiplier = Param(nameof(AtrMultiplier), 2m).SetNotNegative()
.SetDisplay("ATR Multiplier", "ATR multiple for channel width", "Indicators");
_stopLossAtrMultiplier = Param(nameof(StopLossAtrMultiplier), 2m).SetNotNegative()
.SetDisplay("Stop ATR Multiplier", "Frozen entry ATR distance; zero disables the stop", "Protection");
_candleType = Param(nameof(CandleType), TimeSpan.FromMinutes(5).TimeFrame())
.SetDisplay("Candle Type", "Keltner and stop candle timeframe", "General");
OrderRegistering += order => _pendingOrder = order;
}
public override IEnumerable<(Security sec, DataType dt)> GetWorkingSecurities()
=> [(Security, CandleType), (Security, DataType.Level1)];
protected override void OnReseted()
{
base.OnReseted();
_pendingOrder = null;
_stopDistance = null;
_protectionStarted = false;
}
protected override void OnStarted2(DateTime time)
{
base.OnStarted2(time);
_pendingOrder = null;
_stopDistance = null;
_protectionStarted = false;
var ema = new ExponentialMovingAverage { Length = EmaPeriod };
var atr = new AverageTrueRange { Length = AtrPeriod };
foreach (var field in new[] { Level1Fields.BestBidPrice, Level1Fields.BestAskPrice })
{
var quotes = new Subscription(DataType.Level1, Security);
quotes.MarketData.BuildField = field;
SubscribeLevel1(quotes).Bind(ObserveProtectionQuote).Start();
}
var candles = SubscribeCandles(CandleType);
candles.BindEx(ema, atr, ProcessCandle, false).Start();
var area = CreateChartArea();
if (area != null)
{
DrawCandles(area, candles);
DrawIndicator(area, ema);
DrawIndicator(area, atr);
DrawOwnTrades(area);
}
}
private void ObserveProtectionQuote(Level1ChangeMessage quote)
{
// Native protection evaluates executable quotes between finished signal candles.
}
private void ProcessCandle(ICandleMessage candle, IIndicatorValue emaValue, IIndicatorValue atrValue)
{
if (candle.State != CandleStates.Finished || !emaValue.IsFormed ||
!atrValue.IsFormed || !IsFormedAndOnlineAndAllowTrading())
return;
if (_pendingOrder is not null && _pendingOrder.State is not (OrderStates.Done or OrderStates.Failed))
return;
var middle = emaValue.GetValue<decimal>();
var atr = atrValue.GetValue<decimal>();
var lower = middle - atr * AtrMultiplier;
var upper = middle + atr * AtrMultiplier;
var close = candle.ClosePrice;
if (Position > 0m && close >= middle)
SellMarket(Position);
else if (Position < 0m && close <= middle)
BuyMarket(Math.Abs(Position));
else if (Position == 0m && close < lower && close > candle.OpenPrice)
Enter(Sides.Buy, atr);
else if (Position == 0m && close > upper && close < candle.OpenPrice)
Enter(Sides.Sell, atr);
}
private void Enter(Sides side, decimal atr)
{
var distance = atr * StopLossAtrMultiplier;
_stopDistance ??= new Unit(distance);
_stopDistance.Value = distance;
if (!_protectionStarted && distance > 0m)
{
StartProtection(new Unit(), _stopDistance, useMarketOrders: true, isLocalStop: true);
_protectionStarted = true;
}
RegisterOrder(new Order
{
Security = Security,
Portfolio = Portfolio,
Type = OrderTypes.Market,
Side = side,
Volume = Volume,
Comment = "Keltner reversal entry",
});
}
}
import clr
clr.AddReference("StockSharp.Messages")
clr.AddReference("StockSharp.Algo")
clr.AddReference("StockSharp.Algo.Indicators")
clr.AddReference("StockSharp.Algo.Strategies")
clr.AddReference("StockSharp.BusinessEntities")
from System import TimeSpan, Decimal, Math
from StockSharp.Messages import DataType, CandleStates, Unit, OrderStates, OrderTypes, Sides, Level1Fields
from StockSharp.BusinessEntities import Subscription, Order
from StockSharp.Algo.Indicators import ExponentialMovingAverage, AverageTrueRange
from StockSharp.Algo.Strategies import Strategy
class keltner_channel_reversal_strategy(Strategy):
"""Fade a directional outside close with separate EMA/ATR lengths and frozen ATR protection."""
def __init__(self):
super(keltner_channel_reversal_strategy, self).__init__()
self._ema_period = self.Param("EmaPeriod", 20).SetGreaterThanZero().SetDisplay("EMA Period", "Close EMA length for middle band", "Indicators")
self._atr_period = self.Param("AtrPeriod", 14).SetGreaterThanZero().SetDisplay("ATR Period", "Wilder ATR length for channel width and stop", "Indicators")
self._atr_multiplier = self.Param("AtrMultiplier", 2.0).SetNotNegative().SetDisplay("ATR Multiplier", "ATR multiple for channel width", "Indicators")
self._stop_loss_atr_multiplier = self.Param("StopLossAtrMultiplier", 2.0).SetNotNegative().SetDisplay("Stop ATR Multiplier", "Frozen entry ATR distance; zero disables the stop", "Protection")
self._candle_type = self.Param("CandleType", DataType.TimeFrame(TimeSpan.FromMinutes(5))).SetDisplay("Candle Type", "Keltner and stop candle timeframe", "General")
self._pending_order = None
self._stop_distance = None
self._protection_started = False
self.OrderRegistering += self._track_pending
@property
def candle_type(self):
return self._candle_type.Value
def GetWorkingSecurities(self):
return [(self.Security, self.candle_type), (self.Security, DataType.Level1)]
def _track_pending(self, order):
self._pending_order = order
def OnReseted(self):
super(keltner_channel_reversal_strategy, self).OnReseted()
self._pending_order = None
self._stop_distance = None
self._protection_started = False
def OnStarted2(self, time):
super(keltner_channel_reversal_strategy, self).OnStarted2(time)
self._pending_order = None
self._stop_distance = None
self._protection_started = False
ema = ExponentialMovingAverage()
ema.Length = self._ema_period.Value
atr = AverageTrueRange()
atr.Length = self._atr_period.Value
for field in (Level1Fields.BestBidPrice, Level1Fields.BestAskPrice):
quotes = Subscription(DataType.Level1, self.Security)
quotes.MarketData.BuildField = field
self.SubscribeLevel1(quotes).Bind(self._observe_protection_quote).Start()
candles = self.SubscribeCandles(self.candle_type)
candles.BindEx(ema, atr, self._process_candle, False).Start()
area = self.CreateChartArea()
if area is not None:
self.DrawCandles(area, candles)
self.DrawIndicator(area, ema)
self.DrawIndicator(area, atr)
self.DrawOwnTrades(area)
def _observe_protection_quote(self, quote):
# Native protection evaluates executable quotes between finished signal candles.
pass
def _process_candle(self, candle, ema_value, atr_value):
if (candle.State != CandleStates.Finished or not ema_value.IsFormed or
not atr_value.IsFormed or not self.IsFormedAndOnlineAndAllowTrading()):
return
if self._pending_order is not None and self._pending_order.State not in (OrderStates.Done, OrderStates.Failed):
return
middle = ema_value.GetValue[Decimal](None)
atr = atr_value.GetValue[Decimal](None)
lower = middle - atr * Decimal(self._atr_multiplier.Value)
upper = middle + atr * Decimal(self._atr_multiplier.Value)
close = candle.ClosePrice
if self.Position > 0 and close >= middle:
self.SellMarket(self.Position)
elif self.Position < 0 and close <= middle:
self.BuyMarket(Math.Abs(self.Position))
elif self.Position == 0 and close < lower and close > candle.OpenPrice:
self._enter(Sides.Buy, atr)
elif self.Position == 0 and close > upper and close < candle.OpenPrice:
self._enter(Sides.Sell, atr)
def _enter(self, side, atr):
distance = atr * Decimal(self._stop_loss_atr_multiplier.Value)
if self._stop_distance is None:
self._stop_distance = Unit(distance)
self._stop_distance.Value = distance
if not self._protection_started and distance > 0:
self.StartProtection(Unit(), self._stop_distance, useMarketOrders=True, isLocalStop=True)
self._protection_started = True
order = Order()
order.Security = self.Security
order.Portfolio = self.Portfolio
order.Type = OrderTypes.Market
order.Side = side
order.Volume = self.Volume
order.Comment = "Keltner reversal entry"
self.RegisterOrder(order)
def CreateClone(self):
return keltner_channel_reversal_strategy()