← Back to Home
Relative Volume Spike Momentum Strategy with Trailing Stops - Optimization & Backtest

Relative Volume Spike Momentum Strategy with Trailing Stops - Optimization & Backtest

Price and volume are fundamental pillars of technical analysis, with volume often providing crucial confirmation for price movements. A “volume spike” – unusually high volume relative to its recent average – can signal strong conviction behind a price move, indicating potential trend initiation or acceleration. This article introduces the RelativeVolumeSpikeMomentumTrailingStopStrategy, a system that identifies high-probability trade setups by combining significant price momentum with a confirming relative volume spike, and critically, ensures disciplined profit protection and risk management through the consistent application of trailing stops.

RelativeVolumeSpikeStrategy.png

The Strategy’s Core Components

The strategy is built within backtrader, utilizing custom calculations for momentum and relative volume.

import backtrader as bt
import numpy as np

class RelativeVolumeSpikeMomentumTrailingStopStrategy(bt.Strategy):
    params = (
        ('momentum_roc_window', 7),           # Lookback window for Price Rate of Change (Momentum)
        ('momentum_buy_threshold', 0.01),     # Momentum % threshold to consider a long entry
        ('momentum_sell_threshold', -0.01),   # Momentum % threshold to consider a short entry
        ('volume_sma_window', 7),             # Window for Volume Simple Moving Average
        ('rvol_spike_threshold', 2.0),        # Relative Volume threshold for a spike (e.g., 2.0 = 2x average volume)
        ('trail_percent', 0.02),              # Trailing stop percentage (e.g., 2%)
    )
    
    def __init__(self):
        self.order = None # Tracks active entry/exit orders
        self.trailing_stop_order = None # Tracks the active trailing stop order

        # Price Rate of Change (momentum) - (Current_Close / Close_X_Bars_Ago) - 1
        self.price_roc = (self.data.close / self.data.close(-self.params.momentum_roc_window)) - 1
        
        # Volume SMA for relative volume calculation
        self.volume_sma = bt.indicators.SimpleMovingAverage(self.data.volume, period=self.params.volume_sma_window)
        
        # Relative Volume (RVOL) = Current_Volume / Volume_SMA
        self.rvol = self.data.volume / self.volume_sma
        
    def notify_order(self, order):
        # Ignore submitted or accepted orders as they are still pending
        if order.status in [order.Submitted, order.Accepted]:
            return

        # If an order has completed (filled, canceled, rejected, etc.)
        if order.status in [order.Completed]:
            # If the completed order was a buy order that established a long position
            if order.isbuy():
                if self.position.size > 0: # Ensure we are actually long
                    # Place a trailing sell stop immediately after entry
                    self.trailing_stop_order = self.sell(
                        exectype=bt.Order.StopTrail, 
                        trailpercent=self.params.trail_percent
                    )
            # If the completed order was a sell order
            elif order.issell():
                if self.position.size < 0: # If it was a short entry
                    # Place a trailing buy stop immediately after short entry
                    self.trailing_stop_order = self.buy(
                        exectype=bt.Order.StopTrail, 
                        trailpercent=self.params.trail_percent
                    )
                else: # If it was a close order for a long position (either manually closed or by stop hit)
                    if self.trailing_stop_order:
                        self.cancel(self.trailing_stop_order) # Cancel any outstanding trailing stop
                    self.trailing_stop_order = None # Clear the reference

        # Clear the main order reference if the order is no longer active
        if order.status in [order.Completed, order.Canceled, order.Rejected, order.Margin]:
            self.order = None
            if order == self.trailing_stop_order: # If the order that completed/canceled was our trailing stop
                self.trailing_stop_order = None # Clear its reference too
  1. Price Rate of Change (Momentum): The strategy calculates momentum as the percentage change in the closing price over a momentum_roc_window (e.g., 7 periods). This indicates the recent strength and direction of price movement.
  2. Relative Volume (RVOL): RVOL is a powerful confirmation tool, calculated as the current volume divided by its volume_sma_window (e.g., 7-period) Simple Moving Average. A rvol_spike_threshold (e.g., 2.0 for 200% of average volume) identifies instances of significantly higher-than-average volume.
  3. Trailing Stops: Crucially, the strategy adheres to the principle of always using trailing stops. Upon the execution of any entry order, a bt.Order.StopTrail order is immediately placed, set to trail the price by a trail_percent (e.g., 2%). This dynamic stop-loss mechanism protects profits as the trade moves favorably and limits losses if the market reverses.

