Rolling Backtest of a Multi-Timeframe Pivot Point Strategy with Volume and Momentum Confirmation

import backtrader as bt
import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class PivotPointStrategy(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 (near/touch)
        ('breakout_threshold', 0.03), # 3% threshold for breakouts
        ('volume_multiplier', 1.2), # Volume confirmation multiplier (1.2x average)
        ('volume_period', 30),      # Volume average period
        ('rsi_period', 14),         # RSI for momentum confirmation
        ('stop_loss_pct', 0.05),    # 5% stop loss
    )
    
    def __init__(self):
        # Data feeds
        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
        self.volume_sma = bt.indicators.SMA(self.volume, period=self.params.volume_period)
        self.rsi = bt.indicators.RSI(period=self.params.rsi_period)
        
        # Storage for calculated pivot levels (keyed by date/week/month start)
        self.daily_pivots = {}
        self.weekly_pivots = {}
        self.monthly_pivots = {}
        
        # OHLC data accumulators for multi-period pivot calculations
        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 last calculation dates to determine when to reset OHLC for new periods
        self.last_daily_calc = None
        self.last_weekly_calc = None
        self.last_monthly_calc = None
        
        # Order tracking
        self.order = None      # To prevent multiple orders
        self.stop_order = None # To manage the stop-loss order

    def calculate_pivot_levels(self, high, low, close):
        """Calculates classic pivot point levels."""
        pp = (high + low + close) / 3
        s1 = (2 * pp) - high
        s2 = pp - (high - low)
        s3 = low - 2 * (high - pp)
        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 new pivots when a period rolls over.
        """
        current_date = self.data.datetime.date(0)
        current_high = self.high[0]
        current_low = self.low[0]
        current_close = self.close[0]
        
        # Daily OHLC and Pivot Calculation
        if self.last_daily_calc != current_date:
            if self.last_daily_calc is not None:
                # If it's a new day, calculate pivots from the *previous* day's aggregated OHLC
                self.daily_pivots[current_date] = self.calculate_pivot_levels(
                    self.daily_ohlc['high'], self.daily_ohlc['low'], self.daily_ohlc['close']
                )
            # Reset OHLC for the new day
            self.daily_ohlc = {'high': current_high, 'low': current_low, 'close': current_close}
            self.last_daily_calc = current_date
        else:
            # Continue accumulating OHLC for the current day
            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 # Always use current close for daily pivot calc

        # Weekly OHLC and Pivot Calculation (Week starts on Monday, weekday() returns 0 for Monday)
        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

        # Monthly OHLC and Pivot Calculation
        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 and consolidates all active pivot levels for the current bar."""
        current_date = self.data.datetime.date(0)
        levels = []
        
        # Add daily pivots if enabled and available for the current day (yesterday's calculation)
        if self.params.use_daily and (current_date - timedelta(days=1)) in self.daily_pivots:
            daily_levels_date = current_date - timedelta(days=1) # Pivots are for today, based on yesterday's data
            if daily_levels_date in self.daily_pivots:
                 daily = self.daily_pivots[daily_levels_date]
                 for level_name, level_value in daily.items():
                     levels.append((level_value, 'daily', level_name))
        
        # Add weekly pivots if enabled and available
        if self.params.use_weekly:
            prev_week_start = current_date - timedelta(days=current_date.weekday() + 7) # Start of previous week
            if prev_week_start in self.weekly_pivots:
                weekly = self.weekly_pivots[prev_week_start]
                for level_name, level_value in weekly.items():
                    levels.append((level_value, 'weekly', level_name))
        
        # Add monthly pivots if enabled and available
        if self.params.use_monthly:
            prev_month_start = (current_date.replace(day=1) - timedelta(days=1)).replace(day=1) # Start of previous month
            if prev_month_start in self.monthly_pivots:
                monthly = self.monthly_pivots[prev_month_start]
                for level_name, level_value in monthly.items():
                    levels.append((level_value, 'monthly', level_name))
        
        # Sort levels for easier searching (e.g., finding nearest)
        return sorted(levels, key=lambda x: x[0])

    def check_level_interaction(self, price, high, low):
        """
        Determines if the current bar's price action interacts with any pivot levels
        (bounce or breakout).
        """
        levels = self.get_current_pivot_levels()
        
        for level_price, timeframe, level_name in levels:
            # Check for 'touch' or 'near' interaction (potential bounce)
            distance_pct = abs(price - level_price) / level_price
            
            if distance_pct <= self.params.bounce_threshold:
                # If current bar touches or crosses the level
                if (low <= level_price <= high):
                    return 'touch', level_price, timeframe, level_name
                # Or if very close
                elif distance_pct <= self.params.bounce_threshold / 2:
                    return 'near', level_price, timeframe, level_name
            
            # Check for breakouts
            elif distance_pct <= self.params.breakout_threshold:
                # Support breakout: Close below level, and level was within bar range
                if level_name.startswith('S') and price < level_price and low <= level_price <= high:
                    return 'support_break', level_price, timeframe, level_name
                # Resistance breakout: Close above level, and level was within bar range
                elif level_name.startswith('R') and price > level_price and low <= level_price <= high:
                    return 'resistance_break', level_price, timeframe, level_name
        
        return None, None, None, None # No significant interaction

    def volume_confirmation(self):
        """Checks if current volume is above the average volume by a multiplier."""
        # Ensure enough data for volume SMA and handle division by zero
        if np.isnan(self.volume_sma[0]) or self.volume_sma[0] == 0:
            return True # No average, assume confirmation (or add a warning)
        return self.volume[0] > self.volume_sma[0] * self.params.volume_multiplier

    def momentum_confirmation(self, trade_direction):
        """Checks RSI for momentum confirmation."""
        # Ensure RSI has enough data
        if np.isnan(self.rsi[0]):
            return True # No RSI, assume confirmation (or add a warning)
            
        if trade_direction == 'long':
            # For long, RSI should not be overbought (above 70) and not extremely oversold (below 30)
            return self.rsi[0] > 30 and self.rsi[0] < 70 
        elif trade_direction == 'short':
            # For short, RSI should not be oversold (below 30) and not extremely overbought (above 70)
            return self.rsi[0] > 30 and self.rsi[0] < 70 
        return True # Default if direction is not specified

    def notify_order(self, order):
        # Handles order completion (execution) or failure (canceled/rejected)
        if order.status in [order.Completed]:
            if order.isbuy() and self.position.size > 0: # Check if it's an entry buy
                stop_price = order.executed.price * (1 - self.params.stop_loss_pct)
                self.stop_order = self.sell(exectype=bt.Order.Stop, price=stop_price, size=self.position.size)
                self.log(f'BUY EXECUTED, Price: {order.executed.price:.2f}, Setting Stop Loss at {stop_price:.2f}')
            elif order.issell() and self.position.size < 0: # Check if it's an entry sell (short)
                stop_price = order.executed.price * (1 + self.params.stop_loss_pct)
                self.stop_order = self.buy(exectype=bt.Order.Stop, price=stop_price, size=abs(self.position.size))
                self.log(f'SELL EXECUTED, Price: {order.executed.price:.2f}, Setting Stop Loss at {stop_price:.2f}')
            elif order.isbuy() and self.position.size == 0 and self.stop_order and order.ref == self.stop_order.ref: # Stop-loss buy (cover short)
                self.log(f'STOP LOSS (COVER SHORT) EXECUTED, Price: {order.executed.price:.2f}')
            elif order.issell() and self.position.size == 0 and self.stop_order and order.ref == self.stop_order.ref: # Stop-loss sell (exit long)
                self.log(f'STOP LOSS (EXIT LONG) EXECUTED, Price: {order.executed.price:.2f}')

            self.order = None # Clear general order reference
            if self.stop_order and order.ref == self.stop_order.ref: # If the completed order was the stop order itself
                self.stop_order = None # Clear stop order reference
        
        elif order.status in [order.Canceled, order.Rejected, order.Margin]:
            self.log(f'Order Canceled/Rejected/Margin: Status {order.getstatusname()}')
            self.order = None # Clear entry order reference
            if self.stop_order and order.ref == self.stop_order.ref:
                self.log("WARNING: Stop-Loss Order failed!", doprint=True)
                self.stop_order = None


    def next(self):
        # Ensure no pending orders and enough data for indicators
        if self.order is not None:
            return
        
        # Minimum data needed for indicators to warm up
        min_data_needed = max(self.params.volume_period, self.params.rsi_period)
        if len(self.data) < min_data_needed + 1: # +1 for current bar data
            return

        # Update OHLC data and calculate pivots for the *next* day/week/month
        # (These are pivot points for the *current* bar's date, based on previous period's OHLC)
        self.update_ohlc_data() 
        
        current_price = self.close[0]
        current_high = self.high[0]
        current_low = self.low[0]
        
        # Check interaction with pivot levels
        interaction, level_price, timeframe, level_name = self.check_level_interaction(
            current_price, current_high, current_low
        )
        
        if interaction is None: # No significant interaction with any level
            return
        
        # Trading logic based on pivot level interactions
        
        # BOUNCE STRATEGY (Price touches/nears level and reverses)
        if interaction == 'touch' or interaction == 'near':
            if level_name.startswith('S'): # Support level - expect bounce UP (buy signal)
                if self.momentum_confirmation('long') and self.volume_confirmation():
                    if self.position.size < 0: # Close existing short position
                        if self.stop_order is not None: self.cancel(self.stop_order)
                        self.order = self.close()
                        self.log(f'Close SHORT on {level_name} ({timeframe}) bounce signal at {current_price:.2f}')
                    elif not self.position: # Open new long position
                        self.order = self.buy()
                        self.log(f'BUY on {level_name} ({timeframe}) bounce signal at {current_price:.2f}')
            
            elif level_name.startswith('R'): # Resistance level - expect bounce DOWN (sell/short signal)
                if self.momentum_confirmation('short') and self.volume_confirmation():
                    if self.position.size > 0: # Close existing long position
                        if self.stop_order is not None: self.cancel(self.stop_order)
                        self.order = self.close()
                        self.log(f'Close LONG on {level_name} ({timeframe}) bounce signal at {current_price:.2f}')
                    elif not self.position: # Open new short position
                        self.order = self.sell()
                        self.log(f'SELL on {level_name} ({timeframe}) bounce signal at {current_price:.2f}')
        
        # BREAKOUT STRATEGY (Price breaks through level with confirmation)
        elif interaction == 'resistance_break': # Break above resistance - go long
            if self.momentum_confirmation('long') and self.volume_confirmation():
                if self.position.size < 0: # Close existing short position
                    if self.stop_order is not None: self.cancel(self.stop_order)
                    self.order = self.close()
                    self.log(f'Close SHORT on {level_name} ({timeframe}) breakout signal at {current_price:.2f}')
                elif not self.position: # Open new long position
                    self.order = self.buy()
                    self.log(f'BUY on {level_name} ({timeframe}) breakout signal at {current_price:.2f}')
        
        elif interaction == 'support_break': # Break below support - go short
            if self.momentum_confirmation('short') and self.volume_confirmation():
                if self.position.size > 0: # Close existing long position
                    if self.stop_order is not None: self.cancel(self.stop_order)
                    self.order = self.close()
                    self.log(f'Close LONG on {level_name} ({timeframe}) breakout signal at {current_price:.2f}')
                elif not self.position: # Open new short position
                    self.order = self.sell()
                    self.log(f'SELL on {level_name} ({timeframe}) breakout signal at {current_price:.2f}')

    def stop(self):
        # Optional: Log final portfolio value at the end of the backtest
        print(f'Final Portfolio Value: {self.broker.getvalue():,.2f}')