Adapting to Market Trends with Spectral Slope Adaptive Filter Strategy

The Spectral Slope Adaptive Filter Strategy introduces an advanced approach to dynamic indicator adaptation by leveraging spectral analysis to infer the “color” or underlying statistical properties of price movements. Instead of relying on traditional volatility measures, this strategy estimates the spectral slope of the price series to determine whether the market is in a trending (more predictable) or noisy (more random/mean-reverting) regime. This spectral slope then directly influences the smoothing period of an Exponential Moving Average (EMA), making the filter intelligently responsive to current market conditions. The strategy uses this adaptive EMA for trade signals and employs an ATR-based trailing stop for risk management.

What is Power Spectral Density?

The Power Spectral Density (PSD) describes how the power (variance) of a time series is distributed across different frequencies. It helps identify dominant cycles, noise, or trends in the signal.

For a discrete-time signal \(x[n]\):

\[ \text{PSD}(f) = \lim_{T \to \infty} \frac{1}{T} \left| \sum_{n=0}^{T-1} x[n] \, e^{-j2\pi fn} \right|^2 \]

This is the squared magnitude of the Fourier Transform, normalized by time.

Welch’s Method (Practical Estimation)

Used for noisy time series (like price):

1. Split signal into overlapping segments.
2. Apply window function (e.g., Hamming).
3. Compute FFT for each segment.
4. Average the squared FFTs.

\[ \text{PSD}(f) = \frac{1}{K} \sum_{k=1}^{K} \frac{|\text{FFT}(w_k x_k)[f]|^2}{U} \]

Where:

• \(K\): number of segments
• \(x_k\): k-th segment
• \(w_k\): window function
• \(U\): window normalization factor

Log-Log Slope (Spectral Slope)

From PSD output:

• Compute \(\log_{10}(f_i)\) and \(\log_{10}(\text{PSD}_i)\)
• Fit linear regression:

\[ \log_{10}(\text{PSD}_i) = a + b \cdot \log_{10}(f_i) \]

• The slope \(b\) quantifies trend/noise:

  • \(b \approx -3\): strong trend (low frequency)
  • \(b \approx -1\): noise (high frequency)

Strategy Overview

The SpectralSlopeAdaptiveFilter operates on a sophisticated adaptive mechanism and manages trades as follows:

Spectral Slope Analysis for Adaptation: The core innovation is the use of the spectral slope of the log-log power spectral density (PSD) of the price series.

• Concept: Financial time series often exhibit “colored noise” properties, where the power of price fluctuations changes with frequency. A common model is \(S(f) \\propto f^{-\\beta}\), where \(S(f)\) is the power at frequency \(f\), and \(\\beta\) is the spectral slope.
  • \(\\beta \\approx 0\) (slope near 0): White noise (random, no correlation).
  • \(\\beta \\approx 1\) (slope near -1): Pink noise (long-range correlations, “1/f noise”).
  • \(\\beta \\approx 2\) (slope near -2): Brownian motion (random walk, integrated white noise).
• Interpretation: A spectral slope closer to -2 (steeper negative slope) often indicates more trending or persistent behavior, while a slope closer to 0 or -1 might suggest more mean-reverting or random characteristics.
• Calculation: The strategy calculates this slope using Welch’s method for Power Spectral Density estimation, followed by a linear regression on the log-log plot of frequency versus PSD.

Adaptive EMA Calculation: The EMA’s smoothing period is dynamically determined by the calculated spectral slope:

• The spectral_slope is mapped to an EMA period between period_filter_min (for noisier markets, where a faster filter is desired) and period_filter_max (for trending markets, where a slower filter is desired).
• A slope closer to slope_map_noise results in a faster EMA, and a slope closer to slope_map_trend results in a slower EMA.
• This adaptive EMA (adaptive_ema) is then updated using the calculated period.

Entry Logic: The strategy uses a simple crossover for entry signals:

