在 GitHub 上查看
Blockbuster Bollinger Breakout 策略
Blockbuster Bollinger Breakout 策略来自 MetaTrader 4 的 “BLOCKBUSTER EA”。原版通过检测价格突破布林带并超出额外偏移量来寻找反转机会。本移植版保留了关键的信号、止盈和止损设置,同时利用 StockSharp 的高层 API 实现指标绑定、蜡烛订阅和仓位管理。
核心思路
- 使用用户定义的周期和偏差构建布林带。
- 当蜡烛的收盘价高于上轨(加上偏移量)或低于下轨(减去偏移量)时触发交易信号。
- 收盘价高于上轨 + 偏移量 → 开空;收盘价低于下轨 – 偏移量 → 开多。
- 按照点数计算的止盈和止损管理仓位,与 MQL 原版一致。
所有距离和阈值都以品种的“点”为单位,并自动乘以 PriceStep。例如 3 表示 3 个最小变动价位。
细节说明
指标设置
- 使用布林带指标,输入为蜡烛收盘价(原 EA 采用开盘价,本实现选择收盘价以适配 StockSharp 的蜡烛订阅流程)。
- 参数:
BollingerPeriod、BollingerDeviation。
- 额外参数
DistancePoints 控制偏移距离。
入场条件
- 做多:
Close < LowerBand - Distance 且当前净头寸 ≤ 0。
- 做空:
Close > UpperBand + Distance 且当前净头寸 ≥ 0。
- 当方向反转时,订单数量为
TradeVolume + |Position|,确保一次只持有单方向仓位。
出场条件
- 在每根已完成蜡烛上计算未实现盈亏(点数)。
- 止盈:收益 ≥
ProfitTargetPoints。
- 止损:亏损 ≥
LossLimitPoints。
- 设置为 0 即可关闭对应功能。止盈判定优先于止损,和原程序相同。
状态管理
- 信号触发时记录进场价格,用于后续盈利计算。
- 仓位平仓后自动重置内部状态。
参数表
| 参数 |
默认值 |
说明 |
BollingerPeriod |
20 |
布林带计算周期。 |
BollingerDeviation |
2.0 |
标准差倍数。 |
DistancePoints |
3 |
超出布林带的额外距离(点)。 |
ProfitTargetPoints |
3 |
止盈阈值(点)。0 表示禁用。 |
LossLimitPoints |
20 |
止损阈值(点)。0 表示禁用。 |
TradeVolume |
1 |
下单数量,对应 MT4 的 Lots。 |
CandleType |
1 分钟 |
计算所用蜡烛类型。 |
使用建议
适用于拥有明确 PriceStep 的产品,如外汇、指数 CFD、流动性好的期货。
由于指标改用收盘价,建议在目标周期上重新回测,确认与 MT4 的表现一致。
如果界面支持图表,会显示蜡烛、布林带以及策略成交,方便观察。
策略只在完整蜡烛上做判断,保证回测与实时的一致性。
标签
- 策略类型:反转突破
- 方向:双向
- 指标:Bollinger Bands
- 止损:支持(可配置)
- 周期:短周期(默认 1 分钟)
- 复杂度:简单
using System;
using System.Collections.Generic;
using StockSharp.Algo.Indicators;
using StockSharp.Algo.Strategies;
using StockSharp.BusinessEntities;
using StockSharp.Messages;
namespace StockSharp.Samples.Strategies;
/// <summary>
/// Blockbuster Bollinger breakout strategy.
/// Buys when price closes below lower Bollinger band (mean reversion).
/// Sells when price closes above upper Bollinger band.
/// Exits at middle band.
/// </summary>
public class BlockbusterBollingerStrategy : Strategy
{
private readonly StrategyParam<int> _bollingerPeriod;
private readonly StrategyParam<decimal> _bollingerWidth;
private readonly StrategyParam<DataType> _candleType;
public int BollingerPeriod { get => _bollingerPeriod.Value; set => _bollingerPeriod.Value = value; }
public decimal BollingerWidth { get => _bollingerWidth.Value; set => _bollingerWidth.Value = value; }
public DataType CandleType { get => _candleType.Value; set => _candleType.Value = value; }
public BlockbusterBollingerStrategy()
{
_bollingerPeriod = Param(nameof(BollingerPeriod), 20)
.SetDisplay("BB Period", "Bollinger bands period", "Indicators");
_bollingerWidth = Param(nameof(BollingerWidth), 1m)
.SetDisplay("BB Width", "Bollinger bands deviation", "Indicators");
_candleType = Param(nameof(CandleType), TimeSpan.FromHours(4).TimeFrame())
.SetDisplay("Candle Type", "Candle timeframe", "General");
}
public override IEnumerable<(Security sec, DataType dt)> GetWorkingSecurities() => [(Security, CandleType)];
protected override void OnReseted() { base.OnReseted(); }
protected override void OnStarted2(DateTime time)
{
base.OnStarted2(time);
var bb = new BollingerBands { Length = BollingerPeriod, Width = BollingerWidth };
var subscription = SubscribeCandles(CandleType);
subscription
.BindEx(bb, ProcessCandle)
.Start();
}
private void ProcessCandle(ICandleMessage candle, IIndicatorValue value)
{
if (candle.State != CandleStates.Finished)
return;
if (!value.IsFinal || value.IsEmpty)
return;
var bbVal = value as BollingerBandsValue;
if (bbVal == null)
return;
var upper = bbVal.UpBand;
var lower = bbVal.LowBand;
var middle = bbVal.MovingAverage;
if (upper == null || lower == null || middle == null)
return;
var close = candle.ClosePrice;
// Breakout above upper band - momentum buy
if (close > upper.Value && Position <= 0)
{
if (Position < 0)
BuyMarket();
BuyMarket();
}
// Breakout below lower band - momentum sell
else if (close < lower.Value && Position >= 0)
{
if (Position > 0)
SellMarket();
SellMarket();
}
// Exit long at middle
else if (Position > 0 && close < middle.Value)
{
SellMarket();
}
// Exit short at middle
else if (Position < 0 && close > middle.Value)
{
BuyMarket();
}
}
}
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 BollingerBands
from StockSharp.Algo.Strategies import Strategy
class blockbuster_bollinger_strategy(Strategy):
def __init__(self):
super(blockbuster_bollinger_strategy, self).__init__()
self._bollinger_period = self.Param("BollingerPeriod", 20) \
.SetDisplay("BB Period", "Bollinger bands period", "Indicators")
self._bollinger_width = self.Param("BollingerWidth", 1.0) \
.SetDisplay("BB Width", "Bollinger bands deviation", "Indicators")
self._candle_type = self.Param("CandleType", DataType.TimeFrame(TimeSpan.FromHours(4))) \
.SetDisplay("Candle Type", "Candle timeframe", "General")
@property
def bollinger_period(self):
return self._bollinger_period.Value
@property
def bollinger_width(self):
return self._bollinger_width.Value
@property
def candle_type(self):
return self._candle_type.Value
def OnReseted(self):
super(blockbuster_bollinger_strategy, self).OnReseted()
def OnStarted2(self, time):
super(blockbuster_bollinger_strategy, self).OnStarted2(time)
bb = BollingerBands()
bb.Length = self.bollinger_period
bb.Width = self.bollinger_width
subscription = self.SubscribeCandles(self.candle_type)
subscription.BindEx(bb, self.process_candle).Start()
def process_candle(self, candle, bb_value):
if candle.State != CandleStates.Finished:
return
if not bb_value.IsFinal or bb_value.IsEmpty:
return
if bb_value.UpBand is None or bb_value.LowBand is None or bb_value.MovingAverage is None:
return
upper = float(bb_value.UpBand)
lower = float(bb_value.LowBand)
middle = float(bb_value.MovingAverage)
close = float(candle.ClosePrice)
if close > upper and self.Position <= 0:
if self.Position < 0:
self.BuyMarket()
self.BuyMarket()
elif close < lower and self.Position >= 0:
if self.Position > 0:
self.SellMarket()
self.SellMarket()
elif self.Position > 0 and close < middle:
self.SellMarket()
elif self.Position < 0 and close > middle:
self.BuyMarket()
def CreateClone(self):
return blockbuster_bollinger_strategy()