Article

VWAP Anchored Breakout Strategy A Backtrader Implementation and Rolling Backtest

This article explores a trading strategy built with backtrader that identifies and acts on price breakouts, using Volume Weighted Average Price (VWAP) as a key anchoring and confirmation tool. The strategy aims to capture strong directional moves confirmed by volume, trend strength (ADX), and volatility expansion (ATR), while managing risk with trailing stops.

1. Imports and Strategy Class Definition

The strategy begins by importing necessary libraries and defining the VWAPAnchoredBreakoutStrategy class, inheriting from bt.Strategy.

import backtrader as bt
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, time

class VWAPAnchoredBreakoutStrategy(bt.Strategy):
    # Parameters for the strategy
    params = (
        # VWAP Parameters
        ('vwap_session_length', 7),    # Session length for VWAP calculation
        ('vwap_weekly_length', 28),    # Weekly VWAP length
        
        # Breakout Parameters
        ('breakout_lookback', 7),      # Lookback period for prior high/low
        ('adx_threshold', 25),         # ADX > 20 for trend confirmation
        ('adx_period', 7),             # ADX calculation period
        
        # Volume and ATR Confirmation
        ('volume_multiplier', 1.1),    # Volume > 1.5x average
        ('volume_period', 7),          # Volume average period
        ('atr_period', 14),            # ATR period
        ('atr_expansion_threshold', 1.1), # ATR expansion threshold
        ('atr_expansion_period', 7),   # Period to compare ATR expansion
        
        # Trailing Stop Parameters
        ('trailing_stop_atr_multiplier', 5.0), # Trailing stop distance
        ('initial_stop_atr_multiplier', 1.),  # Initial stop loss
        
        # Risk Management
        ('position_size_pct', 0.95),   # Position size percentage
        ('printlog', True),
    )

Explanation:

import backtrader as bt: Imports the core backtrader library.
import yfinance as yf: Used for downloading historical market data.
import pandas as pd, import numpy as np: Standard libraries for data manipulation.
import matplotlib.pyplot as plt: For plotting results (though not directly used in the provided cerebro.run() block, it’s common for analysis).
from datetime import datetime, time: For handling dates and times.
class VWAPAnchoredBreakoutStrategy(bt.Strategy):: Defines our custom strategy.
params: A tuple defining configurable parameters for the strategy. These allow easy tuning without changing the core logic. Parameters include lengths for VWAP, lookback periods for breakouts, thresholds for ADX, multipliers for volume and ATR, and settings for stop losses and position sizing.

2. Logging and Initialization (`init`, `log`)

The __init__ method sets up the data feeds and calculates various technical indicators used in the strategy. The log method provides a consistent way to print messages during backtesting.

    def log(self, txt, dt=None, doprint=False):
        if self.params.printlog or doprint:
            dt = dt or self.datas[0].datetime.date(0)
            print(f'{dt.isoformat()} - {txt}')

    def __init__(self):
        self.dataclose = self.datas[0].close
        self.datahigh = self.datas[0].high
        self.datalow = self.datas[0].low
        self.datavolume = self.datas[0].volume
        self.order = None # To keep track of pending orders
        
        # Technical Indicators
        self.atr = bt.indicators.ATR(period=self.params.atr_period)
        self.adx = bt.indicators.ADx(period=self.params.adx_period) # Note: Backtrader's ADX is ADX by default. Using ADx to be explicit.
        
        # VWAP Calculations
        # Session VWAP (shorter term)
        typical_price = (self.datahigh + self.datalow + self.dataclose) / 3
        self.vwap_session = bt.indicators.WeightedAverage(
            typical_price, self.datavolume, 
            period=self.params.vwap_session_length
        )
        
        # Weekly VWAP (longer term)
        self.vwap_weekly = bt.indicators.WeightedAverage(
            typical_price, self.datavolume, 
            period=self.params.vwap_weekly_length
        )
        
        # Volume indicators
        self.volume_sma = bt.indicators.SMA(self.datavolume, period=self.params.volume_period)
        
        # ATR expansion detection
        self.atr_sma = bt.indicators.SMA(self.atr, period=self.params.atr_expansion_period)
        
        # Prior session high/low tracking
        self.prior_high = bt.indicators.Highest(self.datahigh, period=self.params.breakout_lookback)
        self.prior_low = bt.indicators.Lowest(self.datalow, period=self.params.breakout_lookback)
        
        # Tracking variables for current trade
        self.entry_price = None
        self.stop_price = None
        self.trail_price = None
        self.position_type = None  # 1 for long, -1 for short
        self.breakout_confirmed = False