• Long Signal: If the previous day’s close price is above the current value of the adaptive_ema.
• Short Signal: If the previous day’s close price is below the current value of the adaptive_ema.

Exit Logic (ATR Trailing Stop): All open positions are managed with an ATR-based trailing stop:

• An initial trailing stop is set a multiple (atr_multiplier_sl) of the Average True Range (ATR) away from the entry price.
• For long positions, the stop price moves up as the price moves favorably (trailing the highest price since entry). If the price falls below this stop, the position is closed.
• For short positions, the stop price moves down as the price moves favorably (trailing the lowest price since entry). If the price rises above this stop, the position is closed.

Backtrader Implementation

Here’s the SpectralSlopeAdaptiveFilter class, including its helper methods:

import backtrader as bt
import numpy as np
from scipy import signal, stats
from collections import deque
import warnings

warnings.filterwarnings("ignore", category=stats.ConstantInputWarning)


class SpectralSlopeAdaptiveFilter(bt.Strategy):
    params = (
        ('spectrum_window', 30), # Window size for spectral analysis (price_buffer)
        ('spectrum_nperseg', 10), # Length of segments used in Welch's method (nperseg must be <= spectrum_window)
        ('slope_map_trend', -2.8), # Spectral slope value representing strong trend (steeper negative)
        ('slope_map_noise', -1.2), # Spectral slope value representing noise/mean-reversion (less steep negative)
        ('period_filter_min', 15), # Minimum EMA period (for noisy/mean-reverting markets)
        ('period_filter_max', 150), # Maximum EMA period (for trending markets)
        ('atr_window_sl', 14), # Window for ATR calculation for stop loss
        ('atr_multiplier_sl', 2.0), # Multiplier for ATR to set trailing stop distance
    )
    
    def __init__(self):
        # ATR for stop loss
        self.atr = bt.indicators.AverageTrueRange(self.data, period=self.params.atr_window_sl)
        
        # State variables
        self.price_buffer = deque(maxlen=self.params.spectrum_window) # Buffer to hold price data for spectral analysis
        self.adaptive_ema = None
        self.spectral_slope = -2.0  # Default to Brownian motion slope, typical for financial data
        self.trailing_stop = None
        self.entry_price = None # Stores entry price for initial stop calculation

    def _calculate_spectral_slope(self, price_segment):
        """Calculate log-log slope of price spectrum using Welch's method"""
        # Ensure sufficient data for nperseg and detrending
        if len(price_segment) < self.params.spectrum_nperseg // 2 or len(price_segment) < 2:
            return -2.0  # Default Brownian slope if data is insufficient
            
        try:
            # Detrend the price segment to focus on fluctuations, not overall trend
            detrended = signal.detrend(price_segment)
            
            # If detrended series is almost constant, return default slope
            if np.std(detrended) < 1e-9: # Check for near-zero standard deviation
                return -2.0
                
            # Calculate power spectral density (PSD) using Welch's method
            # fs=1.0 for daily data, nperseg is segment length, scaling='density' gives PSD
            freqs, psd = signal.welch(
                detrended, 
                fs=1.0, # Sampling frequency (e.g., 1.0 for daily data)
                nperseg=min(len(detrended), self.params.spectrum_nperseg), # Ensure nperseg <= data length
                scaling='density',
                nfft=max(self.params.spectrum_nperseg, len(detrended)) # NFFT should be at least nperseg
            )
            
            # Filter out zero frequencies (log10(0) is undefined) and very small PSD values
            valid_indices = np.where((freqs > 1e-6) & (psd > 1e-9))[0]
            
            # Need at least 2 points for a linear regression slope, but 3 is safer for stats.linregress
            if len(valid_indices) < 3:
                return -2.0
                
            log_freqs = np.log10(freqs[valid_indices])
            log_psd = np.log10(psd[valid_indices])
            
            # Check for sufficient variation in log_freqs or log_psd to avoid errors in linregress
            if np.std(log_freqs) < 1e-6 or np.std(log_psd) < 1e-6:
                return -2.0
                
            # Perform linear regression on log-log plot (log(PSD) vs log(freq))
            slope, _, _, _, _ = stats.linregress(log_freqs, log_psd)
            
            return slope
            
        except (ValueError, FloatingPointError):
            # Handle cases where inputs might be problematic for log or linregress
            return -2.0
    
    def _map_slope_to_period(self, slope):
        """Map spectral slope to EMA period"""
        # Clip slope to mapping range [slope_map_trend, slope_map_noise]
        clipped_slope = np.clip(slope, self.params.slope_map_trend, self.params.slope_map_noise)
        
        # Normalize the clipped slope to a 0-1 range
        # 0 corresponds to slope_map_trend (strong trend, max period)
        # 1 corresponds to slope_map_noise (noise/mean-reversion, min period)
        slope_range = self.params.slope_map_noise - self.params.slope_map_trend
        if slope_range == 0: # Avoid division by zero if thresholds are identical
            return (self.params.period_filter_min + self.params.period_filter_max) // 2
            
        norm_slope = (clipped_slope - self.params.slope_map_trend) / slope_range
        
        # Map normalized slope to the desired EMA period range
        # A smaller (more negative) slope -> closer to slope_map_trend (norm_slope close to 0) -> period_filter_max (slower EMA for trending)
        # A larger (less negative) slope -> closer to slope_map_noise (norm_slope close to 1) -> period_filter_min (faster EMA for noisy)
        period = self.params.period_filter_min + (1 - norm_slope) * (self.params.period_filter_max - self.params.period_filter_min)
        
        return int(np.clip(np.round(period), self.params.period_filter_min, self.params.period_filter_max))
    
    def _update_adaptive_ema(self):
        """Update the spectral-slope adaptive EMA"""
        current_price = self.data.close[0]
        
        # Calculate spectral slope only if we have enough data in the buffer
        if len(self.price_buffer) == self.params.spectrum_window:
            self.spectral_slope = self._calculate_spectral_slope(list(self.price_buffer))
        
        # Map the current spectral slope to an adaptive EMA period
        adaptive_period = self._map_slope_to_period(self.spectral_slope)
        
        # Calculate alpha (smoothing factor) for the EMA based on the adaptive period
        alpha = 2 / (adaptive_period + 1)
        
        # Update the adaptive EMA
        if self.adaptive_ema is None:
            self.adaptive_ema = current_price # Initialize with current price on first valid bar
        else:
            self.adaptive_ema = alpha * current_price + (1 - alpha) * self.adaptive_ema
            
    def next(self):
        # Ensure sufficient data for initial calculations (buffers, indicators)
        if len(self.data) < max(self.params.spectrum_window, self.params.atr_window_sl):
            return
            
        current_close = self.data.close[0]
        current_atr = self.atr[0]
        
        # Update price buffer with current close price
        self.price_buffer.append(current_close)
        
        # Update the adaptive EMA for the current bar
        self._update_adaptive_ema()
        
        # Ensure adaptive EMA has been initialized before using it for signals
        if self.adaptive_ema is None:
            return
            
        position = self.position.size # Current position size
        
        # --- Handle existing positions: Check Trailing Stops First ---
        # For long positions
        if position > 0:
            if self.trailing_stop is None: # Initialize stop on the first bar after entry
                self.trailing_stop = self.entry_price - self.params.atr_multiplier_sl * current_atr
            elif self.data.low[0] <= self.trailing_stop: # Check if price hit stop loss
                self.close() # Close position
                self.trailing_stop = None # Reset stop for next trade
                self.entry_price = None # Reset entry price
                return # Exit next() after closing
            else: # Update trailing stop if price moved favorably
                new_stop = current_close - self.params.atr_multiplier_sl * current_atr
                if new_stop > self.trailing_stop: # Stop only moves up
                    self.trailing_stop = new_stop
                    
        # For short positions
        elif position < 0:
            if self.trailing_stop is None: # Initialize stop on the first bar after entry
                self.trailing_stop = self.entry_price + self.params.atr_multiplier_sl * current_atr
            elif self.data.high[0] >= self.trailing_stop: # Check if price hit stop loss
                self.close() # Close position
                self.trailing_stop = None # Reset stop for next trade
                self.entry_price = None # Reset entry price
                return # Exit next() after closing
            else: # Update trailing stop if price moved favorably
                new_stop = current_close + self.params.atr_multiplier_sl * current_atr
                if new_stop < self.trailing_stop: # Stop only moves down
                    self.trailing_stop = new_stop
                    
        # --- Entry signals based on adaptive EMA crossover - only if no position open ---
        if position == 0:
            prev_close = self.data.close[-1]
            
            # Long signal: previous close crosses above adaptive EMA
            if prev_close > self.adaptive_ema and self.data.close[-2] <= self.adaptive_ema: # Added crossover condition
                self.buy() # Place buy order
                self.entry_price = self.data.open[0] # Record entry price for initial stop
                # Initial trailing stop will be set on the next bar in the next() call or through notify_order.
                # It's better to explicitly set it here for immediate effect or in notify_order.
                # For this strategy, setting it here after order is placed is safer as notify_order runs after next().
                self.trailing_stop = self.entry_price - self.params.atr_multiplier_sl * current_atr
                
            # Short signal: previous close crosses below adaptive EMA
            elif prev_close < self.adaptive_ema and self.data.close[-2] >= self.adaptive_ema: # Added crossover condition
                self.sell() # Place sell order
                self.entry_price = self.data.open[0] # Record entry price for initial stop
                # Initial trailing stop set here
                self.trailing_stop = self.entry_price + self.params.atr_multiplier_sl * current_atr

