Introduction
Financial markets never move in a straight line. They go through regimes – periods where price behavior, volatility, and sentiment follow recognizable patterns. Detecting when one regime ends and another begins is one of the most valuable skills for traders and quantitative analysts. It helps with adjusting exposure, managing risk, and deciding when to retrain predictive models.
Traditional methods like moving-average crossovers or volatility filters only capture part of the story. More advanced statistical models such as Hidden Markov Models (HMMs) try to infer hidden “states” of the market, but they often rely on heavy assumptions about return distributions being normal.
A newer approach takes a different route – instead of assuming a model for the data, it treats each segment of market returns as a probability distribution and compares these distributions directly using a concept called the Wasserstein distance. The algorithm built around this idea is known as Wasserstein K-Means, and it offers a powerful way to identify distinct market regimes without any parametric assumptions.
What Is a Market Regime?
A market regime is simply a phase of market behavior that looks internally consistent but clearly different from other phases.
- Bull regimes show sustained positive returns and relatively low volatility.
- Bear regimes bring high volatility, drawdowns, and panic correlations.
- Neutral or transition phases lie somewhere between the two.
Rather than defining these regimes manually, we can let data reveal them by grouping historical return patterns that behave similarly.
The Limitations of Traditional Clustering
The classic K-Means algorithm groups data points in Euclidean space – it measures straight-line distances between points like average returns or volatilities. That’s fine when each observation is just a number or vector, but not when each observation is a distribution of returns.
Financial markets are messy. Two 10-day return segments might have the same average and volatility but completely different shapes – one smooth, another filled with fat tails or skewness. Capturing those differences requires looking at the entire distribution, not just its moments.
Why the Wasserstein Distance Works Better
The Wasserstein distance (sometimes called Earth Mover’s Distance) measures how much “effort” it takes to morph one probability distribution into another. Imagine two piles of sand: the Wasserstein distance tells you how much sand you’d have to move and how far to make one pile look like the other.
This measure is ideal for comparing financial return distributions because:
- It captures differences in shape, spread, and location.
- It works well for non-Gaussian, heavy-tailed data.
- It provides a natural way to compute “average” distributions, called Wasserstein barycenters.
- It’s computationally efficient in one dimension, so it can be used on rolling windows of returns.
How the Wasserstein K-Means Algorithm Works
The algorithm adapts classical K-Means to operate on distributions instead of data points:
- Slice the returns into overlapping windows (for example, 10-day blocks).
Each block becomes an empirical distribution of returns. - Compute distances between all pairs of distributions using the Wasserstein metric.
- Find barycenters (the average shape) for each cluster – these serve as the new centroids.
- Reassign and iterate until cluster changes are minimal.
The result: each time window belongs to a regime whose overall return distribution is most similar. In practice, one cluster often represents a low-variance bullish phase and another a high-variance bearish phase.
Why This Matters
- Model-Free: No assumption about returns being normal or stationary.
- Captures Market Texture: Recognizes changes in skewness, kurtosis, and tail risk.
- Early Warning: Picks up subtle volatility transitions before they appear in common indicators.
- More Robust than HMMs: Works directly on observed data, not hidden states.
- Intuitive Visualization: Regime color bands overlay nicely on candlestick charts to reveal market shifts.
A Practical Implementation for NIFTY
List of Libraries needs to be installed and ensure openalgo is running in the background
pip install numpy pandas scipy plotly requests logging openalgoBelow is a simplified version of the Python implementation. It uses OpenAlgo to fetch NIFTY daily data, computes log returns, builds rolling windows, and applies Wasserstein K-Means to classify market regimes.
"""
NIFTY Market Regime Clustering - TRULY NON-REPAINTING VERSION
Based on "Clustering Market Regimes Using the Wasserstein Distance"
This version ABSOLUTELY DOES NOT REPAINT because:
1. At each bar t, we ONLY use data from bars 0 to t-1 for training
2. We create NON-OVERLAPPING windows from historical data
3. We fit the model on these historical windows ONLY
4. We then classify the current bar t using a window ending at t
5. The label at bar t NEVER changes when new bars arrive
KEY PRINCIPLE: The regime at bar t is determined by:
- A model trained on bars 0 to t-1
- Applied to a window ending at bar t
- This represents: "Based on what we knew before bar t, what regime are we in now?"
"""
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from openalgo import api
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import logging
from typing import List, Tuple, Dict
from scipy import stats
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# ========================================================================================
# CONFIGURATION
# ========================================================================================
# OpenAlgo API Configuration
API_KEY = "0a96410b983a6a1382d8ab162728d25a86945f1b4d3b0f32d62a0abf08e9f0f5"
API_HOST = "http://127.0.0.1:5000"
# Market Data Configuration
SYMBOL = "NIFTY"
EXCHANGE = "NSE_INDEX"
INTERVAL = "D"
# Date Range
START_DATE = "2020-01-01"
END_DATE = "2024-07-15"
# Wasserstein K-Means Parameters
K_CLUSTERS = 2 # Number of regimes (bull/bear)
WINDOW_LENGTH = 10 # Number of returns per distribution
MIN_TRAINING_WINDOWS = 20 # Minimum windows needed to start classification
REFIT_FREQUENCY = 10 # Refit model every N bars (0 = refit every bar)
MAX_ITERATIONS = 100
TOLERANCE = 1e-4
# Visualization
SHOW_PLOT = True
SAVE_HTML = True
HTML_FILENAME = "regimechange.html"
# ========================================================================================
# WASSERSTEIN DISTANCE AND K-MEANS IMPLEMENTATION
# ========================================================================================
class WassersteinKMeans:
"""
Wasserstein K-Means clustering algorithm for probability distributions.
"""
def __init__(self, n_clusters=2, max_iter=100, tolerance=1e-4):
self.n_clusters = n_clusters
self.max_iter = max_iter
self.tolerance = tolerance
self.centroids = None
self.labels = None
self.inertia = None
def wasserstein_distance(self, dist1: np.ndarray, dist2: np.ndarray, p=1) -> float:
"""
Calculate the p-Wasserstein distance between two empirical distributions.
"""
sorted_dist1 = np.sort(dist1)
sorted_dist2 = np.sort(dist2)
distance = np.sum(np.abs(sorted_dist1 - sorted_dist2) ** p) / len(dist1)
return distance ** (1/p)
def wasserstein_barycenter(self, distributions: List[np.ndarray]) -> np.ndarray:
"""
Calculate the Wasserstein barycenter (median) of empirical distributions.
"""
sorted_dists = np.array([np.sort(dist) for dist in distributions])
barycenter = np.median(sorted_dists, axis=0)
return barycenter
def fit(self, distributions: List[np.ndarray]) -> 'WassersteinKMeans':
"""
Fit the Wasserstein K-Means model to the data.
"""
if len(distributions) < self.n_clusters:
raise ValueError(f"Need at least {self.n_clusters} distributions to cluster")
n_distributions = len(distributions)
# Initialize centroids by randomly sampling from distributions
np.random.seed(42) # For reproducibility
random_indices = np.random.choice(n_distributions, self.n_clusters, replace=False)
self.centroids = [distributions[i].copy() for i in random_indices]
for iteration in range(self.max_iter):
# Assignment step
labels = []
for dist in distributions:
distances = [self.wasserstein_distance(dist, centroid)
for centroid in self.centroids]
labels.append(np.argmin(distances))
labels = np.array(labels)
# Update step
new_centroids = []
for k in range(self.n_clusters):
cluster_dists = [distributions[i] for i in range(n_distributions)
if labels[i] == k]
if len(cluster_dists) > 0:
new_centroids.append(self.wasserstein_barycenter(cluster_dists))
else:
new_centroids.append(distributions[np.random.randint(n_distributions)].copy())
# Calculate loss
loss = sum(self.wasserstein_distance(old, new)
for old, new in zip(self.centroids, new_centroids))
self.centroids = new_centroids
if loss < self.tolerance:
break
# Final assignment
self.labels = np.array([np.argmin([self.wasserstein_distance(dist, centroid)
for centroid in self.centroids])
for dist in distributions])
# Calculate inertia
self.inertia = sum(min([self.wasserstein_distance(distributions[i], centroid)
for centroid in self.centroids]) ** 2
for i in range(n_distributions))
# Label clusters by variance (0 = low variance/bull, 1 = high variance/bear)
centroid_vars = [np.var(centroid) for centroid in self.centroids]
if centroid_vars[1] < centroid_vars[0]:
self.labels = 1 - self.labels
self.centroids = self.centroids[::-1]
return self
def predict(self, distribution: np.ndarray) -> int:
"""
Predict cluster label for a single new distribution.
"""
if self.centroids is None:
raise ValueError("Model has not been fitted yet")
distances = [self.wasserstein_distance(distribution, centroid)
for centroid in self.centroids]
return np.argmin(distances)
# ========================================================================================
# TRULY NON-REPAINTING DATA PROCESSING
# ========================================================================================
def fetch_nifty_data(client: api, start_date: str, end_date: str) -> pd.DataFrame:
"""
Fetch historical NIFTY data from OpenAlgo.
"""
logger.info(f"Fetching NIFTY data from {start_date} to {end_date}...")
df = client.history(
symbol=SYMBOL,
exchange=EXCHANGE,
interval=INTERVAL,
start_date=start_date,
end_date=end_date
)
logger.info(f"Fetched {len(df)} candles")
return df
def create_log_returns(df: pd.DataFrame) -> pd.DataFrame:
"""
Calculate log returns from price data.
"""
df = df.copy()
df['log_return'] = np.log(df['close'] / df['close'].shift(1))
df = df.dropna()
logger.info(f"Calculated log returns")
return df
def create_non_overlapping_windows(returns: np.ndarray, window_length: int) -> List[np.ndarray]:
"""
Create NON-OVERLAPPING windows from returns.
This ensures no data point appears in multiple windows.
Args:
returns: Array of returns
window_length: Length of each window
Returns:
List of windows (numpy arrays)
"""
windows = []
n_returns = len(returns)
# Create non-overlapping windows
for i in range(0, n_returns - window_length + 1, window_length):
window = returns[i:i + window_length]
if len(window) == window_length:
windows.append(window.copy())
return windows
def classify_regime_non_repainting(returns: pd.Series,
window_length: int,
min_training_windows: int,
refit_frequency: int,
n_clusters: int) -> pd.Series:
"""
Classify regimes in a TRULY non-repainting manner.
For each bar t:
1. Use ONLY returns from bars 0 to t-1 for training
2. Create non-overlapping windows from this historical data
3. Fit WK-means on these historical windows (or use cached model if refit_frequency > 0)
4. Create current window: returns from bars [t-window_length+1, t]
5. Predict regime for current window
6. Assign regime to bar t
Args:
returns: Series of returns
window_length: Number of returns in each window
min_training_windows: Minimum windows needed before we start classification
refit_frequency: How often to refit (0 = every bar, N = every N bars)
n_clusters: Number of clusters
Returns:
Series of regime labels (0 = low variance, 1 = high variance)
"""
n_returns = len(returns)
labels = np.full(n_returns, -1, dtype=int) # -1 = not yet classified
returns_array = returns.values
# Calculate minimum bars needed
min_bars_needed = window_length + (min_training_windows * window_length)
logger.info(f"Starting regime classification...")
logger.info(f"Window length: {window_length}")
logger.info(f"Min training windows: {min_training_windows}")
logger.info(f"Refit frequency: {refit_frequency} bars")
logger.info(f"Minimum bars needed: {min_bars_needed}")
current_model = None
last_refit_bar = -1
# Process each bar starting from min_bars_needed
for t in range(min_bars_needed, n_returns):
# Decide whether to refit the model
should_refit = (
current_model is None or # Never fitted before
refit_frequency == 0 or # Refit every bar
(t - last_refit_bar) >= refit_frequency # Time to refit
)
if should_refit:
if t % 50 == 0:
logger.info(f"Processing bar {t}/{n_returns} - Refitting model")
# CRITICAL: Use ONLY historical data (bars 0 to t-1)
historical_returns = returns_array[:t] # Excludes bar t!
# Create non-overlapping windows from historical data
training_windows = create_non_overlapping_windows(
historical_returns,
window_length
)
# Ensure we have enough windows
if len(training_windows) < n_clusters:
labels[t] = 0 # Default to first cluster
continue
# Fit model on historical windows ONLY
try:
current_model = WassersteinKMeans(
n_clusters=n_clusters,
max_iter=MAX_ITERATIONS,
tolerance=TOLERANCE
)
current_model.fit(training_windows)
last_refit_bar = t
except Exception as e:
logger.warning(f"Failed to fit model at bar {t}: {e}")
labels[t] = labels[t-1] if t > 0 else 0
continue
# Create current window ending at bar t
# This window uses bars [t-window_length+1, t]
current_window_start = max(0, t - window_length + 1)
current_window = returns_array[current_window_start:t+1]
# Pad if necessary (shouldn't happen after min_bars_needed)
if len(current_window) < window_length:
padding = np.zeros(window_length - len(current_window))
current_window = np.concatenate([padding, current_window])
# Predict regime for current window
try:
current_label = current_model.predict(current_window)
labels[t] = current_label
except Exception as e:
logger.warning(f"Failed to predict at bar {t}: {e}")
labels[t] = labels[t-1] if t > 0 and labels[t-1] >= 0 else 0
# Fill initial bars with default value
labels[labels == -1] = 0
return pd.Series(labels, index=returns.index, dtype=int)
# ========================================================================================
# TESTING FUNCTIONS
# ========================================================================================
def test_for_repainting(returns: pd.Series,
window_length: int,
min_training_windows: int,
n_clusters: int) -> bool:
"""
Test if the algorithm repaints by checking if labels change when new data is added.
Returns:
True if repainting detected, False if clean
"""
logger.info("\n" + "="*80)
logger.info("TESTING FOR REPAINTING")
logger.info("="*80)
# Get labels for first 80% of data
split_point = int(len(returns) * 0.8)
returns_partial = returns.iloc[:split_point]
logger.info(f"First pass: Using {len(returns_partial)} bars")
labels_partial = classify_regime_non_repainting(
returns_partial,
window_length,
min_training_windows,
refit_frequency=0, # Refit every bar for this test
n_clusters=n_clusters
)
# Get labels for full data
logger.info(f"Second pass: Using all {len(returns)} bars")
labels_full = classify_regime_non_repainting(
returns,
window_length,
min_training_windows,
refit_frequency=0,
n_clusters=n_clusters
)
# Compare labels for the overlapping period
# Start from min_bars_needed to avoid comparing unclassified bars
min_bars_needed = window_length + (min_training_windows * window_length)
comparison_start = min_bars_needed
labels_partial_compare = labels_partial.iloc[comparison_start:].values
labels_full_compare = labels_full.iloc[comparison_start:split_point].values
# Check if any labels changed
different_labels = np.sum(labels_partial_compare != labels_full_compare)
total_labels = len(labels_partial_compare)
logger.info(f"\nResults:")
logger.info(f"Labels compared: {total_labels}")
logger.info(f"Labels that changed: {different_labels}")
logger.info(f"Percentage changed: {100*different_labels/total_labels:.2f}%")
if different_labels == 0:
logger.info("✅ NO REPAINTING DETECTED!")
return False
else:
logger.info(f"❌ REPAINTING DETECTED! {different_labels} labels changed!")
return True
# ========================================================================================
# VISUALIZATION
# ========================================================================================
def plot_regime_clustering(df: pd.DataFrame,
labels: pd.Series) -> go.Figure:
"""
Create Plotly candlestick chart with regime lines overlay.
"""
fig = go.Figure()
# Add candlestick chart
fig.add_trace(
go.Candlestick(
x=df.index,
open=df['open'],
high=df['high'],
low=df['low'],
close=df['close'],
name='NIFTY',
increasing_line_color='lightgreen',
decreasing_line_color='lightcoral',
showlegend=False
)
)
# Group consecutive labels for line segments
regime_changes = labels.diff().fillna(0) != 0
regime_periods = []
current_regime = labels.iloc[0]
start_idx = 0
for i in range(1, len(labels)):
if regime_changes.iloc[i]:
regime_periods.append({
'start': start_idx,
'end': i - 1,
'regime': current_regime
})
current_regime = labels.iloc[i]
start_idx = i
# Add last period
regime_periods.append({
'start': start_idx,
'end': len(labels) - 1,
'regime': current_regime
})
# Add regime lines
shown_regimes = set()
for period in regime_periods:
start = period['start']
end = period['end']
regime = period['regime']
period_data = df.iloc[start:end+1]
color = 'green' if regime == 0 else 'red'
regime_name = 'Bull Regime (Low Vol)' if regime == 0 else 'Bear Regime (High Vol)'
show_in_legend = regime not in shown_regimes
if show_in_legend:
shown_regimes.add(regime)
fig.add_trace(
go.Scatter(
x=period_data.index,
y=period_data['close'],
mode='lines',
line=dict(color=color, width=3),
name=regime_name,
showlegend=show_in_legend,
legendgroup=f'regime_{regime}'
)
)
fig.update_layout(
title={
'text': 'NIFTY Market Regime Clustering (TRULY Non-Repainting)',
'x': 0.5,
'xanchor': 'center'
},
xaxis_title='Date',
yaxis_title='Price',
height=700,
showlegend=True,
hovermode='x unified',
xaxis_rangeslider_visible=False,
xaxis=dict(type='category')
)
return fig
# ========================================================================================
# MAIN EXECUTION
# ========================================================================================
def main():
"""
Main execution function.
"""
logger.info("OpenAlgo Python Bot is running.")
logger.info("=" * 80)
logger.info("NIFTY WASSERSTEIN K-MEANS - TRULY NON-REPAINTING VERSION")
logger.info("=" * 80)
# Initialize OpenAlgo client
client = api(api_key=API_KEY, host=API_HOST)
# Fetch data
df = fetch_nifty_data(client, START_DATE, END_DATE)
# Calculate log returns
df = create_log_returns(df)
# Test for repainting (optional - set to False to skip)
RUN_REPAINTING_TEST = True
if RUN_REPAINTING_TEST:
test_for_repainting(
df['log_return'],
window_length=WINDOW_LENGTH,
min_training_windows=MIN_TRAINING_WINDOWS,
n_clusters=K_CLUSTERS
)
# Create regime labels using TRULY non-repainting method
logger.info("\n" + "=" * 80)
logger.info("CREATING REGIME LABELS")
logger.info("=" * 80)
labels = classify_regime_non_repainting(
df['log_return'],
window_length=WINDOW_LENGTH,
min_training_windows=MIN_TRAINING_WINDOWS,
refit_frequency=REFIT_FREQUENCY,
n_clusters=K_CLUSTERS
)
df['regime'] = labels
# Print statistics
logger.info("\n" + "=" * 80)
logger.info("CLUSTERING RESULTS")
logger.info("=" * 80)
for cluster in range(K_CLUSTERS):
cluster_returns = df[df['regime'] == cluster]['log_return']
if len(cluster_returns) > 0:
logger.info(f"\nCluster {cluster} ({'Bull' if cluster == 0 else 'Bear'}):")
logger.info(f" Count: {len(cluster_returns)}")
logger.info(f" Percentage: {100*len(cluster_returns)/len(df):.1f}%")
logger.info(f" Mean Return: {cluster_returns.mean():.6f}")
logger.info(f" Std Dev: {cluster_returns.std():.6f}")
logger.info(f" Variance: {cluster_returns.var():.6f}")
# Count regime changes
regime_changes = (labels.diff() != 0).sum()
logger.info(f"\nTotal regime changes: {regime_changes}")
# Visualize
if SHOW_PLOT:
logger.info("\nCreating visualizations...")
fig = plot_regime_clustering(df, labels)
fig.show()
if SAVE_HTML:
fig.write_html(HTML_FILENAME)
logger.info(f"Saved plot to {HTML_FILENAME}")
logger.info("\n" + "=" * 80)
logger.info("ANALYSIS COMPLETE")
logger.info("=" * 80)
logger.info("\nThis version is TRULY non-repainting because:")
logger.info("1. Each bar's regime uses ONLY data available before that bar")
logger.info("2. Training windows are non-overlapping from historical data")
logger.info("3. Model is fitted on data from bars 0 to t-1 only")
logger.info("4. Current bar t is classified using this historical model")
logger.info("5. Labels do NOT change when new data arrives")
logger.info("=" * 80)
if __name__ == "__main__":
main()The output is a candlestick plot showing bull (green) and bear (red) periods that align closely with historical market events such as the 2020 crash or 2022 volatility phase.
What Researchers Observed
When tested on real and simulated data, the Wasserstein K-Means algorithm:
- Consistently identified well-known crisis periods and recovery phases.
- Produced clusters with high internal similarity and clear separation between regimes.
- Outperformed traditional moment-based K-Means and Gaussian HMMs in detecting subtle volatility shifts.
- Worked even on synthetic data generated from jump-diffusion models, where standard methods often failed.
How Traders and Analysts Can Use It
- Regime Detection Dashboard: Visualize market regimes with confidence scores.
- Dynamic Risk Management: Reduce position sizes during detected high-variance phases.
- Adaptive Model Training: Retrain machine-learning or forecasting models when a new regime emerges.
- Portfolio Hedging: Use regime signals to time protective option strategies or volatility hedges.
Key Takeaways
- Market regime detection is the foundation of adaptive trading systems.
- The Wasserstein K-Means algorithm reframes the task from comparing points to comparing entire distributions.
- It’s robust, interpretable, and works well for financial data where normality assumptions break down.
- With open-source tools like OpenAlgo, traders can implement it easily for Indian indices like NIFTY or BANKNIFTY.
References
- Horváth B., Issa Z., Muguruza A. (2021). Clustering Market Regimes Using the Wasserstein Distance. SSRN 3947905
- OpenAlgo Documentation – https://docs.openalgo.in
