Volatility-Driven Trend Following Rolling Backtest of Volatility Momentum Strategy with Adaptive Stops

import backtrader as bt
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

class SimpleVolatilityMomentumStrategy(bt.Strategy):
    """Simple Volatility Momentum: When vol accelerates, trade with price direction"""
    
    params = (
        ('vol_window', 30),          # Volatility calculation period (for rolling std of returns)
        ('vol_momentum_window', 7),  # Volatility momentum lookback (current vol vs. vol N bars ago)
        ('price_sma_window', 30),    # Price trend SMA period
        ('atr_window', 14),          # ATR period for stop loss
        ('atr_multiplier', 5.0),     # ATR stop multiplier
    )
    
    def __init__(self):
        # Calculate daily percentage returns
        self.returns = bt.indicators.PctChange(self.data.close, period=1)
        
        # Volatility = rolling standard deviation of returns
        self.volatility = bt.indicators.StandardDeviation(self.returns, period=self.params.vol_window)
        
        # Volatility momentum = Current Volatility - Volatility N bars ago
        self.vol_momentum = self.volatility - self.volatility(-self.params.vol_momentum_window)
        
        # Price trend = Simple Moving Average of the close price
        self.price_sma = bt.indicators.SMA(self.data.close, period=self.params.price_sma_window)
        
        # ATR for stop loss calculations
        self.atr = bt.indicators.ATR(self.data, period=self.params.atr_window)
        
        # Trading variables for internal state
        self.stop_price = 0       # Stores the current stop loss price for the active position
        self.trade_count = 0      # Counts the number of trades executed

    def log(self, txt, dt=None):
        ''' Logging function for strategy actions '''
        dt = dt or self.datas[0].datetime.date(0)
        print(f'{dt.isoformat()}: {txt}')

    def notify_order(self, order):
        # Log status of completed orders
        if order.status in [order.Completed]:
            if order.isbuy():
                self.log(f'LONG EXECUTED - Price: {order.executed.price:.2f}')
            elif order.issell():
                # Differentiate between closing a long and opening a short
                if self.position.size == 0: # If position size is now zero, it was a closing order
                    self.log(f'POSITION CLOSED - Price: {order.executed.price:.2f}')
                else: # Otherwise, it was an opening short order
                    self.log(f'SHORT EXECUTED - Price: {order.executed.price:.2f}')
        # No explicit handling for Canceled/Rejected here, relying on default backtrader behavior or allowing next bar to re-evaluate.

    def notify_trade(self, trade):
        # Log profit/loss when a trade is fully closed
        if trade.isclosed:
            self.log(f'TRADE CLOSED - PnL: {trade.pnl:.2f}')
            # Reset stop price and trade count on trade closure (if needed for next trade's stop)
            self.stop_price = 0


    def next(self):
        # Ensure sufficient data for all indicators to be calculated
        # The longest period is max(vol_window, price_sma_window, atr_window) + vol_momentum_window for lookback
        min_bars_needed = max(self.params.vol_window, self.params.price_sma_window, self.params.atr_window) + self.params.vol_momentum_window
        if len(self) < min_bars_needed:
            return
            
        # Get current values of indicators
        vol_momentum = self.vol_momentum[0]
        current_price = self.data.close[0]
        sma_price = self.price_sma[0]
        current_atr = self.atr[0]
        
        # --- Risk Management: Check and update Stop Loss ---
        if self.position: # If we are currently in a position
            # 1. Check if stop loss has been hit (fixed stop based on initial calculation)
            if self.position.size > 0 and current_price <= self.stop_price: # Long position, price falls to stop
                self.close()
                self.log(f'STOP LOSS HIT (Long) - Closed at {current_price:.2f}')
                return # Exit from next() after closing position
            elif self.position.size < 0 and current_price >= self.stop_price: # Short position, price rises to stop
                self.close()
                self.log(f'STOP LOSS HIT (Short) - Closed at {current_price:.2f}')
                return # Exit from next() after closing position
            
            # 2. Trailing Stop Update: Adjust stop price only if it moves favorably
            # This is done *after* checking if the stop was hit, for current bar.
            # If the position is still open after stop check:
            if self.position.size > 0: # Long position
                new_potential_stop = current_price - (current_atr * self.params.atr_multiplier)
                if new_potential_stop > self.stop_price: # Only raise the stop price
                    self.stop_price = new_potential_stop
                    # self.log(f'Updated Long Stop to {self.stop_price:.2f}') # Optional: verbose logging
            elif self.position.size < 0: # Short position
                new_potential_stop = current_price + (current_atr * self.params.atr_multiplier)
                if new_potential_stop < self.stop_price: # Only lower the stop price
                    self.stop_price = new_potential_stop
                    # self.log(f'Updated Short Stop to {self.stop_price:.2f}') # Optional: verbose logging
        
        # --- Exit if Volatility Momentum reverses ---
        # This acts as a primary exit signal based on the strategy's core premise
        if self.position and vol_momentum <= 0:
            self.close()
            self.log(f'VOL MOMENTUM EXIT - Vol momentum: {vol_momentum:.6f} (No longer accelerating)')
            return # Exit from next() after closing position
            
        # --- Entry Signals: Volatility accelerating + price direction ---
        # Only enter if currently flat (no open position)
        if not self.position and vol_momentum > 0: # Volatility must be accelerating
            
            # Long Entry: Price above SMA (uptrend)
            if current_price > sma_price:
                self.buy()
                # Set initial stop loss based on current price and ATR
                self.stop_price = current_price - (current_atr * self.params.atr_multiplier)
                self.trade_count += 1
                self.log(f'LONG ENTRY - Price: {current_price:.2f}, Vol Mom: {vol_momentum:.6f}, Initial Stop: {self.stop_price:.2f}')
                
            # Short Entry: Price below SMA (downtrend)
            elif current_price < sma_price:
                self.sell()
                # Set initial stop loss based on current price and ATR
                self.stop_price = current_price + (current_atr * self.params.atr_multiplier)
                self.trade_count += 1
                self.log(f'SHORT ENTRY - Price: {current_price:.2f}, Vol Mom: {vol_momentum:.6f}, Initial Stop: {self.stop_price:.2f}')
        
    def stop(self):
        ''' Executed at the end of the backtest '''
        print(f'\n=== SIMPLE VOLATILITY MOMENTUM RESULTS ===')
        print(f'Total Trades: {self.trade_count}')
        print(f'Strategy Logic: When volatility accelerates, trade with price trend.')
        print(f'Parameters: Volatility Window={self.params.vol_window}d, Volatility Momentum Window={self.params.vol_momentum_window}d, Price SMA={self.params.price_sma_window}d')
        print(f'Stops: ATR {self.params.atr_window}d × {self.params.atr_multiplier} (Adaptive Trailing)')

