Spectral Slope Adaptive Filter Strategy - Self-Adjusting Averages for Trend Following with Trailing Stops

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

warnings.filterwarnings("ignore", category=DeprecationWarning)
warnings.filterwarnings("ignore", category=RuntimeWarning)

class SpectralSlopeAdaptiveFilterTrailingStopStrategy(bt.Strategy):
    params = (
        ('spectrum_window', 128),     # Window size for spectral analysis
        ('spectrum_nperseg', 64),     # Segment length for Welch's method
        ('slope_map_trend', -2.5),    # Spectral slope considered strongly trending
        ('slope_map_noise', -1.0),    # Spectral slope considered noisy/ranging
        ('period_filter_min', 10),    # Minimum EMA period (for trending)
        ('period_filter_max', 100),   # Maximum EMA period (for noisy)
        ('atr_window_sl', 14),        # ATR window for trailing stop calculation
        ('atr_multiplier_sl', 2.0),   # Multiplier for ATR in trailing stop
    )
    
    def __init__(self):
        self.atr = bt.indicators.AverageTrueRange(self.data, period=self.params.atr_window_sl)
        
        self.price_buffer = deque(maxlen=self.params.spectrum_window)
        self.spectral_slope = -2.0 # Initial default to Brownian (trending)
        self.adaptive_ema = None   # The dynamically smoothed moving average
        self.trailing_stop = None  # Current trailing stop price
        self.entry_price = None    # Price at which position was entered

    def _calculate_spectral_slope(self, price_values):
        """Calculate log-log spectral slope using Welch's method."""
        if len(price_values) < self.params.spectrum_nperseg // 2 or np.std(price_values) < 1e-9:
            return np.nan
        try:
            detrended = signal.detrend(price_values)
            if np.std(detrended) < 1e-9: return np.nan
            
            freqs, psd = signal.welch(
                detrended, fs=1.0, nperseg=min(len(detrended), self.params.spectrum_nperseg),
                scaling='density', nfft=max(self.params.spectrum_nperseg, len(detrended))
            )
            valid_indices = np.where((freqs > 1e-6) & (psd > 1e-9))[0]
            if len(valid_indices) < 2: return np.nan
            
            log_freqs = np.log10(freqs[valid_indices])
            log_psd = np.log10(psd[valid_indices])
            
            if np.std(log_freqs) < 1e-6 or np.std(log_psd) < 1e-6: return np.nan
            
            slope, _, _, _, _ = stats.linregress(log_freqs, log_psd)
            return slope
        except (ValueError, FloatingPointError):
            return np.nan
    
    def _normalize_slope_to_period(self, slope):
        """Map spectral slope to EMA period based on defined ranges."""
        clipped_slope = np.clip(slope, self.params.slope_map_trend, self.params.slope_map_noise)
        range_val = self.params.slope_map_noise - self.params.slope_map_trend
        if range_val == 0: return (self.params.period_filter_min + self.params.period_filter_max) // 2
            
        normalized = (clipped_slope - self.params.slope_map_trend) / range_val
        # Map: trendier (more negative slope) -> shorter period, noisier -> longer period
        period = self.params.period_filter_min + (1 - normalized) * (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 dynamically adjusted EMA."""
        current_price = self.data.close[0]
        
        if len(self.price_buffer) == self.params.spectrum_window:
            # Calculate slope on the full buffer
            slope = self._calculate_spectral_slope(list(self.price_buffer))
            if not np.isnan(slope):
                self.spectral_slope = slope
        
        # Use the latest spectral slope (from previous bar's calculation) for current EMA period
        adaptive_period = self._normalize_slope_to_period(self.spectral_slope)
        alpha = 2 / (adaptive_period + 1)
        
        if self.adaptive_ema is None:
            self.adaptive_ema = current_price
        else:
            self.adaptive_ema = alpha * current_price + (1 - alpha) * self.adaptive_ema

    def next(self):
        # Ensure sufficient data for all calculations
        if len(self.data) < max(self.params.spectrum_window, self.params.atr_window_sl) + 1: # +1 for prev_close
            return
            
        current_close = self.data.close[0]
        prev_close = self.data.close[-1] # Used for EMA crossover check
        current_atr = self.atr[0]
        
        # Update price buffer for spectral analysis
        self.price_buffer.append(current_close)
        
        # Update the adaptive EMA
        self._update_adaptive_ema()
        
        if self.adaptive_ema is None or np.isnan(current_atr) or np.isnan(self.adaptive_ema):
            return # Ensure indicators are ready

        position = self.position.size
        
        # --- Exit Logic: ATR-based Trailing Stop ---
        # This implementation uses a manual trailing stop for demonstration.
        # For production, backtrader's bt.Order.StopTrail is often preferred
        # (as seen in previous examples) for robustness in event handling.
        
        if position > 0: # Currently Long
            # Re-calculate trailing stop: Price - ATR_Multiplier * ATR
            new_stop_level = current_close - self.params.atr_multiplier_sl * current_atr
            # Trailing stop only moves up (for long positions)
            if self.trailing_stop is None or new_stop_level > self.trailing_stop:
                self.trailing_stop = new_stop_level
            
            # Check if price has hit the trailing stop
            if current_close < self.trailing_stop: # Use current_close instead of low for strictness
                self.close()
                self.trailing_stop = None
                self.entry_price = None # Reset
        
        elif position < 0: # Currently Short
            # Re-calculate trailing stop: Price + ATR_Multiplier * ATR
            new_stop_level = current_close + self.params.atr_multiplier_sl * current_atr
            # Trailing stop only moves down (for short positions)
            if self.trailing_stop is None or new_stop_level < self.trailing_stop:
                self.trailing_stop = new_stop_level
            
            # Check if price has hit the trailing stop
            if current_close > self.trailing_stop: # Use current_close instead of high for strictness
                self.close()
                self.trailing_stop = None
                self.entry_price = None # Reset
                
        # --- Entry Logic (only if not in a position) ---
        if position == 0:
            # Long signal: previous close crosses ABOVE adaptive EMA
            if prev_close < self.adaptive_ema and current_close > self.adaptive_ema:
                self.buy()
                self.entry_price = current_close # Use current close for initial stop reference
                # Initialize trailing stop upon entry
                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 current_close < self.adaptive_ema:
                self.sell()
                self.entry_price = current_close
                # Initialize trailing stop upon entry
                self.trailing_stop = self.entry_price + self.params.atr_multiplier_sl * current_atr

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

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)