Parameters (params):

• spectrum_window: The lookback window (number of bars) used to collect price data for the spectral analysis. This determines the segment length for the PSD calculation.
• spectrum_nperseg: The length of segments (in bars) to use for the Welch’s method PSD estimation. It must be less than or equal to spectrum_window.
• slope_map_trend: A spectral slope value (e.g., -2.8) that is mapped to the period_filter_max (slower EMA). Slopes more negative than this suggest strong trending.
• slope_map_noise: A spectral slope value (e.g., -1.2) that is mapped to the period_filter_min (faster EMA). Slopes less negative (closer to 0) suggest noisier/mean-reverting markets.
• period_filter_min: The minimum period for the adaptive EMA, used when the market is identified as noisy/mean-reverting.
• period_filter_max: The maximum period for the adaptive EMA, used when the market is identified as strongly trending.
• atr_window_sl: The lookback window for the Average True Range (ATR) indicator, used to calculate the trailing stop distance.
• atr_multiplier_sl: The multiplier for the ATR, determining how far the trailing stop is placed from the price.

Initialization (__init__):

• self.atr: Initializes the Average True Range indicator, used for calculating the dynamic trailing stop.
• self.price_buffer: A deque (double-ended queue) of fixed size (spectrum_window) to store the recent closing prices. This buffer serves as the input data segment for the spectral analysis.
• self.adaptive_ema: Stores the current value of the dynamically adjusted Exponential Moving Average.
• self.spectral_slope: Stores the latest calculated spectral slope, defaulting to -2.0 (representing Brownian motion/random walk).
• self.trailing_stop: Stores the current price level of the ATR-based trailing stop.
• self.entry_price: Stores the entry price of the current position, used for setting the initial trailing stop.

