Rolling Backtest of a Volatility Breakout Trading Strategy

import backtrader as bt
import numpy as np

class VolatilityBreakoutSystem(bt.Strategy):
    """
    Volatility-Adjusted Breakout System with Trailing Stops
    
    Features:
    - Breakout detection using N-period highs/lows
    - ADX trend strength filter
    - VWAP-based volatility filter
    - ATR-based initial and trailing stops
    """
    
    params = (
        # Breakout Parameters
        ('breakout_period', 7),
        ('adx_period', 14),
        ('adx_threshold', 20),
        
        # VWAP Parameters
        ('vwap_period', 7),
        ('vwap_multiplier', 1.1),  # Price must be X times ATR from VWAP
        
        # ATR Parameters
        ('atr_period', 14),
        ('atr_stop_multiplier', 3.0),  # Initial stop loss distance
        ('atr_trail_multiplier', 1.5), # Trailing stop distance
        
        # Trailing Stop Activation/Step
        ('trail_activation', 1.5),  # Activate trailing after X times ATR profit
        ('trail_step', 0.5), # Trail by X times ATR steps (currently not directly used in 'update_trailing_stop' as it re-calculates from current ATR)
    )
    
    def __init__(self):
        # Data feeds for indicators
        self.high = self.data.high
        self.low = self.data.low
        self.close = self.data.close
        self.volume = self.data.volume
        
        # Technical indicators
        self.atr = bt.indicators.ATR(period=self.params.atr_period)
        self.adx = bt.indicators.ADX(period=self.params.adx_period)
        
        # Breakout indicators (Highest High / Lowest Low over breakout_period)
        self.highest = bt.indicators.Highest(self.high, period=self.params.breakout_period)
        self.lowest = bt.indicators.Lowest(self.low, period=self.params.breakout_period)
        
        # VWAP calculation (simplified for backtrader indicator use)
        self.vwap = self._calculate_vwap_indicator()
        
        # Order tracking variables
        self.order = None          # To track active entry/exit orders
        self.stop_order = None     # To track the initial stop loss order
        self.trail_order = None    # To track the dynamic trailing stop order
        
        # Position tracking variables
        self.entry_price = None        # Price at which the current position was entered
        self.entry_atr = None          # ATR at the time of entry, for profit calculation
        self.highest_profit = 0        # Highest unrealized profit in ATR terms
        self.trailing_active = False   # Flag to indicate if trailing stop is active
        
        # Performance tracking (for internal logging, analyzers are more comprehensive)
        self.trade_count = 0
        self.win_count = 0
        self.total_pnl = 0
    
    def _calculate_vwap_indicator(self):
        """Internal helper to calculate VWAP as a Backtrader indicator."""
        typical_price = (self.high + self.low + self.close) / 3
        volume_price = typical_price * self.volume
        
        # Use Simple Moving Averages for the numerator and denominator
        # For a true cumulative VWAP, you would need to implement it as a custom indicator
        vwap_numerator = bt.indicators.SMA(volume_price, period=self.params.vwap_period)
        vwap_denominator = bt.indicators.SMA(self.volume, period=self.params.vwap_period)
        
        # Avoid division by zero if volume_denominator is zero
        return vwap_numerator / vwap_denominator
    
    def log(self, txt, dt=None):
        """Logging function for strategy events."""
        dt = dt or self.datas[0].datetime.date(0)
        print(f'{dt.isoformat()}: {txt}')
    
    def notify_order(self, order):
        """Handle order notifications for execution and state changes."""
        if order.status in [order.Submitted, order.Accepted]:
            # Order submitted or accepted, nothing to do yet
            return
        
        if order.status in [order.Completed]:
            if order.isbuy():
                self.log(f'BUY EXECUTED: Price: {order.executed.price:.2f}, '
                         f'Size: {order.executed.size:.4f}, Cost: {order.executed.value:.2f}')
                # Initialize position tracking for a new long
                self.entry_price = order.executed.price
                self.entry_atr = self.atr[0] # ATR at entry
                self.highest_profit = 0      # Reset highest profit for new trade
                self.trailing_active = False # Trailing stop not active initially
            else: # Order is a SELL (either entry or exit)
                self.log(f'SELL EXECUTED: Price: {order.executed.price:.2f}, '
                         f'Size: {order.executed.size:.4f}, Cost: {order.executed.value:.2f}')
                # If it's a short entry (size will be negative, but executed.size positive)
                if order.isbuy() == False and self.position.size < 0: # This check indicates a short entry
                     self.entry_price = order.executed.price
                     self.entry_atr = self.atr[0]
                     self.highest_profit = 0
                     self.trailing_active = False
                # If it's an exit for a long or cover for a short, no entry details needed
        
        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log(f'Order {order.getstatusname()}: Ref={order.ref}')
        
        # Clear order tracking references
        if order == self.order:
            self.order = None
        if order == self.stop_order:
            self.stop_order = None
        if order == self.trail_order:
            self.trail_order = None
    
    def notify_trade(self, trade):
        """Handle trade notifications (profit/loss on closed trades)."""
        if not trade.isclosed:
            return
        
        self.trade_count += 1
        self.total_pnl += trade.pnl
        
        if trade.pnl > 0:
            self.win_count += 1
        
        win_rate = (self.win_count / self.trade_count) * 100 if self.trade_count > 0 else 0
        
        self.log(f'TRADE CLOSED: PnL: {trade.pnl:.2f}, Net PnL: {trade.pnlcomm:.2f}, '
                 f'Total Trades: {self.trade_count}, Win Rate: {win_rate:.1f}%')
        
        # Reset position-specific tracking
        self.entry_price = None
        self.entry_atr = None
        self.highest_profit = 0
        self.trailing_active = False
    
        
    def can_open_position(self):
        """Determines if a new position can be opened."""
        # Simple check: no existing position and no pending orders
        return not self.position and not self.order
    
    def is_breakout_long(self):
        """Check for long breakout condition: new highest high."""
        if len(self.highest) < 2: # Need at least current and previous highest
            return False
        
        return self.close[0] > self.highest[-1] # Close above previous highest
    
    def is_breakout_short(self):
        """Check for short breakout condition: new lowest low."""
        if len(self.lowest) < 2: # Need at least current and previous lowest
            return False
        
        return self.close[0] < self.lowest[-1] # Close below previous lowest
    
    def is_adx_strong(self):
        """Check if ADX indicates strong trend."""
        if len(self.adx) < 1:
            return False
        
        return self.adx[0] > self.params.adx_threshold
    
    def is_price_away_from_vwap(self):
        """Check if price is sufficiently away from VWAP, indicating conviction."""
        if len(self.vwap) < 1 or len(self.atr) < 1:
            return False
        
        current_price = self.close[0]
        vwap_value = self.vwap[0]
        atr_value = self.atr[0]
        
        # Price should be at least X * ATR away from VWAP
        distance_threshold = atr_value * self.params.vwap_multiplier
        
        distance_from_vwap = abs(current_price - vwap_value)
        
        return distance_from_vwap >= distance_threshold
    
    def is_atr_rising(self):
        """Check if ATR is rising, indicating increasing volatility."""
        if len(self.atr) < 3: # Need at least 3 periods to check for rising pattern
            return False
        
        # ATR should be rising over the last 2 periods (current > prev > prev_prev)
        return (self.atr[0] > self.atr[-1] and 
                self.atr[-1] > self.atr[-2])
    
    def update_trailing_stop(self):
        """Update trailing stop loss for an existing position."""
        if not self.position or not self.entry_price or len(self.atr) < 1:
            return # No position or insufficient data
        
        current_price = self.close[0]
        current_atr = self.atr[0]
        
        # Calculate unrealized profit in terms of ATR at entry
        if self.position.size > 0:  # Long position
            unrealized_profit_atr = (current_price - self.entry_price) / self.entry_atr
            # Update highest profit for potential trail activation
            self.highest_profit = max(self.highest_profit, unrealized_profit_atr)
            
            # Activate trailing stop if sufficient profit is reached
            if self.highest_profit >= self.params.trail_activation:
                self.trailing_active = True
            
            if self.trailing_active:
                # Calculate new trailing stop level
                new_trail_stop_price = current_price - (current_atr * self.params.atr_trail_multiplier)
                
                # If there's no existing trail order or the new stop is higher, update it
                if (not self.trail_order or 
                    new_trail_stop_price > self.trail_order.price):
                    
                    # Cancel existing trailing stop order if it exists and is active
                    if self.trail_order and self.trail_order.alive():
                        self.cancel(self.trail_order)
                    
                    # Place a new stop sell order to close the long position
                    self.trail_order = self.sell(
                        exectype=bt.Order.Stop,
                        price=new_trail_stop_price,
                        size=self.position.size # Ensure it closes the full position
                    )
                    self.log(f'LONG TRAILING STOP UPDATED: New Stop: {new_trail_stop_price:.2f}')
        
        elif self.position.size < 0: # Short position
            unrealized_profit_atr = (self.entry_price - current_price) / self.entry_atr
            self.highest_profit = max(self.highest_profit, unrealized_profit_atr)
            
            if self.highest_profit >= self.params.trail_activation:
                self.trailing_active = True
            
            if self.trailing_active:
                new_trail_stop_price = current_price + (current_atr * self.params.atr_trail_multiplier)
                
                if (not self.trail_order or 
                    new_trail_stop_price < self.trail_order.price):
                    
                    if self.trail_order and self.trail_order.alive():
                        self.cancel(self.trail_order)
                    
                    self.trail_order = self.buy(
                        exectype=bt.Order.Stop,
                        price=new_trail_stop_price,
                        size=abs(self.position.size) # Ensure it covers the full position
                    )
                    self.log(f'SHORT TRAILING STOP UPDATED: New Stop: {new_trail_stop_price:.2f}')
            
    def next(self):
        """Main strategy logic executed on each bar."""
        # Ensure all indicators have warmed up with enough data
        if (len(self.atr) < self.params.atr_period or
            len(self.adx) < self.params.adx_period or
            len(self.vwap) < self.params.vwap_period or
            len(self.highest) < self.params.breakout_period or
            len(self.lowest) < self.params.breakout_period): # Check for lowest as well for short entries
            return
        
        # First, manage existing positions (update trailing stop)
        self.update_trailing_stop()
        
        # If an order is pending (entry or initial stop), do nothing else
        if self.order or self.stop_order: 
            return
        
        # Check if we are allowed to open new positions (e.g., if flat)
        if not self.can_open_position():
            return
        
        # Current market data and indicator values
        current_price = self.close[0]
        current_atr = self.atr[0]
        
        # Apply market filters
        adx_strong = self.is_adx_strong()
        price_away_vwap = self.is_price_away_from_vwap()
        atr_rising = self.is_atr_rising()
        
        # Entry conditions for new positions
        if not self.position: # Only consider new entries if currently flat
            # Long breakout entry conditions
            if (self.is_breakout_long() and adx_strong and price_away_vwap and atr_rising):
                stop_distance = current_atr * self.params.atr_stop_multiplier
                stop_price = current_price - stop_distance
                
                # Enter long position (sizing handled by Cerebro's sizer)
                self.order = self.buy()
                
                # Set the initial stop loss
                self.stop_order = self.sell(
                    exectype=bt.Order.Stop,
                    price=stop_price,
                    # size is automatically handled by backtrader if buy order is implied for sizing
                )
                
                self.log(f'LONG BREAKOUT: Price: {current_price:.2f}, '
                         f'Initial Stop: {stop_price:.2f}, ADX: {self.adx[0]:.2f}, '
                         f'VWAP Away: {price_away_vwap}, ATR Rising: {atr_rising}')
            
            # Short breakout entry conditions
            elif (self.is_breakout_short() and adx_strong and price_away_vwap and atr_rising):
                stop_distance = current_atr * self.params.atr_stop_multiplier
                stop_price = current_price + stop_distance
                
                # Enter short position (sizing handled by Cerebro's sizer)
                self.order = self.sell()
                
                # Set the initial stop loss
                self.stop_order = self.buy(
                    exectype=bt.Order.Stop,
                    price=stop_price,
                )
                
                self.log(f'SHORT BREAKOUT: Price: {current_price:.2f}, '
                         f'Initial Stop: {stop_price:.2f}, ADX: {self.adx[0]:.2f}, '
                         f'VWAP Away: {price_away_vwap}, ATR Rising: {atr_rising}')
    
    def stop(self):
        """Called at the end of the backtest."""
        self.log(f'Final Portfolio Value: {self.broker.getvalue():.2f}')