Explanation:

log(self, txt, dt=None, doprint=False): A utility function to print logs, prepending the current date from the data feed. doprint allows overriding printlog parameter for specific important messages.
__init__(self):
- self.dataclose, self.datahigh, self.datalow, self.datavolume: Aliases to access the current bar’s close, high, low, and volume data more easily. self.datas[0] refers to the first (and in this case, only) data feed.
- self.order = None: Initializes a variable to hold any active order, preventing new orders while one is pending.
- self.atr = bt.indicators.ATR(...): Calculates the Average True Range (ATR), a measure of market volatility.
- self.adx = bt.indicators.ADX(...): Calculates the Average Directional Index (ADX), which measures the strength of a trend.
- VWAP Calculations:
  - typical_price = (self.datahigh + self.datalow + self.dataclose) / 3: Computes the typical price for each bar.
  - self.vwap_session and self.vwap_weekly: Calculate Volume Weighted Average Price over a shorter (“session”) and longer (“weekly”) period. bt.indicators.WeightedAverage calculates VWAP when fed typical_price and volume.
- self.volume_sma: A Simple Moving Average (SMA) of volume, used to compare current volume against its recent average.
- self.atr_sma: An SMA of ATR, used to detect if current volatility is expanding relative to recent volatility.
- self.prior_high, self.prior_low: bt.indicators.Highest and bt.indicators.Lowest track the highest high and lowest low over a specified lookback period, excluding the current bar (accessed with [-1]).
- Tracking Variables: entry_price, stop_price, trail_price, position_type, breakout_confirmed are initialized to keep track of trade-specific information.

3. Order and Trade Notifications (`notify_order`, `notify_trade`)

These methods are backtrader callbacks that get invoked when an order’s status changes or a trade is closed. They are useful for logging and debugging.

    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]:
            return # Order submitted/accepted - nothing to do yet

        if order.status in [order.Completed]:
            if order.isbuy():
                self.log(f'BUY EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}')
            elif order.issell():
                self.log(f'SELL EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}')

        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log('Order Canceled/Margin/Rejected')

        self.order = None # Clear pending order
    
    def notify_trade(self, trade):
        if not trade.isclosed:
            return # Trade is not yet closed

        self.log(f'TRADE CLOSED - P&L: {trade.pnlcomm:.2f}')

Explanation:

notify_order(self, order):
- Checks the order.status.
- If Submitted or Accepted, it means the order is in progress, so we do nothing.
- If Completed, it logs the execution details (price, cost) for buy or sell orders.
- If Canceled, Margin, or Rejected, it logs the failure.
- Finally, self.order = None clears the tracking variable, allowing new orders to be placed.
notify_trade(self, trade):
- Called when a trade (not just an order) is opened or closed.
- if not trade.isclosed: return ensures we only process closed trades.
- Logs the profit/loss (pnlcomm - P&L with commission) of the closed trade.

4. Breakout and Confirmation Checks

