Fractal Analysis: Market Self-Similarity and Hurst Exponent
Financial markets exhibit fractal properties—patterns that repeat across different time scales. A 5-minute chart of the S&P 500 can look remarkably similar to a daily or weekly chart. This self-similarity suggests that markets follow fractal dynamics rather than purely random walks. Fractal analysis, particularly the Hurst exponent, provides quantitative measures of this behavior, enabling traders to distinguish trending markets from mean-reverting ones.Understanding Fractals and Market Dynamics
A fractal is a pattern that repeats at different scales. In markets, this manifests as similar price behavior across timeframes—what Benoit Mandelbrot called the "fractal nature of market time."The Hurst Exponent
The Hurst exponent (H) measures long-term memory in time series: ``
H = 0.5: Random walk (Brownian motion)
H > 0.5: Persistent (trending) behavior
H < 0.5: Anti-persistent (mean-reverting) behavior
`
The relationship between standard deviation and time:
`
σ(τ) ∝ τ^H
`
Where τ is the time interval.
Rescaled Range (R/S) Analysis
R/S analysis calculates H by examining how range scales with time:
`
R/S = (max_cumsum - min_cumsum) / std_dev
E[R/S] = (aH)τ^H
`
Taking logarithms:
`
log(E[R/S]) = log(aH) + H·log(τ)
`
The slope of log(R/S) vs log(τ) gives the Hurst exponent.
Key Takeaways
- Markets exhibit self-similar (fractal) behavior across time scales
- Hurst exponent quantifies persistence (H > 0.5) or mean reversion (H < 0.5)
- H = 0.5 indicates random walk (standard financial theory)
- Most markets show H ≈ 0.5-0.7 (weak persistence)
- Fractal dimension reveals complexity and information content
- Time-varying H captures regime shifts between trending and ranging
Why Fractal Analysis Matters for Trading
1. Regime Classification
Identify whether current market exhibits trending or mean-reverting behavior:
`python
import numpy as np
from hurst import compute_Hc, random_walk
def calculate_hurst_exponent(prices, lags=range(2, 100)):
"""
Calculate Hurst exponent using R/S analysis
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Rescaled range analysis
lags_array = np.array(list(lags))
tau = []
rs = []
for lag in lags:
# Subset into non-overlapping windows
n_windows = len(returns) // lag
if n_windows < 1:
continue
rs_values = []
for i in range(n_windows):
window = returns.iloc[ilag:(i+1)lag].values
if len(window) < lag:
continue
# Mean-adjusted cumulative sum
mean_centered = window - window.mean()
cumsum = np.cumsum(mean_centered)
# Range
R = cumsum.max() - cumsum.min()
# Standard deviation
S = window.std()
if S > 0:
rs_values.append(R / S)
if rs_values:
tau.append(lag)
rs.append(np.mean(rs_values))
# Linear regression: log(R/S) vs log(lag)
tau = np.array(tau)
rs = np.array(rs)
poly = np.polyfit(np.log(tau), np.log(rs), 1)
hurst = poly[0]
return hurst, tau, rs
Example
hurst, lags, rs = calculate_hurst_exponent(spy_prices)
if hurst > 0.55:
print(f"Trending market (H={hurst:.3f}) - use momentum strategies")
elif hurst < 0.45:
print(f"Mean-reverting market (H={hurst:.3f}) - use reversal strategies")
else:
print(f"Random walk (H={hurst:.3f}) - market is efficient")
`
2. Strategy Selection
Choose strategies based on measured persistence:
`python
def adaptive_strategy_selection(prices, window=252):
"""
Select strategy based on rolling Hurst exponent
"""
signals = []
for i in range(window, len(prices)):
window_prices = prices.iloc[i-window:i]
# Calculate Hurst for this window
hurst, _, _ = calculate_hurst_exponent(window_prices)
# Strategy selection
if hurst > 0.55:
# Trending: momentum strategy
returns = window_prices.pct_change()
signal = np.sign(returns.iloc[-20:].mean()) # 20-day momentum
elif hurst < 0.45:
# Mean-reverting: reversal strategy
current_price = window_prices.iloc[-1]
ma = window_prices.rolling(20).mean().iloc[-1]
signal = -np.sign(current_price - ma) # Fade deviations
else:
# Random walk: no signal
signal = 0
signals.append(signal)
return pd.Series(signals, index=prices.index[window:])
`
3. Fractal Dimension
Measure market complexity using fractal dimension:
`python
def fractal_dimension(prices, method='box_counting'):
"""
Calculate fractal dimension of price series
D = 2 - H (for self-affine fractals)
"""
if method == 'hurst_based':
hurst, _, _ = calculate_hurst_exponent(prices)
D = 2 - hurst
elif method == 'box_counting':
# Box-counting method
returns = np.log(prices / prices.shift(1)).dropna().values
# Normalize to [0, 1]
normalized = (returns - returns.min()) / (returns.max() - returns.min())
# Count boxes at different scales
scales = np.logspace(0.5, 3, num=20, dtype=int)
counts = []
for scale in scales:
# Grid resolution
bins = np.linspace(0, 1, scale)
counts.append(len(np.unique(np.digitize(normalized, bins))))
# Log-log regression
scales = scales[np.array(counts) > 0]
counts = np.array(counts)[np.array(counts) > 0]
poly = np.polyfit(np.log(scales), np.log(counts), 1)
D = poly[0]
return D
`
Practical Trading Applications
Rolling Hurst for Regime Detection
Track regime shifts in real-time:
`python
def rolling_hurst(prices, window=252, step=21):
"""
Calculate rolling Hurst exponent
"""
hurst_values = []
dates = []
for i in range(window, len(prices), step):
window_prices = prices.iloc[i-window:i]
try:
hurst, _, _ = calculate_hurst_exponent(window_prices)
hurst_values.append(hurst)
dates.append(prices.index[i])
except:
continue
return pd.Series(hurst_values, index=dates)
Visualization
rolling_h = rolling_hurst(spy_prices)
import matplotlib.pyplot as plt
plt.figure(figsize=(15, 6))
plt.plot(rolling_h.index, rolling_h.values)
plt.axhline(y=0.5, color='r', linestyle='--', label='Random Walk')
plt.axhline(y=0.55, color='g', linestyle='--', label='Persistence Threshold')
plt.axhline(y=0.45, color='b', linestyle='--', label='Mean Reversion Threshold')
plt.ylabel('Hurst Exponent')
plt.xlabel('Date')
plt.title('Rolling Hurst Exponent - Market Regime')
plt.legend()
plt.grid(True)
`
Hurst-Adjusted Position Sizing
Scale position size based on market persistence:
`python
def hurst_position_sizing(prices, base_size=1.0):
"""
Adjust position size based on Hurst exponent
"""
hurst, _, _ = calculate_hurst_exponent(prices)
# Persistence scaling
if hurst > 0.55:
# Trending: increase size (higher Sharpe potential)
multiplier = 1.0 + 2 * (hurst - 0.55)
elif hurst < 0.45:
# Mean reverting: moderate size (faster cycles)
multiplier = 1.0 + (0.45 - hurst)
else:
# Random: base size
multiplier = 1.0
position_size = base_size * multiplier
return position_size, hurst
`
Detrended Fluctuation Analysis (DFA)
Alternative method to estimate Hurst exponent:
`python
def detrended_fluctuation_analysis(prices, scales=None):
"""
DFA: robust Hurst estimation
More robust to non-stationarity than R/S
"""
returns = np.log(prices / prices.shift(1)).dropna().values
# Cumulative sum (profile)
mean_return = returns.mean()
profile = np.cumsum(returns - mean_return)
if scales is None:
scales = np.unique(np.logspace(0.5, 2.5, num=20).astype(int))
fluctuations = []
for scale in scales:
# Split into non-overlapping segments
n_segments = len(profile) // scale
if n_segments < 1:
continue
segment_fluctuations = []
for i in range(n_segments):
segment = profile[iscale:(i+1)scale]
# Detrend (linear fit)
x = np.arange(scale)
poly = np.polyfit(x, segment, 1)
trend = np.polyval(poly, x)
# Fluctuation
detrended = segment - trend
fluctuation = np.sqrt(np.mean(detrended**2))
segment_fluctuations.append(fluctuation)
fluctuations.append(np.mean(segment_fluctuations))
# Log-log regression
scales = np.array(scales[:len(fluctuations)])
fluctuations = np.array(fluctuations)
poly = np.polyfit(np.log(scales), np.log(fluctuations), 1)
hurst = poly[0] # Slope is Hurst exponent
return hurst, scales, fluctuations
`
Multi-Fractal Analysis
Measure scale-dependent Hurst exponents:
`python
def multifractal_spectrum(prices, q_range=np.arange(-5, 6)):
"""
Multi-fractal spectrum: H varies with moment order q
Monofractal: H(q) constant
Multifractal: H(q) varies
"""
returns = np.log(prices / prices.shift(1)).dropna().values
hurst_spectrum = {}
for q in q_range:
if q == 0:
continue # Skip q=0 (special case)
# Generalized Hurst exponent
scales = np.unique(np.logspace(0.5, 2.5, num=15).astype(int))
fluctuations = []
for scale in scales:
n_segments = len(returns) // scale
if n_segments < 1:
continue
segment_values = []
for i in range(n_segments):
segment = returns[iscale:(i+1)scale]
segment_values.append(np.abs(segment).sum())
# q-th moment
if q > 0:
fluctuation = np.mean(np.array(segment_values)q)(1/q)
else:
fluctuation = np.exp(np.mean(np.log(segment_values + 1e-10)))
fluctuations.append(fluctuation)
# Scaling exponent
scales = np.array(scales[:len(fluctuations)])
fluctuations = np.array(fluctuations)
poly = np.polyfit(np.log(scales), np.log(fluctuations + 1e-10), 1)
hurst_spectrum[q] = poly[0] / q if q != 0 else poly[0]
return hurst_spectrum
If H(q) is constant → monofractal
If H(q) decreases with q → multifractal (complex dynamics)
`
Advanced Techniques
Time-Varying Hurst with Wavelets
Combine wavelets and Hurst for multi-scale persistence:
`python
import pywt
def wavelet_hurst(prices, wavelet='db4', levels=5):
"""
Calculate Hurst exponent at each wavelet scale
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Wavelet decomposition
coeffs = pywt.wavedec(returns, wavelet, level=levels)
# Hurst at each scale
scale_hurst = {}
for i, detail in enumerate(coeffs[1:]):
scale = 2 ** (levels - i)
# Reconstruct this scale
coeffs_single = [np.zeros_like(c) for c in coeffs]
coeffs_single[i+1] = detail
reconstructed = pywt.waverec(coeffs_single, wavelet)
reconstructed = reconstructed[:len(returns)]
# Calculate Hurst for this scale
# Simplified: use variance scaling
var = np.var(detail)
scale_hurst[f'{scale}d'] = var
return scale_hurst
`
Hurst-Based Volatility Forecasting
Forecast volatility using persistence:
`python
def hurst_volatility_forecast(prices, horizon=21):
"""
Forecast volatility using Hurst scaling
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Current volatility (daily)
current_vol = returns.std()
# Hurst exponent
hurst, _, _ = calculate_hurst_exponent(prices)
# Scale volatility to horizon
# σ(T) = σ(1) * T^H
forecast_vol = current_vol (horizon * hurst)
return forecast_vol, hurst
Compare to standard scaling (sqrt(T) for random walk)
If H > 0.5: volatility grows faster than sqrt(T)
If H < 0.5: volatility grows slower than sqrt(T)
`
Optimal Holding Period
Determine optimal holding period based on autocorrelation decay:
`python
def optimal_holding_period(prices, max_lag=100):
"""
Find horizon where autocorrelation decays to threshold
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Autocorrelation at different lags
autocorr = [returns.autocorr(lag=lag) for lag in range(1, max_lag)]
# Find where autocorr crosses threshold (e.g., 0.1)
threshold = 0.1
optimal_lag = next((i+1 for i, ac in enumerate(autocorr) if abs(ac) < threshold),
max_lag)
# Relate to Hurst
hurst, _, _ = calculate_hurst_exponent(prices)
return {
'optimal_lag': optimal_lag,
'hurst': hurst,
'interpretation': 'Use momentum' if hurst > 0.55 else 'Use mean reversion'
}
`
Implementation Best Practices
1. Minimum Sample Size
Hurst estimation requires sufficient data:
`python
def validate_sample_size(prices, min_size=256):
"""
Check if sample size is sufficient for Hurst estimation
"""
if len(prices) < min_size:
raise ValueError(f"Need at least {min_size} observations. Have {len(prices)}.")
# Rule of thumb: need at least 2^8 = 256 points
# More data improves reliability
recommended_size = 2 ** int(np.log2(len(prices)))
return {
'actual_size': len(prices),
'recommended_size': recommended_size,
'sufficient': len(prices) >= min_size
}
`
2. Removing Trends
Detrend before Hurst calculation:
`python
from scipy.signal import detrend
def robust_hurst(prices, detrend_method='linear'):
"""
Calculate Hurst with proper detrending
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Detrend
if detrend_method == 'linear':
returns_detrended = detrend(returns, type='linear')
elif detrend_method == 'constant':
returns_detrended = returns - returns.mean()
else:
returns_detrended = returns.values
# Calculate Hurst on detrended data
prices_detrended = pd.Series(np.exp(np.cumsum(returns_detrended)),
index=returns.index)
hurst, lags, rs = calculate_hurst_exponent(prices_detrended)
return hurst
`
3. Confidence Intervals
Bootstrap confidence intervals for Hurst:
`python
def hurst_confidence_interval(prices, n_bootstrap=100, confidence=0.95):
"""
Bootstrap confidence intervals for Hurst exponent
"""
returns = np.log(prices / prices.shift(1)).dropna().values
hurst_estimates = []
for _ in range(n_bootstrap):
# Bootstrap sample
sample = np.random.choice(returns, size=len(returns), replace=True)
# Reconstruct prices
sample_prices = pd.Series(np.exp(np.cumsum(sample)))
# Calculate Hurst
h, _, _ = calculate_hurst_exponent(sample_prices)
hurst_estimates.append(h)
# Confidence interval
lower = np.percentile(hurst_estimates, (1 - confidence) * 50)
upper = np.percentile(hurst_estimates, (1 + confidence) * 50)
mean = np.mean(hurst_estimates)
return {
'mean': mean,
'lower': lower,
'upper': upper,
'confidence': confidence
}
`
Real-World Case Study
Adaptive Fractal Trading System
`python
class FractalTradingSystem:
def __init__(self, window=252, rebalance_freq=21):
self.window = window
self.rebalance_freq = rebalance_freq
def calculate_regime(self, prices):
"""Determine market regime via Hurst"""
hurst, _, _ = calculate_hurst_exponent(prices)
if hurst > 0.55:
return 'trending', hurst
elif hurst < 0.45:
return 'reverting', hurst
else:
return 'random', hurst
def generate_signals(self, prices, regime):
"""Generate signals based on regime"""
returns = prices.pct_change()
if regime == 'trending':
# Momentum: 20/50 crossover
ma20 = prices.rolling(20).mean()
ma50 = prices.rolling(50).mean()
signal = np.where(ma20 > ma50, 1, -1)
elif regime == 'reverting':
# Mean reversion: Bollinger Band
ma = prices.rolling(20).mean()
std = prices.rolling(20).std()
upper = ma + 2 * std
lower = ma - 2 * std
signal = np.where(prices < lower, 1,
np.where(prices > upper, -1, 0))
else:
# Random: no position
signal = np.zeros(len(prices))
return pd.Series(signal, index=prices.index)
def backtest(self, prices):
"""Full backtest with regime adaptation"""
returns = prices.pct_change()
portfolio_returns = []
regimes_log = []
for i in range(self.window, len(prices)):
# Determine regime
if i % self.rebalance_freq == 0:
window_prices = prices.iloc[i-self.window:i]
regime, hurst = self.calculate_regime(window_prices)
regimes_log.append({'date': prices.index[i], 'regime': regime, 'hurst': hurst})
# Generate signal
history = prices.iloc[:i]
signals = self.generate_signals(history, regime)
# Current position
position = signals.iloc[-1] if len(signals) > 0 else 0
# Return
port_return = position * returns.iloc[i]
portfolio_returns.append(port_return)
results = pd.Series(portfolio_returns, index=prices.index[self.window:])
regimes_df = pd.DataFrame(regimes_log)
return results, regimes_df
Usage
system = FractalTradingSystem(window=252, rebalance_freq=21)
strategy_returns, regimes = system.backtest(spy_prices)
Performance
sharpe = strategy_returns.mean() / strategy_returns.std() * np.sqrt(252)
cumulative = (1 + strategy_returns).cumprod()
print(f"Sharpe Ratio: {sharpe:.2f}")
print(f"Cumulative Return: {cumulative.iloc[-1] - 1:.2%}")
print(f"\nRegime Distribution:")
print(regimes['regime'].value_counts())
``