Build a Production ML Signal Pipeline

Build a quant ML pipeline with a feature store, XGBoost, and drift monitoring.

Introduction

30 Second Summary

Most machine learning models at financial firms don't fail because of bad predictions. They fail because the system around them leaks future information into training, serves outdated data, or silently breaks without anyone noticing.

In this project, you will build a production-grade ML signal pipeline that mirrors what senior AI engineers build at quantitative hedge funds. The model is intentionally simple because this project focuses on the production engineering that prevents silent failures.

What You'll Build

A fully local ML pipeline that ingests market data, computes technical indicators, stores features in a feature store, trains a stock-ranking model with walk-forward validation, and monitors for data drift using KL Divergence.

By the end of this project, you'll have:

  • A data ingestion pipeline that downloads market data and computes technical indicators with strict temporal correctness.
  • A Feast feature store ensuring training-serving consistency and point-in-time correctness for your ML model.
  • A trained XGBoost signal model validated with walk-forward methodology to prevent look-ahead bias.
  • An automated drift monitoring system that detects when live market data diverges from training data.
  • Secret Mission: Add a backtesting engine with Sharpe ratio, max drawdown, and realistic transaction cost modeling.

Are there any prerequisites?

This project assumes you know basic Python syntax (variables, functions, loops) and have a general understanding of stock market OHLCV data (open, high, low, close, volume). No prior machine learning or quantitative finance experience is required.

Before We Start

You're about to build the production ML infrastructure layer that separates senior AI engineers from junior data scientists at hedge funds. The model itself (XGBoost) is intentionally simple.

The engineering around it is what makes this production-worthy: a feature store, point-in-time correctness, walk-forward validation, and drift monitoring. In this step, you'll install Python 3.11.9 and set up a clean project environment with all the packages you need.

In this step, get ready to:
  • Install Python 3.11.9 and verify it's on your PATH.
  • Create a project directory with a virtual environment and install all dependencies.
  • Verify that every package imports correctly.
Install Python 3.11.9

This project requires Python 3.11.9 specifically because pandas 3.0.3 requires Python 3.11 or higher. Older versions like 3.8 or 3.9 will fail silently when installing dependencies.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Visual Studio Code and press Enter to open it.
  • Open VS Code's built-in terminal by pressing Ctrl+` (both macOS and Windows).
  • Check your Python version by running this command:
python --version

Which command should I use?

On macOS, if python --version shows Python 2 or fails, try python3 --version instead. On Windows, python --version should work after installation.

✔️ I see Python 3.11.9

Python 3.11.9 is ready to go. Skip ahead to the next substep.

ⓧ I see an older version

You have Python installed but it's an older version (like 3.8 or 3.9) that won't work with this project's dependencies. You need to install 3.11.9 alongside your existing version.

Windows

  • Open the Python 3.11.9 release page in your browser:
  • https://www.python.org/downloads/release/python-3119/
  • Scroll down and click the Windows installer (64-bit) link to download python-3.11.9-amd64.exe.
  • Run the installer.
  • On the first screen, check Add python.exe to PATH at the bottom.
  • Click Install Now.
  • On the final screen, click Disable path length limit if it appears.
  • Close your terminal in VS Code and open a new one (Ctrl+` twice) so the updated PATH takes effect.
  • Verify the new version by running:
python --version

You should see Python 3.11.9.

macOS

  • Open the Python 3.11.9 release page in your browser:
  • https://www.python.org/downloads/release/python-3119/
  • Click the macOS 64-bit universal2 installer link to download the .pkg file.
  • Double-click the downloaded file and follow the installer prompts.
  • After installation, open the Applications/Python 3.11 folder and double-click Install Certificates.command.
  • Close your terminal in VS Code and open a new one (Ctrl+` twice) so the updated PATH takes effect.
  • Verify the new version by running:
python3 --version

You should see Python 3.11.9. On macOS, use python3 for all commands in this project.

ⓧ Command not found

Python is not installed on your machine. Follow the steps below to install 3.11.9.

Windows

  • Open the Python 3.11.9 release page in your browser:
  • https://www.python.org/downloads/release/python-3119/
  • Scroll down and click the Windows installer (64-bit) link to download python-3.11.9-amd64.exe.
  • Run the installer.
  • On the first screen, check Add python.exe to PATH at the bottom.
  • Click Install Now.
  • On the final screen, click Disable path length limit if it appears.
  • Close your terminal in VS Code and open a new one (Ctrl+` twice) so the updated PATH takes effect.
  • Verify by running:
python --version

You should see Python 3.11.9.

macOS

  • Open the Python 3.11.9 release page in your browser:
  • https://www.python.org/downloads/release/python-3119/
  • Click the macOS 64-bit universal2 installer link to download the .pkg file.
  • Double-click the downloaded file and follow the installer prompts.
  • After installation, open the Applications/Python 3.11 folder and double-click Install Certificates.command.
  • Close your terminal in VS Code and open a new one (Ctrl+` twice) so the updated PATH takes effect.
  • Verify by running:
python3 --version

You should see Python 3.11.9. On macOS, use python3 for all commands in this project.

Why Python 3.11.9 specifically?

The pandas 3.0.3 library requires Python 3.11 or higher. XGBoost 3.2.0 requires Python 3.10 or higher.

Python 3.11.9 satisfies both and is the last version of 3.11 with binary installers on python.org.

Create the project directory and install dependencies

Now you'll create a dedicated folder for this project and set up an isolated Python environment inside it. This keeps your project's packages separate from any other Python work on your machine.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir quant-signal-pipeline
cd quant-signal-pipeline
  • Create a virtual environment by running:
python -m venv venv

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here won't affect any other projects on your machine.

On macOS, replace python with python3 in all commands throughout this project.

  • Activate the virtual environment by running:

Windows

venv\Scripts\activate

macOS

source venv/bin/activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

Make sure you ran the activate command from inside the quant-signal-pipeline folder. On Windows, if you see an execution policy error, run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned in PowerShell first, then try activating again.

help me activate my Python virtual environment

Now you'll create the file that lists all the packages this project needs.

  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop and select the quant-signal-pipeline folder, then click Open.
  • Click the new-file icon in the Explorer sidebar (or press Ctrl+N).
  • Name the file requirements.txt and paste the following content:
pandas==3.0.3
yfinance==1.4.1
feast==0.62.0
xgboost==3.2.0
scikit-learn
numpy
scipy
pyarrow

What does this file do?

A requirements.txt file lists every Python package your project depends on. Pinning versions (like pandas==3.0.3) ensures everyone gets the exact same working versions.

