Hidden Markov Models for Market Regime Detection
Financial markets do not behave uniformly over time. Bull markets exhibit different statistical properties than bear markets: returns are higher, volatility is lower, and correlations shift. A strategy that performs well in one regime may fail catastrophically in another. Hidden Markov Models (HMMs) provide a principled probabilistic framework for detecting these regime changes and adapting trading behavior accordingly.
An HMM assumes that the observed market data is generated by a process that switches between a finite number of hidden states (regimes). The model learns the statistical properties of each regime and the transition probabilities between them, then uses Bayes' theorem to infer which regime is most likely at each point in time.
Key Takeaways
- Market regimes are real. Volatility clustering, momentum persistence, and correlation shifts all differ across regimes.
- HMMs provide probabilistic regime labels, not binary classifications, allowing graduated position sizing.
- Two or three regimes typically suffice for most trading applications: low-vol trending, high-vol trending, and high-vol mean-reverting.
- Regime detection is inherently lagged. HMMs identify regime changes with a delay, which must be accounted for in strategy design.
HMM Fundamentals
An HMM with N states is defined by three sets of parameters: initial state probabilities, transition probabilities, and emission distributions.
import numpy as np
import pandas as pd
from hmmlearn.hmm import GaussianHMM
from sklearn.preprocessing import StandardScaler
class MarketRegimeHMM:
"""
Hidden Markov Model for market regime detection.
Uses Gaussian emissions fitted to return/volatility features.
"""
def __init__(
self,
n_regimes: int = 3,
n_iter: int = 200,
covariance_type: str = "full",
random_state: int = 42,
):
self.n_regimes = n_regimes
self.model = GaussianHMM(
n_components=n_regimes,
covariance_type=covariance_type,
n_iter=n_iter,
random_state=random_state,
tol=1e-4,
)
self.scaler = StandardScaler()
self.fitted = False
def prepare_features(
self, returns: pd.Series, vol_window: int = 21
) -> np.ndarray:
"""
Create feature matrix for HMM estimation.
Features: returns, rolling volatility, rolling skewness.
"""
features = pd.DataFrame({
"return": returns,
"volatility": returns.rolling(vol_window).std(),
"skewness": returns.rolling(vol_window).skew(),
"abs_return": returns.abs(),
}).dropna()
return features
def fit(self, returns: pd.Series) -> "MarketRegimeHMM":
"""Fit the HMM to historical return data."""
features = self.prepare_features(returns)
X = self.scaler.fit_transform(features.values)
self.model.fit(X)
self.fitted = True
self.feature_index = features.index
# Sort regimes by mean return (regime 0 = lowest return)
self._sort_regimes(features.values)
return self
def _sort_regimes(self, raw_features: np.ndarray):
"""Sort regimes so regime 0 = bearish, regime N = bullish."""
means = self.model.means_[:, 0] # Mean return per regime
order = np.argsort(means)
# Reorder model parameters
self.model.means_ = self.model.means_[order]
self.model.covars_ = self.model.covars_[order]
self.model.startprob_ = self.model.startprob_[order]
self.model.transmat_ = self.model.transmat_[order][:, order]
def predict_regimes(self, returns: pd.Series) -> pd.DataFrame:
"""
Predict regime probabilities for each time step.
Uses the forward-backward algorithm for smoothed estimates.
"""
if not self.fitted:
raise RuntimeError("Model must be fitted first")
features = self.prepare_features(returns)
X = self.scaler.transform(features.values)
# Most likely regime sequence (Viterbi)
regimes = self.model.predict(X)
# Regime probabilities (forward-backward)
regime_probs = self.model.predict_proba(X)
results = pd.DataFrame(index=features.index)
results["regime"] = regimes
regime_names = self._get_regime_names()
for i, name in enumerate(regime_names):
results[f"prob_{name}"] = regime_probs[:, i]
results["regime_name"] = results["regime"].map(
{i: name for i, name in enumerate(regime_names)}
)
return results
def _get_regime_names(self) -> list[str]:
"""Auto-name regimes based on fitted parameters."""
names = []
for i in range(self.n_regimes):
mean_ret = self.model.means_[i, 0]
mean_vol = self.model.means_[i, 1] if self.model.means_.shape[1] > 1 else 0
if mean_ret < -0.5 and mean_vol > 0.5:
names.append("crisis")
elif mean_ret < 0:
names.append("bearish")
elif mean_vol > 0.5:
names.append("volatile")
else:
names.append("bullish")
return names
def get_regime_statistics(self) -> pd.DataFrame:
"""Return the statistical properties of each regime."""
stats = []
regime_names = self._get_regime_names()
for i in range(self.n_regimes):
mean = self.model.means_[i]
if self.model.covariance_type == "full":
var = np.diag(self.model.covars_[i])
else:
var = self.model.covars_[i]
stats.append({
"regime": regime_names[i],
"mean_return": mean[0],
"mean_volatility": mean[1] if len(mean) > 1 else None,
"return_std": np.sqrt(var[0]),
"stationary_prob": self._stationary_distribution()[i],
})
return pd.DataFrame(stats)
def _stationary_distribution(self) -> np.ndarray:
"""Compute the stationary distribution of the Markov chain."""
A = self.model.transmat_.T - np.eye(self.n_regimes)
A[-1, :] = 1 # Replace last row with normalization constraint
b = np.zeros(self.n_regimes)
b[-1] = 1
pi = np.linalg.solve(A, b)
return pi
Regime-Adaptive Trading Strategy
Different regimes call for different trading approaches. Use the HMM's output to switch between strategies dynamically.
class RegimeAdaptiveStrategy:
"""
Adapt trading behavior based on detected regime.
Regime mappings (example):
- Bullish: Trend following, full position size
- Bearish: Mean reversion or reduce exposure
- Crisis: Risk-off, minimal exposure
"""
def __init__(
self,
hmm: MarketRegimeHMM,
regime_configs: dict | None = None,
):
self.hmm = hmm
self.regime_configs = regime_configs or {
"bullish": {
"strategy": "trend_following",
"position_scale": 1.0,
"stop_loss_mult": 2.0,
},
"bearish": {
"strategy": "mean_reversion",
"position_scale": 0.5,
"stop_loss_mult": 1.5,
},
"crisis": {
"strategy": "risk_off",
"position_scale": 0.1,
"stop_loss_mult": 1.0,
},
"volatile": {
"strategy": "vol_selling",
"position_scale": 0.3,
"stop_loss_mult": 1.5,
},
}
def generate_signals(
self,
prices: pd.Series,
regimes: pd.DataFrame,
sma_short: int = 10,
sma_long: int = 50,
rsi_period: int = 14,
) -> pd.DataFrame:
"""
Generate regime-adaptive signals.
"""
returns = prices.pct_change()
# Base indicators
sma_s = prices.rolling(sma_short).mean()
sma_l = prices.rolling(sma_long).mean()
delta = prices.diff()
gain = delta.clip(lower=0).rolling(rsi_period).mean()
loss = (-delta.clip(upper=0)).rolling(rsi_period).mean()
rsi = 100 - (100 / (1 + gain / loss))
signals = pd.DataFrame(index=prices.index)
signals["price"] = prices
signals["regime"] = regimes["regime_name"].reindex(prices.index).ffill()
signals["raw_signal"] = 0.0
signals["position_scale"] = 0.0
signals["final_signal"] = 0.0
for i in range(max(sma_long, rsi_period), len(prices)):
idx = prices.index[i]
regime = signals.loc[idx, "regime"]
if pd.isna(regime):
continue
config = self.regime_configs.get(regime, self.regime_configs.get("bullish"))
if config["strategy"] == "trend_following":
# Long when short MA > long MA
if sma_s.iloc[i] > sma_l.iloc[i]:
raw = 1.0
else:
raw = -0.5
elif config["strategy"] == "mean_reversion":
# Buy oversold, sell overbought
if rsi.iloc[i] < 30:
raw = 1.0
elif rsi.iloc[i] > 70:
raw = -1.0
else:
raw = 0.0
elif config["strategy"] == "risk_off":
raw = 0.0
elif config["strategy"] == "vol_selling":
raw = -0.3 if rsi.iloc[i] > 60 else 0.3
else:
raw = 0.0
signals.loc[idx, "raw_signal"] = raw
signals.loc[idx, "position_scale"] = config["position_scale"]
signals.loc[idx, "final_signal"] = raw * config["position_scale"]
return signals
Model Selection: Number of Regimes
Choosing the right number of regimes is critical. Too few and you miss important market states; too many and you overfit.
def select_n_regimes(
returns: pd.Series,
max_regimes: int = 6,
n_fits: int = 5,
) -> pd.DataFrame:
"""
Compare HMMs with different numbers of regimes using BIC.
Multiple fits per n_regimes to handle local optima.
"""
results = []
for n in range(2, max_regimes + 1):
best_bic = np.inf
best_ll = -np.inf
for seed in range(n_fits):
try:
hmm = MarketRegimeHMM(n_regimes=n, random_state=seed)
features = hmm.prepare_features(returns)
X = hmm.scaler.fit_transform(features.values)
hmm.model.fit(X)
score = hmm.model.score(X)
bic = -2 score len(X) + n (n + 2 features.shape[1]) * np.log(len(X))
if bic < best_bic:
best_bic = bic
best_ll = score
except Exception:
continue
results.append({
"n_regimes": n,
"bic": best_bic,
"log_likelihood": best_ll,
})
results_df = pd.DataFrame(results)
best_n = results_df.loc[results_df["bic"].idxmin(), "n_regimes"]
print(f"Optimal number of regimes: {int(best_n)} (by BIC)")
return results_df
Backtesting the Regime Strategy
def backtest_regime_strategy(
prices: pd.Series,
n_regimes: int = 3,
training_window: int = 504, # 2 years
) -> pd.DataFrame:
"""
Walk-forward backtest of regime-adaptive strategy.
Refit HMM periodically on expanding window.
"""
returns = prices.pct_change().dropna()
all_results = []
refit_freq = 63 # Quarterly refit
hmm = None
for i in range(training_window, len(returns)):
# Refit HMM periodically
if i % refit_freq == 0 or hmm is None:
train_returns = returns.iloc[:i]
try:
hmm = MarketRegimeHMM(n_regimes=n_regimes)
hmm.fit(train_returns)
except Exception:
continue
# Predict current regime
recent = returns.iloc[max(0, i-252):i+1]
try:
regime_info = hmm.predict_regimes(recent)
current_regime = regime_info["regime_name"].iloc[-1]
except Exception:
current_regime = "bullish" # Default
# Position sizing based on regime
if current_regime == "crisis":
position = 0.0
elif current_regime == "bearish":
position = 0.3
elif current_regime == "volatile":
position = 0.5
else:
position = 1.0
all_results.append({
"date": returns.index[i],
"return": returns.iloc[i],
"regime": current_regime,
"position": position,
"strategy_return": returns.iloc[i] * position,
})
results = pd.DataFrame(all_results).set_index("date")
# Performance
strat_ret = results["strategy_return"]
buy_hold = results["return"]
strat_sharpe = strat_ret.mean() / strat_ret.std() * np.sqrt(252)
bh_sharpe = buy_hold.mean() / buy_hold.std() * np.sqrt(252)
strat_dd = (
(1 + strat_ret).cumprod().cummax()
- (1 + strat_ret).cumprod()
) / (1 + strat_ret).cumprod().cummax()
bh_dd = (
(1 + buy_hold).cumprod().cummax()
- (1 + buy_hold).cumprod()
) / (1 + buy_hold).cumprod().cummax()
print(f"Strategy Sharpe: {strat_sharpe:.2f} vs Buy-Hold: {bh_sharpe:.2f}")
print(f"Strategy Max DD: {strat_dd.max():.2%} vs Buy-Hold: {bh_dd.max():.2%}")
return results
Transition Probability Analysis
The transition matrix reveals regime persistence and switch probabilities.
def analyze_transitions(hmm: MarketRegimeHMM) -> None:
"""Print and analyze the transition probability matrix."""
names = hmm._get_regime_names()
transmat = hmm.model.transmat_
stationary = hmm._stationary_distribution()
print("Transition Probability Matrix:")
print(f"{'':>12}", end="")
for name in names:
print(f"{name:>12}", end="")
print()
for i, name in enumerate(names):
print(f"{name:>12}", end="")
for j in range(len(names)):
print(f"{transmat[i,j]:>12.3f}", end="")
print()
print(f"\nStationary Distribution:")
for name, prob in zip(names, stationary):
bar = "#" int(prob 50)
print(f" {name:>10}: {prob:.3f} {bar}")
print(f"\nExpected Regime Durations:")
for i, name in enumerate(names):
duration = 1 / (1 - transmat[i, i])
print(f" {name:>10}: {duration:.1f} days")
FAQ
How many regimes should I use for stock market data?
Two or three regimes work best for most applications. A two-regime model captures the fundamental bull/bear distinction. A three-regime model adds a "crisis" or "high-volatility" state that is useful for tail risk management. Models with more than four regimes tend to overfit and produce unstable estimates. Use BIC to select the optimal number, and validate on out-of-sample data.
How quickly does an HMM detect regime changes?
HMMs typically detect regime changes with a lag of 5-15 trading days, depending on how distinct the regimes are and the noise level. The detection is fastest when regimes have very different volatility characteristics (e.g., a sudden volatility spike at the start of a crisis). It is slowest for gradual transitions between similar regimes. Use the regime probability (not the hard classification) for faster response.
Can I use an HMM for intraday regime detection?
Yes, but the feature set must be adapted. Intraday data has different characteristics: microstructure noise, volume seasonality, and bid-ask bounce. Use features like realized variance (1-minute or 5-minute), order flow imbalance, and spread width. The number of regimes may also differ: intraday markets often have distinct open, midday, and close regimes driven by institutional trading patterns.
How do I validate that HMM regimes are economically meaningful?
After fitting, examine the regime statistics: mean returns, volatility, and duration. Overlay regime labels on a price chart to visually verify they correspond to recognized market episodes (e.g., the 2020 COVID crash should be classified as "crisis"). Compare regime-conditional Sharpe ratios. If all regimes have similar statistics, the HMM is not capturing meaningful structure.