Bollinger Bands Breakout

An intraday strategy that places BUY STOP and SELL STOP orders at the Bollinger Bands each morning, catching breakouts in either direction. All positions are closed at 21:00. This example demonstrates several intermediate patterns you’ll use in real strategies.

Difficulty

Intermediate

Timeframe

Hourly (H1) + Daily (D1)

Order Type

Pending STOP Orders

Concepts Covered

Multi-timeframe, pending orders, time-based exits, daily state tracking

Strategy Logic

1. Each morning at 08:00, calculate the Bollinger Bands (20-period, 2.5 std dev) on H1 bars
2. Place a BUY STOP at the upper band and a SELL STOP at the lower band
3. If price breaks out in either direction, the pending order fills automatically
4. At 21:00, close all open positions and cancel any unfilled pending orders
5. Reset the daily state and repeat the next morning

Step-by-Step Walkthrough

1. Multi-timeframe setup

signal_timeframe = StrategyTimeframes.ONE_HOUR
daily_timeframe = StrategyTimeframes.ONE_DAY
strategy_timeframes = [signal_timeframe, daily_timeframe]

Registering multiple timeframes means your signal engine gets called on every H1 bar and every D1 bar. Use event.timeframe to filter which timeframe you want to act on:

if event.timeframe == signal_timeframe:
    # Only run indicator logic on H1 bars

Why Register Multiple Timeframes?

Even though the indicator logic only runs on H1, we register D1 as well so the signal engine gets called at the daily close. This lets us use time-based logic (like the 21:00 exit) that fires on every bar regardless of timeframe.

2. Daily state tracking with module-level variables

orders_placed_today: dict[str, bool] = {
    symbol: False for symbol in symbols_to_trade
}
current_trading_date: dict[str, datetime | None] = {
    symbol: None for symbol in symbols_to_trade
}

Module-level dictionaries persist across bar events. On each bar, check if the date changed and reset:

if current_trading_date[symbol] != current_date:
    current_trading_date[symbol] = current_date
    orders_placed_today[symbol] = False

This pattern is essential for any strategy that needs “once per day” behavior — order placement, daily rebalancing, etc.

3. End-of-day exit: time-based position management

if current_time >= time(close_hour, close_minute):
    if open_positions['TOTAL'] > 0:
        modules.EXECUTION_ENGINE.close_all_strategy_positions()
    if pending_orders['TOTAL'] > 0:
        modules.EXECUTION_ENGINE.cancel_all_strategy_pending_orders()
    return

Use Python’s time() objects for intraday time comparisons. The return after the close block prevents any new orders from being placed after the exit window.

Pending Order Cleanup

Always cancel unfilled pending orders when you close positions. If you forget, orphaned STOP orders can trigger unexpected entries.

4. Morning entry: calculating Bollinger Bands

close = indicator_bars.select('close').to_numpy().flatten()
upper, middle, lower = BollingerBands.compute(close, bb_period, bb_std_dev)

current_upper = upper[-1]
current_lower = lower[-1]

BollingerBands.compute() returns three NumPy arrays — upper band, middle band (SMA), and lower band. We use the latest values as our breakout levels.

5. Pending STOP orders

Unlike market orders (which execute immediately), STOP orders sit in the order book until price reaches the trigger level:

# BUY STOP at upper band
signal_events.append(SignalEvent(
    symbol=symbol,
    ...
    signal_type=SignalType.BUY,
    order_type=OrderType.STOP,       # Pending order
    order_price=upper_breakout,      # Triggers when price reaches this level
    ...
))

# SELL STOP at lower band
signal_events.append(SignalEvent(
    symbol=symbol,
    ...
    signal_type=SignalType.SELL,
    order_type=OrderType.STOP,
    order_price=lower_breakout,
    ...
))

PyEventBT supports three order types:

Order Type	Behavior
OrderType.MARKET	Execute immediately at current bid/ask
OrderType.LIMIT	Execute when price reaches target (better price)
OrderType.STOP	Execute when price breaks through level (breakout)

6. Returning multiple signals

A single call to the signal engine can return multiple SignalEvents. In this strategy, we return both a BUY STOP and a SELL STOP — only the one that gets triggered by price will execute.

signal_events = []
signal_events.append(SignalEvent(..., signal_type=SignalType.BUY, ...))
signal_events.append(SignalEvent(..., signal_type=SignalType.SELL, ...))
return signal_events

Adapt This Strategy

Variation	What to change
Tighter/wider bands	Adjust bb_std_dev (try 1.5 for tighter, 3.0 for wider)
Different trading window	Change order_placement_hour and close_hour
Add stop-loss	Set sl to a non-zero value based on middle band or ATR
Cancel opposite on fill	After a BUY fills, cancel the pending SELL (use a hook or state tracking)
Multiple symbols	Add to symbols_to_trade — the per-symbol dicts handle state automatically

Full Code

Copy this into a .py file and run it directly.

from pyeventbt import (
    Strategy,
    BarEvent,
    SignalEvent,
    Modules,
    StrategyTimeframes,
    PassthroughRiskConfig,
    MinSizingConfig
)
from pyeventbt.events.events import OrderType, SignalType
from pyeventbt.indicators.indicators import BollingerBands