import backtrader as bt
import yfinance as yf
import pandas as pd
import numpy as np
import dateutil.relativedelta as rd
import matplotlib.pyplot as plt
import seaborn as sns

# Assuming VolatilityBreakoutSystem class is defined above
# from VolatilityBreakoutSystem import VolatilityBreakoutSystem 

strategy = VolatilityBreakoutSystem # Set the strategy to be tested

def run_rolling_backtest(
    ticker="ETH-USD",
    start="2018-01-01",
    end="2025-06-24", # Current date as per your prompt
    window_months=3,
    strategy_params=None
):
    strategy_params = strategy_params or {}
    all_results = []
    start_dt = pd.to_datetime(start)
    end_dt = pd.to_datetime(end)
    current_start = start_dt

    while True:
        current_end = current_start + rd.relativedelta(months=window_months)
        # Adjust the end of the current window if it exceeds the overall end date
        if current_end > end_dt:
            current_end = end_dt
            if current_start >= current_end: # No valid period left
                break

        print(f"\nROLLING BACKTEST: {current_start.date()} to {current_end.date()}")

        # Data download using yfinance, respecting the user's preference for auto_adjust=False and droplevel
        # Using the saved preference: yfinance download with auto_adjust=False and droplevel(axis=1, level=1)
        data = yf.download(ticker, start=current_start, end=current_end, auto_adjust=False, progress=False)
        
        # Apply droplevel if data is a MultiIndex, as per user's preference
        if isinstance(data.columns, pd.MultiIndex):
            data = data.droplevel(1, axis=1)

        # Check for sufficient data after droplevel
        # Calculate min data needed based on strategy parameters
        # These are default values if not overridden in strategy_params
        min_atr_period = strategy_params.get('atr_period', 14)
        min_adx_period = strategy_params.get('adx_period', 14)
        min_vwap_period = strategy_params.get('vwap_period', 7)
        min_breakout_period = strategy_params.get('breakout_period', 7)

        # The strategy requires all indicators to have sufficient data for their calculations
        # The longest period among them plus a buffer (e.g., 2-3 bars for safe comparisons like ATR rising)
        min_data_for_indicators = max(min_atr_period, min_adx_period, min_vwap_period, min_breakout_period) + 3 

        if data.empty or len(data) < min_data_for_indicators:
            print(f"Not enough data for period {current_start.date()} to {current_end.date()} (requires {min_data_for_indicators} bars). Skipping.")
            # Move to the next window. If moving to the next window makes us pass overall end, break.
            if current_end == end_dt: # If the current window already reached overall end_dt
                 break
            current_start = current_end # Move to the end of the current (insufficient) period
            continue

        feed = bt.feeds.PandasData(dataname=data)
        cerebro = bt.Cerebro()
        cerebro.addstrategy(strategy, **strategy_params) # Pass strategy params to the strategy
        cerebro.adddata(feed)
        cerebro.broker.setcash(100000)
        cerebro.broker.setcommission(commission=0.001)
        cerebro.addsizer(bt.sizers.PercentSizer, percents=95)

        start_val = cerebro.broker.getvalue()
        cerebro.run()
        final_val = cerebro.broker.getvalue()
        ret = (final_val - start_val) / start_val * 100

        all_results.append({
            'start': current_start.date(),
            'end': current_end.date(),
            'return_pct': ret,
            'final_value': final_val,
        })

        print(f"Return: {ret:.2f}% | Final Value: {final_val:.2f}")
        
        # Move to the next window. If current_end already reached overall end_dt, then break.
        if current_end == end_dt:
            break
        current_start = current_end # For non-overlapping windows, next start is current end

    return pd.DataFrame(all_results)

