regime detection - walk-forward validation improving accuracy

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from datetime import datetime, timedelta

class WalkForwardRegimeDetector:
    def __init__(self, train_days=90, test_days=30, roll_days=30):
        self.train_days = train_days
        self.test_days = test_days
        self.roll_days = roll_days
        self.models = {}
        self.performance_log = []

    def prepare_features(self, df, lookback=20):
        """
        Extract regime features from market data
        """
        # Volatility features
        df['returns'] = df['close'].pct_change()
        df['realized_vol'] = df['returns'].rolling(lookback).std() * np.sqrt(252)
        df['vol_trend'] = df['realized_vol'] - df['realized_vol'].shift(lookback)

        # Volume features
        df['volume_ratio'] = df['volume'] / df['volume'].rolling(lookback).mean()
        df['volume_trend'] = df['volume'].pct_change(lookback)

        # Correlation features (SPX vs sector ETFs)
        df['correlation'] = df['returns'].rolling(lookback).corr(df['sector_returns'])
        df['corr_trend'] = df['correlation'] - df['correlation'].shift(lookback)

        # Momentum features
        df['momentum_20'] = df['close'].pct_change(20)
        df['momentum_60'] = df['close'].pct_change(60)

        # VIX features
        df['vix_regime'] = pd.cut(df['VIX'], bins=[0, 15, 20, 100], labels=[0, 1, 2])
        df['vix_change'] = df['VIX'].pct_change(5)

        return df.dropna()

    def assign_regime_labels(self, df):
        """
        Label historical data with regime classifications
        Based on realized future performance
        """
        # Look forward 10 days to classify regime
        df['future_sharpe'] = (
            df['returns'].shift(-10).rolling(10).mean() /
            df['returns'].shift(-10).rolling(10).std()
        )

        # Classify regimes
        conditions = [
            (df['future_sharpe'] > 0.5) & (df['realized_vol'] < 0.15),  # Low vol favorable
            (df['future_sharpe'] > 0.3) & (df['realized_vol'] < 0.25),  # Med vol favorable
            (df['future_sharpe'] < -0.3) | (df['realized_vol'] > 0.30), # High vol unfavorable
        ]
        choices = ['low_vol_favorable', 'med_vol_favorable', 'unfavorable']
        df['regime'] = np.select(conditions, choices, default='neutral')

        return df

    def walk_forward_validate(self, market_data):
        """
        Walk-forward validation across entire dataset
        """
        results = []

        start_date = market_data.index[0]
        end_date = market_data.index[-1]
        current_date = start_date + timedelta(days=self.train_days)

        while current_date + timedelta(days=self.test_days) <= end_date:
            # Training window
            train_start = current_date - timedelta(days=self.train_days)
            train_end = current_date
            train_data = market_data[train_start:train_end]

            # Test window
            test_start = train_end
            test_end = test_start + timedelta(days=self.test_days)
            test_data = market_data[test_start:test_end]

            # Train model
            model = self.train_regime_model(train_data)

            # Validate on test set
            accuracy, predictions = self.validate_model(model, test_data)

            results.append({
                'train_start': train_start,
                'train_end': train_end,
                'test_start': test_start,
                'test_end': test_end,
                'accuracy': accuracy,
                'predictions': predictions
            })

            # Roll forward
            current_date += timedelta(days=self.roll_days)

        return pd.DataFrame(results)

    def train_regime_model(self, train_data):
        """
        Train RandomForest classifier on regime detection
        """
        features = ['realized_vol', 'vol_trend', 'volume_ratio',
                   'correlation', 'corr_trend', 'momentum_20',
                   'vix_regime', 'vix_change']

        X = train_data[features]
        y = train_data['regime']

        model = RandomForestClassifier(
            n_estimators=100,
            max_depth=8,
            min_samples_split=20,
            min_samples_leaf=10,
            random_state=42
        )

        model.fit(X, y)
        return model

    def validate_model(self, model, test_data):
        """
        Validate trained model on test set
        """
        features = ['realized_vol', 'vol_trend', 'volume_ratio',
                   'correlation', 'corr_trend', 'momentum_20',
                   'vix_regime', 'vix_change']

        X_test = test_data[features]
        y_test = test_data['regime']

        predictions = model.predict(X_test)
        accuracy = (predictions == y_test).mean()

        return accuracy, predictions

# Usage example
if __name__ == "__main__":
    # Load market data
    market_data = load_market_data('2020-01-01', '2024-09-12')

    # Prepare features and labels
    market_data = prepare_features(market_data)
    market_data = assign_regime_labels(market_data)

    # Initialize walk-forward detector
    detector = WalkForwardRegimeDetector(
        train_days=90,
        test_days=30,
        roll_days=30
    )

    # Run walk-forward validation
    results = detector.walk_forward_validate(market_data)

    print(f"Average accuracy: {results['accuracy'].mean():.2%}")
    print(f"Accuracy std: {results['accuracy'].std():.2%}")
    print(f"Min accuracy: {results['accuracy'].min():.2%}")
    print(f"Max accuracy: {results['accuracy'].max():.2%}")