← Back to Home
Multi-Timeframe Pivot Point Trading - A Confirmed Bounce & Breakout Strategy

Multi-Timeframe Pivot Point Trading - A Confirmed Bounce & Breakout Strategy

Pivot points are foundational technical analysis tools that provide dynamic support and resistance levels based on previous period’s price action. By incorporating pivot points from multiple timeframes (daily, weekly, monthly), traders can gain a more comprehensive view of potential turning points in the market. This article explores a backtrader strategy that leverages these multi-timeframe pivot points, combining them with volume and Relative Strength Index (RSI) for trade confirmation, and always ensuring disciplined exits with trailing stops.

Pasted image 20250729231126.png

The Essence of Pivot Points

The standard pivot point calculation is based on the previous period’s high, low, and close prices. From the central Pivot Point (PP), several support (S1, S2, S3) and resistance (R1, R2, R3) levels are derived:

These levels act as potential areas where price might reverse (bounce) or continue its move (breakout).

PivotPointVolumeRSIConfirmationStrategy in backtrader

The PivotPointStrategy (renamed to PivotPointVolumeRSIConfirmationStrategy to reflect its comprehensive logic) is designed to identify and trade these interactions.

import backtrader as bt
from datetime import timedelta # Needed for weekly/monthly calculations
import numpy as np # For np.isnan check