# ... (all code as above) ...

def report_stats(df):
    returns = df['return_pct']
    stats = {
        'Mean Return %': np.mean(returns),
        'Median Return %': np.median(returns),
        'Std Dev %': np.std(returns),
        'Min Return %': np.min(returns),
        'Max Return %': np.max(returns),
        'Sharpe Ratio': np.mean(returns) / np.std(returns) if np.std(returns) > 0 else np.nan
    }
    print("\n=== ROLLING BACKTEST STATISTICS ===")
    for k, v in stats.items():
        print(f"{k}: {v:.2f}")
    return stats

def plot_return_distribution(df):
    sns.set(style="whitegrid")
    plt.figure(figsize=(10, 5))
    sns.histplot(df['return_pct'], bins=20, kde=True, color='dodgerblue')
    plt.axvline(df['return_pct'].mean(), color='black', linestyle='--', label='Mean')
    plt.title('Rolling Backtest Return Distribution')
    plt.xlabel('Return %')
    plt.ylabel('Frequency')
    plt.legend()
    plt.tight_layout()
    plt.show()

def plot_four_charts(df, rolling_sharpe_window=4):
    """
    Generates four analytical plots for rolling backtest results.
    """
    fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(12, 8)) # Adjusted figsize for clarity

    # Calculate period numbers (0, 1, 2, 3, ...)
    periods = list(range(len(df)))
    returns = df['return_pct']

    # 1. Period Returns (Top Left)
    colors = ['green' if r >= 0 else 'red' for r in returns]
    ax1.bar(periods, returns, color=colors, alpha=0.7)
    ax1.set_title('Period Returns', fontsize=14, fontweight='bold')
    ax1.set_xlabel('Period')
    ax1.set_ylabel('Return %')
    ax1.axhline(y=0, color='black', linestyle='-', alpha=0.3)
    ax1.grid(True, alpha=0.3)

    # 2. Cumulative Returns (Top Right)
    cumulative_returns = (1 + returns / 100).cumprod() * 100 - 100
    ax2.plot(periods, cumulative_returns, marker='o', linewidth=2, markersize=4, color='blue') # Smaller markers
    ax2.set_title('Cumulative Returns', fontsize=14, fontweight='bold')
    ax2.set_xlabel('Period')
    ax2.set_ylabel('Cumulative Return %')
    ax2.grid(True, alpha=0.3)

    # 3. Rolling Sharpe Ratio (Bottom Left)
    rolling_sharpe = returns.rolling(window=rolling_sharpe_window).apply(
        lambda x: x.mean() / x.std() if x.std() > 0 else np.nan, raw=False # Added raw=False for lambda
    )
    # Only plot where we have valid rolling calculations
    valid_mask = ~rolling_sharpe.isna()
    valid_periods = [i for i, valid in enumerate(valid_mask) if valid]
    valid_sharpe = rolling_sharpe[valid_mask]

    ax3.plot(valid_periods, valid_sharpe, marker='o', linewidth=2, markersize=4, color='orange') # Smaller markers
    ax3.axhline(y=0, color='red', linestyle='--', alpha=0.5)
    ax3.set_title(f'Rolling Sharpe Ratio ({rolling_sharpe_window}-period)', fontsize=14, fontweight='bold')
    ax3.set_xlabel('Period')
    ax3.set_ylabel('Sharpe Ratio')
    ax3.grid(True, alpha=0.3)

    # 4. Return Distribution (Bottom Right)
    bins = min(15, max(5, len(returns)//2))
    ax4.hist(returns, bins=bins, alpha=0.7, color='steelblue', edgecolor='black')
    mean_return = returns.mean()
    ax4.axvline(mean_return, color='red', linestyle='--', linewidth=2,
                label=f'Mean: {mean_return:.2f}%')
    ax4.set_title('Return Distribution', fontsize=14, fontweight='bold')
    ax4.set_xlabel('Return %')
    ax4.set_ylabel('Frequency')
    ax4.legend()
    ax4.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.show()

if __name__ == '__main__':
    # Run the rolling backtest with default parameters (ETH-USD, 3-month windows)
    df = run_rolling_backtest(end="2025-06-24") # End date updated to current date

    print("\n=== ROLLING BACKTEST RESULTS ===")
    print(df) # Display the results DataFrame

    stats = report_stats(df) # Print aggregated statistics
    plot_four_charts(df) # Display the four analytical plots