HMM Regime Detection - Market State Classification

Location: /Volumes/LegacySafe/SS_III/core/ml/regime_detector.py

Impact: Improved regime forecasting (verified 2025)

Sources (2025):

• MDPI Bitcoin Regime Shifts (May 2025): Bayesian HMM analysis 2016-2024
• AJPAS HMM Applications (July 2025): "HMMs outperform other models in forecasting regime shifts"
• QuantStart: 56%→24% max drawdown in backtest

What It Does

Classifies market into regimes to prevent wrong-strategy-for-wrong-market:

┌──────────────────────────────────────────────────────────────┐
│                    MARKET REGIMES                            │
├───────────┬────────────────┬────────────────┬───────────────┤
│ TRENDING  │ MEAN_REVERTING │   VOLATILE     │    CRISIS     │
│  (40%)    │     (35%)      │    (20%)       │     (5%)      │
│ ↗↗↗ or ↘↘↘│    ↔↔↔↔↔       │   ↕↕↕↕↕        │  💀💀💀       │
│  FOLLOW   │    FADE        │  REDUCE        │   FLAT        │
└───────────┴────────────────┴────────────────┴───────────────┘

Core Implementation

import numpy as np
from hmmlearn import GaussianHMM
from typing import Tuple, List
import pandas as pd

class RegimeDetector:
    """HMM-based market regime classifier."""

    REGIMES = {
        0: "TRENDING",
        1: "MEAN_REVERTING",
        2: "VOLATILE",
        3: "CRISIS"
    }

    def __init__(self, n_regimes: int = 4, lookback: int = 100):
        self.n_regimes = n_regimes
        self.lookback = lookback
        self.model = GaussianHMM(
            n_components=n_regimes,
            covariance_type="full",
            n_iter=100,
            random_state=42
        )
        self.fitted = False

    def prepare_features(self, df: pd.DataFrame) -> np.ndarray:
        """Extract regime-relevant features from OHLCV data."""
        features = pd.DataFrame()

        # Returns (momentum indicator)
        features['returns'] = df['close'].pct_change()

        # Volatility (20-period rolling std)
        features['volatility'] = features['returns'].rolling(20).std()

        # Volume ratio (relative to 20-period average)
        features['volume_ratio'] = df['volume'] / df['volume'].rolling(20).mean()

        # Trend strength (ADX proxy: abs return / volatility)
        features['trend_strength'] = (
            features['returns'].rolling(14).mean().abs() /
            features['volatility']
        )

        # Mean reversion indicator (price vs 20 SMA)
        features['mean_reversion'] = (
            (df['close'] - df['close'].rolling(20).mean()) /
            df['close'].rolling(20).std()
        )

        return features.dropna().values

    def fit(self, df: pd.DataFrame) -> 'RegimeDetector':
        """Train HMM on historical data."""
        X = self.prepare_features(df)
        self.model.fit(X)
        self.fitted = True
        return self

    def predict(self, df: pd.DataFrame) -> Tuple[str, float]:
        """Predict current regime with confidence."""
        if not self.fitted:
            raise ValueError("Model not fitted. Call fit() first.")

        X = self.prepare_features(df)

        # Get state probabilities
        probs = self.model.predict_proba(X[-1:])
        state = np.argmax(probs)
        confidence = probs[0, state]

        return self.REGIMES[state], confidence

    def get_strategy_filter(self, regime: str) -> dict:
        """Return strategy parameters for current regime."""
        filters = {
            "TRENDING": {
                "strategies": ["momentum", "breakout", "trend_following"],
                "position_multiplier": 1.0,
                "stop_loss_multiplier": 1.5,  # Wider stops for trends
                "take_profit_multiplier": 2.0
            },
            "MEAN_REVERTING": {
                "strategies": ["mean_reversion", "rsi_oversold", "bollinger_bounce"],
                "position_multiplier": 0.8,
                "stop_loss_multiplier": 1.0,
                "take_profit_multiplier": 1.0
            },
            "VOLATILE": {
                "strategies": ["volatility_breakout"],
                "position_multiplier": 0.5,  # Half size
                "stop_loss_multiplier": 2.0,  # Much wider stops
                "take_profit_multiplier": 1.5
            },
            "CRISIS": {
                "strategies": [],  # NO TRADING
                "position_multiplier": 0.0,
                "stop_loss_multiplier": 0,
                "take_profit_multiplier": 0
            }
        }
        return filters.get(regime, filters["VOLATILE"])

Integration Points

With ECO_SYSTEM_4 Pipeline

# In ECO_SYSTEM_4/stages/signal_stage.py
from core.ml.regime_detector import RegimeDetector

detector = RegimeDetector()
detector.fit(historical_data)

async def filter_signal(signal, market_data):
    regime, confidence = detector.predict(market_data)
    filter_params = detector.get_strategy_filter(regime)

    # Block signals in CRISIS mode
    if regime == "CRISIS":
        return None

    # Check if signal strategy matches regime
    if signal.strategy not in filter_params["strategies"]:
        return None  # Wrong strategy for this regime

    # Adjust position size
    signal.position_size *= filter_params["position_multiplier"]

    return signal

With Risk Management

# In core/risk/position_sizer.py
def adjust_for_regime(base_size: float, regime: str) -> float:
    multipliers = {
        "TRENDING": 1.0,
        "MEAN_REVERTING": 0.8,
        "VOLATILE": 0.5,
        "CRISIS": 0.0
    }
    return base_size * multipliers.get(regime, 0.5)

Configuration

Add to BRAIN.json:

{
  "regime_detection": {
    "enabled": true,
    "n_regimes": 4,
    "lookback_periods": 100,
    "retrain_frequency": "daily",
    "confidence_threshold": 0.6,
    "crisis_indicators": {
      "vix_threshold": 35,
      "correlation_breakdown": 0.3,
      "volume_spike": 3.0
    }
  }
}

Required Dependencies

pip install hmmlearn scikit-learn pandas numpy

Testing

cd /Volumes/LegacySafe/SS_III/core/ml

python -c "
from regime_detector import RegimeDetector
import pandas as pd

# Load sample BTC data
df = pd.read_csv('sample_btc_ohlcv.csv')
detector = RegimeDetector()
detector.fit(df)
regime, conf = detector.predict(df)
print(f'Current Regime: {regime} ({conf:.1%} confidence)')
"

Visual Dashboard

# For ds_star visualization
def plot_regimes(df, regimes):
    import plotly.graph_objects as go

    colors = {
        "TRENDING": "green",
        "MEAN_REVERTING": "blue",
        "VOLATILE": "orange",
        "CRISIS": "red"
    }

    fig = go.Figure()
    # Plot price with regime background colors
    # ... visualization code

Research Sources (Verified 2025)

Confidence: HIGH - Multiple 2025 academic papers confirm effectiveness.

Status

• Implementation: NOT STARTED
• Priority: HIGH (drawdown protection)
• Dependencies: hmmlearn, numpy, pandas