Execution Logic

The next method processes signals for entries, while notify_order manages the placement and cancellation of trailing stops.

    def next(self):
        # Ensure sufficient data for all indicators to be calculated
        if len(self.data) < max(self.params.momentum_roc_window, self.params.volume_sma_window) + 1: # +1 for lagged values
            return
            
        # Do not issue new orders if an order is already pending
        if self.order:
            return
            
        position = self.position.size # Current position size
        
        # Get previous values for signal generation to avoid lookahead bias
        # Using [-1] for previous bar's calculated value
        prev_roc = self.price_roc[-1]
        prev_rvol = self.rvol[-1]
        
        # Check for NaN values in indicator outputs
        if np.isnan(prev_roc) or np.isnan(prev_rvol):
            return
            
        # --- Entry Logic (only if currently flat) ---
        if position == 0:
            volume_spike_confirmed = prev_rvol > self.params.rvol_spike_threshold
            
            if not volume_spike_confirmed: # Only consider entries if there's a volume spike
                return
                
            # Long entry: Positive momentum AND confirmed by a volume spike
            if prev_roc > self.params.momentum_buy_threshold:
                self.order = self.buy() # Place buy order
                
            # Short entry: Negative momentum AND confirmed by a volume spike
            elif prev_roc < self.params.momentum_sell_threshold:
                self.order = self.sell() # Place sell order

        # --- Exit Logic ---
        # All exits are handled by the trailing stop orders placed in notify_order.
        # There is no additional momentum-based exit in this version.

The strategy generates entry signals only when there is no open position. A trade is initiated if:

  1. A significant relative volume spike occurs on the previous bar (prev_rvol > rvol_spike_threshold).
  2. Confirmed by a strong momentum signal on the previous bar:
    • For a long entry: prev_roc is greater than momentum_buy_threshold.
    • For a short entry: prev_roc is less than momentum_sell_threshold.

Upon successful entry, the trailing stop automatically takes effect, allowing the strategy to capture profits during trend continuation and exit gracefully upon reversal.

Parameter Optimization: Finding the Best Fit

Parameter optimization systematically tests various combinations of a strategy’s input parameters to find those that yield the best historical performance according to a chosen metric (e.g., Sharpe Ratio, total return). This process helps in identifying the most effective settings for a given strategy on a specific dataset.

import backtrader as bt
import numpy as np
import pandas as pd
import yfinance as yf # Assuming yfinance is used for data fetching

def optimize_parameters(strategy_class, opt_params, ticker, start_date, end_date):
    """Run optimization to find best parameters with diagnostics"""
    print("="*60)
    print(f"OPTIMIZING: {strategy_class.__name__} on {ticker}")
    print("="*60)

    # Fetch data for optimization
    print(f"Fetching data from {start_date} to {end_date}...")
    # User-specified: yfinance download with auto_adjust=False and droplevel(axis=1, level=1)
    df = yf.download(ticker, start=start_date, end=end_date, auto_adjust=False, progress=False)
    if isinstance(df.columns, pd.MultiIndex):
        df = df.droplevel(1, axis=1)

    if df.empty:
        print("No data fetched for optimization. Exiting.")
        return None

    print(f"Data shape: {df.shape}")
    print(f"Date range: {df.index[0].date()} to {df.index[-1].date()}")

    # Set up optimization
    cerebro = bt.Cerebro()
    data = bt.feeds.PandasData(dataname=df)
    cerebro.adddata(data)
    
    start_cash = 10000.0
    cerebro.broker.setcash(start_cash)
    cerebro.broker.setcommission(commission=0.001)
    
    # Add analyzers for performance metrics
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe', timeframe=bt.TimeFrame.Days, annualize=True, riskfreerate=0.0)
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
    cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')

    print("Testing parameter combinations...")
    cerebro.optstrategy(strategy_class, **opt_params) # Run the optimization

    stratruns = cerebro.run()
    print(f"Optimization complete! Tested {len(stratruns)} combinations.")

    # Collect and analyze results
    results = []
    for i, run in enumerate(stratruns):
        strategy = run[0]
        sharpe_analysis = strategy.analyzers.sharpe.get_analysis()
        returns_analysis = strategy.analyzers.returns.get_analysis()
        trades_analysis = strategy.analyzers.trades.get_analysis()
        
        rtot = returns_analysis.get('rtot', 0.0)
        final_value = start_cash * (1 + rtot)
        sharpe_ratio = sharpe_analysis.get('sharperatio', -999.0) # Default to a low number
        total_trades = trades_analysis.get('total', {}).get('total', 0)

        if sharpe_ratio is None or np.isnan(sharpe_ratio):
            sharpe_ratio = -999.0

        result = {
            'sharpe_ratio': sharpe_ratio,
            'final_value': final_value,
            'return_pct': rtot * 100,
            'total_trades': total_trades,
        }
        
        # Dynamically add parameter values to the results
        param_values = {p: getattr(strategy.p, p) for p in opt_params.keys()}
        result.update(param_values)
        
        results.append(result)

    # Filter for valid results (at least one trade) and sort
    valid_results = [r for r in results if r['total_trades'] > 0]
    
    if not valid_results:
        print("\nNo combinations resulted in any trades. Cannot determine best parameters.")
        return None
        
    results_sorted = sorted(valid_results, key=lambda x: x['sharpe_ratio'], reverse=True)
    
    print(f"\n{'='*120}")
    print("TOP 5 PARAMETER COMBINATIONS BY SHARPE RATIO")
    print(f"{'='*120}")
    
    top_5_df = pd.DataFrame(results_sorted[:5])
    print(top_5_df.to_string())
    
    best_params = results_sorted[0]
    print(f"\nBest Parameters Found: {best_params}")
    
    return best_params

