← Back to Home
Rolling Window Backtesting of a Composite Momentum Strategy

Rolling Window Backtesting of a Composite Momentum Strategy

Effective algorithmic trading relies on thorough backtesting. While a single backtest over a long period can provide an overall picture, it might hide periods of poor performance or indicate that a strategy is over-optimized for specific historical conditions. Rolling backtests address this by evaluating a strategy’s performance over successive, overlapping (or non-overlapping) time windows, providing a more dynamic and reliable assessment of its consistency and adaptability.

Here we use a Python framework for performing rolling backtests using backtrader and yfinance, exploring the idea of a “Composite Price-Volume Momentum Strategy”.

1. Essential Imports and Strategy Selection

We start by importing all necessary libraries for data handling, plotting, and backtrader functionalities. The CompositePVMomentumStrategy is chosen as our subject for the rolling backtest.

import backtrader as bt
import yfinance as yf
import pandas as pd
import numpy as np
import dateutil.relativedelta as rd # For easier date arithmetic in rolling windows
import matplotlib.pyplot as plt
import seaborn as sns # For enhanced plotting

# The strategy to be tested in the rolling backtest
from CompositePVMomentumStrategy import CompositePVMomentumStrategy
strategy = CompositePVMomentumStrategy

Explanation:

2. The CompositePVMomentumStrategy Explained

This strategy combines price-based and volume-based momentum signals, along with RSI and trailing stops, to identify entry and exit points.

import backtrader as bt
import yfinance as yf
import pandas as pd

class CompositePVMomentumStrategy(bt.Strategy):
    params = (
        ('rsi_period', 30),
        ('momentum_period', 30),
        ('volume_ma_period', 30),
        ('price_ma_period', 30),
        ('trailing_percent', 5.0), # Trailing stop percentage
        ('momentum_threshold', 0.01), # Price change threshold for momentum
        ('volume_threshold', 1.2), # Volume ratio threshold
        ('rsi_oversold', 30),
        ('rsi_overbought', 70),
    )
    
    def __init__(self):
        # Data references
        self.data_close = self.data.close
        self.data_volume = self.data.volume

        # Price indicators
        self.rsi = bt.indicators.RSI(period=self.params.rsi_period)
        self.momentum = bt.indicators.Momentum(period=self.params.momentum_period) # Backtrader's Momentum (current price / price N periods ago - 1)
        self.price_ma = bt.indicators.SMA(self.data_close, period=self.params.price_ma_period)
        
        # Volume indicators
        self.volume_ma = bt.indicators.SMA(self.data_volume, period=self.params.volume_ma_period)
        # Ratio of current volume to its moving average
        self.volume_ratio = self.data_volume / self.volume_ma
        
        # Custom price momentum calculation (percentage change over momentum_period)
        self.price_momentum_calc = (self.data_close - self.data_close(-self.params.momentum_period)) / self.data_close(-self.params.momentum_period)
        
        # Trailing stop variables
        self.trailing_stop_long = None
        self.trailing_stop_short = None
        self.highest_price = None # Tracks highest price since long entry
        self.lowest_price = None  # Tracks lowest price since short entry
        
    def next(self):
        current_price = self.data_close[0]
        
        # Ensure enough data points for all indicators to be calculated
        # The longest period amongst all indicators
        min_required_len = max(self.params.rsi_period, 
                               self.params.momentum_period, 
                               self.params.volume_ma_period, 
                               self.params.price_ma_period)
        
        if len(self.data_close) < min_required_len:
            return # Not enough data for indicators to be valid

        # Update trailing stops for existing positions
        if self.position.size > 0:  # Long position
            # Update highest price seen since entry
            if self.highest_price is None or current_price > self.highest_price:
                self.highest_price = current_price
            # Calculate and update trailing stop
            self.trailing_stop_long = self.highest_price * (1 - self.params.trailing_percent / 100)
            
            # Check trailing stop exit
            if current_price <= self.trailing_stop_long:
                self.close()
                self.highest_price = None # Reset for next trade
                self.trailing_stop_long = None
                
        elif self.position.size < 0:  # Short position
            # Update lowest price seen since entry
            if self.lowest_price is None or current_price < self.lowest_price:
                self.lowest_price = current_price
            # Calculate and update trailing stop
            self.trailing_stop_short = self.lowest_price * (1 + self.params.trailing_percent / 100)
            
            # Check trailing stop exit
            if current_price >= self.trailing_stop_short:
                self.close()
                self.lowest_price = None # Reset for next trade
                self.trailing_stop_short = None
                
        # Entry signals when no position
        if self.position.size == 0:
            # Composite momentum conditions using current bar values [0]
            price_above_ma = current_price > self.price_ma[0]
            price_below_ma = current_price < self.price_ma[0]
            strong_volume = self.volume_ratio[0] > self.params.volume_threshold
            positive_momentum = self.price_momentum_calc[0] > self.params.momentum_threshold
            negative_momentum = self.price_momentum_calc[0] < -self.params.momentum_threshold
            
            # Long entry conditions
            long_signal = (
                price_above_ma and
                positive_momentum and
                strong_volume and
                self.rsi[0] > self.params.rsi_oversold and # RSI not overbought, not oversold
                self.rsi[0] < self.params.rsi_overbought
            )
            
            # Short entry conditions  
            short_signal = (
                price_below_ma and
                negative_momentum and
                strong_volume and
                self.rsi[0] < self.params.rsi_overbought and # RSI not overbought, not oversold
                self.rsi[0] > self.params.rsi_oversold
            )
            
            if long_signal:
                self.buy()
                self.highest_price = current_price # Set initial high for trailing stop
                self.trailing_stop_long = current_price * (1 - self.params.trailing_percent / 100)
                
            elif short_signal:
                self.sell()
                self.lowest_price = current_price # Set initial low for trailing stop
                self.trailing_stop_short = current_price * (1 + self.params.trailing_percent / 100)

