1. Learning Objectives

By the end of this lesson, you will be able to:

  • Design and implement a production-grade financial data pipeline with expandable windows.

  • Construct point-in-time feature matrices that completely eliminate look-ahead bias.

  • Compute technical indicators (RSI, MACD, Bollinger Bands, ATR) using vectorised Pandas.

  • Implement feature normalisation using expanding or rolling windows.

  • Create lagged features and interaction features for AI models.

  • Build an efficient backtesting framework with walk-forward validation.

  • Handle corporate actions (splits, dividends) using Pandas adjustment factors.

  • Implement a feature store for reproducible AI model training.

2. The Financial Data Pipeline Architecture

A production-grade pipeline has four stages: Ingestion, Cleaning, Feature Engineering, and Output.

Stage 1: Ingestion

text
import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta

def ingest_data(tickers, start_date, end_date):
    """
    Download adjusted close prices for given tickers.
    """
    data = yf.download(tickers, start=start_date, end=end_date)['Adj Close']
    data = data.rename(columns={t: t.replace(' ', '_') for t in data.columns})
    return data

tickers = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'META', 'TSLA']
start = '2015-01-01'
end = '2024-01-01'
raw_data = ingest_data(tickers, start, end)

Stage 2: Cleaning and Corporate Actions

text
def clean_data(df):
    """
    Handle missing values, adjust for corporate actions.
    """
    # Check for missing data
    missing_pct = df.isna().sum() / len(df) * 100
    print(f"Missing data: {missing_pct}")

    # Forward fill for weekends/holidays (only if < 5% missing)
    if df.isna().sum().max() < 0.05 * len(df):
        df = df.fillna(method='ffill')

    # Drop any remaining NaN rows
    df = df.dropna()

    # Adjust for splits (if not already adjusted)
    # Note: yfinance returns adjusted close automatically
    return df

cleaned_data = clean_data(raw_data)

3. Point-in-Time Feature Construction – The Cardinal Rule

The Principle: For any timestamp t, all features must be computed using ONLY data available strictly before t. No future data can leak into the training set.

3.1 Expanding Window vs Rolling Window

text
def construct_features_point_in_time(df, lookback_days=252):
    """
    Construct features using only past data (expanding window).
    This is the SAFEST method for live deployment.
    """
    features = pd.DataFrame(index=df.index)

    for t in range(lookback_days, len(df)):
        # Only use data up to t-1 (strictly past)
        past_data = df.iloc[:t-1]

        # Compute features using past_data only
        features.iloc[t] = compute_features(past_data.iloc[-lookback_days:])

    return features

3.2 Time Series Cross-Validation

text
from sklearn.model_selection import TimeSeriesSplit

def time_series_cv(X, y, n_splits=5, test_size=252):
    """
    Walk-forward cross-validation for financial time series.
    """
    tscv = TimeSeriesSplit(n_splits=n_splits, test_size=test_size, gap=1)

    for fold, (train_idx, val_idx) in enumerate(tscv.split(X)):
        X_train, X_val = X.iloc[train_idx], X.iloc[val_idx]
        y_train, y_val = y.iloc[train_idx], y.iloc[val_idx]

        # Train model on X_train, validate on X_val
        print(f"Fold {fold}: train {len(train_idx)}, val {len(val_idx)}")
        # ... train and evaluate ...

3.3 Feature Lagging – Preventing Look-Ahead

text
def create_lagged_features(df, lags=[1, 2, 3, 5, 10, 20]):
    """
    Create lagged features. Feature at time t uses price at t-lag.
    """
    df_lagged = df.copy()
    for lag in lags:
        df_lagged[f'Return_Lag_{lag}'] = df['Return'].shift(lag)
        df_lagged[f'Price_Lag_{lag}'] = df['Close'].shift(lag)
    return df_lagged

4. Technical Indicators – Vectorised Implementation

4.1 Relative Strength Index (RSI)

text
def compute_rsi(data, window=14):
    """
    RSI = 100 - 100 / (1 + RS)
    RS = Average Gain / Average Loss
    """
    delta = data.diff()
    gain = delta.clip(lower=0)
    loss = -delta.clip(upper=0)

    avg_gain = gain.rolling(window=window, min_periods=window).mean()
    avg_loss = loss.rolling(window=window, min_periods=window).mean()

    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

