Hidden Markov Model Regime-Adaptive Momentum Strategy with Dynamic Lookbacks and Trailing Stops

import backtrader as bt
import numpy as np
import pandas as pd
from collections import deque
try:
    from hmmlearn import hmm
    HMM_AVAILABLE = True
except ImportError:
    HMM_AVAILABLE = False
    print("Warning: hmmlearn not available. Using simplified regime detection.")
import warnings

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


class HMMRegimeAdaptiveMomentumTrailingStopStrategy(bt.Strategy):
    params = (
        ('n_hmm_regimes', 2),                          # Number of hidden states for HMM
        ('hmm_train_ratio', 0.4),                      # Ratio of data for HMM training
        ('regime_momentum_lookbacks', {0: 63, 1: 252}), # Momentum lookbacks for each regime (e.g., 0=fast, 1=slow)
        ('absolute_momentum_threshold', 0.0),          # Momentum threshold for entry/exit
        ('retrain_frequency', 63),                     # Bars after which to retrain HMM
        ('min_data_for_hmm', 100),                     # Minimum data points to train HMM
        ('trail_percent', 0.02),                       # Trailing stop percentage (Added as per user preference)
    )
    
    def __init__(self):
        self.returns = bt.indicators.PctChange(self.data.close, period=1) # Daily returns
        
        # State variables for HMM and regime tracking
        self.returns_buffer = deque(maxlen=1000) # Stores returns for HMM training/prediction
        self.current_regime = 0                  # Default regime
        self.hmm_model = None
        self.last_retrain = 0
        self.hmm_trained = False
        
        # Order and position tracking
        self.order = None
        self.trailing_stop_order = None # To manage the trailing stop order

    def _simple_regime_detection(self, returns_data):
        """Fallback regime detection when HMM is not available or fails."""
        if len(returns_data) < 50:
            return 0 # Default to a safe regime if not enough data
            
        # Simple volatility-based regime: high vol = regime 0 (faster momentum), low vol = regime 1 (slower momentum)
        recent_vol = np.std(returns_data[-30:]) if len(returns_data) >= 30 else np.std(returns_data)
        historical_vol = np.std(returns_data)
        
        return 0 if recent_vol > historical_vol else 1
    
    def _train_hmm(self, returns_data):
        """Train HMM model on a portion of the returns data."""
        if not HMM_AVAILABLE:
            return False # HMM library not found
            
        if len(returns_data) < self.params.min_data_for_hmm:
            return False # Not enough data for robust HMM training
            
        try:
            # Prepare data for HMM (needs to be 2D array)
            hmm_data = np.array(returns_data).reshape(-1, 1)
            
            # Use a portion of data for training to save computation
            train_size = int(len(hmm_data) * self.params.hmm_train_ratio)
            train_data = hmm_data[max(0, len(hmm_data) - train_size):] # Use most recent data for training
            
            if len(train_data) < self.params.min_data_for_hmm: # Double check min data after slicing
                return False

            # Initialize and train HMM
            self.hmm_model = hmm.GaussianHMM(
                n_components=self.params.n_hmm_regimes,
                covariance_type="diag", # Diagonal covariance matrix
                n_iter=100,             # Number of iterations for EM algorithm
                random_state=42         # For reproducibility
            )
            
            self.hmm_model.fit(train_data)
            self.hmm_trained = True
            return True
            
        except Exception as e:
            # Log HMM training failures
            # print(f"HMM training failed: {e}") 
            self.hmm_trained = False
            return False
    
    def _predict_regime(self, returns_data):
        """Predict the current market regime using the trained HMM or fallback."""
        if not returns_data:
            return 0 # Default regime if no data
            
        if HMM_AVAILABLE and self.hmm_trained and self.hmm_model is not None:
            try:
                # Use recent data for prediction (e.g., last 50 bars)
                recent_data = np.array(returns_data[-50:]).reshape(-1, 1)
                if len(recent_data) > 0:
                    predicted_regimes = self.hmm_model.predict(recent_data)
                    return predicted_regimes[-1] # Return the most recent predicted regime
            except Exception:
                pass # HMM prediction might fail if model is ill-conditioned or data is poor
        
        # Fallback to simple regime detection if HMM is not available/trained or prediction fails
        return self._simple_regime_detection(returns_data)
    
    def _calculate_momentum_signal(self, regime):
        """Calculate momentum signal based on the current regime's lookback."""
        # Get regime-specific lookback period from parameters
        lookback = self.params.regime_momentum_lookbacks.get(regime, 63) # Default to 63 if regime not found
        
        # Ensure sufficient data for the chosen lookback period
        if len(self.data) < lookback + 1:
            return 0 # No signal if not enough data
            
        current_price = self.data.close[0]
        past_price = self.data.close[-lookback] # Price 'lookback' bars ago
        
        # Handle potential NaN values or division by zero
        if np.isnan(current_price) or np.isnan(past_price) or past_price == 0:
            return 0
            
        momentum_return = (current_price / past_price) - 1 # Calculate simple return over lookback
        
        # Generate signal based on absolute momentum threshold
        if momentum_return > self.params.absolute_momentum_threshold:
            return 1 # Positive momentum -> Long signal
        else:
            return 0 # Non-positive momentum -> Flat signal (or exit long)

    def next(self):
        # Skip if insufficient data for initial calculations
        if len(self.data) < 10:
            return
            
        current_return = self.returns[0]
        
        # Update returns buffer for HMM
        if not np.isnan(current_return):
            self.returns_buffer.append(current_return)
        
        # Train/retrain HMM periodically or if not yet trained
        # Only retrain if enough historical data is buffered for HMM
        if ((len(self.data) - self.last_retrain >= self.params.retrain_frequency) or 
            not self.hmm_trained) and len(self.returns_buffer) >= self.params.min_data_for_hmm:
            
            if self._train_hmm(list(self.returns_buffer)):
                self.last_retrain = len(self.data) # Update last retrain bar index
        
        # Predict current regime using buffered returns
        if len(self.returns_buffer) >= 20: # Ensure enough recent data for prediction
            self.current_regime = self._predict_regime(list(self.returns_buffer))
        
        # Calculate momentum signal based on the determined regime's lookback period
        signal = self._calculate_momentum_signal(self.current_regime)
        
        # Execute trades based on signal and current position
        position = self.position.size
        
        if signal == 1: # Long signal
            if position == 0: # If currently flat, enter long
                self.buy() # Execute buy order
                self.order = True # Mark order as pending (simplified tracking)
                # Trailing stop will be placed in notify_order upon completion
        
        elif signal == 0: # Flat/Exit signal
            if position > 0: # If currently long, exit position
                self.close() # Close current long position
                self.order = True # Mark order as pending
                # Trailing stop will be canceled in notify_order upon completion

    def notify_order(self, order):
        """Order notification handler for managing trailing stops."""
        if order.status in [order.Submitted, order.Accepted]:
            return # Ignore pending orders

        if order.status == order.Completed:
            # If a primary (buy/sell) order completed successfully and we now have a position
            if order.isbuy():
                if self.position.size > 0 and self.trailing_stop_order is None: # Just entered long
                    self.trailing_stop_order = self.sell(exectype=bt.Order.StopTrail, 
                                                         trailpercent=self.p.trail_percent)
            elif order.issell(): # Could be a short entry or a long exit
                if self.position.size < 0 and self.trailing_stop_order is None: # Just entered short
                    # Note: Original strategy only had long entries. If short entries were intended
                    # with trailing stops, this logic would apply for short positions too.
                    # For a strategy that only goes long and then flattens, this 'issell'
                    # implies a close order, and we should cancel the trailing stop.
                    pass # Handled below for closing positions
                elif self.position.size == 0 and self.trailing_stop_order: # Position closed (e.g., by manual close or trailing stop hit)
                    self.cancel(self.trailing_stop_order)
                    self.trailing_stop_order = None
        
        # If an order (including trailing stop) is canceled, margin call, or rejected, clear its reference
        if order.status in [order.Canceled, order.Margin, order.Rejected]:
            if order == self.order:
                self.order = None
            if order == self.trailing_stop_order:
                self.trailing_stop_order = None

    def notify_trade(self, trade):
        """Trade notification (for logging or more advanced logic if needed)"""
        if trade.isclosed:
            pass # Trade completed, might log details here

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)