from datetime import datetime, time
from decimal import Decimal
import logging
import numpy as np

logger = logging.getLogger("pyeventbt")

# ── Configuration ────────────────────────────────────────────────
strategy_id = "bbands_breakout"
strategy = Strategy(logging_level=logging.INFO)

# This strategy listens on TWO timeframes
signal_timeframe = StrategyTimeframes.ONE_HOUR
daily_timeframe = StrategyTimeframes.ONE_DAY
strategy_timeframes = [signal_timeframe, daily_timeframe]

symbols_to_trade = ['EURUSD']
starting_capital = 100000

# Strategy Parameters
bb_period = 20
bb_std_dev = 2.5
close_hour = 21
close_minute = 0
order_placement_hour = 8
order_placement_minute = 0

# Daily state tracking (per symbol)
orders_placed_today: dict[str, bool] = {
    symbol: False for symbol in symbols_to_trade
}
current_trading_date: dict[str, datetime | None] = {
    symbol: None for symbol in symbols_to_trade
}


# ── Signal Engine ────────────────────────────────────────────────
@strategy.custom_signal_engine(
    strategy_id=strategy_id,
    strategy_timeframes=strategy_timeframes
)
def bbands_breakout(event: BarEvent, modules: Modules):
    """
    Bollinger Bands Breakout:
    - Place pending STOP orders at upper/lower bands each morning
    - Close everything at 21:00
    """

    symbol = event.symbol
    signal_events = []

    current_time = event.datetime.time()
    current_date = event.datetime.date()

    # ── Reset daily state on new day ─────────────────────────
    if current_trading_date[symbol] != current_date:
        current_trading_date[symbol] = current_date
        orders_placed_today[symbol] = False

    # ── Check current state ──────────────────────────────────
    open_positions = modules.PORTFOLIO \
        .get_number_of_strategy_open_positions_by_symbol(symbol)
    pending_orders = modules.PORTFOLIO \
        .get_number_of_strategy_pending_orders_by_symbol(symbol)

    # ── End-of-day exit: close everything at 21:00 ───────────
    if current_time >= time(close_hour, close_minute):
        if open_positions['TOTAL'] > 0:
            modules.EXECUTION_ENGINE.close_all_strategy_positions()

        if pending_orders['TOTAL'] > 0:
            modules.EXECUTION_ENGINE.cancel_all_strategy_pending_orders()

        return

    # ── Morning entry: place breakout orders ─────────────────
    if (current_time >= time(order_placement_hour, order_placement_minute)
            and not orders_placed_today[symbol]
            and pending_orders['TOTAL'] == 0
            and event.timeframe == signal_timeframe):

        # Get bars for Bollinger Bands calculation
        bars_needed = bb_period + 10
        indicator_bars = modules.DATA_PROVIDER.get_latest_bars(
            symbol, signal_timeframe, bars_needed
        )

        if indicator_bars is None or indicator_bars.height < bars_needed:
            return

        # Calculate Bollinger Bands
        close = indicator_bars.select('close').to_numpy().flatten()
        upper, middle, lower = BollingerBands.compute(
            close, bb_period, bb_std_dev
        )

        current_upper = upper[-1]
        current_lower = lower[-1]

        if np.isnan(current_upper) or np.isnan(current_lower):
            return

        upper_breakout = Decimal(str(current_upper))
        lower_breakout = Decimal(str(current_lower))

        # Signal timing
        if modules.TRADING_CONTEXT == "BACKTEST":
            time_generated = event.datetime + signal_timeframe.to_timedelta()
        else:
            time_generated = datetime.now()

        # BUY STOP at upper band
        signal_events.append(SignalEvent(
            symbol=symbol,
            time_generated=time_generated,
            strategy_id=strategy_id,
            signal_type=SignalType.BUY,
            order_type=OrderType.STOP,
            order_price=upper_breakout,
            sl=Decimal('0.0'),
            tp=Decimal('0.0'),
        ))

        # SELL STOP at lower band
        signal_events.append(SignalEvent(
            symbol=symbol,
            time_generated=time_generated,
            strategy_id=strategy_id,
            signal_type=SignalType.SELL,
            order_type=OrderType.STOP,
            order_price=lower_breakout,
            sl=Decimal('0.0'),
            tp=Decimal('0.0'),
        ))

        orders_placed_today[symbol] = True

    return signal_events


# ── Sizing & Risk ────────────────────────────────────────────────
strategy.configure_predefined_sizing_engine(MinSizingConfig())
strategy.configure_predefined_risk_engine(PassthroughRiskConfig())


# ── Run Backtest ─────────────────────────────────────────────────
backtest = strategy.backtest(
    strategy_id=strategy_id,
    initial_capital=starting_capital,
    symbols_to_trade=symbols_to_trade,
    csv_dir=None,
    backtest_name=strategy_id,
    start_date=datetime(2020, 1, 1),
    end_date=datetime(2023, 12, 1),
    export_backtest_parquet=False,
    account_currency='USD'
)

backtest.plot()