Estrategia de Reversión a la Media por Ancho de Keltner
La estrategia de Reversión a la Media por Ancho de Keltner se centra en lecturas extremas de las Keltner para explotar la reversión. Las grandes desviaciones del nivel normal rara vez perduran.
Las pruebas indican un rendimiento anual promedio de aproximadamente 160%. Funciona mejor en el mercado de divisas.
Las operaciones se activan cuando el indicador se aleja mucho de su media y luego comienza a revertirse. Tanto las configuraciones largas como cortas incluyen un stop protector.
Adecuada para operadores de swing que esperan oscilaciones, la estrategia cierra las posiciones una vez que las Keltner regresan al equilibrio. Parámetro inicial EmaPeriod = 20.
Detalles
- Criterios de entrada: El indicador cruza de regreso hacia la media.
- Largo/Corto: Ambas direcciones.
- Criterios de salida: El indicador revierte al promedio.
- Stops: Sí.
- Valores predeterminados:
EmaPeriod= 20AtrPeriod= 14KeltnerMultiplier= 2.0mWidthLookbackPeriod= 20WidthDeviationMultiplier= 2.0mAtrStopMultiplier= 2.0mCandleType= TimeSpan.FromMinutes(5)
- Filtros:
- Categoría: Mean Reversion
- Dirección: Ambos
- Indicadores: Keltner
- Stops: Sí
- Complejidad: Intermedio
- Marco temporal: Corto plazo
- Estacionalidad: No
- Redes neuronales: No
- Divergencia: No
- Nivel de riesgo: Medio
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>
/// Keltner Channel width mean reversion.
/// Enters when the channel width is beyond its average by a standard deviation multiplier and starts
/// turning back toward the average: long on an extreme contraction, short on an extreme expansion.
/// Exits when the width returns to its average or the ATR stop is hit.
/// </summary>
public class KeltnerWidthMeanReversionStrategy : Strategy
{
private readonly StrategyParam<int> _emaPeriod;
private readonly StrategyParam<int> _atrPeriod;
private readonly StrategyParam<decimal> _keltnerMultiplier;
private readonly StrategyParam<int> _widthLookbackPeriod;
private readonly StrategyParam<decimal> _widthDeviationMultiplier;
private readonly StrategyParam<decimal> _atrStopMultiplier;
private readonly StrategyParam<DataType> _candleType;
private SimpleMovingAverage _widthAverage;
private StandardDeviation _widthStdDev;
private decimal? _prevWidth;
private decimal _stopPrice;
/// <summary>
/// EMA period for the channel middle line.
/// </summary>
public int EmaPeriod
{
get => _emaPeriod.Value;
set => _emaPeriod.Value = value;
}
/// <summary>
/// ATR period for the channel bands and the stop.
/// </summary>
public int AtrPeriod
{
get => _atrPeriod.Value;
set => _atrPeriod.Value = value;
}
/// <summary>
/// ATR multiplier for the channel bands.
/// </summary>
public decimal KeltnerMultiplier
{
get => _keltnerMultiplier.Value;
set => _keltnerMultiplier.Value = value;
}
/// <summary>
/// Lookback period for width statistics.
/// </summary>
public int WidthLookbackPeriod
{
get => _widthLookbackPeriod.Value;
set => _widthLookbackPeriod.Value = value;
}
/// <summary>
/// Standard deviation multiplier for extreme width.
/// </summary>
public decimal WidthDeviationMultiplier
{
get => _widthDeviationMultiplier.Value;
set => _widthDeviationMultiplier.Value = value;
}
/// <summary>
/// Stop-loss distance in ATR multiples.
/// </summary>
public decimal AtrStopMultiplier
{
get => _atrStopMultiplier.Value;
set => _atrStopMultiplier.Value = value;
}
/// <summary>
/// Candle type.
/// </summary>
public DataType CandleType
{
get => _candleType.Value;
set => _candleType.Value = value;
}
/// <summary>
/// Initialize <see cref="KeltnerWidthMeanReversionStrategy"/>.
/// </summary>
public KeltnerWidthMeanReversionStrategy()
{
_emaPeriod = Param(nameof(EmaPeriod), 20)
.SetGreaterThanZero()
.SetDisplay("EMA Period", "EMA period for Keltner Channel", "Indicators");
_atrPeriod = Param(nameof(AtrPeriod), 14)
.SetGreaterThanZero()
.SetDisplay("ATR Period", "ATR period for Keltner Channel", "Indicators");
_keltnerMultiplier = Param(nameof(KeltnerMultiplier), 2.0m)
.SetGreaterThanZero()
.SetDisplay("Keltner Multiplier", "ATR multiplier for Keltner Channel", "Indicators");
_widthLookbackPeriod = Param(nameof(WidthLookbackPeriod), 20)
.SetGreaterThanZero()
.SetDisplay("Width Lookback", "Period for width statistics", "Strategy");
_widthDeviationMultiplier = Param(nameof(WidthDeviationMultiplier), 2.0m)
.SetGreaterThanZero()
.SetDisplay("Width Deviation Multiplier", "Standard deviation multiplier for extreme width", "Strategy");
_atrStopMultiplier = Param(nameof(AtrStopMultiplier), 2.0m)
.SetNotNegative()
.SetDisplay("ATR Stop Multiplier", "Stop-loss distance in ATR multiples", "Risk Management");
_candleType = Param(nameof(CandleType), TimeSpan.FromMinutes(5).TimeFrame())
.SetDisplay("Candle Type", "Type of candles to use", "General");
}
/// <inheritdoc />
public override IEnumerable<(Security sec, DataType dt)> GetWorkingSecurities()
{
return [(Security, CandleType)];
}
/// <inheritdoc />
protected override void OnReseted()
{
base.OnReseted();
_widthAverage = null;
_widthStdDev = null;
_prevWidth = null;
_stopPrice = 0m;
}
/// <inheritdoc />
protected override void OnStarted2(DateTime time)
{
base.OnStarted2(time);
var ema = new ExponentialMovingAverage { Length = EmaPeriod };
var atr = new AverageTrueRange { Length = AtrPeriod };
_widthAverage = new SimpleMovingAverage { Length = WidthLookbackPeriod };
_widthStdDev = new StandardDeviation { Length = WidthLookbackPeriod };
var subscription = SubscribeCandles(CandleType);
subscription
.Bind(ema, atr, ProcessCandle)
.Start();
var area = CreateChartArea();
if (area != null)
{
DrawCandles(area, subscription);
DrawIndicator(area, ema);
DrawOwnTrades(area);
}
}
private void ProcessCandle(ICandleMessage candle, decimal emaValue, decimal atrValue)
{
if (candle.State != CandleStates.Finished)
return;
// Width of the channel: (EMA + k*ATR) - (EMA - k*ATR).
var width = 2m * KeltnerMultiplier * atrValue;
var avgWidth = _widthAverage.Process(width, candle.ServerTime, true).ToDecimal();
var stdWidth = _widthStdDev.Process(width, candle.ServerTime, true).ToDecimal();
var prevWidth = _prevWidth;
_prevWidth = width;
if (!_widthAverage.IsFormed || !_widthStdDev.IsFormed || prevWidth is not decimal prev)
return;
if (!IsFormedAndOnlineAndAllowTrading())
return;
if (CheckStop(candle))
return;
var close = candle.ClosePrice;
var stopDistance = AtrStopMultiplier * atrValue;
// Extreme reading that has started to turn back toward the average.
if (width < avgWidth - WidthDeviationMultiplier * stdWidth && width > prev && Position <= 0)
{
BuyMarket(Volume + Math.Abs(Position));
_stopPrice = stopDistance > 0 ? close - stopDistance : 0m;
}
else if (width > avgWidth + WidthDeviationMultiplier * stdWidth && width < prev && Position >= 0)
{
SellMarket(Volume + Math.Abs(Position));
_stopPrice = stopDistance > 0 ? close + stopDistance : 0m;
}
else if ((Position > 0 && width >= avgWidth) || (Position < 0 && width <= avgWidth))
{
ExitPosition();
}
}
private bool CheckStop(ICandleMessage candle)
{
if (_stopPrice == 0m)
return false;
if ((Position > 0 && candle.LowPrice <= _stopPrice) || (Position < 0 && candle.HighPrice >= _stopPrice))
{
ExitPosition();
return true;
}
return false;
}
private void ExitPosition()
{
if (Position > 0)
SellMarket(Position);
else if (Position < 0)
BuyMarket(-Position);
_stopPrice = 0m;
}
}
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 ExponentialMovingAverage, AverageTrueRange, SimpleMovingAverage, StandardDeviation
from StockSharp.Algo.Strategies import Strategy
from indicator_extensions import *
class keltner_width_mean_reversion_strategy(Strategy):
"""
Keltner Channel width mean reversion.
Enters when the channel width is beyond its average by a standard deviation multiplier and starts
turning back toward the average: long on an extreme contraction, short on an extreme expansion.
Exits when the width returns to its average or the ATR stop is hit.
"""
def __init__(self):
super(keltner_width_mean_reversion_strategy, self).__init__()
self._ema_period = self.Param("EmaPeriod", 20) \
.SetGreaterThanZero() \
.SetDisplay("EMA Period", "EMA period for Keltner Channel", "Indicators")
self._atr_period = self.Param("AtrPeriod", 14) \
.SetGreaterThanZero() \
.SetDisplay("ATR Period", "ATR period for Keltner Channel", "Indicators")
self._keltner_multiplier = self.Param("KeltnerMultiplier", 2.0) \
.SetGreaterThanZero() \
.SetDisplay("Keltner Multiplier", "ATR multiplier for Keltner Channel", "Indicators")
self._width_lookback_period = self.Param("WidthLookbackPeriod", 20) \
.SetGreaterThanZero() \
.SetDisplay("Width Lookback", "Period for width statistics", "Strategy")
self._width_deviation_multiplier = self.Param("WidthDeviationMultiplier", 2.0) \
.SetGreaterThanZero() \
.SetDisplay("Width Deviation Multiplier", "Standard deviation multiplier for extreme width", "Strategy")
self._atr_stop_multiplier = self.Param("AtrStopMultiplier", 2.0) \
.SetNotNegative() \
.SetDisplay("ATR Stop Multiplier", "Stop-loss distance in ATR multiples", "Risk Management")
self._candle_type = self.Param("CandleType", DataType.TimeFrame(TimeSpan.FromMinutes(5))) \
.SetDisplay("Candle Type", "Type of candles to use", "General")
self._width_average = None
self._width_std_dev = None
self._prev_width = None
self._stop_price = 0.0
@property
def CandleType(self):
return self._candle_type.Value
def GetWorkingSecurities(self):
return [(self.Security, self.CandleType)]
def OnReseted(self):
super(keltner_width_mean_reversion_strategy, self).OnReseted()
self._width_average = None
self._width_std_dev = None
self._prev_width = None
self._stop_price = 0.0
def OnStarted2(self, time):
super(keltner_width_mean_reversion_strategy, self).OnStarted2(time)
ema = ExponentialMovingAverage()
ema.Length = self._ema_period.Value
atr = AverageTrueRange()
atr.Length = self._atr_period.Value
self._width_average = SimpleMovingAverage()
self._width_average.Length = self._width_lookback_period.Value
self._width_std_dev = StandardDeviation()
self._width_std_dev.Length = self._width_lookback_period.Value
subscription = self.SubscribeCandles(self.CandleType)
subscription.Bind(ema, atr, self._process_candle).Start()
area = self.CreateChartArea()
if area is not None:
self.DrawCandles(area, subscription)
self.DrawIndicator(area, ema)
self.DrawOwnTrades(area)
def _process_candle(self, candle, ema_value, atr_value):
if candle.State != CandleStates.Finished:
return
atr_value = float(atr_value)
# Width of the channel: (EMA + k*ATR) - (EMA - k*ATR).
width = 2.0 * float(self._keltner_multiplier.Value) * atr_value
avg_width = float(process_float(self._width_average, width, candle.ServerTime, True))
std_width = float(process_float(self._width_std_dev, width, candle.ServerTime, True))
prev = self._prev_width
self._prev_width = width
if not self._width_average.IsFormed or not self._width_std_dev.IsFormed or prev is None:
return
if not self.IsFormedAndOnlineAndAllowTrading():
return
if self._check_stop(candle):
return
close = float(candle.ClosePrice)
k = float(self._width_deviation_multiplier.Value)
stop_distance = float(self._atr_stop_multiplier.Value) * atr_value
# Extreme reading that has started to turn back toward the average.
if width < avg_width - k * std_width and width > prev and self.Position <= 0:
self.BuyMarket(self.Volume + abs(self.Position))
self._stop_price = close - stop_distance if stop_distance > 0 else 0.0
elif width > avg_width + k * std_width and width < prev and self.Position >= 0:
self.SellMarket(self.Volume + abs(self.Position))
self._stop_price = close + stop_distance if stop_distance > 0 else 0.0
elif (self.Position > 0 and width >= avg_width) or (self.Position < 0 and width <= avg_width):
self._exit_position()
def _check_stop(self, candle):
if self._stop_price == 0.0:
return False
if (self.Position > 0 and float(candle.LowPrice) <= self._stop_price) or \
(self.Position < 0 and float(candle.HighPrice) >= self._stop_price):
self._exit_position()
return True
return False
def _exit_position(self):
if self.Position > 0:
self.SellMarket(self.Position)
elif self.Position < 0:
self.BuyMarket(-self.Position)
self._stop_price = 0.0
def CreateClone(self):
return keltner_width_mean_reversion_strategy()