Feast is the open-source feature store you'll use for training/serving consistency. yfinance downloads free market data from Yahoo Finance with no API key required.

  • Save requirements.txt (Cmd+S on macOS or Ctrl+S on Windows).
  • Reopen your terminal (Ctrl+`) and make sure you still see (venv) in the prompt.
  • Install all the dependencies by running:
pip install -r requirements.txt

This will download and install all eight packages plus their sub-dependencies. It may take 1-2 minutes.

Installation failing on XGBoost?

On Windows, XGBoost requires the Visual C++ Redistributable DLLs. If you see an error about missing DLLs, download and install the latest x64 Visual C++ Redistributable from Microsoft's official page. Then retry the pip install command.

If you already have Visual Studio installed, you likely have these DLLs already and won't see this error.

help me fix XGBoost installation errors on Windows

✔️ Awesome, I've got everything!

Great. Double check you've saved requirements.txt and the install completed without errors.

ⓧ I'd like to double check the full code

Here's what your requirements.txt should look like:

pandas==3.0.3
yfinance==1.4.1
feast==0.62.0
xgboost==3.2.0
scikit-learn
numpy
scipy
pyarrow
Verify all packages installed correctly

A quick import test confirms that every package is properly installed and accessible from your virtual environment.

  • Run this verification command in your terminal:
python -c "import feast; import xgboost; import yfinance; import pandas; print('All packages ready')"

What should I see?

If everything installed correctly, you'll see All packages ready printed in your terminal. On macOS, use python3 instead of python in the command above.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is active (you should see (venv) in the prompt). If you opened a new terminal, you need to re-activate it.

If a specific package fails, try reinstalling it individually:

pip install feast

help me debug a ModuleNotFoundError in my Python virtual environment

Your environment is ready. Next up, you'll build the data ingestion pipeline that downloads market data and engineers features with strict temporal correctness.

Ingest Data and Engineer Features

Your environment is ready. Now it's time to build the first real component of the pipeline: the data layer.

Before any ML model can generate signals, it needs clean features computed from raw market data. In this step, you'll build a data ingestion script that downloads historical stock prices and computes six technical indicators. Every indicator uses only past data at each point, which prevents a critical flaw called look-ahead bias.

In this step, get ready to:
  • Create technical indicator functions that compute features using only historical data.
  • Build a data pipeline that downloads OHLCV data for 10 S&P 500 stocks.
  • Save the complete feature matrix as a timestamped Parquet file.
Create the ingestion script with technical indicator functions

Technical indicators transform raw price data into signals that capture momentum, trend, and volatility. You'll write four helper functions that each compute one family of indicators using only data available up to that point in time.

  • In VS Code, create a new file called ingest_data.py in the root of your quant-signal-pipeline folder.
  • Add the imports, constants, and the RSI computation function by pasting this code:
import os
import numpy as np
import pandas as pd
import yfinance as yf
from datetime import datetime, timedelta

# 10 liquid S&P 500 stocks as our investment universe
TICKERS = ["AAPL", "MSFT", "GOOGL", "AMZN", "NVDA",
            "META", "JPM", "JNJ", "V", "PG"]
END_DATE = datetime.now()
START_DATE = END_DATE - timedelta(days=730)  # 2 years of history
OUTPUT_DIR = "feature_repo/data"


def compute_rsi(series, period=14):
    # RSI measures momentum by comparing average gains to average losses
    delta = series.diff()
    gain = delta.where(delta > 0, 0.0)
    loss = -delta.where(delta < 0, 0.0)
    avg_gain = gain.rolling(window=period).mean()
    avg_loss = loss.rolling(window=period).mean()
    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

What does this code do?

The imports bring in yfinance for downloading market data and pandas for data manipulation.

The constants define a basket of 10 stocks and a 2-year lookback window (730 days). The compute_rsi function calculates the Relative Strength Index, a momentum oscillator that ranges from 0 to 100. Values above 70 suggest overbought conditions. Values below 30 suggest oversold.

  • Below the compute_rsi function, add the MACD computation function:
def compute_macd(series, fast=12, slow=26):
    # MACD captures trend direction using the gap between fast and slow EMAs
    ema_fast = series.ewm(span=fast, adjust=False).mean()
    ema_slow = series.ewm(span=slow, adjust=False).mean()
    return ema_fast - ema_slow

What is MACD?

MACD (Moving Average Convergence Divergence) measures the difference between a short-term and long-term exponential moving average. When the fast EMA crosses above the slow EMA, it signals upward momentum.

  • Below compute_macd, add the two volatility indicator functions:
def compute_bollinger_width(series, period=20):
    # Bollinger Width measures price volatility relative to its moving average
    sma = series.rolling(window=period).mean()
    std = series.rolling(window=period).std()
    upper = sma + 2 * std
    lower = sma - 2 * std
    width = (upper - lower) / sma
    return width


def compute_atr(high, low, close, period=14):
    # ATR measures true range (including gaps) averaged over a window
    tr1 = high - low
    tr2 = (high - close.shift(1)).abs()
    tr3 = (low - close.shift(1)).abs()
    tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
    atr = tr.rolling(window=period).mean()
    return atr

What do these volatility indicators capture?

Bollinger Band width expands when a stock becomes volatile and contracts during calm periods. Average True Range (ATR) accounts for price gaps between trading days, giving a more complete picture of volatility than simple high-minus-low.

Both use rolling windows, which means they only look at data up to the current point. This prevents any future information from leaking into the features.

Add the feature engineering and data pipeline functions

Now you'll combine the indicator functions into an orchestrator that applies all six features to a single stock's data, then build the main pipeline that downloads and processes all 10 tickers.

  • Below the compute_atr function, add the engineer_features function that applies all indicators and creates the forward return label:
def engineer_features(df):
    # Apply all technical indicators to a single stock's DataFrame
    df = df.copy()
    df["sma_20"] = df["Close"].rolling(window=20).mean()
    df["sma_50"] = df["Close"].rolling(window=50).mean()
    df["rsi_14"] = compute_rsi(df["Close"], period=14)
    df["macd"] = compute_macd(df["Close"])
    df["bollinger_width"] = compute_bollinger_width(df["Close"])
    df["atr_14"] = compute_atr(df["High"], df["Low"], df["Close"])
    # 5-day forward return is the supervised learning target
    df["forward_return_5d"] = df["Close"].shift(-5) / df["Close"] - 1
    return df

Why compute forward returns here?

The forward_return_5d column is the label for supervised learning. It tells the model what actually happened 5 trading days after each row. The model learns to predict which stocks will outperform based on the technical indicators available at that point.

Using shift(-5) looks forward in time, but that's fine for the label. The features (SMA, RSI, MACD, etc.) only use past data. The walk-forward validation in a later step ensures the model never trains on future labels.

  • Below engineer_features, add the main() function that downloads data for all tickers and saves the output:
def main():
    os.makedirs(OUTPUT_DIR, exist_ok=True)
    all_features = []

    for ticker in TICKERS:
        print(f"Downloading {ticker}...")
        data = yf.download(ticker, start=START_DATE, end=END_DATE, progress=False)

        if data.empty:
            print(f"  No data for {ticker}, skipping.")
            continue

        # Handle yfinance MultiIndex columns from single-ticker downloads
        data = data.reset_index()
        if isinstance(data.columns, pd.MultiIndex):
            data.columns = data.columns.get_level_values(0)

        data = engineer_features(data)
        data["ticker"] = ticker
        # Drop rows where the longest indicator (SMA_50) hasn't warmed up
        data = data.dropna(subset=["sma_50", "forward_return_5d"])
        all_features.append(data)

    # Combine all tickers into one DataFrame
    combined = pd.concat(all_features, ignore_index=True)
    combined = combined.rename(columns={"Date": "event_timestamp"})
    combined["event_timestamp"] = pd.to_datetime(combined["event_timestamp"], utc=True)

    # Select only the columns needed for the feature store
    output_cols = [
        "event_timestamp", "ticker",
        "sma_20", "sma_50", "rsi_14", "macd", "bollinger_width", "atr_14",
        "forward_return_5d"
    ]
    combined = combined[output_cols]

    output_path = os.path.join(OUTPUT_DIR, "stock_features.parquet")
    combined.to_parquet(output_path, index=False)
    print(f"\nSaved {len(combined)} rows to {output_path}")
    print(f"Date range: {combined['event_timestamp'].min()} to {combined['event_timestamp'].max()}")
    print(f"Tickers: {combined['ticker'].nunique()}")


if __name__ == "__main__":
    main()

What does this pipeline do?

The main() function loops through each ticker, downloads 2 years of daily OHLCV data from Yahoo Finance, applies all technical indicators, and drops rows where the slowest indicator (50-day SMA) hasn't accumulated enough history.

It also handles a quirk of yfinance: even single-ticker downloads return a MultiIndex on columns. The code flattens this to a simple column layout. Finally, everything is saved as a Parquet file with UTC timestamps for Feast compatibility.

  • Save ingest_data.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before running it in the next substep.

ⓧ I'd like to double check the full code

import os
import numpy as np
import pandas as pd
import yfinance as yf
from datetime import datetime, timedelta


TICKERS = ["AAPL", "MSFT", "GOOGL", "AMZN", "NVDA",
            "META", "JPM", "JNJ", "V", "PG"]
END_DATE = datetime.now()
START_DATE = END_DATE - timedelta(days=730)
OUTPUT_DIR = "feature_repo/data"


def compute_rsi(series, period=14):
    delta = series.diff()
    gain = delta.where(delta > 0, 0.0)
    loss = -delta.where(delta < 0, 0.0)
    avg_gain = gain.rolling(window=period).mean()
    avg_loss = loss.rolling(window=period).mean()
    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi


def compute_macd(series, fast=12, slow=26):
    ema_fast = series.ewm(span=fast, adjust=False).mean()
    ema_slow = series.ewm(span=slow, adjust=False).mean()
    return ema_fast - ema_slow


def compute_bollinger_width(series, period=20):
    sma = series.rolling(window=period).mean()
    std = series.rolling(window=period).std()
    upper = sma + 2 * std
    lower = sma - 2 * std
    width = (upper - lower) / sma
    return width


def compute_atr(high, low, close, period=14):
    tr1 = high - low
    tr2 = (high - close.shift(1)).abs()
    tr3 = (low - close.shift(1)).abs()
    tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
    atr = tr.rolling(window=period).mean()
    return atr


def engineer_features(df):
    df = df.copy()
    df["sma_20"] = df["Close"].rolling(window=20).mean()
    df["sma_50"] = df["Close"].rolling(window=50).mean()
    df["rsi_14"] = compute_rsi(df["Close"], period=14)
    df["macd"] = compute_macd(df["Close"])
    df["bollinger_width"] = compute_bollinger_width(df["Close"])
    df["atr_14"] = compute_atr(df["High"], df["Low"], df["Close"])
    df["forward_return_5d"] = df["Close"].shift(-5) / df["Close"] - 1
    return df


def main():
    os.makedirs(OUTPUT_DIR, exist_ok=True)
    all_features = []

    for ticker in TICKERS:
        print(f"Downloading {ticker}...")
        data = yf.download(ticker, start=START_DATE, end=END_DATE, progress=False)

        if data.empty:
            print(f"  No data for {ticker}, skipping.")
            continue

        data = data.reset_index()
        if isinstance(data.columns, pd.MultiIndex):
            data.columns = data.columns.get_level_values(0)

        data = engineer_features(data)
        data["ticker"] = ticker
        data = data.dropna(subset=["sma_50", "forward_return_5d"])
        all_features.append(data)

    combined = pd.concat(all_features, ignore_index=True)
    combined = combined.rename(columns={"Date": "event_timestamp"})
    combined["event_timestamp"] = pd.to_datetime(combined["event_timestamp"], utc=True)

    output_cols = [
        "event_timestamp", "ticker",
        "sma_20", "sma_50", "rsi_14", "macd", "bollinger_width", "atr_14",
        "forward_return_5d"
    ]
    combined = combined[output_cols]

    output_path = os.path.join(OUTPUT_DIR, "stock_features.parquet")
    combined.to_parquet(output_path, index=False)
    print(f"\nSaved {len(combined)} rows to {output_path}")
    print(f"Date range: {combined['event_timestamp'].min()} to {combined['event_timestamp'].max()}")
    print(f"Tickers: {combined['ticker'].nunique()}")


if __name__ == "__main__":
    main()
Run the script and verify the output

With the script complete, run it to download live market data and produce your feature matrix.

  • In your terminal (with the virtual environment still active), run the ingestion script:
python ingest_data.py
  • Wait for the download to complete. You should see each ticker being downloaded, followed by a summary.

What should I see?

The script prints each ticker as it downloads, then a final summary showing approximately 4800 rows saved to feature_repo/data/stock_features.parquet. The exact row count depends on market holidays and ticker availability but should be close to 4800 (10 tickers x ~480 trading days).

  • Verify the Parquet file exists and has the right shape by running:
python -c "import pandas as pd; df = pd.read_parquet('feature_repo/data/stock_features.parquet'); print(f'Shape: {df.shape}'); print(f'Columns: {list(df.columns)}'); print(f'NaN count: {df.isna().sum().sum()}')"
  • Confirm the output shows a shape around (4800, 9), the 9 expected columns, and 0 NaN values.

Script failing or returning empty data?

  • If you see a network error, check your internet connection. yfinance downloads from Yahoo Finance's public API.
  • If a specific ticker returns empty data, the script handles this gracefully and skips it. Your final row count may be slightly lower.
  • If you see a ModuleNotFoundError, make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.

Help me debug my ingest_data.py script. I'm getting an error when running it.

Your raw market data is now transformed into a clean feature matrix ready for the feature store. Next up, you'll register these features in Feast to guarantee the same features are served consistently during both training and live prediction.

Set Up the Feast Feature Store

Your data pipeline is producing clean, timestamped features. Now you need a system that guarantees those features are served consistently between training and production.

Without a feature store, teams often compute features differently in training notebooks versus production serving code. This causes training-serving skew, one of the most common (and hardest to debug) failures in production ML. In this step, you will configure Feast as your feature store so that training and serving always use the exact same feature logic.

In this step, get ready to:
  • Define your feature store configuration and feature definitions.
  • Register features with Feast and materialize them to the online store.
  • Verify point-in-time correctness with a retrieval test.
Create the Feast configuration and feature definitions

Feast needs two files to operate: a project configuration file that tells it where to store data, and a Python file that defines what features exist and where they come from.

  • In VS Code, create a new file called feature_store.yaml inside the feature_repo/ directory.
  • Paste the following configuration into feature_store.yaml:
project: quant_signals
provider: local
registry: data/registry.db
online_store:
  type: sqlite
  path: data/online_store.db
offline_store:
  type: file
entity_key_serialization_version: 3

What does this configuration do?

This file tells Feast how to operate locally without any cloud infrastructure.

  • provider: local means Feast runs entirely on your machine.
  • registry: data/registry.db stores entity and feature view definitions in a local SQLite database.
  • online_store uses SQLite for low-latency feature serving.
  • offline_store: type: file uses Parquet files directly for batch retrieval (training data generation).
  • Save feature_store.yaml.
  • Create another new file called features.py inside the same feature_repo/ directory.
  • Paste the following feature definitions into features.py:
from datetime import timedelta

from feast import Entity, FeatureView, Field, FileSource
from feast.types import Float32, String

stock = Entity(
    name="ticker",
    join_keys=["ticker"],
    description="Stock ticker symbol",
)

stock_features_source = FileSource(
    path="data/stock_features.parquet",
    timestamp_field="event_timestamp",
)

stock_features_view = FeatureView(
    name="stock_technical_features",
    entities=[stock],
    schema=[
        Field(name="sma_20", dtype=Float32),
        Field(name="sma_50", dtype=Float32),
        Field(name="rsi_14", dtype=Float32),
        Field(name="macd", dtype=Float32),
        Field(name="bollinger_width", dtype=Float32),
        Field(name="atr_14", dtype=Float32),
        Field(name="forward_return_5d", dtype=Float32),
    ],
    online=True,
    source=stock_features_source,
    ttl=timedelta(days=7),
)

What does this code do?

This file teaches Feast the shape of your data.

  • Entity defines what you look up features by. Here, each stock ticker is an entity.
  • FileSource points Feast at your Parquet file and tells it which column holds the timestamp.
  • FeatureView defines the group of features that belong together. The schema lists each feature column and its type.
  • ttl=timedelta(days=7) sets a 7-day time-to-live, meaning the online store only serves features from the last 7 days.
  • Save features.py.

Not sure where to create these files?

Both files go inside the feature_repo/ folder that already exists in your quant-signal-pipeline/ project. Your folder structure should look like this: quant-signal-pipeline/feature_repo/feature_store.yaml and quant-signal-pipeline/feature_repo/features.py.

help me find the right folder for my Feast files

✔️ Awesome, I've got everything!

Great. Double check you have saved both files before continuing.

ⓧ I'd like to double check the full code

Here is the complete feature_store.yaml:

project: quant_signals
provider: local
registry: data/registry.db
online_store:
  type: sqlite
  path: data/online_store.db
offline_store:
  type: file
entity_key_serialization_version: 3

Here is the complete features.py:

from datetime import timedelta

from feast import Entity, FeatureView, Field, FileSource
from feast.types import Float32, String

stock = Entity(
    name="ticker",
    join_keys=["ticker"],
    description="Stock ticker symbol",
)

stock_features_source = FileSource(
    path="data/stock_features.parquet",
    timestamp_field="event_timestamp",
)

stock_features_view = FeatureView(
    name="stock_technical_features",
    entities=[stock],
    schema=[
        Field(name="sma_20", dtype=Float32),
        Field(name="sma_50", dtype=Float32),
        Field(name="rsi_14", dtype=Float32),
        Field(name="macd", dtype=Float32),
        Field(name="bollinger_width", dtype=Float32),
        Field(name="atr_14", dtype=Float32),
        Field(name="forward_return_5d", dtype=Float32),
    ],
    online=True,
    source=stock_features_source,
    ttl=timedelta(days=7),
)
Register features and materialize to the online store

With your definitions in place, you need to register them with Feast and then load feature values into the online store for low-latency serving.

  • Navigate into the feature_repo/ directory by running:
cd feature_repo
  • Register your entity and feature view definitions by running:
feast apply

You should see output confirming that the entity and feature view were created.

What does feast apply do?

This command scans the current directory for Python files containing Feast object definitions (entities, feature views, data sources). It registers them in the registry.db file so Feast knows what features exist and where to find them.

  • Verify your feature view is registered by running:
feast feature-views list

You should see a table showing stock_technical_features with entity ticker and type FeatureView.

Seeing a 'No module named feast' error?

Make sure your virtual environment is still activated. You should see (venv) at the beginning of your terminal prompt. If not, run venv\Scripts\activate (Windows) or source venv/bin/activate (macOS/Linux) from your quant-signal-pipeline/ directory.

help me fix my Feast installation

Now you need to materialize features into the online store. This copies the latest feature values from your Parquet file into SQLite for fast serving.

  • Materialize your features by running this Python script from the feature_repo/ directory:
python -c "
from datetime import datetime, timedelta
from feast import FeatureStore

fs = FeatureStore(repo_path='.')
end_date = datetime.now()
start_date = end_date - timedelta(days=730)
fs.materialize(start_date=start_date, end_date=end_date)
print('Materialization complete!')
"

You should see output indicating that features are being materialized, followed by the confirmation message.

Why materialize to the online store?

The offline store (Parquet files) is designed for bulk retrieval when generating training datasets. The online store (SQLite) is designed for low-latency lookups when serving predictions in production.

Materialization copies the latest feature values per entity into the online store. In production at a hedge fund, this runs on a schedule (e.g., after market close) so the serving layer always has fresh data.

Materialization failing with a timestamp error?

Make sure you are running the command from inside the feature_repo/ directory, not from the project root. Feast reads feature_store.yaml from the current working directory.

help me fix my Feast materialization

Verify point-in-time correctness with a retrieval test

The core guarantee of a feature store is point-in-time correctness. When you request features for a given entity at a given timestamp, Feast returns only data that was available at that moment. No future data leaks in.

  • Run this test script to verify point-in-time retrieval works correctly:
python -c "
import pandas as pd
from feast import FeatureStore

fs = FeatureStore(repo_path='.')

# Request features for 2 tickers at a specific historical timestamp
entity_df = pd.DataFrame({
    'ticker': ['AAPL', 'MSFT'],
    'event_timestamp': [pd.Timestamp('2025-06-01', tz='UTC'),
                        pd.Timestamp('2025-06-01', tz='UTC')]
})

features = fs.get_historical_features(
    entity_df=entity_df,
    features=[
        'stock_technical_features:sma_20',
        'stock_technical_features:rsi_14',
        'stock_technical_features:macd',
    ],
).to_df()

print('Point-in-time retrieval result:')
print(features.to_string())
print(f'\nRows returned: {len(features)}')
print('No future data leakage - Feast only returned features available at the requested timestamp.')
"

You should see a DataFrame with 2 rows (one per ticker) showing feature values that existed on or before June 1, 2025. No data from after that date appears.

Why does point-in-time correctness matter?

In quantitative finance, using future data during training is called look-ahead bias. It makes models appear to perform well in backtests but fail completely in live trading.

Feast prevents this by design. When you pass a timestamp in your entity DataFrame, it automatically filters out any feature values that did not exist yet at that time. This is the #1 reason production ML teams use feature stores instead of ad-hoc SQL queries.

Getting empty results or NaN values?

If the timestamp you request falls outside your data range, Feast returns null values. Make sure the date in the test (June 2025) falls within your dataset's date range. You can check by opening your Parquet file and looking at the minimum and maximum event_timestamp values.

help me debug my Feast point-in-time retrieval

Your feature store is live. Feast now guarantees that training and serving use the exact same features, computed the same way, with no future data leaking in. Next up, you will train an XGBoost model using these features with walk-forward validation.

Train the Signal Model with Walk-Forward Validation

Your Feast feature store is live, serving point-in-time-correct features from both the online and offline stores. Now it's time to put those features to work by training a model that generates trading signals.

The model itself is intentionally simple. The real engineering challenge is validating it without leaking future data. Standard cross-validation shuffles time-series data, which lets future information bleed into training. Walk-forward validation trains only on past data and tests on the next unseen window, exactly how the model would run in production.

In this step, get ready to:
  • Retrieve training features from the Feast offline store using point-in-time joins.
  • Train an XGBClassifier across 4 sequential walk-forward folds.
  • Save the final model and evaluate whether it shows consistent edge over random.
Set up data retrieval and label creation

Before training a model, you need two things: a clean dataset retrieved from the feature store, and labels that define what "good" and "bad" stocks look like. You'll rank stocks cross-sectionally at each timestamp, labeling the top 20% of performers as BUY and the bottom 20% as SELL.

  • Create a new file called train_model.py in the root of your quant-signal-pipeline folder (the same level as ingest_data.py).
  • Add the imports, constants, and data-loading helper by pasting this code:
import json
import numpy as np
import pandas as pd
from feast import FeatureStore
from xgboost import XGBClassifier
from sklearn.metrics import accuracy_score, classification_report


# Feature references matching the Feast FeatureView definition
FEATURE_REFS = [
    "stock_technical_features:sma_20",
    "stock_technical_features:sma_50",
    "stock_technical_features:rsi_14",
    "stock_technical_features:macd",
    "stock_technical_features:bollinger_width",
    "stock_technical_features:atr_14",
]
# Column names for the feature matrix
FEATURE_COLS = ["sma_20", "sma_50", "rsi_14", "macd", "bollinger_width", "atr_14"]


def create_entity_df(parquet_path):
    # Load the raw feature data and ensure timestamps are UTC
    df = pd.read_parquet(parquet_path)
    df["event_timestamp"] = pd.to_datetime(df["event_timestamp"], utc=True)
    return df

What does this code do?

FEATURE_REFS lists the exact Feast feature references (in view_name:feature_name format) that you'll request from the feature store. This ensures training and serving always pull the same features.

create_entity_df() loads the Parquet file and ensures timestamps are UTC. Feast requires timezone-aware timestamps for point-in-time joins.

  • Next, add the label creation function below create_entity_df:
def create_labels(df):
    # Cross-sectional ranking: label top/bottom quintile per timestamp
    df = df.copy()
    df["label"] = 0
    grouped = df.groupby("event_timestamp")["forward_return_5d"]
    for ts, group in grouped:
        if len(group) < 5:
            continue
        top_quintile = group.quantile(0.8)
        bottom_quintile = group.quantile(0.2)
        # Top performers = BUY signal (1), bottom = SELL signal (-1)
        df.loc[group[group >= top_quintile].index, "label"] = 1
        df.loc[group[group <= bottom_quintile].index, "label"] = -1
    return df

What does this code do?

At each timestamp, this function ranks all 10 stocks by their 5-day forward return. The top 20% get label 1 (BUY), the bottom 20% get label -1 (SELL), and the middle 60% get 0 (HOLD).

This is cross-sectional labeling. Instead of asking "will this stock go up?", you're asking "which stocks will outperform the group?" This is how most quant ranking models frame the problem.

Why cross-sectional ranking instead of absolute prediction?

Predicting absolute returns is extremely noisy. In a market crash, every stock falls. Cross-sectional ranking asks a much easier question: which stocks will do relatively better than others? This signal is more stable across market regimes.

  • Save train_model.py.

Confused about where to add the function?

Each function goes directly below the previous one in the same file. create_labels sits below create_entity_df with a single blank line separating them.

help me structure train_model.py correctly

Build walk-forward validation with XGBoost training

Standard k-fold cross-validation randomly shuffles data points into folds. For time-series, this means future data leaks into training, producing artificially inflated accuracy that won't hold in production.

Walk-forward validation respects time. Each fold trains on everything before a cutoff date and tests on the next unseen window. This is how quant funds validate models before deploying capital.

  • Add the walk-forward validation function below create_labels in train_model.py:
def walk_forward_validation(df, n_folds=4):
    # Sort by time and split into sequential folds
    df = df.sort_values("event_timestamp").reset_index(drop=True)
    dates = df["event_timestamp"].unique()
    dates = np.sort(dates)
    fold_size = len(dates) // (n_folds + 1)

    results = []
    for fold in range(n_folds):
        # Each fold expands the training window forward in time
        train_end_idx = (fold + 1) * fold_size
        test_end_idx = min((fold + 2) * fold_size, len(dates))

        train_dates = dates[:train_end_idx]
        test_dates = dates[train_end_idx:test_end_idx]

        train_df = df[df["event_timestamp"].isin(train_dates)]
        test_df = df[df["event_timestamp"].isin(test_dates)]

        # Only train on stocks with clear BUY/SELL signals (not HOLD)
        train_df = train_df[train_df["label"] != 0]
        test_df = test_df[test_df["label"] != 0]

        if len(train_df) < 50 or len(test_df) < 10:
            continue

        X_train = train_df[FEATURE_COLS].values
        y_train = (train_df["label"] == 1).astype(int).values
        X_test = test_df[FEATURE_COLS].values
        y_test = (test_df["label"] == 1).astype(int).values

        # Train XGBoost classifier for this fold
        model = XGBClassifier(
            n_estimators=100,
            max_depth=4,
            learning_rate=0.05,
            eval_metric="logloss",
            random_state=42,
        )
        model.fit(X_train, y_train, verbose=False)

        preds = model.predict(X_test)
        acc = accuracy_score(y_test, preds)
        hit_rate = np.mean(preds == y_test)

        results.append({
            "fold": fold + 1,
            "train_size": len(train_df),
            "test_size": len(test_df),
            "accuracy": acc,
            "hit_rate": hit_rate,
        })
        print(f"Fold {fold+1}: accuracy={acc:.4f}, train={len(train_df)}, test={len(test_df)}")

    return results, model

What does this code do?

The function splits your ~480 trading days into 5 equal segments. Fold 1 trains on segment 1 and tests on segment 2. Fold 2 trains on segments 1-2 and tests on segment 3. Each fold expands the training window forward, never looking at future data.

The XGBClassifier is trained fresh on each fold with conservative hyperparameters (100 trees, max depth 4, low learning rate). It predicts whether a stock belongs in the top quintile (label=1) or bottom quintile (label=0 in binary terms).

Why walk-forward instead of k-fold?

Imagine training on January-December data, then "testing" on March. The model already saw April through December during training. It's not really predicting the future. Walk-forward makes every test window genuinely unseen, just like production.

  • Now add the main() function that orchestrates the full training pipeline below walk_forward_validation:
def main():
    # Connect to Feast and retrieve features using point-in-time joins
    fs = FeatureStore(repo_path="feature_repo/")

    entity_df = create_entity_df("feature_repo/data/stock_features.parquet")
    entity_df_for_feast = entity_df[["event_timestamp", "ticker"]].copy()

    training_data = fs.get_historical_features(
        entity_df=entity_df_for_feast,
        features=FEATURE_REFS,
    ).to_df()

    # Merge forward returns back in for label creation
    training_data = training_data.merge(
        entity_df[["event_timestamp", "ticker", "forward_return_5d"]],
        on=["event_timestamp", "ticker"],
        how="left",
    )

    training_data = training_data.dropna(subset=FEATURE_COLS + ["forward_return_5d"])
    training_data = create_labels(training_data)

    print(f"Training data shape: {training_data.shape}")
    print(f"Label distribution: {training_data['label'].value_counts().to_dict()}")
    print("\n--- Walk-Forward Validation ---")

    results, final_model = walk_forward_validation(training_data)

    print("\n--- Summary ---")
    avg_acc = np.mean([r["accuracy"] for r in results])
    print(f"Average accuracy across folds: {avg_acc:.4f}")
    print(f"Random baseline: 0.5000")
    print(f"Edge over random: {(avg_acc - 0.5) * 100:.2f} percentage points")

    # Save the final fold's model for serving
    final_model.save_model("model.json")
    print("\nModel saved to model.json")


if __name__ == "__main__":
    main()

What does this code do?

The main() function is the orchestration layer. It connects to Feast, retrieves features using get_historical_features() (which guarantees point-in-time correctness), merges forward returns back in for labeling, then runs walk-forward validation.

After validation, it prints a summary comparing average accuracy against the 50% random baseline. The final model (from the last fold, trained on the most data) is saved as model.json for serving in the next step.

  • Save train_model.py.

Not sure all the pieces are in the right order?

The file should have: imports at the top, then FEATURE_REFS and FEATURE_COLS constants, then four functions in order: create_entity_df, create_labels, walk_forward_validation, main. Use the double-check tab below to compare your full file.

help me fix my train_model.py structure

✔️ Awesome, I've got everything!

Great. Double check you've saved train_model.py before running it in the next substep.

ⓧ I'd like to double check the full code

import json
import numpy as np
import pandas as pd
from feast import FeatureStore
from xgboost import XGBClassifier
from sklearn.metrics import accuracy_score, classification_report


FEATURE_REFS = [
    "stock_technical_features:sma_20",
    "stock_technical_features:sma_50",
    "stock_technical_features:rsi_14",
    "stock_technical_features:macd",
    "stock_technical_features:bollinger_width",
    "stock_technical_features:atr_14",
]
FEATURE_COLS = ["sma_20", "sma_50", "rsi_14", "macd", "bollinger_width", "atr_14"]


def create_entity_df(parquet_path):
    df = pd.read_parquet(parquet_path)
    df["event_timestamp"] = pd.to_datetime(df["event_timestamp"], utc=True)
    return df


def create_labels(df):
    df = df.copy()
    df["label"] = 0
    grouped = df.groupby("event_timestamp")["forward_return_5d"]
    for ts, group in grouped:
        if len(group) < 5:
            continue
        top_quintile = group.quantile(0.8)
        bottom_quintile = group.quantile(0.2)
        df.loc[group[group >= top_quintile].index, "label"] = 1
        df.loc[group[group <= bottom_quintile].index, "label"] = -1
    return df


def walk_forward_validation(df, n_folds=4):
    df = df.sort_values("event_timestamp").reset_index(drop=True)
    dates = df["event_timestamp"].unique()
    dates = np.sort(dates)
    fold_size = len(dates) // (n_folds + 1)

    results = []
    for fold in range(n_folds):
        train_end_idx = (fold + 1) * fold_size
        test_end_idx = min((fold + 2) * fold_size, len(dates))

        train_dates = dates[:train_end_idx]
        test_dates = dates[train_end_idx:test_end_idx]

        train_df = df[df["event_timestamp"].isin(train_dates)]
        test_df = df[df["event_timestamp"].isin(test_dates)]

        train_df = train_df[train_df["label"] != 0]
        test_df = test_df[test_df["label"] != 0]

        if len(train_df) < 50 or len(test_df) < 10:
            continue

        X_train = train_df[FEATURE_COLS].values
        y_train = (train_df["label"] == 1).astype(int).values
        X_test = test_df[FEATURE_COLS].values
        y_test = (test_df["label"] == 1).astype(int).values

        model = XGBClassifier(
            n_estimators=100,
            max_depth=4,
            learning_rate=0.05,
            eval_metric="logloss",
            random_state=42,
        )
        model.fit(X_train, y_train, verbose=False)

        preds = model.predict(X_test)
        acc = accuracy_score(y_test, preds)
        hit_rate = np.mean(preds == y_test)

        results.append({
            "fold": fold + 1,
            "train_size": len(train_df),
            "test_size": len(test_df),
            "accuracy": acc,
            "hit_rate": hit_rate,
        })
        print(f"Fold {fold+1}: accuracy={acc:.4f}, train={len(train_df)}, test={len(test_df)}")

    return results, model


def main():
    fs = FeatureStore(repo_path="feature_repo/")

    entity_df = create_entity_df("feature_repo/data/stock_features.parquet")
    entity_df_for_feast = entity_df[["event_timestamp", "ticker"]].copy()

    training_data = fs.get_historical_features(
        entity_df=entity_df_for_feast,
        features=FEATURE_REFS,
    ).to_df()

    training_data = training_data.merge(
        entity_df[["event_timestamp", "ticker", "forward_return_5d"]],
        on=["event_timestamp", "ticker"],
        how="left",
    )

    training_data = training_data.dropna(subset=FEATURE_COLS + ["forward_return_5d"])
    training_data = create_labels(training_data)

    print(f"Training data shape: {training_data.shape}")
    print(f"Label distribution: {training_data['label'].value_counts().to_dict()}")
    print("\n--- Walk-Forward Validation ---")

    results, final_model = walk_forward_validation(training_data)

    print("\n--- Summary ---")
    avg_acc = np.mean([r["accuracy"] for r in results])
    print(f"Average accuracy across folds: {avg_acc:.4f}")
    print(f"Random baseline: 0.5000")
    print(f"Edge over random: {(avg_acc - 0.5) * 100:.2f} percentage points")

    final_model.save_model("model.json")
    print("\nModel saved to model.json")


if __name__ == "__main__":
    main()
Run the training pipeline and interpret results

Time to see whether the model finds a real signal. A consistent accuracy above 50% across all folds means the technical indicators carry predictive information. Large fold-to-fold variance would indicate regime sensitivity.

  • Run the training script by executing this command in your terminal:
python train_model.py
  • Wait for the script to finish. It will take 10-30 seconds as it retrieves features from Feast and trains 4 separate models.
  • You should see output showing the training data shape, label distribution, per-fold accuracy, and a summary comparing average accuracy against the 0.5000 random baseline.

What should I see?

Your output will show 4 fold results, each with an accuracy value. A healthy model shows consistent accuracy between roughly 52-55% across all folds. This looks small, but even 2-3 percentage points above random (50%) represents a tradeable edge at scale.

The final line confirms model.json was saved to disk. This is the trained model you'll load for serving in the next step.

  • Confirm model.json was created by running:
ls model.json
  • You should see model.json listed in the output.

Seeing an error about missing features or NaN values?

Make sure you ran feast apply in the previous step and that feature_repo/data/stock_features.parquet exists. The training script reads features through Feast, which needs both the registry and the data file.

If you see an XGBoost error on Windows, you may need the Visual C++ Redistributable installed (the same fix from Step 1).

help me debug my train_model.py errors

Extra for Experts: What if accuracy is below 50%?

If one fold dips below 50%, that's a sign the market regime shifted during that window. This is normal. What matters is the average across all folds. If the average is also below 50%, the technical indicators don't carry predictive power for this stock universe over this time period.

Your signal model is trained and validated with production-grade methodology. Next up, you'll build the serving layer that turns these predictions into live trading signals, plus a drift monitor that detects when the model's assumptions start breaking down.

Serve Predictions and Monitor for Drift

In the previous step, you trained an XGBoost signal model and validated it with walk-forward methodology. The model proves it can pick stocks better than random across multiple time folds.

Training a model is only half the battle. In production, market regimes shift constantly. A model trained during a bull run may silently fail during a correction. In this step, you will build the production serving layer that generates live trading signals and the monitoring system that detects when the model's assumptions break down.

In this step, get ready to:
  • Build the signal serving path that loads your trained model and generates BUY/SELL/HOLD signals from Feast online features.
  • Implement KL Divergence drift monitoring that compares training distributions against recent market data.
  • Run the full end-to-end pipeline and verify the signal table and drift report output.
Build the signal serving path

The serving path demonstrates how a production system would use your trained model. It fetches the latest features from Feast's online store (low-latency, pre-computed values) and converts model probabilities into actionable trading signals.

  • Create a new file called serve_and_monitor.py in your quant-signal-pipeline folder (the same level as train_model.py).
  • Add the imports, configuration constants, and the signal generation function by pasting this code:
import numpy as np
import pandas as pd
from scipy.stats import entropy
from feast import FeatureStore
from xgboost import XGBClassifier


# Stock tickers to generate signals for
TICKERS = ["AAPL", "MSFT", "GOOGL", "AMZN", "NVDA",
            "META", "JPM", "JNJ", "V", "PG"]
# Feature columns matching the Feast FeatureView schema
FEATURE_COLS = ["sma_20", "sma_50", "rsi_14", "macd", "bollinger_width", "atr_14"]
# KL divergence above this threshold triggers a drift alert
DRIFT_THRESHOLD = 0.1
# Number of histogram bins for distribution comparison
N_BINS = 20


def generate_signals(model, features_df):
    # Extract feature values and get model probabilities
    X = features_df[FEATURE_COLS].values
    probas = model.predict_proba(X)[:, 1]

    # Convert probabilities to BUY/SELL/HOLD signals
    signals = []
    for i, ticker in enumerate(features_df["ticker"].values):
        prob = probas[i]
        if prob > 0.6:
            signal = "BUY"
        elif prob < 0.4:
            signal = "SELL"
        else:
            signal = "HOLD"
        signals.append({
            "ticker": ticker,
            "signal": signal,
            "confidence": round(abs(prob - 0.5) * 2, 4),
        })
    return pd.DataFrame(signals)

What does this code do?

The imports bring in scipy.stats.entropy for KL Divergence computation, FeatureStore for online feature retrieval, and XGBClassifier for loading the trained model.

The generate_signals() function takes the model's probability output (probability of being a top-quintile stock) and converts it into discrete trading signals. A probability above 0.6 means BUY, below 0.4 means SELL, and anything in between means HOLD. The confidence score scales from 0 to 1 based on how far the probability is from the 0.5 decision boundary.

You will not see a change yet. The next chunk adds the drift monitoring functions that complete the script.

Implement KL Divergence drift monitoring

A model can silently degrade when the market regime changes. KL Divergence measures how much a recent feature distribution has shifted from the training distribution. If any feature drifts beyond a threshold, the system flags an alert so you know to retrain.

  • In serve_and_monitor.py, add the drift monitoring functions below the generate_signals() function:
def compute_kl_divergence(train_values, recent_values, n_bins=N_BINS):
    # Create shared bin edges across both distributions
    min_val = min(train_values.min(), recent_values.min())
    max_val = max(train_values.max(), recent_values.max())
    bins = np.linspace(min_val, max_val, n_bins + 1)

    # Build normalized histograms for each window
    train_hist, _ = np.histogram(train_values, bins=bins, density=True)
    recent_hist, _ = np.histogram(recent_values, bins=bins, density=True)

    # Add epsilon to avoid division by zero in log
    train_hist = train_hist + 1e-10
    recent_hist = recent_hist + 1e-10

    # Normalize to valid probability distributions
    train_hist = train_hist / train_hist.sum()
    recent_hist = recent_hist / recent_hist.sum()

    # Compute KL divergence: D(recent || train)
    kl_div = entropy(recent_hist, train_hist)
    return kl_div


def monitor_drift(full_data, recent_window_days=20):
    # Split data into training window and recent window
    full_data = full_data.sort_values("event_timestamp")
    cutoff = full_data["event_timestamp"].max() - pd.Timedelta(days=recent_window_days)

    train_data = full_data[full_data["event_timestamp"] < cutoff]
    recent_data = full_data[full_data["event_timestamp"] >= cutoff]

    # Compare each feature's distribution between windows
    drift_report = []
    for col in FEATURE_COLS:
        train_vals = train_data[col].dropna()
        recent_vals = recent_data[col].dropna()

        if len(train_vals) < 50 or len(recent_vals) < 10:
            continue

        kl_div = compute_kl_divergence(train_vals, recent_vals)
        alert = kl_div > DRIFT_THRESHOLD

        drift_report.append({
            "feature": col,
            "kl_divergence": round(kl_div, 6),
            "alert": alert,
            "status": "DRIFT DETECTED" if alert else "OK",
        })

    return pd.DataFrame(drift_report)

How does the drift detection work?

The compute_kl_divergence() function discretizes both the training and recent feature values into 20 histogram bins across the same range. It adds a tiny epsilon (1e-10) to avoid log-of-zero errors, normalizes both histograms to proper probability distributions, then calls scipy.stats.entropy(recent_hist, train_hist) which computes D(recent || train) = sum(recent * log(recent / train)).

The monitor_drift() function splits the full dataset into a training window (everything before the last 20 days) and a recent window (the last 20 days). It checks each of the 6 technical indicators independently. If any feature's KL Divergence exceeds the threshold of 0.1, the system flags a drift alert for that feature.

Why KL Divergence over simpler metrics?

You could compare means or standard deviations, but those miss shape changes in the distribution. KL Divergence captures any change in the full probability distribution. A feature whose mean stays the same but becomes bimodal (like RSI during a volatile regime switch) triggers an alert because the distribution shape changed. This is the institutional standard for detecting data drift in production ML systems.

You will not see a change yet. The next chunk adds the main() function that runs the full pipeline.

Run the full pipeline end-to-end

The main() function ties together the serving path and drift monitoring into a single end-to-end production run.

  • In serve_and_monitor.py, add the main() function below monitor_drift():
def main():
    # Initialize Feast and load the trained model
    fs = FeatureStore(repo_path="feature_repo/")

    model = XGBClassifier()
    model.load_model("model.json")

    # Fetch latest features from the online store for all tickers
    entity_rows = [{"ticker": t} for t in TICKERS]
    online_features = fs.get_online_features(
        features=[
            "stock_technical_features:sma_20",
            "stock_technical_features:sma_50",
            "stock_technical_features:rsi_14",
            "stock_technical_features:macd",
            "stock_technical_features:bollinger_width",
            "stock_technical_features:atr_14",
        ],
        entity_rows=entity_rows,
    ).to_df()

    # Generate and print trading signals
    print("=" * 60)
    print("SIGNAL GENERATION")
    print("=" * 60)
    signals = generate_signals(model, online_features)
    print(signals.to_string(index=False))

    # Run drift monitoring on the full historical dataset
    print("\n" + "=" * 60)
    print("DRIFT MONITORING")
    print("=" * 60)
    full_data = pd.read_parquet("feature_repo/data/stock_features.parquet")
    full_data["event_timestamp"] = pd.to_datetime(full_data["event_timestamp"], utc=True)
    drift_report = monitor_drift(full_data)
    print(drift_report.to_string(index=False))

    # Print summary alert status
    alerts = drift_report[drift_report["alert"] == True]
    if len(alerts) > 0:
        print(f"\n[WARNING] {len(alerts)} feature(s) showing drift. Consider model retraining.")
    else:
        print("\n[OK] All features within expected distribution. Model safe to serve.")


if __name__ == "__main__":
    main()

What does the main function do?

  • It loads the trained model from model.json and fetches the latest materialized features from Feast's online store using get_online_features().
  • It passes those features through generate_signals() to produce a BUY/SELL/HOLD table for each ticker.
  • It reads the full Parquet file and runs monitor_drift() to compare training-period distributions against the most recent 20-day window.
  • It prints a final verdict: either all features are stable or specific features have drifted past the 0.1 threshold.
  • Save serve_and_monitor.py.
  • Run the full pipeline by executing this command:
python serve_and_monitor.py
  • You should see two sections of output. The first is a SIGNAL GENERATION table showing each ticker with a BUY, SELL, or HOLD signal and a confidence value between 0 and 1. The second is a DRIFT MONITORING table showing each feature's KL divergence value and whether it triggered an alert.

Seeing a KeyError or empty features?

  • If you see KeyError on a feature column, make sure you ran feast apply and the materialization script from Step 3. The online store must contain current feature values.
  • If signals all show HOLD with 0.0 confidence, the online store may have stale or missing values. Re-run your materialization script from Step 3 to refresh the online store.
  • If the script cannot find model.json, make sure you are running the command from the quant-signal-pipeline directory where the model file was saved in Step 4.

Why isn't my serve_and_monitor.py producing the expected output?

✔️ Awesome, I've got everything!

Great. Double check you have saved serve_and_monitor.py and that your terminal output shows both the signal table and drift report with no errors.

ⓧ I'd like to double check the full code

import numpy as np
import pandas as pd
from scipy.stats import entropy
from feast import FeatureStore
from xgboost import XGBClassifier


TICKERS = ["AAPL", "MSFT", "GOOGL", "AMZN", "NVDA",
            "META", "JPM", "JNJ", "V", "PG"]
FEATURE_COLS = ["sma_20", "sma_50", "rsi_14", "macd", "bollinger_width", "atr_14"]
DRIFT_THRESHOLD = 0.1
N_BINS = 20


def compute_kl_divergence(train_values, recent_values, n_bins=N_BINS):
    min_val = min(train_values.min(), recent_values.min())
    max_val = max(train_values.max(), recent_values.max())
    bins = np.linspace(min_val, max_val, n_bins + 1)

    train_hist, _ = np.histogram(train_values, bins=bins, density=True)
    recent_hist, _ = np.histogram(recent_values, bins=bins, density=True)

    train_hist = train_hist + 1e-10
    recent_hist = recent_hist + 1e-10

    train_hist = train_hist / train_hist.sum()
    recent_hist = recent_hist / recent_hist.sum()

    kl_div = entropy(recent_hist, train_hist)
    return kl_div


def generate_signals(model, features_df):
    X = features_df[FEATURE_COLS].values
    probas = model.predict_proba(X)[:, 1]

    signals = []
    for i, ticker in enumerate(features_df["ticker"].values):
        prob = probas[i]
        if prob > 0.6:
            signal = "BUY"
        elif prob < 0.4:
            signal = "SELL"
        else:
            signal = "HOLD"
        signals.append({
            "ticker": ticker,
            "signal": signal,
            "confidence": round(abs(prob - 0.5) * 2, 4),
        })
    return pd.DataFrame(signals)


def monitor_drift(full_data, recent_window_days=20):
    full_data = full_data.sort_values("event_timestamp")
    cutoff = full_data["event_timestamp"].max() - pd.Timedelta(days=recent_window_days)

    train_data = full_data[full_data["event_timestamp"] < cutoff]
    recent_data = full_data[full_data["event_timestamp"] >= cutoff]

    drift_report = []
    for col in FEATURE_COLS:
        train_vals = train_data[col].dropna()
        recent_vals = recent_data[col].dropna()

        if len(train_vals) < 50 or len(recent_vals) < 10:
            continue

        kl_div = compute_kl_divergence(train_vals, recent_vals)
        alert = kl_div > DRIFT_THRESHOLD

        drift_report.append({
            "feature": col,
            "kl_divergence": round(kl_div, 6),
            "alert": alert,
            "status": "DRIFT DETECTED" if alert else "OK",
        })

    return pd.DataFrame(drift_report)


def main():
    fs = FeatureStore(repo_path="feature_repo/")

    model = XGBClassifier()
    model.load_model("model.json")

    entity_rows = [{"ticker": t} for t in TICKERS]
    online_features = fs.get_online_features(
        features=[
            "stock_technical_features:sma_20",
            "stock_technical_features:sma_50",
            "stock_technical_features:rsi_14",
            "stock_technical_features:macd",
            "stock_technical_features:bollinger_width",
            "stock_technical_features:atr_14",
        ],
        entity_rows=entity_rows,
    ).to_df()

    print("=" * 60)
    print("SIGNAL GENERATION")
    print("=" * 60)
    signals = generate_signals(model, online_features)
    print(signals.to_string(index=False))

    print("\n" + "=" * 60)
    print("DRIFT MONITORING")
    print("=" * 60)
    full_data = pd.read_parquet("feature_repo/data/stock_features.parquet")
    full_data["event_timestamp"] = pd.to_datetime(full_data["event_timestamp"], utc=True)
    drift_report = monitor_drift(full_data)
    print(drift_report.to_string(index=False))

    alerts = drift_report[drift_report["alert"] == True]
    if len(alerts) > 0:
        print(f"\n[WARNING] {len(alerts)} feature(s) showing drift. Consider model retraining.")
    else:
        print("\n[OK] All features within expected distribution. Model safe to serve.")


if __name__ == "__main__":
    main()

Secret mission

Backtesting Engine with Performance Analytics

Your pipeline generates signals and monitors drift. But can it actually make money? Build a backtesting engine that simulates trading the model's signals with realistic transaction costs, then evaluates performance using institutional-grade metrics like Sharpe ratio and max drawdown.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your files on disk for future experimentation or delete them to reclaim space.

Resources you used:

  • Local Parquet files (~5MB of market data) in feature_repo/data/.
  • SQLite database files for Feast online store and registry (~2MB) in feature_repo/data/.
  • Trained model file (model.json, ~100KB).
  • Python virtual environment (venv/ directory) with all installed packages.

All resources live inside your quant-signal-pipeline/ directory. Total disk usage is approximately 10MB.

Keep everything running

No action needed. Choose this if you plan to continue experimenting with different model parameters, adding new tickers, or extending the pipeline.

  • Your files remain on disk at no cost.
  • You can reactivate the virtual environment anytime by navigating to the project folder and running the activate command.
  • Python 3.11.9 remains installed on your system.

Pause - I'll come back to this later

Shut down the active virtual environment and close your editor. Your project files stay on disk so you can pick up where you left off.

  • Deactivate the virtual environment by running:
deactivate
  • Close VS Code.
  • When you return, reactivate the environment from the quant-signal-pipeline/ directory.

Delete - I don't want to use this again

Remove all project files to reclaim ~10MB of disk space. This deletes your scripts, data, trained model, and virtual environment.

  • Deactivate the virtual environment first by running:
deactivate
  • Navigate up one level and delete the project directory.

macOS

cd ~/Desktop
rm -rf quant-signal-pipeline

Windows

cd ~\Desktop
rmdir /s /q quant-signal-pipeline
  • Confirm the folder is gone by running ls (macOS) or dir (Windows) on your Desktop. You should no longer see quant-signal-pipeline in the output.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade ML signal generation pipeline that mirrors what senior AI engineers build at quantitative hedge funds.

You've learned how to:

  • Configured Feast as a feature store ensuring training-serving consistency and point-in-time correctness, the number one source of silent bugs in quant ML.
  • Trained an XGBoost signal model with walk-forward validation, the gold standard for preventing overfitting on time-series data.
  • Implemented KL Divergence drift monitoring that detects when market regime changes invalidate the model's assumptions.
  • Secret Mission: Built a backtesting engine with Sharpe ratio, max drawdown, and transaction cost modeling to validate whether the signal generates real alpha.

Ready to quiz yourself?