Helper Methods for Spectral Analysis and Adaptive EMA:

• _calculate_spectral_slope(self, price_segment):
  • Purpose: This crucial method estimates the spectral slope (\(\\beta\)) of a given price_segment. The slope indicates the underlying “color” of the price movement.
  • Detrending: It first removes any linear trend from the price_segment using scipy.signal.detrend. This focuses the analysis on fluctuations rather than the overall directional bias, which is handled by the EMA itself.
  • Power Spectral Density (PSD): It then applies scipy.signal.welch to compute the PSD of the detrended price data. welch breaks the signal into overlapping segments, computes a modified periodogram for each, and averages them to get the PSD.
  • Log-Log Regression: It performs a linear regression on the log of frequencies and the log of their corresponding PSD values (log_freqs vs log_psd). The slope of this regression line is the spectral slope.
  • Error Handling: Includes checks for insufficient data, constant input (zero standard deviation), zero frequencies, or problematic PSD values to prevent errors and return a default Brownian slope (-2.0) in such cases.
• _map_slope_to_period(self, slope):
  • Purpose: This method translates the calculated spectral_slope into a suitable period for the adaptive EMA.
  • Clipping: The input slope is clipped between slope_map_trend (more negative, indicating strong trend) and slope_map_noise (less negative, indicating more noise/mean-reversion).
  • Normalization: The clipped slope is normalized to a 0-1 range. A slope closer to slope_map_trend (e.g., -2.8) results in a normalized value close to 0, which maps to period_filter_max (a slower EMA for trending markets). A slope closer to slope_map_noise (e.g., -1.2) results in a normalized value closer to 1, which maps to period_filter_min (a faster EMA for noisy markets).
  • Period Mapping: The normalized slope is then scaled to the desired range of EMA periods (period_filter_min to period_filter_max). The mapping ensures that trending markets (steeper negative slope) get a slower EMA, and noisy markets (less steep negative slope) get a faster EMA.
