Стратегия HMA Crossover ATR Curvature
HMA Crossover ATR Curvature — трендовая стратегия, комбинирующая пересечение быстрой и медленной Hull Moving Average с фильтром кривизны (вторая производная). Размер позиции рассчитывается по ATR и доле риска, выход осуществляется по трейлинг-стопу на основе ATR.
Подробности
- Данные: ценовые свечи.
- Условия входа:
- Лонг: быстрая HMA пересекает медленную снизу вверх, кривизна выше порога.
- Шорт: быстрая HMA пересекает медленную сверху вниз, кривизна ниже отрицательного порога.
- Условия выхода: трейлинг-стоп на основе ATR.
- Стопы: ATR трейлинг-стоп.
- Параметры по умолчанию:
FastLength= 15SlowLength= 34AtrLength= 14RiskPercent= 1AtrMultiplier= 1.5TrailMultiplier= 1CurvatureThreshold= 0
- Фильтры:
- Категория: трендовая
- Направление: лонг и шорт
- Индикаторы: HMA, ATR
- Сложность: низкая
- Уровень риска: средний
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>
/// HMA Crossover ATR Curvature strategy.
/// Goes long when the fast HMA crosses above the slow HMA while the curvature (second difference) of the fast HMA is above
/// CurvatureThreshold, and short on the opposite cross with curvature below -CurvatureThreshold, reversing an opposite position.
/// The order size risks RiskPercent of the portfolio over a distance of AtrMultiplier ATRs, and an ATR trailing stop of
/// TrailMultiplier ATRs closes the position.
/// </summary>
public class HmaCrossoverAtrCurvatureStrategy : Strategy
{
private readonly StrategyParam<int> _fastLength;
private readonly StrategyParam<int> _slowLength;
private readonly StrategyParam<int> _atrLength;
private readonly StrategyParam<decimal> _riskPercent;
private readonly StrategyParam<decimal> _atrMultiplier;
private readonly StrategyParam<decimal> _trailMultiplier;
private readonly StrategyParam<decimal> _curvatureThreshold;
private readonly StrategyParam<DataType> _candleType;
private decimal? _prevFast;
private decimal? _prevFast2;
private decimal? _prevSlow;
private decimal? _trailStop;
/// <summary>
/// Fast HMA length.
/// </summary>
public int FastLength
{
get => _fastLength.Value;
set => _fastLength.Value = value;
}
/// <summary>
/// Slow HMA length.
/// </summary>
public int SlowLength
{
get => _slowLength.Value;
set => _slowLength.Value = value;
}
/// <summary>
/// ATR period.
/// </summary>
public int AtrLength
{
get => _atrLength.Value;
set => _atrLength.Value = value;
}
/// <summary>
/// Percent of the portfolio risked per trade.
/// </summary>
public decimal RiskPercent
{
get => _riskPercent.Value;
set => _riskPercent.Value = value;
}
/// <summary>
/// ATR multiple used as the risk distance for position sizing.
/// </summary>
public decimal AtrMultiplier
{
get => _atrMultiplier.Value;
set => _atrMultiplier.Value = value;
}
/// <summary>
/// ATR multiple of the trailing stop distance.
/// </summary>
public decimal TrailMultiplier
{
get => _trailMultiplier.Value;
set => _trailMultiplier.Value = value;
}
/// <summary>
/// Minimum curvature of the fast HMA.
/// </summary>
public decimal CurvatureThreshold
{
get => _curvatureThreshold.Value;
set => _curvatureThreshold.Value = value;
}
/// <summary>
/// Candle type.
/// </summary>
public DataType CandleType
{
get => _candleType.Value;
set => _candleType.Value = value;
}
/// <summary>
/// Constructor.
/// </summary>
public HmaCrossoverAtrCurvatureStrategy()
{
_fastLength = Param(nameof(FastLength), 15)
.SetGreaterThanZero()
.SetDisplay("Fast Length", "Fast HMA length", "Indicators");
_slowLength = Param(nameof(SlowLength), 34)
.SetGreaterThanZero()
.SetDisplay("Slow Length", "Slow HMA length", "Indicators");
_atrLength = Param(nameof(AtrLength), 14)
.SetGreaterThanZero()
.SetDisplay("ATR Length", "ATR period", "Indicators");
_riskPercent = Param(nameof(RiskPercent), 1m)
.SetGreaterThanZero()
.SetDisplay("Risk %", "Percent of the portfolio risked per trade", "Risk");
_atrMultiplier = Param(nameof(AtrMultiplier), 1.5m)
.SetGreaterThanZero()
.SetDisplay("ATR Multiplier", "ATR multiple used as risk distance for sizing", "Risk");
_trailMultiplier = Param(nameof(TrailMultiplier), 1m)
.SetGreaterThanZero()
.SetDisplay("Trail Multiplier", "ATR multiple of the trailing stop distance", "Risk");
_curvatureThreshold = Param(nameof(CurvatureThreshold), 0m)
.SetDisplay("Curvature Threshold", "Minimum curvature of the fast HMA", "Signals");
_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();
ResetState();
}
private void ResetState()
{
_prevFast = null;
_prevFast2 = null;
_prevSlow = null;
_trailStop = null;
}
/// <inheritdoc />
protected override void OnStarted2(DateTime time)
{
base.OnStarted2(time);
ResetState();
var fastHma = new HullMovingAverage { Length = FastLength };
var slowHma = new HullMovingAverage { Length = SlowLength };
var atr = new AverageTrueRange { Length = AtrLength };
var subscription = SubscribeCandles(CandleType);
subscription
.Bind(fastHma, slowHma, atr, ProcessCandle)
.Start();
var area = CreateChartArea();
if (area != null)
{
DrawCandles(area, subscription);
DrawIndicator(area, fastHma);
DrawIndicator(area, slowHma);
DrawOwnTrades(area);
}
}
private void ProcessCandle(ICandleMessage candle, decimal fast, decimal slow, decimal atr)
{
if (candle.State != CandleStates.Finished)
return;
var prevFast = _prevFast;
var prevFast2 = _prevFast2;
var prevSlow = _prevSlow;
_prevFast2 = _prevFast;
_prevFast = fast;
_prevSlow = slow;
if (prevFast is not decimal pf || prevFast2 is not decimal pf2 || prevSlow is not decimal ps || atr <= 0)
return;
if (!IsFormedAndOnlineAndAllowTrading())
return;
var trailDistance = atr * TrailMultiplier;
if (Position > 0)
{
var stop = candle.ClosePrice - trailDistance;
_trailStop = _trailStop is decimal s ? Math.Max(s, stop) : stop;
if (candle.LowPrice <= _trailStop)
{
SellMarket(Position);
_trailStop = null;
return;
}
}
else if (Position < 0)
{
var stop = candle.ClosePrice + trailDistance;
_trailStop = _trailStop is decimal s ? Math.Min(s, stop) : stop;
if (candle.HighPrice >= _trailStop)
{
BuyMarket(-Position);
_trailStop = null;
return;
}
}
var curvature = fast - 2m * pf + pf2;
var crossUp = pf <= ps && fast > slow;
var crossDown = pf >= ps && fast < slow;
if (crossUp && curvature > CurvatureThreshold && Position <= 0)
{
BuyMarket(GetOrderVolume(atr) + Math.Abs(Position));
_trailStop = candle.ClosePrice - trailDistance;
}
else if (crossDown && curvature < -CurvatureThreshold && Position >= 0)
{
SellMarket(GetOrderVolume(atr) + Math.Abs(Position));
_trailStop = candle.ClosePrice + trailDistance;
}
}
private decimal GetOrderVolume(decimal atr)
{
var equity = Portfolio?.CurrentValue ?? 0m;
var riskDistance = atr * AtrMultiplier;
if (equity <= 0 || riskDistance <= 0)
return Volume;
var step = Security?.VolumeStep ?? 1m;
if (step <= 0)
step = 1m;
var volume = Math.Floor(equity * RiskPercent / 100m / riskDistance / step) * step;
return volume > 0 ? volume : Volume;
}
}
import clr
clr.AddReference("StockSharp.Messages")
clr.AddReference("StockSharp.Algo")
clr.AddReference("StockSharp.Algo.Indicators")
clr.AddReference("StockSharp.Algo.Strategies")
import math
from System import TimeSpan, Decimal
from StockSharp.Messages import DataType, CandleStates
from StockSharp.Algo.Indicators import HullMovingAverage, AverageTrueRange
from StockSharp.Algo.Strategies import Strategy
class hma_crossover_atr_curvature_strategy(Strategy):
"""
HMA Crossover ATR Curvature strategy.
Goes long when the fast HMA crosses above the slow HMA while the curvature (second difference) of the fast HMA is above
CurvatureThreshold, and short on the opposite cross with curvature below -CurvatureThreshold, reversing an opposite position.
The order size risks RiskPercent of the portfolio over a distance of AtrMultiplier ATRs, and an ATR trailing stop of
TrailMultiplier ATRs closes the position.
"""
def __init__(self):
super(hma_crossover_atr_curvature_strategy, self).__init__()
self._fast_length = self.Param("FastLength", 15).SetGreaterThanZero().SetDisplay("Fast Length", "Fast HMA length", "Indicators")
self._slow_length = self.Param("SlowLength", 34).SetGreaterThanZero().SetDisplay("Slow Length", "Slow HMA length", "Indicators")
self._atr_length = self.Param("AtrLength", 14).SetGreaterThanZero().SetDisplay("ATR Length", "ATR period", "Indicators")
self._risk_percent = self.Param("RiskPercent", 1.0).SetGreaterThanZero().SetDisplay("Risk %", "Percent of the portfolio risked per trade", "Risk")
self._atr_multiplier = self.Param("AtrMultiplier", 1.5).SetGreaterThanZero().SetDisplay("ATR Multiplier", "ATR multiple used as risk distance for sizing", "Risk")
self._trail_multiplier = self.Param("TrailMultiplier", 1.0).SetGreaterThanZero().SetDisplay("Trail Multiplier", "ATR multiple of the trailing stop distance", "Risk")
self._curvature_threshold = self.Param("CurvatureThreshold", 0.0).SetDisplay("Curvature Threshold", "Minimum curvature of the fast HMA", "Signals")
self._candle_type = self.Param("CandleType", DataType.TimeFrame(TimeSpan.FromMinutes(5))).SetDisplay("Candle Type", "Type of candles to use", "General")
self._reset_state()
@property
def candle_type(self):
return self._candle_type.Value
def _reset_state(self):
self._prev_fast = None
self._prev_fast2 = None
self._prev_slow = None
self._trail_stop = None
def OnReseted(self):
super(hma_crossover_atr_curvature_strategy, self).OnReseted()
self._reset_state()
def OnStarted2(self, time):
super(hma_crossover_atr_curvature_strategy, self).OnStarted2(time)
self._reset_state()
fast_hma = HullMovingAverage()
fast_hma.Length = self._fast_length.Value
slow_hma = HullMovingAverage()
slow_hma.Length = self._slow_length.Value
atr = AverageTrueRange()
atr.Length = self._atr_length.Value
subscription = self.SubscribeCandles(self.candle_type)
subscription.Bind(fast_hma, slow_hma, atr, self._process_candle).Start()
area = self.CreateChartArea()
if area is not None:
self.DrawCandles(area, subscription)
self.DrawIndicator(area, fast_hma)
self.DrawIndicator(area, slow_hma)
self.DrawOwnTrades(area)
def _process_candle(self, candle, fast_value, slow_value, atr_value):
if candle.State != CandleStates.Finished:
return
fast = float(fast_value)
slow = float(slow_value)
atr = float(atr_value)
prev_fast = self._prev_fast
prev_fast2 = self._prev_fast2
prev_slow = self._prev_slow
self._prev_fast2 = self._prev_fast
self._prev_fast = fast
self._prev_slow = slow
if prev_fast is None or prev_fast2 is None or prev_slow is None or atr <= 0:
return
if not self.IsFormedAndOnlineAndAllowTrading():
return
close = float(candle.ClosePrice)
trail_distance = atr * float(self._trail_multiplier.Value)
if self.Position > 0:
stop = close - trail_distance
self._trail_stop = stop if self._trail_stop is None else max(self._trail_stop, stop)
if float(candle.LowPrice) <= self._trail_stop:
self.SellMarket(self.Position)
self._trail_stop = None
return
elif self.Position < 0:
stop = close + trail_distance
self._trail_stop = stop if self._trail_stop is None else min(self._trail_stop, stop)
if float(candle.HighPrice) >= self._trail_stop:
self.BuyMarket(-self.Position)
self._trail_stop = None
return
threshold = float(self._curvature_threshold.Value)
curvature = fast - 2.0 * prev_fast + prev_fast2
cross_up = prev_fast <= prev_slow and fast > slow
cross_down = prev_fast >= prev_slow and fast < slow
if cross_up and curvature > threshold and self.Position <= 0:
self.BuyMarket(self._get_order_volume(atr) + abs(self.Position))
self._trail_stop = close - trail_distance
elif cross_down and curvature < -threshold and self.Position >= 0:
self.SellMarket(self._get_order_volume(atr) + abs(self.Position))
self._trail_stop = close + trail_distance
def _get_order_volume(self, atr):
equity = 0.0
if self.Portfolio is not None and self.Portfolio.CurrentValue is not None:
equity = float(self.Portfolio.CurrentValue)
risk_distance = atr * float(self._atr_multiplier.Value)
if equity <= 0 or risk_distance <= 0:
return self.Volume
step = 1.0
if self.Security is not None and self.Security.VolumeStep is not None:
step = float(self.Security.VolumeStep)
if step <= 0:
step = 1.0
volume = math.floor(equity * float(self._risk_percent.Value) / 100.0 / risk_distance / step) * step
return Decimal(volume) if volume > 0 else self.Volume
def CreateClone(self):
return hma_crossover_atr_curvature_strategy()