Estrategia de MA Ponderada por Volumen con Desviación Estándar
Esta estrategia aplica una Media Móvil Ponderada por Volumen (VWMA) con un filtro de desviación estándar. Mide el impulso de la VWMA y abre una posición larga cuando el movimiento al alza supera un umbral de desviación configurable. Se abre una posición corta cuando el movimiento a la baja cruza el umbral negativo. El enfoque intenta capturar movimientos direccionales fuertes confirmados por el volumen.
Parámetros
- Tipo de vela
- Longitud de VWMA
- Período de StdDev
- K1
- K2
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>
/// Volume Weighted Moving Average with Standard Deviation filter.
/// Opens long when VWMA momentum exceeds threshold, short on opposite.
/// </summary>
public class VolumeWeightedMaStDevStrategy : Strategy
{
private readonly StrategyParam<DataType> _candleType;
private readonly StrategyParam<int> _vwmaLength;
private readonly StrategyParam<int> _stdPeriod;
private readonly StrategyParam<decimal> _k1;
private VolumeWeightedMovingAverage _vwma;
private StandardDeviation _stdDev;
private decimal? _prevVwma;
public DataType CandleType { get => _candleType.Value; set => _candleType.Value = value; }
public int VwmaLength { get => _vwmaLength.Value; set => _vwmaLength.Value = value; }
public int StdPeriod { get => _stdPeriod.Value; set => _stdPeriod.Value = value; }
public decimal K1 { get => _k1.Value; set => _k1.Value = value; }
public VolumeWeightedMaStDevStrategy()
{
_candleType = Param(nameof(CandleType), TimeSpan.FromHours(4).TimeFrame())
.SetDisplay("Candle Type", "Time frame for analysis", "General");
_vwmaLength = Param(nameof(VwmaLength), 12)
.SetDisplay("VWMA Length", "Period for Volume Weighted MA", "Indicators");
_stdPeriod = Param(nameof(StdPeriod), 9)
.SetDisplay("StdDev Period", "Period for standard deviation", "Indicators");
_k1 = Param(nameof(K1), 0.5m)
.SetDisplay("K1", "Deviation multiplier for signal threshold", "Signal");
}
/// <inheritdoc />
public override IEnumerable<(Security sec, DataType dt)> GetWorkingSecurities()
{
return [(Security, CandleType)];
}
/// <inheritdoc />
protected override void OnReseted()
{
base.OnReseted();
_vwma = null;
_stdDev = null;
_prevVwma = null;
}
/// <inheritdoc />
protected override void OnStarted2(DateTime time)
{
base.OnStarted2(time);
_prevVwma = null;
_vwma = new VolumeWeightedMovingAverage { Length = VwmaLength };
_stdDev = new StandardDeviation { Length = StdPeriod };
var subscription = SubscribeCandles(CandleType);
subscription
.Bind(_vwma, ProcessCandle)
.Start();
var area = CreateChartArea();
if (area != null)
{
DrawCandles(area, subscription);
DrawIndicator(area, _vwma);
DrawOwnTrades(area);
}
}
private void ProcessCandle(ICandleMessage candle, decimal vwmaValue)
{
if (candle.State != CandleStates.Finished)
return;
if (!_vwma.IsFormed)
return;
var t = candle.ServerTime;
if (_prevVwma is null)
{
_prevVwma = vwmaValue;
return;
}
var diff = vwmaValue - _prevVwma.Value;
var stdResult = _stdDev.Process(new DecimalIndicatorValue(_stdDev, diff, t) { IsFinal = true });
if (!_stdDev.IsFormed)
{
_prevVwma = vwmaValue;
return;
}
var stdValue = stdResult.ToDecimal();
var filter = K1 * stdValue;
if (diff > filter && Position <= 0)
BuyMarket();
else if (diff < -filter && Position >= 0)
SellMarket();
_prevVwma = vwmaValue;
}
}
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 VolumeWeightedMovingAverage, StandardDeviation
from StockSharp.Algo.Strategies import Strategy
from indicator_extensions import *
class volume_weighted_ma_st_dev_strategy(Strategy):
def __init__(self):
super(volume_weighted_ma_st_dev_strategy, self).__init__()
self._candle_type = self.Param("CandleType", DataType.TimeFrame(TimeSpan.FromHours(4))) \
.SetDisplay("Candle Type", "Time frame for analysis", "General")
self._vwma_length = self.Param("VwmaLength", 12) \
.SetDisplay("VWMA Length", "Period for Volume Weighted MA", "Indicators")
self._std_period = self.Param("StdPeriod", 9) \
.SetDisplay("StdDev Period", "Period for standard deviation", "Indicators")
self._k1 = self.Param("K1", 0.5) \
.SetDisplay("K1", "Deviation multiplier for signal threshold", "Signal")
self._vwma = None
self._std_dev = None
self._prev_vwma = None
@property
def candle_type(self):
return self._candle_type.Value
@property
def vwma_length(self):
return self._vwma_length.Value
@property
def std_period(self):
return self._std_period.Value
@property
def k1(self):
return self._k1.Value
def OnReseted(self):
super(volume_weighted_ma_st_dev_strategy, self).OnReseted()
self._vwma = None
self._std_dev = None
self._prev_vwma = None
def OnStarted2(self, time):
super(volume_weighted_ma_st_dev_strategy, self).OnStarted2(time)
self._prev_vwma = None
self._vwma = VolumeWeightedMovingAverage()
self._vwma.Length = self.vwma_length
self._std_dev = StandardDeviation()
self._std_dev.Length = self.std_period
subscription = self.SubscribeCandles(self.candle_type)
subscription.Bind(self._vwma, self.process_candle).Start()
area = self.CreateChartArea()
if area is not None:
self.DrawCandles(area, subscription)
self.DrawIndicator(area, self._vwma)
self.DrawOwnTrades(area)
def process_candle(self, candle, vwma_value):
if candle.State != CandleStates.Finished:
return
if not self._vwma.IsFormed:
return
vwma_value = float(vwma_value)
if self._prev_vwma is None:
self._prev_vwma = vwma_value
return
diff = vwma_value - self._prev_vwma
std_result = process_float(self._std_dev, diff, candle.ServerTime, True)
if not self._std_dev.IsFormed:
self._prev_vwma = vwma_value
return
std_value = float(std_result)
k1 = float(self.k1)
filter_val = k1 * std_value
if diff > filter_val and self.Position <= 0:
self.BuyMarket()
elif diff < -filter_val and self.Position >= 0:
self.SellMarket()
self._prev_vwma = vwma_value
def CreateClone(self):
return volume_weighted_ma_st_dev_strategy()