• _update_adaptive_ema():
  • Purpose: Updates the value of self.adaptive_ema for the current bar using the dynamically determined period.
  • It first calls _calculate_spectral_slope (if price_buffer is full) to get the spectral_slope.
  • Then, it uses _map_slope_to_period to convert this slope into an adaptive_period.
  • The standard EMA smoothing factor alpha = 2 / (period + 1) is calculated using this adaptive_period.
  • Finally, the self.adaptive_ema is updated using the current price and the new alpha.

Main Logic (next):

The next method is executed on each new bar of data and orchestrates the strategy’s operations:

1. Data Sufficiency Check: Ensures enough historical data is available in the data feed for the spectral analysis window and ATR calculation to warm up.
2. Price Buffer Update: The current closing price is appended to self.price_buffer.
3. Adaptive EMA Update: self._update_adaptive_ema() is called to compute the current value of the adaptive EMA.
4. Position Management (Trailing Stop): This section, executed if a position is open, is responsible for managing the trailing stop:
   • Initialization: If a position is just opened (i.e., self.trailing_stop is None), the initial stop price is calculated using the entry_price (recorded upon order placement) and the current ATR.
   • Stop Loss Check: For long positions, if the current data.low[0] falls below or equals self.trailing_stop, the position is closed. For short positions, if data.high[0] rises above or equals self.trailing_stop, the position is closed. The trailing_stop and entry_price are reset, and the method returns.
   • Stop Update: If the stop loss is not triggered, the trailing_stop is updated. For long positions, it only moves up if the price moves favorably; for short positions, it only moves down.
5. Entry Signals (if no position open):
   • If no position is currently open (position == 0), the strategy looks for a crossover of the previous bar’s close price and the adaptive_ema:
     • Long Entry: If prev_close crosses above self.adaptive_ema (i.e., prev_close > self.adaptive_ema and data.close[-2] <= self.adaptive_ema), a buy order is placed. The entry_price is recorded for the subsequent trailing stop initialization.
     • Short Entry: If prev_close crosses below self.adaptive_ema (i.e., prev_close < self.adaptive_ema and data.close[-2] >= self.adaptive_ema), a sell order is placed. The entry_price is recorded.
   • The initial trailing_stop for the new position is set immediately after the order is placed, using the recorded entry_price and current ATR.

Rolling Backtesting

Running the strategy over multiple, sequential time windows (e.g., yearly) across a broad historical dataset. This helps assess the strategy’s robustness and consistency under different market conditions.