4.2 Moving Average Convergence Divergence (MACD)

text
def compute_macd(data, fast=12, slow=26, signal=9):
    """
    MACD = EMA(fast) - EMA(slow)
    Signal = EMA(signal, MACD)
    Histogram = MACD - Signal
    """
    ema_fast = data.ewm(span=fast, adjust=False).mean()
    ema_slow = data.ewm(span=slow, adjust=False).mean()
    macd = ema_fast - ema_slow
    signal_line = macd.ewm(span=signal, adjust=False).mean()
    histogram = macd - signal_line
    return macd, signal_line, histogram

4.3 Bollinger Bands

text
def compute_bollinger_bands(data, window=20, num_std=2):
    """
    Middle = SMA(window)
    Upper = Middle + num_std * Std(window)
    Lower = Middle - num_std * Std(window)
    %B = (Close - Lower) / (Upper - Lower)
    """
    middle = data.rolling(window=window).mean()
    std = data.rolling(window=window).std()
    upper = middle + num_std * std
    lower = middle - num_std * std
    pct_b = (data - lower) / (upper - lower)
    return upper, middle, lower, pct_b

4.4 Average True Range (ATR)

text
def compute_atr(high, low, close, window=14):
    """
    TR = max(High - Low, |High - Close_prev|, |Low - Close_prev|)
    ATR = SMA(TR, window)
    """
    tr1 = high - low
    tr2 = (high - close.shift()).abs()
    tr3 = (low - close.shift()).abs()
    true_range = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
    atr = true_range.rolling(window=window).mean()
    return atr

4.5 On-Balance Volume (OBV)

text
def compute_obv(close, volume):
    """
    OBV = cumulative volume with sign based on price direction
    """
    sign = np.sign(close.diff())
    obv = (sign * volume).cumsum()
    return obv

5. Feature Normalisation with Expanding Windows

text
def expanding_normalise(data, min_periods=252):
    """
    Normalise features using expanding mean and std.
    This prevents look-ahead bias.
    """
    mean = data.expanding(min_periods=min_periods).mean()
    std = data.expanding(min_periods=min_periods).std()
    normalised = (data - mean) / std
    return normalised, mean, std

def rolling_normalise(data, window=252):
    """
    Normalise features using rolling window.
    This is more adaptive to regime changes.
    """
    mean = data.rolling(window=window).mean()
    std = data.rolling(window=window).std()
    normalised = (data - mean) / std
    return normalised, mean, std

6. Target Engineering – Triple-Barrier Labelling

text
def triple_barrier_labels(close, upper_threshold=0.02, lower_threshold=0.02, horizon=20):
    """
    Implement the Triple-Barrier labelling method (Lopez de Prado).
    Returns:
        - 1 if upper barrier hit first
        - -1 if lower barrier hit first
        - 0 if time barrier expires first
    """
    labels = np.zeros(len(close))
    for i in range(len(close) - horizon):
        for j in range(1, horizon + 1):
            price_change = (close.iloc[i+j] / close.iloc[i]) - 1
            if price_change >= upper_threshold:
                labels[i] = 1
                break
            elif price_change <= -lower_threshold:
                labels[i] = -1
                break
            else:
                labels[i] = 0
    return labels

7. Feature Store – Reproducible Feature Engineering

A feature store stores computed features for reuse, ensuring that training and inference use identical transformations.

text
class FeatureStore:
    def __init__(self, base_path='./feature_store'):
        self.base_path = base_path
        os.makedirs(base_path, exist_ok=True)

    def compute_features(self, df, feature_set='technical'):
        """
        Compute a pre-defined feature set.
        """
        features = pd.DataFrame(index=df.index)

        if feature_set == 'technical':
            # Price-based
            features['Return'] = df['Close'].pct_change()
            features['Log_Return'] = np.log(df['Close'] / df['Close'].shift(1))
            features['High_Low_Ratio'] = df['High'] / df['Low']
            features['Close_Open_Ratio'] = df['Close'] / df['Open']

            # Technical indicators
            features['RSI_14'] = compute_rsi(df['Close'], 14)
            features['SMA_20'] = df['Close'].rolling(20).mean()
            features['SMA_50'] = df['Close'].rolling(50).mean()
            features['SMA_200'] = df['Close'].rolling(200).mean()
            features['Price_SMA_20'] = df['Close'] / features['SMA_20'] - 1
            features['Price_SMA_50'] = df['Close'] / features['SMA_50'] - 1

            # Volatility
            features['Vol_20'] = features['Return'].rolling(20).std()
            features['Vol_50'] = features['Return'].rolling(50).std()
            features['Vol_Ratio'] = features['Vol_20'] / features['Vol_50']

            # Volume-based
            features['Volume_Change'] = df['Volume'].pct_change()
            features['Volume_SMA_20'] = df['Volume'].rolling(20).mean()
            features['Volume_Ratio'] = df['Volume'] / features['Volume_SMA_20']

        return features

    def save_features(self, features, name='features.parquet'):
        features.to_parquet(f'{self.base_path}/{name}')

    def load_features(self, name='features.parquet'):
        return pd.read_parquet(f'{self.base_path}/{name}')