# ... (imports from top of the rolling backtest script) ...
import dateutil.relativedelta as rd # Added import for rolling backtest
import seaborn as sns # Added import for plotting
from datetime import datetime # Added import for current date

# Define the strategy for the rolling backtest
strategy = SimpleVolatilityMomentumStrategy

def run_rolling_backtest(
    ticker="SOL-USD", # Default ticker for article's example
    start="2018-01-01",
    end=datetime.now().date(), # Set end date to current date for live testing
    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)
        if current_end > end_dt:
            current_end = end_dt 
            if current_start >= current_end:
                break

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

        # Data download using yfinance, respecting user's preference
        # 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 for strategy warm-up
        # Longest period is max(vol_window, price_sma_window, atr_window) + vol_momentum_window for lookback
        vol_window = strategy_params.get('vol_window', SimpleVolatilityMomentumStrategy.params.vol_window)
        vol_momentum_window = strategy_params.get('vol_momentum_window', SimpleVolatilityMomentumStrategy.params.vol_momentum_window)
        price_sma_window = strategy_params.get('price_sma_window', SimpleVolatilityMomentumStrategy.params.price_sma_window)
        atr_window = strategy_params.get('atr_window', SimpleVolatilityMomentumStrategy.params.atr_window)
        
        min_bars_needed = max(vol_window, price_sma_window, atr_window) + vol_momentum_window + 1 # +1 for current bar's data
        
        if data.empty or len(data) < min_bars_needed:
            print(f"Not enough data for period {current_start.date()} to {current_end.date()} (requires at least {min_bars_needed} bars). Skipping.")
            if current_end == end_dt:
                break
            current_start = current_end
            continue

        feed = bt.feeds.PandasData(dataname=data)
        cerebro = bt.Cerebro()
        cerebro.addstrategy(strategy, **strategy_params)
        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}")
        
        if current_end == end_dt:
            break
        current_start = current_end

    return pd.DataFrame(all_results)


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_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
    
    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')
    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
    )
    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')
    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__':
    # Use current date for the end of the backtest for a more "live" simulation
    current_date = datetime.now().date() 
    
    df = run_rolling_backtest(
        ticker="SOL-USD", # Default ticker for article's example
        start="2018-01-01",
        end=current_date, # Use the current date
        window_months=3,
        # strategy_params={ # Example of how to override default parameters
        #     'vol_window': 20,
        #     'vol_momentum_window': 10,
        #     'price_sma_window': 40,
        #     'atr_window': 20,
        #     'atr_multiplier': 6.0,
        # }
    )

    print("\n=== ROLLING BACKTEST RESULTS ===")
    print(df)

    stats = report_stats(df)
    plot_four_charts(df)