These are the core logic functions that determine if the market conditions align with the strategy’s entry criteria.

    def check_vwap_alignment(self, breakout_direction):
        """Check if current price vs VWAP supports the breakout direction"""
        if len(self.vwap_session) < 1 or len(self.vwap_weekly) < 1:
            return False
            
        current_price = self.dataclose[0]
        session_vwap = self.vwap_session[0]
        weekly_vwap = self.vwap_weekly[0]
        
        if breakout_direction == "LONG":
            # For long breakouts, price should be above both VWAPs (or at least session VWAP)
            return current_price > session_vwap # A stricter check could be current_price > weekly_vwap as well
        elif breakout_direction == "SHORT":
            # For short breakouts, price should be below both VWAPs (or at least session VWAP)
            return current_price < session_vwap # A stricter check could be current_price < weekly_vwap as well
            
        return False

    def check_volume_confirmation(self):
        """Check if volume > 1.1x average"""
        if len(self.datavolume) < 1 or len(self.volume_sma) < 1:
            return False
            
        current_volume = self.datavolume[0]
        avg_volume = self.volume_sma[0]
        
        return current_volume > (avg_volume * self.params.volume_multiplier)

    def check_atr_expansion(self):
        """Check if ATR is expanding (current ATR > recent average)"""
        if len(self.atr) < 1 or len(self.atr_sma) < 1:
            return False
            
        current_atr = self.atr[0]
        avg_atr = self.atr_sma[0]
        
        return current_atr > (avg_atr * self.params.atr_expansion_threshold)

    def check_breakout_conditions(self):
        """Check for breakout setup conditions"""
        if len(self.dataclose) < self.params.breakout_lookback:
            return False, None
            
        current_price = self.dataclose[0]
        current_high = self.datahigh[0]
        current_low = self.datalow[0]
        
        # Get prior session high/low (from yesterday, not including today)
        prior_session_high = self.prior_high[-1]  # Previous bar's highest
        prior_session_low = self.prior_low[-1]    # Previous bar's lowest
        
        breakout_direction = None
        
        # Check for breakout of prior high
        if current_high > prior_session_high:
            breakout_direction = "LONG"
            self.log(f'HIGH BREAKOUT DETECTED: Current High {current_high:.2f} > Prior High {prior_session_high:.2f}')
            
        # Check for breakout of prior low
        elif current_low < prior_session_low:
            breakout_direction = "SHORT"
            self.log(f'LOW BREAKOUT DETECTED: Current Low {current_low:.2f} < Prior Low {prior_session_low:.2f}')
            
        if breakout_direction is None:
            return False, None
            
        # Check VWAP alignment
        if not self.check_vwap_alignment(breakout_direction):
            self.log(f'VWAP ALIGNMENT FAILED for {breakout_direction}')
            return False, None
            
        # Check ADX > threshold
        if len(self.adx) < 1 or self.adx[0] <= self.params.adx_threshold:
            self.log(f'ADX TOO LOW: {self.adx[0]:.1f} <= {self.params.adx_threshold}')
            return False, None
            
        # Check volume confirmation
        if not self.check_volume_confirmation():
            self.log(f'VOLUME CONFIRMATION FAILED: {self.datavolume[0]} vs {self.volume_sma[0] * self.params.volume_multiplier:.0f}')
            return False, None
            
        # Check ATR expansion
        if not self.check_atr_expansion():
            self.log(f'ATR EXPANSION FAILED: {self.atr[0]:.4f} vs {self.atr_sma[0] * self.params.atr_expansion_threshold:.4f}')
            return False, None
            
        return True, breakout_direction

Explanation:

check_vwap_alignment(self, breakout_direction):
- Ensures enough data is available for VWAP calculations.
- For a “LONG” breakout, it checks if the current close price is above the session VWAP, confirming upward momentum aligned with volume-weighted price.
- For a “SHORT” breakout, it checks if the current close price is below the session VWAP.
check_volume_confirmation(self):
- Checks if the current bar’s volume is significantly higher (e.g., 1.1 times) than its recent average, indicating strong participation in the breakout.
check_atr_expansion(self):
- Determines if the current ATR is expanding (e.g., 1.1 times its recent average), suggesting increasing volatility, which often accompanies true breakouts.
check_breakout_conditions(self): This is the main entry condition checker.
- It first checks if there’s enough historical data for the lookback period.
- It identifies a potential breakout by comparing the current high/low with the prior_high/prior_low from the lookback period (using [-1] to get the value from the previous bar, representing the prior session’s extreme).
- It then sequentially calls the confirmation functions:
  - check_vwap_alignment()
  - Checks ADX (trend strength) against a threshold. A high ADX (e.g., > 25) indicates a strong trend.
  - check_volume_confirmation()
  - check_atr_expansion()
- Only if all conditions are met does it return True along with the breakout_direction.

5. Exit Conditions and Position Sizing (`update_trailing_stop`, `check_exit_conditions`, `calculate_position_size`)