8. Building a Production-Ready Pipeline

text
class FinancialDataPipeline:
    def __init__(self, tickers, start_date, end_date):
        self.tickers = tickers
        self.start_date = start_date
        self.end_date = end_date
        self.raw_data = None
        self.cleaned_data = None
        self.features = None
        self.labels = None

    def run(self):
        """
        Execute the full pipeline.
        """
        print("1. Ingesting data...")
        self.raw_data = self.ingest()

        print("2. Cleaning data...")
        self.cleaned_data = self.clean()

        print("3. Computing features...")
        self.features = self.compute_features()

        print("4. Engineering targets...")
        self.labels = self.create_labels()

        print("5. Creating train/val/test split...")
        X_train, X_val, X_test, y_train, y_val, y_test = self.split_data()

        return X_train, X_val, X_test, y_train, y_val, y_test

    def ingest(self):
        return yf.download(self.tickers, start=self.start_date, end=self.end_date)['Adj Close']

    def clean(self):
        df = self.raw_data.fillna(method='ffill').dropna()
        return df

    def compute_features(self):
        fs = FeatureStore()
        features = {}
        for ticker in self.tickers:
            ticker_data = pd.DataFrame({
                'Open': self.cleaned_data[ticker + '_Open'],
                'High': self.cleaned_data[ticker + '_High'],
                'Low': self.cleaned_data[ticker + '_Low'],
                'Close': self.cleaned_data[ticker + '_Close'],
                'Volume': self.cleaned_data[ticker + '_Volume']
            })
            features[ticker] = fs.compute_features(ticker_data)
        return pd.concat(features, axis=1)

    def create_labels(self):
        labels = pd.DataFrame()
        for ticker in self.tickers:
            close = self.cleaned_data[ticker + '_Close']
            labels[ticker] = triple_barrier_labels(close)
        return labels

    def split_data(self, train_ratio=0.7, val_ratio=0.15):
        n = len(self.features)
        train_end = int(n * train_ratio)
        val_end = int(n * (train_ratio + val_ratio))

        X_train = self.features.iloc[:train_end]
        X_val = self.features.iloc[train_end:val_end]
        X_test = self.features.iloc[val_end:]

        y_train = self.labels.iloc[:train_end]
        y_val = self.labels.iloc[train_end:val_end]
        y_test = self.labels.iloc[val_end:]

        return X_train, X_val, X_test, y_train, y_val, y_test

9. Summary for the AI Practitioner

  1. Point-in-Time is non-negotiable. Never use KFold; always use TimeSeriesSplit.

  2. Feature lagging prevents look-ahead bias. Features at time t must use data up to t-1.

  3. Technical indicators are computed using vectorised Pandas operations. This is orders of magnitude faster than Python loops.

  4. Normalisation must be done with expanding or rolling windows, not the entire dataset. This prevents data leakage.

  5. Triple-Barrier labelling is superior to simple binary classification. It accounts for realistic stop-loss/take-profit constraints.

  6. Feature stores ensure reproducibility. Save features to Parquet files for reuse across experiments.

  7. Pipeline design should be modular: Ingestion → Cleaning → Feature Engineering → Labelling → Splitting.

  8. Walk-forward validation is the gold standard for financial AI backtesting.

End of Lessons 3.1 and 3.2

In Lessons 3.3 and 3.4, we will cover Deep Learning with PyTorch/TensorFlow, focusing on building neural networks for financial time series, handling sequential data with LSTMs and Transformers, and implementing custom loss functions for financial objectives.

 
 
 
PreviousNext