class PivotPointVolumeRSIConfirmationStrategy(bt.Strategy):
    params = (
        ('use_daily', True),       # Use daily pivot points
        ('use_weekly', True),      # Use weekly pivot points
        ('use_monthly', False),    # Use monthly pivot points
        ('bounce_threshold', 0.01), # 1% threshold for level interaction (e.g., 1% of level price)
        ('breakout_threshold', 0.03), # 3% threshold for breakouts
        ('volume_multiplier', 1.2), # Volume confirmation multiplier (e.g., 20% above average)
        ('volume_period', 7),      # Volume average period
        ('rsi_period', 14),        # RSI for momentum confirmation
        ('stop_loss_pct', 0.05),   # Initial stop loss percentage
        ('trail_percent', 0.02),   # Trailing stop percentage for exits
    )
    
    def __init__(self):
        # Price data
        self.high = self.data.high
        self.low = self.data.low
        self.close = self.data.close
        self.open = self.data.open
        self.volume = self.data.volume
        
        # Technical indicators for confirmation
        self.volume_sma = bt.indicators.SMA(self.volume, period=self.params.volume_period)
        self.rsi = bt.indicators.RSI(period=self.params.rsi_period)
        
        # Pivot point levels storage for different timeframes
        self.daily_pivots = {}
        self.weekly_pivots = {}
        self.monthly_pivots = {}
        
        # Current active levels (sorted list of tuples: (value, timeframe, name))
        self.current_levels = []
        
        # Track last calculation dates to determine new periods
        self.last_daily_calc = None
        self.last_weekly_calc = None
        self.last_monthly_calc = None
        
        # Store recent OHLC for calculations across periods
        self.daily_ohlc = {'high': 0, 'low': 0, 'close': 0}
        self.weekly_ohlc = {'high': 0, 'low': 0, 'close': 0}
        self.monthly_ohlc = {'high': 0, 'low': 0, 'close': 0}
        
        # Track orders to prevent multiple orders
        self.order = None
        self.stop_order = None # For fixed stop loss
        self.trail_order = None # For trailing stop

    def calculate_pivot_levels(self, high, low, close):
        """Calculate pivot point and support/resistance levels based on classic formula."""
        pp = (high + low + close) / 3
        
        # Support levels
        s1 = (2 * pp) - high
        s2 = pp - (high - low)
        s3 = low - 2 * (high - pp)
        
        # Resistance levels
        r1 = (2 * pp) - low
        r2 = pp + (high - low)
        r3 = high + 2 * (pp - low)
        
        return {
            'PP': pp,
            'S1': s1, 'S2': s2, 'S3': s3,
            'R1': r1, 'R2': r2, 'R3': r3
        }

    def update_ohlc_data(self):
        """
        Aggregates OHLC data for daily, weekly, and monthly pivot calculations.
        Calculates pivot levels when a new period begins.
        """
        current_date = self.data.datetime.date(0)
        current_high = self.high[0]
        current_low = self.low[0]
        current_close = self.close[0]
        
        # Update daily OHLC and calculate pivots if a new day starts
        if self.last_daily_calc != current_date:
            if self.last_daily_calc is not None:
                # Calculate pivots from previous day's accumulated OHLC
                self.daily_pivots[current_date] = self.calculate_pivot_levels(
                    self.daily_ohlc['high'], 
                    self.daily_ohlc['low'], 
                    self.daily_ohlc['close']
                )
            
            # Reset for new day
            self.daily_ohlc = {'high': current_high, 'low': current_low, 'close': current_close}
            self.last_daily_calc = current_date
        else:
            # Update current day's high/low
            self.daily_ohlc['high'] = max(self.daily_ohlc['high'], current_high)
            self.daily_ohlc['low'] = min(self.daily_ohlc['low'], current_low)
            self.daily_ohlc['close'] = current_close # Update close as it's the last price of the bar
        
        # Update weekly OHLC (Monday = 0, Sunday = 6) and calculate pivots if a new week starts
        week_start = current_date - timedelta(days=current_date.weekday())
        if self.last_weekly_calc != week_start:
            if self.last_weekly_calc is not None:
                self.weekly_pivots[week_start] = self.calculate_pivot_levels(
                    self.weekly_ohlc['high'], 
                    self.weekly_ohlc['low'], 
                    self.weekly_ohlc['close']
                )
            
            self.weekly_ohlc = {'high': current_high, 'low': current_low, 'close': current_close}
            self.last_weekly_calc = week_start
        else:
            self.weekly_ohlc['high'] = max(self.weekly_ohlc['high'], current_high)
            self.weekly_ohlc['low'] = min(self.weekly_ohlc['low'], current_low)
            self.weekly_ohlc['close'] = current_close
        
        # Update monthly OHLC and calculate pivots if a new month starts
        month_start = current_date.replace(day=1)
        if self.last_monthly_calc != month_start:
            if self.last_monthly_calc is not None:
                self.monthly_pivots[month_start] = self.calculate_pivot_levels(
                    self.monthly_ohlc['high'], 
                    self.monthly_ohlc['low'], 
                    self.monthly_ohlc['close']
                )
            
            self.monthly_ohlc = {'high': current_high, 'low': current_low, 'close': current_close}
            self.last_monthly_calc = month_start
        else:
            self.monthly_ohlc['high'] = max(self.monthly_ohlc['high'], current_high)
            self.monthly_ohlc['low'] = min(self.monthly_ohlc['low'], current_low)
            self.monthly_ohlc['close'] = current_close

    def get_current_pivot_levels(self):
        """
        Retrieves all currently active pivot levels (daily, weekly, monthly)
        and returns them as a sorted list.
        """
        current_date = self.data.datetime.date(0)
        levels = []
        
        # Daily pivots
        if self.params.use_daily and current_date in self.daily_pivots:
            daily = self.daily_pivots[current_date]
            for level_name, level_value in daily.items():
                levels.append((level_value, 'daily', level_name))
        
        # Weekly pivots
        if self.params.use_weekly:
            week_start = current_date - timedelta(days=current_date.weekday())
            if week_start in self.weekly_pivots:
                weekly = self.weekly_pivots[week_start]
                for level_name, level_value in weekly.items():
                    levels.append((level_value, 'weekly', level_name))
        
        # Monthly pivots
        if self.params.use_monthly:
            month_start = current_date.replace(day=1)
            if month_start in self.monthly_pivots:
                monthly = self.monthly_pivots[month_start]
                for level_name, level_value in monthly.items():
                    levels.append((level_value, 'monthly', level_name))
        
        return sorted(levels, key=lambda x: x[0]) # Sort by price

    def find_nearest_levels(self, price):
        """Find the nearest support and resistance levels to the current price."""
        levels = self.get_current_pivot_levels()
        
        resistance_levels = [(level, timeframe, name) for level, timeframe, name in levels if level > price]
        support_levels = [(level, timeframe, name) for level, timeframe, name in levels if level < price]
        
        nearest_resistance = min(resistance_levels, key=lambda x: x[0] - price) if resistance_levels else None
        nearest_support = max(support_levels, key=lambda x: x[0] - price) if support_levels else None
        
        return nearest_support, nearest_resistance

    def check_level_interaction(self, price, high, low):
        """
        Checks if the current price is interacting with any pivot levels,
        distinguishing between 'touch/near' for bounces and 'breakout'.
        """
        levels = self.get_current_pivot_levels()
        
        for level_price, timeframe, level_name in levels:
            # Check if price is near the level for bounce consideration
            distance_pct = abs(price - level_price) / level_price
            
            if distance_pct <= self.params.bounce_threshold:
                # Check if we touched or crossed the level within the current bar
                if (low <= level_price <= high):
                    return 'touch', level_price, timeframe, level_name
                elif distance_pct <= self.params.bounce_threshold / 2: # Very close proximity
                    return 'near', level_price, timeframe, level_name
            
            # Check for breakouts - current close *past* the level with significant momentum
            elif distance_pct <= self.params.breakout_threshold:
                if level_name.startswith('S') and price < level_price: # Price closed below support
                    # Check if the low of the bar broke below the support
                    if low <= level_price:
                        return 'support_break', level_price, timeframe, level_name
                elif level_name.startswith('R') and price > level_price: # Price closed above resistance
                    # Check if the high of the bar broke above the resistance
                    if high >= level_price:
                        return 'resistance_break', level_price, timeframe, level_name
                
        return None, None, None, None

    def volume_confirmation(self):
        """Checks if current volume is significantly higher than average."""
        if np.isnan(self.volume_sma[0]) or self.volume_sma[0] == 0:
            return True # Not enough data for SMA, or SMA is zero
        return self.volume[0] > self.volume_sma[0] * self.params.volume_multiplier

    def momentum_confirmation(self, trade_direction):
        """Checks RSI for momentum confirmation, avoiding overbought/oversold extremes."""
        if np.isnan(self.rsi[0]):
            return True # Not enough data for RSI
        
        if trade_direction == 'long':
            # Looking for a long: RSI should be rising from oversold or in a healthy range
            return self.rsi[0] > 30 # Avoid extreme oversold
        elif trade_direction == 'short':
            # Looking for a short: RSI should be falling from overbought or in a healthy range
            return self.rsi[0] < 70 # Avoid extreme overbought
        return True # Default to true if no specific direction

    def notify_order(self, order):
        if order.status in [order.Completed]:
            if order.isbuy(): # If a buy order completed
                # Set initial fixed stop loss
                stop_price_fixed = order.executed.price * (1 - self.params.stop_loss_pct)
                self.stop_order = self.sell(exectype=bt.Order.Stop, price=stop_price_fixed)
                # Set trailing stop
                self.trail_order = self.sell(exectype=bt.Order.StopTrail, trailpercent=self.p.trail_percent)
            elif order.issell(): # If a sell (short) order completed
                # Set initial fixed stop loss
                stop_price_fixed = order.executed.price * (1 + self.params.stop_loss_pct)
                self.stop_order = self.buy(exectype=bt.Order.Stop, price=stop_price_fixed)
                # Set trailing stop
                self.trail_order = self.buy(exectype=bt.Order.StopTrail, trailpercent=self.p.trail_percent)
        
        # Reset order tracking variables
        if order.status in [order.Completed, order.Canceled, order.Rejected]:
            self.order = None
            if order == self.stop_order:
                self.stop_order = None
            if order == self.trail_order:
                self.trail_order = None

    def next(self):
        if self.order is not None: # If an order is already pending, wait
            return
            
        # Update OHLC data and calculate pivots for the current bar
        self.update_ohlc_data()
        
        # Get current price action
        current_price = self.close[0]
        current_high = self.high[0]
        current_low = self.low[0]
        
        # Check how the price is interacting with pivot levels
        interaction, level_price, timeframe, level_name = self.check_level_interaction(
            current_price, current_high, current_low
        )
        
        # If no significant interaction, do nothing
        if interaction is None:
            return
            
        # Trading logic based on pivot level interactions and confirmations
        if interaction in ['touch', 'near']:
            # Bounce strategy - expect reversal at key levels
            if level_name.startswith('S'): # Support level - expect bounce up
                if (self.momentum_confirmation('long') and 
                    self.volume_confirmation()):
                    
                    if self.position.size < 0: # If currently short, close short position
                        # Cancel existing stop/trail orders before closing
                        if self.stop_order: self.cancel(self.stop_order)
                        if self.trail_order: self.cancel(self.trail_order)
                        self.order = self.close()
                    elif not self.position: # If no position, go long
                        self.order = self.buy()
            
            elif level_name.startswith('R'): # Resistance level - expect bounce down
                if (self.momentum_confirmation('short') and 
                    self.volume_confirmation()):
                    
                    if self.position.size > 0: # If currently long, close long position
                        # Cancel existing stop/trail orders before closing
                        if self.stop_order: self.cancel(self.stop_order)
                        if self.trail_order: self.cancel(self.trail_order)
                        self.order = self.close()
                    elif not self.position: # If no position, go short
                        self.order = self.sell()
        
        elif interaction == 'resistance_break':
            # Resistance breakout - go long
            if (self.momentum_confirmation('long') and 
                self.volume_confirmation()):
                
                if self.position.size < 0: # Close short if existing
                    if self.stop_order: self.cancel(self.stop_order)
                    if self.trail_order: self.cancel(self.trail_order)
                    self.order = self.close()
                elif not self.position: # Go long
                    self.order = self.buy()
        
        elif interaction == 'support_break':
            # Support breakdown - go short
            if (self.momentum_confirmation('short') and 
                self.volume_confirmation()):
                
                if self.position.size > 0: # Close long if existing
                    if self.stop_order: self.cancel(self.stop_order)
                    if self.trail_order: self.cancel(self.trail_order)
                    self.order = self.close()
                elif not self.position: # Go short
                    self.order = self.sell()

