AbcWsCci 策略
概述
AbcWsCci 策略 将日本蜡烛图中的 三白兵 与 三只黑乌鸦 形态,结合 商品通道指数(CCI) 进行信号确认。策略仅处理已经收盘的K线,通过均线判断实体是否足够强势,在多根K线动量与CCI极值同步时才入场。当CCI脱离极值区间时立即离场。
交易思路
- 计算K线实体大小的简单移动平均,用于衡量“长实体”的最低标准。
- 识别三白兵(连续三根实体强劲且中点逐步抬高的阳线)。
- 识别三只黑乌鸦(连续三根实体强劲且中点逐步降低的阴线)。
- 当出现三白兵且 CCI 低于 -50 时做多;当出现三只黑乌鸦且 CCI 高于 50 时做空。
- 当 CCI 从 -80/80 区间外返回时,平掉对应方向的仓位。
参数
| 名称 | 说明 | 默认值 |
|---|---|---|
CciPeriod |
CCI 指标的周期长度。 | 37 |
BodyAveragePeriod |
计算实体平均值的样本数量。 | 13 |
CandleType |
用于识别形态的K线周期。 | 1 小时 |
指标
- CCI:用于确认入场信号并判断离场时机。
- 实体均线:确保形态中的每根K线实体都足够强劲。
仓位管理
- 当没有多头仓位且满足三白兵 + CCI < -50 时买入开仓。
- 当没有空头仓位且满足三只黑乌鸦 + CCI > 50 时卖出开仓。
- 多头仓位在 CCI 脱离 -80/80 区间时立即平仓。
- 空头仓位在 CCI 脱离 +80/-80 区间时立即平仓。
使用说明
- 仅处理收盘后的完整K线,避免使用未完成数据。
- 适用于具有趋势或动量特征的品种。
- 建议结合止损、头寸规模等额外的风险控制措施,以适应不同的交易环境。
namespace StockSharp.Samples.Strategies;
using System;
using Ecng.Common;
using StockSharp.Algo.Indicators;
using StockSharp.Algo.Strategies;
using StockSharp.Messages;
/// <summary>
/// ABC WS CCI strategy: 3 white soldiers / 3 black crows pattern with CCI confirmation.
/// Buys after 3 bullish candles with CCI below 100, sells after 3 bearish candles with CCI above -100.
/// </summary>
public class AbcWsCciStrategy : Strategy
{
private readonly StrategyParam<DataType> _candleType;
private readonly StrategyParam<int> _cciPeriod;
private readonly StrategyParam<int> _signalCooldownCandles;
private int _bullCount;
private int _bearCount;
private int _candlesSinceTrade;
public DataType CandleType { get => _candleType.Value; set => _candleType.Value = value; }
public int CciPeriod { get => _cciPeriod.Value; set => _cciPeriod.Value = value; }
public int SignalCooldownCandles { get => _signalCooldownCandles.Value; set => _signalCooldownCandles.Value = value; }
public AbcWsCciStrategy()
{
_candleType = Param(nameof(CandleType), TimeSpan.FromMinutes(60).TimeFrame())
.SetDisplay("Candle Type", "Candle timeframe", "General");
_cciPeriod = Param(nameof(CciPeriod), 14)
.SetGreaterThanZero()
.SetDisplay("CCI Period", "CCI period for confirmation", "Indicators");
_signalCooldownCandles = Param(nameof(SignalCooldownCandles), 6)
.SetGreaterThanZero()
.SetDisplay("Signal Cooldown", "Bars to wait between trades", "Trading");
}
/// <inheritdoc />
protected override void OnReseted()
{
base.OnReseted();
_bullCount = 0;
_bearCount = 0;
_candlesSinceTrade = SignalCooldownCandles;
}
/// <inheritdoc />
protected override void OnStarted2(DateTime time)
{
base.OnStarted2(time);
_bullCount = 0;
_bearCount = 0;
_candlesSinceTrade = SignalCooldownCandles;
var cci = new CommodityChannelIndex { Length = CciPeriod };
var subscription = SubscribeCandles(CandleType);
subscription.Bind(cci, ProcessCandle).Start();
}
private void ProcessCandle(ICandleMessage candle, decimal cci)
{
if (candle.State != CandleStates.Finished) return;
if (_candlesSinceTrade < SignalCooldownCandles)
_candlesSinceTrade++;
if (candle.ClosePrice > candle.OpenPrice)
{
_bullCount++;
_bearCount = 0;
}
else if (candle.ClosePrice < candle.OpenPrice)
{
_bearCount++;
_bullCount = 0;
}
else
{
_bullCount = 0;
_bearCount = 0;
}
// Exit on CCI reversal
if (Position > 0 && cci > 200 && _candlesSinceTrade >= SignalCooldownCandles)
{
SellMarket();
_candlesSinceTrade = 0;
}
else if (Position < 0 && cci < -200 && _candlesSinceTrade >= SignalCooldownCandles)
{
BuyMarket();
_candlesSinceTrade = 0;
}
// Entry on 3 soldiers/crows pattern
if (_bullCount >= 3 && cci < 100 && Position <= 0 && _candlesSinceTrade >= SignalCooldownCandles)
{
BuyMarket();
_bullCount = 0;
_candlesSinceTrade = 0;
}
else if (_bearCount >= 3 && cci > -100 && Position >= 0 && _candlesSinceTrade >= SignalCooldownCandles)
{
SellMarket();
_bearCount = 0;
_candlesSinceTrade = 0;
}
}
}
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 CommodityChannelIndex
from StockSharp.Algo.Strategies import Strategy
class abc_ws_cci_strategy(Strategy):
def __init__(self):
super(abc_ws_cci_strategy, self).__init__()
self._cci_period = self.Param("CciPeriod", 14) \
.SetDisplay("CCI Period", "CCI period for confirmation", "Indicators")
self._signal_cooldown = self.Param("SignalCooldownCandles", 6) \
.SetDisplay("Signal Cooldown", "Bars to wait between trades", "Trading")
self._cci = None
self._bull_count = 0
self._bear_count = 0
self._candles_since_trade = 0
@property
def cci_period(self):
return self._cci_period.Value
@property
def signal_cooldown(self):
return self._signal_cooldown.Value
def OnReseted(self):
super(abc_ws_cci_strategy, self).OnReseted()
self._cci = None
self._bull_count = 0
self._bear_count = 0
self._candles_since_trade = self.signal_cooldown
def OnStarted2(self, time):
super(abc_ws_cci_strategy, self).OnStarted2(time)
self._cci = CommodityChannelIndex()
self._cci.Length = self.cci_period
self._bull_count = 0
self._bear_count = 0
self._candles_since_trade = self.signal_cooldown
subscription = self.SubscribeCandles(DataType.TimeFrame(TimeSpan.FromMinutes(60)))
subscription.Bind(self._cci, self._process_candle)
subscription.Start()
def _process_candle(self, candle, cci_value):
if candle.State != CandleStates.Finished:
return
if not self._cci.IsFormed:
return
cci_val = float(cci_value)
if self._candles_since_trade < self.signal_cooldown:
self._candles_since_trade += 1
close = float(candle.ClosePrice)
open_p = float(candle.OpenPrice)
if close > open_p:
self._bull_count += 1
self._bear_count = 0
elif close < open_p:
self._bear_count += 1
self._bull_count = 0
else:
self._bull_count = 0
self._bear_count = 0
if self.Position > 0 and cci_val > 200.0 and self._candles_since_trade >= self.signal_cooldown:
self.SellMarket()
self._candles_since_trade = 0
elif self.Position < 0 and cci_val < -200.0 and self._candles_since_trade >= self.signal_cooldown:
self.BuyMarket()
self._candles_since_trade = 0
if self._bull_count >= 3 and cci_val < 100.0 and self.Position <= 0 and self._candles_since_trade >= self.signal_cooldown:
self.BuyMarket()
self._bull_count = 0
self._candles_since_trade = 0
elif self._bear_count >= 3 and cci_val > -100.0 and self.Position >= 0 and self._candles_since_trade >= self.signal_cooldown:
self.SellMarket()
self._bear_count = 0
self._candles_since_trade = 0
def CreateClone(self):
return abc_ws_cci_strategy()