Maximus vX Lite 策略
本策略尝试交易短期盘整区域的突破。它分析 15 分钟K线,当价格离开这些紧凑区间一定距离时开仓。
策略逻辑
- 对每根已完成的 15 分钟K线,计算最近 40 根K线的最高价和最低价。
- 如果高低差小于 Range 参数,则认为形成了盘整区间。
- 在没有新交易的 Delay Open 小时后,价格突破上边界并超过 Distance 点时做多;突破下边界并超过 Distance 点时做空。
- 开仓后设置固定的 Stop Loss,并应用 Trail 点的移动止损。
- 经过 Period 小时后重新计算盘整区间。
参数
DelayOpen– 开启新交易前等待的小时数。Distance– 突破盘整边界的点数。Period– 重新计算盘整区间的小时数。Range– 盘整区间允许的最大宽度(点)。StopLoss– 初始止损距离(点)。Trail– 移动止损距离(点)。
说明
该策略仅使用高级 API:通过 SubscribeCandles 订阅K线,并使用 Bind 绑定指标。订单通过 BuyMarket 和 SellMarket 方法发送。源代码中的注释均为英文。
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>
/// Breakout strategy based on consolidation zones.
/// </summary>
public class MaximusVxLiteStrategy : Strategy
{
private readonly StrategyParam<int> _lookback;
private readonly StrategyParam<DataType> _candleType;
private decimal _prevHigh;
private decimal _prevLow;
private bool _hasPrev;
public int Lookback { get => _lookback.Value; set => _lookback.Value = value; }
public DataType CandleType { get => _candleType.Value; set => _candleType.Value = value; }
public MaximusVxLiteStrategy()
{
_lookback = Param(nameof(Lookback), 20)
.SetDisplay("Lookback", "Highest/Lowest lookback period", "General");
_candleType = Param(nameof(CandleType), TimeSpan.FromHours(4).TimeFrame())
.SetDisplay("Candle Type", "Time frame for candles", "General");
}
public override IEnumerable<(Security sec, DataType dt)> GetWorkingSecurities()
=> [(Security, CandleType)];
protected override void OnReseted()
{
base.OnReseted();
_prevHigh = 0;
_prevLow = 0;
_hasPrev = false;
}
protected override void OnStarted2(DateTime time)
{
base.OnStarted2(time);
var highest = new Highest { Length = Lookback };
var lowest = new Lowest { Length = Lookback };
SubscribeCandles(CandleType).Bind(highest, lowest, ProcessCandle).Start();
}
private void ProcessCandle(ICandleMessage candle, decimal highest, decimal lowest)
{
if (candle.State != CandleStates.Finished) return;
if (!_hasPrev)
{
_prevHigh = highest;
_prevLow = lowest;
_hasPrev = true;
return;
}
var close = candle.ClosePrice;
// Breakout above previous highest
if (close > _prevHigh && Position <= 0)
{
if (Position < 0) BuyMarket();
BuyMarket();
}
// Breakout below previous lowest
else if (close < _prevLow && Position >= 0)
{
if (Position > 0) SellMarket();
SellMarket();
}
_prevHigh = highest;
_prevLow = lowest;
}
}
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 Highest, Lowest
from StockSharp.Algo.Strategies import Strategy
class maximus_vx_lite_strategy(Strategy):
def __init__(self):
super(maximus_vx_lite_strategy, self).__init__()
self._lookback = self.Param("Lookback", 20) \
.SetDisplay("Lookback", "Highest/Lowest lookback period", "General")
self._candle_type = self.Param("CandleType", DataType.TimeFrame(TimeSpan.FromHours(4))) \
.SetDisplay("Candle Type", "Time frame for candles", "General")
self._prev_high = 0.0
self._prev_low = 0.0
self._has_prev = False
@property
def lookback(self):
return self._lookback.Value
@property
def candle_type(self):
return self._candle_type.Value
def OnReseted(self):
super(maximus_vx_lite_strategy, self).OnReseted()
self._prev_high = 0.0
self._prev_low = 0.0
self._has_prev = False
def OnStarted2(self, time):
super(maximus_vx_lite_strategy, self).OnStarted2(time)
highest = Highest()
highest.Length = self.lookback
lowest = Lowest()
lowest.Length = self.lookback
self.SubscribeCandles(self.candle_type).Bind(highest, lowest, self.process_candle).Start()
def process_candle(self, candle, highest, lowest):
if candle.State != CandleStates.Finished:
return
hv = float(highest)
lv = float(lowest)
if not self._has_prev:
self._prev_high = hv
self._prev_low = lv
self._has_prev = True
return
close = float(candle.ClosePrice)
if close > self._prev_high and self.Position <= 0:
if self.Position < 0:
self.BuyMarket()
self.BuyMarket()
elif close < self._prev_low and self.Position >= 0:
if self.Position > 0:
self.SellMarket()
self.SellMarket()
self._prev_high = hv
self._prev_low = lv
def CreateClone(self):
return maximus_vx_lite_strategy()