Financial Time Series

Financial markets are among the hardest prediction targets. They're noisy, non-stationary, and efficient - meaning predictable patterns get arbitraged away. Yet ML has proven value in volatility forecasting, risk management, and alternative data integration. The key is realistic expectations and rigorous evaluation.

Where It Is Used

Financial ML powers critical applications:

• Quantitative trading: Alpha generation, signal extraction
• Risk management: VaR, stress testing, portfolio optimization
• Volatility forecasting: Options pricing, hedging strategies
• Alternative data: Sentiment, satellite imagery, transaction data
• Fraud and compliance: Anomaly detection, AML screening

Products: Bloomberg, Refinitiv, Kensho, Two Sigma, Citadel, hedge funds.

Typical Data

What You Do in Practice

1. Define target clearly - Returns? Volatility? Direction?
2. Establish proper timeline - When would you have known this?
3. Handle look-ahead bias - Point-in-time data only
4. Build walk-forward validation - No future information leakage
5. Account for transaction costs - Slippage, market impact
6. Test across regimes - Bull, bear, high-vol, low-vol
7. Implement risk constraints - Position limits, drawdown controls
8. Monitor for decay - Alpha deteriorates over time

Baseline First

Start with proven statistical models before ML:

import pandas as pd
import numpy as np
from arch import arch_model

class FinancialBaseline:
    def __init__(self, returns):
        """Initialize with return series."""
        self.returns = returns

    def garch_volatility(self, p=1, q=1, horizon=1):
        """GARCH model for volatility forecasting."""
        model = arch_model(
            self.returns * 100,  # Scale for numerical stability
            vol='Garch',
            p=p, q=q
        )
        fitted = model.fit(disp='off')

        # Forecast volatility
        forecast = fitted.forecast(horizon=horizon)
        variance_forecast = forecast.variance.iloc[-1].values / 10000

        return {
            'volatility_forecast': np.sqrt(variance_forecast),
            'model': fitted,
            'aic': fitted.aic,
            'bic': fitted.bic
        }

    def momentum_signal(self, lookback=20):
        """Simple momentum signal."""
        momentum = self.returns.rolling(lookback).mean()
        signal = np.sign(momentum)
        return signal

    def mean_reversion_signal(self, lookback=20, threshold=2):
        """Mean reversion signal based on z-score."""
        rolling_mean = self.returns.rolling(lookback).mean()
        rolling_std = self.returns.rolling(lookback).std()

        zscore = (self.returns - rolling_mean) / rolling_std
        signal = -np.sign(zscore) * (np.abs(zscore) > threshold)

        return signal

# Walk-forward validation
def walk_forward_backtest(data, model_fn, train_window=252, test_window=21):
    """Proper walk-forward backtesting."""
    results = []

    for i in range(train_window, len(data) - test_window, test_window):
        train_data = data.iloc[i-train_window:i]
        test_data = data.iloc[i:i+test_window]

        # Train model on historical data only
        model = model_fn(train_data)

        # Generate predictions for test period
        predictions = model.predict(test_data)

        results.append({
            'start_date': test_data.index[0],
            'end_date': test_data.index[-1],
            'predictions': predictions,
            'actuals': test_data
        })

    return results

Why GARCH baseline: - Well-understood statistical properties - Captures volatility clustering - Fast to fit, easy to interpret - Hard to beat for volatility forecasting

Regime Detection

Markets behave differently in different regimes:

from hmmlearn import hmm
import numpy as np

class RegimeDetector:
    def __init__(self, n_regimes=2):
        self.n_regimes = n_regimes
        self.model = None

    def fit(self, returns, volatility=None):
        """Fit Hidden Markov Model for regime detection."""
        # Prepare features
        features = returns.values.reshape(-1, 1)

        if volatility is not None:
            features = np.column_stack([features, volatility.values])

        # Fit HMM
        self.model = hmm.GaussianHMM(
            n_components=self.n_regimes,
            covariance_type='full',
            n_iter=100,
            random_state=42
        )
        self.model.fit(features)

        return self

    def predict_regime(self, returns, volatility=None):
        """Predict current regime."""
        features = returns.values.reshape(-1, 1)

        if volatility is not None:
            features = np.column_stack([features, volatility.values])

        states = self.model.predict(features)
        probabilities = self.model.predict_proba(features)

        return {
            'current_regime': states[-1],
            'regime_probabilities': probabilities[-1],
            'regime_history': states
        }

    def get_regime_characteristics(self):
        """Get mean and volatility of each regime."""
        means = self.model.means_
        covars = self.model.covars_

        regimes = {}
        for i in range(self.n_regimes):
            regimes[f'regime_{i}'] = {
                'mean_return': means[i][0],
                'volatility': np.sqrt(covars[i][0][0])
            }

        return regimes

Feature Engineering for Finance

