Skip to content
All library documents

Bollinger Channel and CCI Entries with ATR Trailing Stops

Article Strategy library · Author: str =

Summary

This strategy combines Bollinger channels and the Commodity Channel Index (CCI) to place entries on 15-minute bars. When flat, a positive CCI prompts a buy order at the upper Bollinger level, while a negative CCI prompts a short order at the lower level. The supplied defaults are an 18-bar channel with a 3.4 deviation, a 10-bar CCI, and a 30-bar ATR. Position size defaults to one unit.

For an open long, the strategy tracks the highest price since entry and places a stop five-point-two ATRs below that high. For a short, it tracks the lowest price and places a stop the same ATR multiple above it. The document is source code and describes the mechanics, but provides no backtest settings or performance evidence. Its behavior therefore cannot establish profitability; the wide ATR multiple and indicator settings would need evaluation across instruments and market conditions.

Key ideas

  • Entries are conditioned on the CCI sign and use the corresponding upper or lower Bollinger channel as the order level.
  • The strategy processes signals on 15-minute bars and uses an 18-bar, 3.4-deviation Bollinger channel by default.
  • Long and short exits use ATR-based trailing stops that follow the trade's favorable price extreme.
  • The default ATR window is 30 bars and the stop distance is 5.2 times ATR.
  • The document gives implementation details but no backtest results or evidence of profitability.

Tags

Full text
# BollChannelStrategy


# BollChannelStrategy









布林通道策略。

用布林通道和CCI开仓,并用ATR止损的策略。

## Source (MIT)

```python
"""布林通道策略。"""

from vnpy_ctastrategy import (
    CtaTemplate,
    StopOrder,
    TickData,
    BarData,
    TradeData,
    OrderData,
    BarGenerator,
    ArrayManager,
)


class BollChannelStrategy(CtaTemplate):
    """用布林通道和CCI开仓,并用ATR止损的策略。"""

    author: str = "用Python的交易员"

    boll_window: int = 18
    boll_dev: float = 3.4
    cci_window: int = 10
    atr_window: int = 30
    sl_multiplier: float = 5.2
    fixed_size: int = 1

    boll_up: float = 0
    boll_down: float = 0
    cci_value: float = 0
    atr_value: float = 0
    intra_trade_high: float = 0
    intra_trade_low: float = 0
    long_stop: float = 0
    short_stop: float = 0

    parameters: list[str] = [
        "boll_window",
        "boll_dev",
        "cci_window",
        "atr_window",
        "sl_multiplier",
        "fixed_size"
    ]
    variables: list[str] = [
        "boll_up",
        "boll_down",
        "cci_value",
        "atr_value",
        "intra_trade_high",
        "intra_trade_low",
        "long_stop",
        "short_stop"
    ]

    def on_init(self) -> None:
        """
        策略初始化完成时的回调。
        """
        self.write_log("策略初始化")

        self.bg: BarGenerator = BarGenerator(self.on_bar, 15, self.on_15min_bar)
        self.am: ArrayManager = ArrayManager()

        self.load_bar(10)

    def on_start(self) -> None:
        """
        策略启动时的回调。
        """
        self.write_log("策略启动")

    def on_stop(self) -> None:
        """
        策略停止时的回调。
        """
        self.write_log("策略停止")

    def on_tick(self, tick: TickData) -> None:
        """
        新 Tick 数据更新时的回调。
        """
        self.bg.update_tick(tick)

    def on_bar(self, bar: BarData) -> None:
        """
        新 K 线数据更新时的回调。
        """
        self.bg.update_bar(bar)

    def on_15min_bar(self, bar: BarData) -> None:
        """在15分钟K线上按布林通道和CCI开仓,并用ATR跟踪止损。"""
        self.cancel_all()

        am: ArrayManager = self.am
        am.update_bar(bar)
        if not am.inited:
            return

        self.boll_up, self.boll_down = am.boll(self.boll_window, self.boll_dev)
        self.cci_value = am.cci(self.cci_window)
        self.atr_value = am.atr(self.atr_window)

        if self.pos == 0:
            self.intra_trade_high = bar.high_price
            self.intra_trade_low = bar.low_price

            if self.cci_value > 0:
                self.buy(self.boll_up, self.fixed_size, True)
            elif self.cci_value < 0:
                self.short(self.boll_down, self.fixed_size, True)

        elif self.pos > 0:
            self.intra_trade_high = max(self.intra_trade_high, bar.high_price)
            self.intra_trade_low = bar.low_price

            self.long_stop = self.intra_trade_high - self.atr_value * self.sl_multiplier
            self.sell(self.long_stop, abs(self.pos), True)

        elif self.pos < 0:
            self.intra_trade_high = bar.high_price
            self.intra_trade_low = min(self.intra_trade_low, bar.low_price)

            self.short_stop = self.intra_trade_low + self.atr_value * self.sl_multiplier
            self.cover(self.short_stop, abs(self.pos), True)

        self.put_event()

    def on_order(self, order: OrderData) -> None:
        """
        新委托数据更新时的回调。
        """
        pass

    def on_trade(self, trade: TradeData) -> None:
        """
        新成交数据更新时的回调。
        """
        self.put_event()

    def on_stop_order(self, stop_order: StopOrder) -> None:
        """
        停止单更新时的回调。
        """
        pass

```

Shown in full with attribution under the source's licence. Licence: MIT

This summary was written by Stratmill's research agent from the original; it is not a copy of the source.