These functions manage open positions, including stop-loss and trailing stop mechanisms, and handle position sizing.

    def update_trailing_stop(self):
        """Update trailing stop based on ATR"""
        if not self.position or self.trail_price is None:
            return
            
        current_price = self.dataclose[0]
        atr_value = self.atr[0]
        trail_distance = self.params.trailing_stop_atr_multiplier * atr_value
        
        if self.position_type == 1:  # Long position
            new_trail = current_price - trail_distance
            if new_trail > self.trail_price: # Move stop up if price moves favorably
                self.trail_price = new_trail
                self.log(f'TRAIL UPDATED (Long): New trail stop at {self.trail_price:.2f}')
                
        elif self.position_type == -1:  # Short position
            new_trail = current_price + trail_distance
            if new_trail < self.trail_price: # Move stop down if price moves favorably
                self.trail_price = new_trail
                self.log(f'TRAIL UPDATED (Short): New trail stop at {self.trail_price:.2f}')

    def check_exit_conditions(self):
        """Check for exit conditions"""
        if not self.position:
            return False, None
            
        current_price = self.dataclose[0]
        
        # Check initial stop loss
        if self.stop_price is not None:
            if self.position_type == 1 and current_price <= self.stop_price:
                return True, "STOP_LOSS"
            elif self.position_type == -1 and current_price >= self.stop_price:
                return True, "STOP_LOSS"
                
        # Check trailing stop
        if self.trail_price is not None:
            if self.position_type == 1 and current_price <= self.trail_price:
                return True, "TRAILING_STOP"
            elif self.position_type == -1 and current_price >= self.trail_price:
                return True, "TRAILING_STOP"
                
        # Check VWAP mean reversion (optional exit condition)
        if len(self.vwap_session) >= 1:
            session_vwap = self.vwap_session[0]
            if self.position_type == 1 and current_price < session_vwap: # Price falls below session VWAP for long
                return True, "VWAP_REVERSION"
            elif self.position_type == -1 and current_price > session_vwap: # Price rises above session VWAP for short
                return True, "VWAP_REVERSION"
                
        return False, None

    def calculate_position_size(self):
        """Position sizing handled by PercentSizer"""
        return None  # Not used - PercentSizer handles this

Explanation:

update_trailing_stop(self):
- If a position is open, it calculates a new trailing stop price based on the current price and a multiple of ATR.
- For a long position, the trailing stop is moved higher if the price goes up.
- For a short position, the trailing stop is moved lower if the price goes down. This locks in profits.
check_exit_conditions(self):
- Checks if the current price has hit the initial_stop_atr_multiplier based stop loss.
- Checks if the current price has hit the trailing_stop_atr_multiplier based trailing stop.
- Includes an optional VWAP mean reversion exit: if the price crosses back below (for long) or above (for short) the session VWAP, it triggers an exit, assuming the breakout momentum has faded.
calculate_position_size(self): This strategy uses bt.sizers.PercentSizer externally, so this method is a placeholder and returns None.

6. The `next` Method: The Heart of the Strategy