def create_financial_features(df):
    """Create features for financial time series."""
    features = pd.DataFrame(index=df.index)

    # Returns at different horizons
    for lag in [1, 5, 21, 63]:
        features[f'return_{lag}d'] = df['close'].pct_change(lag)

    # Volatility measures
    features['volatility_21d'] = df['close'].pct_change().rolling(21).std()
    features['volatility_63d'] = df['close'].pct_change().rolling(63).std()

    # Price relative to moving averages
    for window in [20, 50, 200]:
        ma = df['close'].rolling(window).mean()
        features[f'price_to_ma{window}'] = df['close'] / ma - 1

    # RSI
    delta = df['close'].diff()
    gain = delta.where(delta > 0, 0).rolling(14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
    rs = gain / loss
    features['rsi_14'] = 100 - (100 / (1 + rs))

    # Volume features
    if 'volume' in df.columns:
        features['volume_ratio'] = df['volume'] / df['volume'].rolling(21).mean()
        features['dollar_volume'] = df['close'] * df['volume']

    # Volatility of volatility
    features['vol_of_vol'] = features['volatility_21d'].rolling(21).std()

    return features

def create_alternative_features(prices, sentiment, macro):
    """Combine price data with alternative data."""
    features = pd.DataFrame(index=prices.index)

    # Align sentiment data
    if sentiment is not None:
        features['sentiment_score'] = sentiment.reindex(prices.index, method='ffill')
        features['sentiment_change'] = features['sentiment_score'].diff(5)

    # Macro features (lagged to avoid look-ahead)
    if macro is not None:
        for col in macro.columns:
            # Use last available value (point-in-time)
            features[f'macro_{col}'] = macro[col].reindex(
                prices.index, method='ffill'
            ).shift(1)  # Extra lag for publication delay

    return features

Evaluation Metrics

def calculate_trading_metrics(returns, benchmark_returns=None, rf_rate=0.02):
    """Calculate comprehensive trading metrics."""
    # Annualization factor (assuming daily returns)
    ann_factor = 252

    # Basic metrics
    total_return = (1 + returns).prod() - 1
    ann_return = (1 + total_return) ** (ann_factor / len(returns)) - 1
    ann_vol = returns.std() * np.sqrt(ann_factor)

    # Sharpe ratio
    sharpe = (ann_return - rf_rate) / ann_vol if ann_vol > 0 else 0

    # Maximum drawdown
    cumulative = (1 + returns).cumprod()
    rolling_max = cumulative.expanding().max()
    drawdowns = cumulative / rolling_max - 1
    max_drawdown = drawdowns.min()

    # Calmar ratio
    calmar = ann_return / abs(max_drawdown) if max_drawdown != 0 else 0

    # Hit rate
    hit_rate = (returns > 0).mean()

    # Information ratio (if benchmark provided)
    info_ratio = None
    if benchmark_returns is not None:
        excess = returns - benchmark_returns
        info_ratio = excess.mean() / excess.std() * np.sqrt(ann_factor)

    return {
        'total_return': total_return,
        'annual_return': ann_return,
        'annual_volatility': ann_vol,
        'sharpe_ratio': sharpe,
        'max_drawdown': max_drawdown,
        'calmar_ratio': calmar,
        'hit_rate': hit_rate,
        'information_ratio': info_ratio
    }

Failure Modes and Debugging Checklist

• [ ] Look-ahead bias: Using future information in features
• [ ] Survivorship bias: Only using stocks that survived
• [ ] Overfitting: Strategy works only on training period
• [ ] Transaction costs ignored: Unrealistic profitability
• [ ] Regime change: Model trained in different market conditions
• [ ] Data snooping: Too many strategy variations tested
• [ ] Point-in-time violations: Using revised data
• [ ] Market impact: Not accounting for slippage

Production Checklist

Before deployment: - [ ] Walk-forward validation across multiple regimes - [ ] Transaction cost sensitivity analysis - [ ] Drawdown limits and position sizing rules - [ ] Point-in-time data verified - [ ] Latency requirements met

After deployment: - [ ] P&L attribution and monitoring - [ ] Regime detection for strategy switching - [ ] Decay monitoring - is alpha fading? - [ ] Risk limits enforced in real-time - [ ] Benchmark comparison daily

Common Pitfalls

# WRONG: Using future data
df['target'] = df['close'].shift(-1) / df['close'] - 1
df['ma_50'] = df['close'].rolling(50).mean()  # Includes future in rolling

# RIGHT: Proper point-in-time
df['target'] = df['close'].shift(-1) / df['close'] - 1
df['ma_50'] = df['close'].shift(1).rolling(50).mean()  # Only past data

# WRONG: Training on all data
model.fit(X, y)
predictions = model.predict(X)  # In-sample "predictions"

# RIGHT: Walk-forward
for train_end in train_dates:
    X_train = X[X.index < train_end]
    y_train = y[y.index < train_end]
    model.fit(X_train, y_train)
    # Predict only forward

Tools and Stack

Example Mini-Case

Scenario: Volatility forecasting for options market-making.

Approach: 1. Target: 5-day realized volatility 2. Features: Historical volatility, implied vol, regime indicators 3. Model: Ensemble of GARCH and gradient boosting 4. Validation: Walk-forward with 2 years training, 1 month test

Results: - Outperformed GARCH baseline by 15% in RMSE - Especially better during regime transitions - Used for volatility surface adjustments

Key learnings: - GARCH still best for normal conditions - ML adds value in regime changes and with alternative data - Ensemble more robust than single model - Must retrain monthly to avoid decay

Related Pages

• Index: ML Field Guide
• Time Series Anomaly Detection
• Demand Forecasting
• Anomaly Detection

Further Reading

• Advances in Financial ML - Marcos Lopez de Prado
• Machine Learning for Asset Managers
• Quantitative Trading - Ernest Chan

Try it live

Everything above runs in your browser — open GARCH Volatility Clustering Lab and change the parameters while it is running. Nothing is installed, nothing is uploaded, the whole model lives in one tab.