Market Regime Classifier

Overview

Classifies current market state into regimes (trending-up, trending-down, ranging, volatile, quiet) using multiple features. Maps each regime to optimal strategy types and risk parameters.

1. Regime Detection Engine

import pandas as pd
import numpy as np
from typing import Literal

RegimeType = Literal["trending_up", "trending_down", "ranging", "volatile", "quiet"]

class RegimeClassifier:
    """Multi-feature regime classification."""

    @staticmethod
    def classify(df: pd.DataFrame, lookback: int = 50) -> dict:
        """
        Classify current market regime using:
        - ADX for trend strength
        - Bollinger bandwidth for volatility
        - Directional bias from MA alignment
        - Range ratio (ATR / price range)
        """
        recent = df.tail(lookback)
        close = recent["close"]
        returns = close.pct_change().dropna()

        # Trend strength via ADX proxy
        plus_dm = recent["high"].diff().clip(lower=0)
        minus_dm = (-recent["low"].diff()).clip(lower=0)
        atr = (recent["high"] - recent["low"]).rolling(14).mean()
        plus_di = plus_dm.rolling(14).mean() / atr.replace(0, np.nan) * 100
        minus_di = minus_dm.rolling(14).mean() / atr.replace(0, np.nan) * 100
        dx = abs(plus_di - minus_di) / (plus_di + minus_di).replace(0, np.nan) * 100
        adx = dx.rolling(14).mean().iloc[-1]

        # Volatility via Bollinger bandwidth
        sma20 = close.rolling(20).mean()
        std20 = close.rolling(20).std()
        bb_width = ((std20 * 2) / sma20).iloc[-1] * 100

        # MA alignment
        sma_10 = close.rolling(10).mean().iloc[-1]
        sma_20 = sma20.iloc[-1]
        sma_50 = close.rolling(50).mean().iloc[-1] if len(close) >= 50 else sma_20
        ma_aligned_up = sma_10 > sma_20 > sma_50
        ma_aligned_down = sma_10 < sma_20 < sma_50

        # Efficiency ratio: net move / total path
        net_move = abs(close.iloc[-1] - close.iloc[0])
        total_path = returns.abs().sum() * close.mean()
        efficiency = net_move / max(total_path, 1e-10)

        # Classify
        regime = RegimeClassifier._determine_regime(adx, bb_width, ma_aligned_up, ma_aligned_down, efficiency)
        strategy = RegimeClassifier._map_strategy(regime)

        return {
            "regime": regime,
            "confidence": RegimeClassifier._regime_confidence(adx, bb_width, efficiency),
            "metrics": {
                "adx": round(adx, 2) if not np.isnan(adx) else 0,
                "bb_width": round(bb_width, 4),
                "efficiency_ratio": round(efficiency, 4),
                "ma_aligned_up": ma_aligned_up,
                "ma_aligned_down": ma_aligned_down,
            },
            "recommended_strategy": strategy,
            "risk_adjustment": RegimeClassifier._risk_adjustment(regime),
        }

    @staticmethod
    def _determine_regime(adx, bb_width, ma_up, ma_down, efficiency) -> RegimeType:
        if adx > 25 and ma_up and efficiency > 0.3:
            return "trending_up"
        if adx > 25 and ma_down and efficiency > 0.3:
            return "trending_down"
        if bb_width > 3.0 and adx < 20:
            return "volatile"
        if bb_width < 1.0 and adx < 15:
            return "quiet"
        return "ranging"

    @staticmethod
    def _map_strategy(regime: RegimeType) -> dict:
        strategies = {
            "trending_up": {
                "primary": "Trend following — buy pullbacks to support/MAs",
                "avoid": "Selling, mean reversion, counter-trend",
                "indicators": "MA crossovers, ADX, trailing stops",
                "entry_style": "Buy dips to EMA20, add on breakouts",
            },
            "trending_down": {
                "primary": "Trend following — sell rallies to resistance/MAs",
                "avoid": "Buying, catching falling knives",
                "indicators": "MA crossovers, ADX, trailing stops",
                "entry_style": "Sell rallies to EMA20, add on breakdowns",
            },
            "ranging": {
                "primary": "Mean reversion — buy support, sell resistance",
                "avoid": "Breakout trades (most will be false)",
                "indicators": "RSI, Stochastic, Bollinger Bands, S/R levels",
                "entry_style": "Fade extremes, tight stops beyond range",
            },
            "volatile": {
                "primary": "Reduce size or stay flat — wait for regime clarity",
                "avoid": "Large positions, tight stops (will get stopped out)",
                "indicators": "ATR for wide stops, wait for structure",
                "entry_style": "Only high-probability setups with wide stops",
            },
            "quiet": {
                "primary": "Breakout preparation — range compression precedes expansion",
                "avoid": "Intraday trading (no movement to capture)",
                "indicators": "Bollinger squeeze, ATR contraction, volume dry-up",
                "entry_style": "Set breakout orders above/below range",
            },
        }
        return strategies.get(regime, strategies["ranging"])

    @staticmethod
    def _risk_adjustment(regime: RegimeType) -> dict:
        adjustments = {
            "trending_up": {"position_size_mult": 1.0, "stop_width_mult": 1.0, "hold_longer": True},
            "trending_down": {"position_size_mult": 1.0, "stop_width_mult": 1.0, "hold_longer": True},
            "ranging": {"position_size_mult": 0.8, "stop_width_mult": 0.8, "hold_longer": False},
            "volatile": {"position_size_mult": 0.5, "stop_width_mult": 1.5, "hold_longer": False},
            "quiet": {"position_size_mult": 0.5, "stop_width_mult": 0.5, "hold_longer": False},
        }
        return adjustments.get(regime, adjustments["ranging"])

    @staticmethod
    def _regime_confidence(adx, bb_width, efficiency) -> float:
        score = 0
        if adx > 30: score += 0.3
        elif adx < 15: score += 0.2
        if bb_width > 3 or bb_width < 1: score += 0.2
        if efficiency > 0.4 or efficiency < 0.1: score += 0.2
        return round(min(score + 0.3, 0.95), 2)

    @staticmethod
    def regime_history(df: pd.DataFrame, window: int = 50, step: int = 10) -> pd.DataFrame:
        """Rolling regime classification to detect regime transitions."""
        results = []
        for i in range(window, len(df), step):
            chunk = df.iloc[i - window:i]
            r = RegimeClassifier.classify(chunk, window)
            results.append({"time": df.index[i], "regime": r["regime"], "confidence": r["confidence"],
                           "adx": r["metrics"]["adx"]})
        return pd.DataFrame(results).set_index("time")

Usage: Always check regime before deciding strategy type. A trend-following setup in a ranging market will fail.