Key Components and Logic

  1. Multi-Timeframe Pivot Calculation:

    • The update_ohlc_data() method intelligently aggregates high, low, and close prices for daily, weekly, and monthly periods.
    • Crucially, it triggers the calculate_pivot_levels() function only when a new daily, weekly, or monthly period begins (e.g., at the start of a new day, Monday for a new week, or the 1st of the month for a new month). This ensures that pivot levels are derived from the previous completed period’s data, which is the standard practice.
    • The get_current_pivot_levels() method then gathers all active daily, weekly, and monthly pivot levels and sorts them by price, making it easy to identify nearby support/resistance.

  2. Interaction Detection (check_level_interaction):

    • This function determines if the current price is “touching” or “near” a pivot level (indicating a potential bounce) or “breaking out” from it.
    • It uses bounce_threshold and breakout_threshold parameters to define these proximity zones, allowing for flexible sensitivity.
    • It differentiates between support and resistance levels for breakout identification.

  3. Confirmation Filters:

    • Volume Confirmation (volume_confirmation): This checks if the current trading volume exceeds a multiple (volume_multiplier) of the average volume over a volume_period. High volume often confirms the strength of a price move.
    • Momentum Confirmation (momentum_confirmation): It uses the RSI indicator to gauge momentum. For long trades, it seeks RSI above 30 (not oversold), and for short trades, RSI below 70 (not overbought). This aims to ensure trades are taken when there’s healthy momentum, avoiding entries into exhausted moves.

  4. Trading Logic (next):

    • The strategy first updates the OHLC data and pivot levels for the current bar.
    • It then identifies the type of interaction with any active pivot level.
    • Bounce Strategy: If the price interacts with a support level (S1, S2, S3) and confirms with volume and RSI, the strategy attempts to go long. If it interacts with a resistance level (R1, R2, R3) with confirmation, it goes short. It will also close opposing existing positions.
    • Breakout Strategy: If price breaks above a resistance level or below a support level, and is confirmed by volume and RSI, the strategy initiates a trade in the direction of the breakout (long for resistance break, short for support break). Again, it closes opposing positions first.

  5. Risk Management (Stop Loss and Trailing Stop):

    • As per your preference for always using trailing stops, the notify_order method immediately places both an initial fixed stop-loss (stop_loss_pct) and a trailing stop (trail_percent) after an entry order is completed.
    • The fixed stop-loss provides immediate protection, while the trailing stop allows for profits to run while still protecting against reversals. When closing positions, any active stop or trailing orders are first canceled to prevent unintended exits.

This PivotPointVolumeRSIConfirmationStrategy provides a comprehensive framework for trading pivot points, enhancing signal reliability through multi-timeframe analysis and additional confirmation indicators, and implementing robust risk management with trailing stops. It reflects a nuanced approach to capturing both bounce and breakout opportunities.