AI Market Regime Detection: Hidden Markov Models & Quantitative Context Awareness

import numpy as np
import pandas as pd
from hmmlearn.hmm import GaussianHMM

def fit_market_regime_hmm(df: pd.DataFrame, n_components: int = 3):
    """
    Fits a 3-State Gaussian Hidden Markov Model (HMM) on crypto log-returns 
    and Parkinson volatility to identify latent market regimes.
    
    States:
    0: Low-Vol Mean Reverting Range
    1: High-Vol Trending Expansion
    2: Extreme Volatility / Toxic Distribution
    """
    # 1. Feature Engineering: Compute Log Returns and Parkinson Volatility
    df['log_return'] = np.log(df['close'] / df['close'].shift(1))
    
    # Parkinson Volatility using High/Low prices
    high_low_ratio = np.log(df['high'] / df['low']) ** 2
    df['parkinson_vol'] = np.sqrt(high_low_ratio / (4 * np.log(2)))
    
    # Clean dataset
    features = df[['log_return', 'parkinson_vol']].dropna()
    
    # 2. Fit Gaussian HMM with full covariance control
    model = GaussianHMM(
        n_components=n_components, 
        covariance_type="full", 
        n_iter=1000, 
        random_state=42
    )
    model.fit(features)
    
    # 3. Predict hidden states and state posterior probabilities
    hidden_states = model.predict(features)
    posterior_probs = model.predict_proba(features)
    
    # 4. Map states by variance (lowest variance -> Range, highest variance -> Extreme Vol)
    state_variances = [np.trace(cov) for cov in model.covars_]
    sorted_state_map = {old_idx: new_idx for new_idx, old_idx in enumerate(np.argsort(state_variances))}
    
    mapped_states = np.array([sorted_state_map[s] for s in hidden_states])
    
    return model, mapped_states, posterior_probs

import numpy as np
import pandas as pd
from sklearn.mixture import GaussianMixture

def compute_hurst_exponent(time_series: np.ndarray, max_lag: int = 20) -> float:
    """
    Calculates the Hurst Exponent (H) using Rescaled Range (R/S) analysis.
    H > 0.5 -> Persistent / Trending Regime
    H = 0.5 -> Random Walk / Brownian Motion
    H < 0.5 -> Anti-persistent / Mean-Reverting Regime
    """
    lags = range(2, max_lag)
    tau = [np.std(time_series[lag:] - time_series[:-lag]) for lag in lags]
    poly = np.polyfit(np.log(lags), np.log(tau), 1)
    return poly[0]

def extract_regime_features(df: pd.DataFrame) -> pd.DataFrame:
    """
    Extracts stationary quantitative features for Unsupervised GMM Clustering.
    """
    feat_df = pd.DataFrame(index=df.index)
    
    # 1. Orderbook Imbalance (OBI) proxy: Volume Delta ratio
    feat_df['obi'] = (df['buy_volume'] - df['sell_volume']) / (df['buy_volume'] + df['sell_volume'] + 1e-8)
    
    # 2. Rolling Hurst Exponent (window = 100 bars)
    feat_df['hurst'] = df['close'].rolling(100).apply(lambda x: compute_hurst_exponent(x.values), raw=False)
    
    # 3. Volatility Risk Premium Proxy (Normalized ATR / Historical Vol)
    atr = (df['high'] - df['low']).rolling(14).mean()
    feat_df['atr_ratio'] = atr / df['close']
    
    # 4. Fit Gaussian Mixture Model for Regime Clustering
    clean_feats = feat_df.dropna()
    gmm = GaussianMixture(n_components=3, covariance_type='full', random_state=42)
    clean_feats['cluster_regime'] = gmm.fit_predict(clean_feats)
    
    return clean_feats

{
  "role": "Quantitative Regime Validation & Macro Risk Engine",
  "task": "Cross-validate technical HMM regime transition signals against unstructured macroeconomic news and orderflow sentiment.",
  "inputs": {
    "symbol": "BTCUSDT",
    "technical_hmm_signal": {
      "predicted_regime": "BULLISH_EXPANSION",
      "state_confidence": 0.88,
      "hurst_exponent": 0.67,
      "orderbook_imbalance_ratio": 0.42
    },
    "macro_news_feed": [
      "Federal Reserve announces unexpected liquidity injection.",
      "Major spot Bitcoin ETF records $450M net daily inflows.",
      "Derivatives leverage open interest surges by 18% in 4 hours."
    ]
  },
  "validation_rules": {
    "require_macro_confluence": true,
    "max_allowed_leverage_fragility_risk": "MEDIUM",
    "enforce_json_format": true
  },
  "expected_output_schema": {
    "macro_regime_validation": "CONFIRMED | FRAGILE_FAKEOUT | INSUFFICIENT_DATA",
    "narrative_alignment_score": 0.92,
    "liquidation_cascade_risk": "LOW",
    "strategy_action": "ACTIVATE_TREND_FOLLOWING",
    "max_allowed_leverage": "3x",
    "risk_rationale": "High HMM confidence (88%) aligned with positive spot ETF inflows. Leverage risk remains low, supporting aggressive trend allocation."
  }
}

class DynamicRegimeStrategyRouter:
    """
    Automated Capital Allocation Router that dynamically shifts strategy weights 
    and adjusts risk management parameters based on AI HMM Regime Probabilities.
    """
    def __init__(self, initial_capital: float = 100000.0):
        self.capital = initial_capital
        self.active_regime = "UNKNOWN"
        self.allocation = {"trend_bot": 0.0, "grid_bot": 0.0, "cash_hedge": 1.0}
        
    def evaluate_and_route(self, regime_probs: dict, atr_pct: float, hurst: float):
        p_range = regime_probs.get("RANGE", 0.0)
        p_trend = regime_probs.get("TREND", 0.0)
        p_crash = regime_probs.get("DISTRIBUTION", 0.0)
        
        # Scenario 1: High-Vol Trend Expansion (p_trend >= 0.65 and Hurst > 0.55)
        if p_trend >= 0.65 and hurst > 0.55:
            self.active_regime = "HIGH_VOL_TREND"
            self.allocation = {"trend_bot": 0.80, "grid_bot": 0.00, "cash_hedge": 0.20}
            stop_loss_multiplier = 2.5  # Wider trailing stop to ride trend expansion
            
        # Scenario 2: Low-Vol Mean Reverting Range (p_range >= 0.60 and Hurst < 0.48)
        elif p_range >= 0.60 and hurst < 0.48:
            self.active_regime = "MEAN_REVERSION_RANGE"
            self.allocation = {"trend_bot": 0.00, "grid_bot": 0.70, "cash_hedge": 0.30}
            stop_loss_multiplier = 1.0  # Tight boundaries for grid scalping
            
        # Scenario 3: Toxic Distribution / Crisis Spill (p_crash > 0.40)
        else:
            self.active_regime = "DEFENSIVE_RISK_OFF"
            self.allocation = {"trend_bot": 0.00, "grid_bot": 0.00, "cash_hedge": 1.00}
            stop_loss_multiplier = 0.5  # Tight risk-off protective limits
            
        return {
            "regime": self.active_regime,
            "capital_allocation": {k: v * self.capital for k, v in self.allocation.items()},
            "stop_loss_atr_mult": stop_loss_multiplier
        }