The next method is called by backtrader for each bar of data. It contains the main trading logic.

    def next(self):
        if self.order:
            return # A pending order exists, do nothing

        # Skip if not enough data for all indicators to calculate
        required_data = max(
            self.params.breakout_lookback,
            self.params.vwap_session_length,
            self.params.adx_period,
            self.params.atr_period,
            self.params.volume_period, # Ensure volume_period is included here for correct indexing
            self.params.atr_expansion_period # Ensure atr_expansion_period is included here for correct indexing
        )
        # Added a check for sufficient bars available for all indicators to be calculated
        if len(self.dataclose) < required_data + max(
            self.params.vwap_session_length,
            self.params.vwap_weekly_length,
            self.params.adx_period,
            self.params.atr_period,
            self.params.volume_period,
            self.params.atr_expansion_period,
            self.params.breakout_lookback # Ensure this is also included for prior_high/low
        ):
            return

        current_price = self.dataclose[0]

        # 1. If in position, manage it (update stop, check exits)
        if self.position:
            self.update_trailing_stop()
            
            should_exit, exit_reason = self.check_exit_conditions()
            if should_exit:
                self.log(f'EXIT SIGNAL ({exit_reason}): Closing position at {current_price:.2f}')
                self.order = self.close()
                # Reset tracking variables
                self.entry_price = None
                self.stop_price = None
                self.trail_price = None
                self.position_type = None
                self.breakout_confirmed = False
            return # Do not look for new entries if already in position

        # 2. If not in position, look for entry signals
        if not self.position:
            breakout_valid, direction = self.check_breakout_conditions()
            
            if breakout_valid and direction:
                atr_value = self.atr[0]
                
                if direction == "LONG":
                    # Enter long position
                    self.log(f'VWAP BREAKOUT LONG SETUP:')
                    self.log(f'  Price: {current_price:.2f}, Session VWAP: {self.vwap_session[0]:.2f}')
                    self.log(f'  ADX: {self.adx[0]:.1f}, Volume: {self.datavolume[0]:.0f} (Avg: {self.volume_sma[0]:.0f})')
                    self.log(f'  ATR: {self.atr[0]:.4f} (Avg: {self.atr_sma[0]:.4f})')
                    
                    # Calculate stops
                    self.stop_price = current_price - (self.params.initial_stop_atr_multiplier * atr_value)
                    self.trail_price = current_price - (self.params.trailing_stop_atr_multiplier * atr_value)
                    
                    self.log(f'LONG ENTRY: Stop={self.stop_price:.2f}, Trail={self.trail_price:.2f}')
                    self.order = self.buy()
                    self.entry_price = current_price
                    self.position_type = 1
                    self.breakout_confirmed = True
                    
                elif direction == "SHORT":
                    # Enter short position
                    self.log(f'VWAP BREAKOUT SHORT SETUP:')
                    self.log(f'  Price: {current_price:.2f}, Session VWAP: {self.vwap_session[0]:.2f}')
                    self.log(f'  ADX: {self.adx[0]:.1f}, Volume: {self.datavolume[0]:.0f} (Avg: {self.volume_sma[0]:.0f})')
                    self.log(f'  ATR: {self.atr[0]:.4f} (Avg: {self.atr_sma[0]:.4f})')
                    
                    # Calculate stops
                    self.stop_price = current_price + (self.params.initial_stop_atr_multiplier * atr_value)
                    self.trail_price = current_price + (self.params.trailing_stop_atr_multiplier * atr_value)
                    
                    self.log(f'SHORT ENTRY: Stop={self.stop_price:.2f}, Trail={self.trail_price:.2f}')
                    self.order = self.sell()
                    self.entry_price = current_price
                    self.position_type = -1
                    self.breakout_confirmed = True

Explanation:

if self.order: return: Prevents the strategy from issuing new orders if an existing order is still being processed.
required_data check: Ensures that there are enough historical bars for all indicators to be fully calculated before attempting to trade. This prevents IndexError when indicators try to access data that doesn’t yet exist.
if self.position:: If the strategy currently holds an open position:
- It calls self.update_trailing_stop() to adjust the trailing stop.
- It then calls self.check_exit_conditions() to see if any exit criteria (stop-loss, trailing stop, VWAP reversion) are met.
- If an exit is triggered, it logs the reason, closes the position (self.close()), sets self.order to prevent re-entry, and resets all trade tracking variables.
- return ensures that no new entry signals are processed on the same bar if an exit occurred.
if not self.position:: If the strategy is not in a position:
- It calls self.check_breakout_conditions() to see if an entry signal is present.
- If a valid breakout is detected:
  - It logs detailed information about the setup.
  - It calculates the initial stop-loss and the initial trailing stop level based on ATR.
  - It places a buy() or sell() order.
  - It updates the entry_price, position_type, and breakout_confirmed tracking variables.

7. Backtesting Setup and Execution

Adding the strategy to Backtester and running a single backtest as well as a rolling backtest, we get the following results:

This comprehensive breakdown covers the purpose and functionality of each part of the provided backtrader strategy, explaining how it implements the VWAP Anchored Breakout logic and how it’s backtested.

1. Imports and Strategy Class Definition

2. Logging and Initialization (__init__, log)

3. Order and Trade Notifications (notify_order, notify_trade)

4. Breakout and Confirmation Checks

5. Exit Conditions and Position Sizing (update_trailing_stop, check_exit_conditions, calculate_position_size)

6. The next Method: The Heart of the Strategy

7. Backtesting Setup and Execution

2. Logging and Initialization (`init`, `log`)

3. Order and Trade Notifications (`notify_order`, `notify_trade`)

5. Exit Conditions and Position Sizing (`update_trailing_stop`, `check_exit_conditions`, `calculate_position_size`)

6. The `next` Method: The Heart of the Strategy