Key Features of optimize_parameters:

Generalized Rolling Backtesting: Assessing Out-of-Sample Performance

Once optimal parameters are identified from an in-sample optimization period, a rolling backtest (also known as walk-forward optimization) assesses the strategy’s stability and performance on unseen data. This method simulates how a strategy would perform in live trading by iteratively optimizing on one period and testing on a subsequent, out-of-sample period.

import dateutil.relativedelta as rd # Needed for date calculations in rolling backtest

def run_rolling_backtest(strategy_class, strategy_params, ticker, start, end, window_months):
    """Generalized rolling backtest function"""
    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:
            break
            
        print(f"\nROLLING BACKTEST: {current_start.date()} to {current_end.date()}")
        
        # Fetch data for the current window
        # User-specified: 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)
        
        if data.empty or len(data) < 30: # Need at least some data for indicators to warm up
            print("Not enough data for this period. Skipping window.")
            current_start += rd.relativedelta(months=window_months)
            continue
            
        if isinstance(data.columns, pd.MultiIndex):
            data = data.droplevel(1, 1)
        
        # Calculate Buy & Hold return for the period as a benchmark
        start_price = data['Close'].iloc[0]
        end_price = data['Close'].iloc[-1]
        benchmark_ret = (end_price - start_price) / start_price * 100
        
        # Setup and run Cerebro for the current window
        feed = bt.feeds.PandasData(dataname=data)
        cerebro = bt.Cerebro()
        
        cerebro.addstrategy(strategy_class, **strategy_params) # Use the optimized parameters
        cerebro.adddata(feed)
        cerebro.broker.setcash(100000) # Initial cash for the window
        cerebro.broker.setcommission(commission=0.001)
        cerebro.addsizer(bt.sizers.PercentSizer, percents=95) # Allocate 95% of capital per trade
        cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')
        
        start_val = cerebro.broker.getvalue()
        results_run = cerebro.run()
        final_val = cerebro.broker.getvalue()
        strategy_ret = (final_val - start_val) / start_val * 100
        
        # Get trade statistics
        trades_analysis = results_run[0].analyzers.trades.get_analysis()
        total_trades = trades_analysis.get('total', {}).get('total', 0)
        
        all_results.append({
            'start': current_start.date(),
            'end': current_end.date(),
            'return_pct': strategy_ret,
            'benchmark_pct': benchmark_ret,
            'trades': total_trades,
        })
        
        print(f"Strategy Return: {strategy_ret:.2f}% | Buy & Hold Return: {benchmark_ret:.2f}% | Trades: {total_trades}")
        current_start = current_end # Move to the next window
        
    return pd.DataFrame(all_results)

Key Features of run_rolling_backtest:

Conclusion

The RelativeVolumeSpikeMomentumTrailingStopStrategy offers a focused yet robust approach to systematic trading. By combining the power of volume confirmation with clear momentum signals for entries and implementing disciplined trailing stops for risk management, it provides a compelling framework for capturing short-to-medium term trends in a dynamic market environment.