Explanation of CompositePVMomentumStrategy:

3. Rolling Backtest Function (run_rolling_backtest)

This function orchestrates the rolling backtest process, dividing the total backtesting period into smaller, sequential windows.

def run_rolling_backtest(
    ticker="SOL-USD",
    start="2018-01-01",
    end="2025-06-24",
    window_months=3, # Length of each rolling window in months
    strategy_params=None
):
    strategy_params = strategy_params or {} # Use provided params or an empty dict
    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)
        if current_end > end_dt:
            # Adjust final window end to not exceed overall end date
            current_end = end_dt 
            if current_start >= current_end: # Break if the window becomes invalid (e.g., current_start is already past end_dt)
                break

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

        # Download data for the current window
        # Using auto_adjust=False and droplevel(axis=1, level=1) as per user's saved preference
        data = yf.download(ticker, start=current_start, end=current_end, auto_adjust=False, progress=False)
        
        # Handle multi-level columns from yfinance as per user preference
        if isinstance(data.columns, pd.MultiIndex):
            data.columns = data.columns.droplevel(1)
            # Ensure correct column names after droplevel for backtrader
            data.rename(columns={'Open': 'open', 'High': 'high', 'Low': 'low', 'Close': 'close', 'Volume': 'volume'}, inplace=True)


        if data.empty or len(data) < 90: # Minimum data points required for meaningful indicators
            print(f"Not enough data for {current_start.date()} to {current_end.date()}. Skipping window.")
            current_start += rd.relativedelta(months=window_months)
            continue # Skip to the next window

        feed = bt.feeds.PandasData(dataname=data)
        cerebro = bt.Cerebro()
        cerebro.addstrategy(strategy, **strategy_params) # Pass strategy parameters
        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}")
        current_start += rd.relativedelta(months=window_months) # Move to the start of the next window

    return pd.DataFrame(all_results)

Explanation of run_rolling_backtest:

4. Reporting and Visualization Functions

These functions process the results from the rolling backtest to provide statistical summaries and insightful plots.

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 plots for rolling backtest analysis:
    1. Period Returns
    2. Cumulative Returns
    3. Rolling Sharpe Ratio
    4. Return Distribution
    """
    fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(12, 10)) # Adjusted figsize for better layout
    
    # 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
    ) # Use raw=False for proper pandas series handling
    
    # 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)) # Dynamic bin calculation for histogram
    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() # Adjust subplots to give a nice fit
    plt.show()

Explanation of Reporting and Visualization:

5. Main Execution Block

This if __name__ == '__main__': block demonstrates how to run the rolling backtest and display the results.

if __name__ == '__main__':
    # Run the rolling backtest
    df = run_rolling_backtest(ticker="SOL-USD", start="2018-01-01", end="2025-06-24", window_months=6) # Changed to 6-month window for example

    print("\n=== ROLLING BACKTEST RESULTS ===")
    print(df) # Print the DataFrame with results from each window

    stats = report_stats(df) # Report overall statistics
    plot_four_charts(df) # Plot the four charts for visual analysis

    # Example of a single backtest run with CompositePVMomentumStrategy
    # This part shows how the CompositePVMomentumStrategy would be run independently
    # run_strategy() # This function is defined below for a single backtest, not part of the rolling framework directly

Explanation:

Pasted image 20250628232405.png

This comprehensive setup provides a robust way to evaluate trading strategies over varying market conditions, moving beyond a single historical snapshot to provide a more reliable measure of performance.