Feature Store in MLOps: Feature Server, Registry & Pipeline

Calculating read time…

Imagine you work at a huge restaurant chain with 500 branches. Every branch needs fresh ingredients every single day — tomatoes, cheese, dough, spices.

Now imagine each branch orders from a different supplier, stores ingredients differently, and uses different measurements. Branch A's "one cup of flour" is Branch B's "half a cup." Chaos. Inconsistency. Wasted effort. Bad food. 🍕💥

Now imagine a Central Ingredient Warehouse: one source, perfectly prepared, consistent measurements, available fresh to every branch on demand. Every branch gets the exact same quality ingredient, every time.

💡 That Central Ingredient Warehouse = A Feature Store!

In Machine Learning, features are the "ingredients" your model needs to make predictions — things like a user's age, purchase history, location, last login time, or account balance.

A Feature Store is the central, organized warehouse for all these ingredients. It ensures every ML model — whether training or serving — gets the same, fresh, consistent features. 🎯

📚 What We'll Cover

🔹 What Are Features? Why Do They Matter So Much?
🔹 The 5 Big Problems Without a Feature Store
🔹 What is a Feature Store?
🔹 The 4 Core Components of a Feature Store
🔹 Feature Pipeline — Where Features Are Born
🔹 Feature Registry — The Master Catalog
🔹 Online Store (Feature Server) — Real-Time Serving
🔹 Offline Store — Historical Training Data
🔹 The Training-Serving Skew Problem (And How Feature Stores Fix It)
🔹 Point-in-Time Correct Joins — The Most Critical Concept
🔹 Popular Feature Store Platforms
🔹 Feature Monitoring — Keeping Features Healthy
🔹 Feature Store Best Practices & Anti-Patterns
🔹 High-Level Summary & Next Steps

🧪 Section 1: What Are Features? Why Do They Matter So Much?

Before we understand a Feature Store, let's be crystal clear about what a feature actually is.

🎯 Features = Inputs to Your ML Model

A feature is any piece of information that helps your model make a better prediction. They're the signals your model uses to "understand" the world.

  Example: Fraud Detection Model

  Raw Data: Transaction record

  Features extracted from that data:

  ├── transaction_amount          → $450.00

  ├── merchant_country            → "NG"

  ├── user_avg_spend_last_30_days → $42.50

  ├── user_txn_count_last_1_hour  → 8

  ├── is_new_device               → True

  ├── time_since_last_login_mins  → 2

  └── distance_from_home_km       → 4200

  Model output: fraud_score = 0.94  →  🚨 Block transaction!

Notice that features are computed from raw data. user_avg_spend_last_30_days doesn't exist in one row — it requires aggregating the user's last 30 days of transactions. This computation is called feature engineering, and it's often the most time-consuming part of ML.

💡 Industry Fact:
Data scientists spend 60–80% of their time on feature engineering — collecting, cleaning, computing, and managing features. A Feature Store automates and centralizes this entire process, giving that time back to focus on actual model innovation. ⏱️

🔥 Section 2: The 5 Big Problems Without a Feature Store

Most companies start ML without a Feature Store and quickly run into the same painful problems. Let's understand each one.

Problem 1: Duplicate Feature Engineering Work 🔁

Team A (Fraud) computes user_txn_count_last_7_days. Team B (Recommendations) also computes user_activity_last_week. They're computing the same thing — just calling it different names! Massive waste of engineering effort across the organization.

Problem 2: Training-Serving Skew 💀

This is the #1 silent killer of ML systems. During training, you compute features one way (in Python/Pandas). During serving (inference), you compute the same features a different way (in Java/SQL). Even tiny differences cause the model to see different data in production than it was trained on. Predictions degrade. Nobody knows why.

Problem 3: Data Leakage (Cheating in Training) 🃏

Imagine training a model to predict if a user will churn next week. You accidentally include a feature that captures data from after the prediction date. Your model achieves 99% accuracy in training and 52% in production. That's data leakage — the model "cheated" using future information it won't have at prediction time.

Problem 4: No Reusability 🗑️

Without a central store, features live in individual notebooks, pipelines, and data scientist laptops. When that person leaves, the knowledge of how to compute customer_lifetime_value_score leaves with them. The next team spends months recomputing it from scratch.

Problem 5: Slow Time-to-Production ⏳

Every new ML model has to rebuild its feature pipeline from zero. Feature discovery, engineering, testing, and deployment adds weeks or months to every project cycle.

❌ Without a Feature Store, you get:

❌ The same features computed 10 different ways by 10 different teams
❌ Models that work in training but fail in production
❌ Data leakage that inflates training metrics and crashes production
❌ Feature knowledge locked in individual silos
❌ 6-month delays to deploy a model that needs 3 new features

🏪 Section 3: What is a Feature Store?

A Feature Store is a centralized data system specifically designed to manage the full lifecycle of ML features — from creation and storage to retrieval and monitoring.

Think of it as a smart, versioned warehouse for ML features that serves two masters simultaneously:

ML Training: Needs large amounts of historical features with exact timestamps — to train models correctly.
ML Serving (Inference): Needs the latest feature values for a specific entity (user, product) in milliseconds — to make real-time predictions.

  THE FEATURE STORE PROMISE:

  "Write your feature logic ONCE.

   Use it for BOTH training AND serving.

   Get identical results in both places.

   Share it with the ENTIRE organization.

   Retrieve it INSTANTLY at inference time."

✅ A Feature Store gives you:

✅ One definition of every feature — no more duplicates
✅ Identical features in training and serving — no more skew
✅ Time-travel queries — see what a feature's value was at any past moment
✅ Feature discovery — browse and reuse features across all teams
✅ Sub-millisecond feature serving — power real-time ML products
✅ Feature monitoring — detect when features drift in production

🏗️ Section 4: The 4 Core Components of a Feature Store

A modern Feature Store has four distinct, cooperating components. Understanding each one is the key to mastering Feature Stores.

  ┌──────────────────────────────────────────────────────────────────┐

  │                    FEATURE STORE                                 │

  │                                                                  │

  │  ┌─────────────────┐    ┌─────────────────────────────────────┐  │

  │  │  1. FEATURE     │    │  2. FEATURE REGISTRY                │  │

  │  │  PIPELINE       │───▶│  (Master Catalog & Metadata)        │  │

  │  │  (ETL + Compute)│    │  "What features exist & how?"       │  │

  │  └────────┬────────┘    └─────────────────────────────────────┘  │

  │           │                                                       │

  │           ├──────────────────────┐                                │

  │           ▼                      ▼                                │

  │  ┌─────────────────┐    ┌─────────────────────────────────────┐  │

  │  │  3. OFFLINE     │    │  4. ONLINE STORE                    │  │

  │  │  STORE          │    │  (Feature Server)                   │  │

  │  │  (Historical)   │    │  "Latest values, <10ms serving"     │  │

  │  │  S3, BigQuery,  │    │  Redis, DynamoDB, Cassandra         │  │

  │  │  Hive, Parquet  │    │                                     │  │

  │  └────────┬────────┘    └───────────────┬─────────────────────┘  │

  │           │                             │                         │

  └───────────┼─────────────────────────────┼─────────────────────────┘

              │                             │

              ▼                             ▼

        ML TRAINING                  ML SERVING

        (batch, historical)          (real-time inference)

⚙️ Section 5: Feature Pipeline — Where Features Are Born

The Feature Pipeline is the engine room of the Feature Store. It's a data processing system that continuously transforms raw data into clean, computed feature values and loads them into the store.

🏭 The Factory Analogy

Think of raw data (server logs, database tables, event streams) as raw materials — iron ore, crude oil, wheat. The Feature Pipeline is the factory that refines these raw materials into finished products (features) ready to use. The Feature Store is the warehouse where finished products are stored.

Two Types of Feature Pipelines

1️⃣ Batch Pipeline (Offline Processing)

Runs on a schedule (hourly, daily, weekly) to process large amounts of historical data and compute aggregate features.

  Schedule: Every day at 2 AM

  Raw data: 90 days of transaction records (50GB)

  Computes:

  ├── user_total_spend_last_30_days     (GROUP BY user, SUM)

  ├── user_avg_transaction_last_7_days  (GROUP BY user, AVG)

  ├── user_favorite_merchant_category   (GROUP BY user, MODE)

  └── user_days_since_first_transaction (DATEDIFF)

  Output: 5M rows of feature values → written to Offline Store + Online Store

2️⃣ Streaming Pipeline (Real-Time Processing)

Processes data as it arrives — in real time — using tools like Apache Kafka, Apache Flink, or Apache Spark Streaming. Generates features with minimal latency (seconds to milliseconds).

  Trigger: New transaction arrives → Kafka topic → Flink job

  Computes (in real time, <1 second):

  ├── user_txn_count_last_1_hour  (sliding window aggregate)

  ├── user_txn_amount_last_5_mins (micro-batch rolling sum)

  └── velocity_flag               (sudden spending spike detection)

  Output → immediately written to Online Store (Redis) → ready for fraud model

💡 Most production Feature Stores use BOTH:

Batch pipeline for heavy historical aggregations (run nightly).
Streaming pipeline for time-sensitive real-time features (run continuously).

The two pipelines complement each other — batch for depth, streaming for freshness. Together they power both training and real-time serving. 🔄

📌 Code Purpose — Batch Feature Pipeline (PySpark)

What this code does: Reads 90 days of raw transaction data from S3, computes 7 user-level aggregate features per user (total spend, transaction count, average spend, max spend, unique merchants, 7-day count, spend velocity), and writes the results as a timestamped Parquet file back to the offline store.

Why it matters: This is the "factory" that produces clean, consistent features for every user. Without this pipeline, each model team would have to recompute these aggregates independently — wasting effort and risking inconsistency.

Key concept introduced: event_timestamp is stamped on every row so Point-in-Time joins work later.

"""
feature_pipeline.py
Batch feature pipeline that computes user-level features
and writes them to a Feature Store.
"""

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from datetime import datetime, timedelta


def create_spark_session():
    return SparkSession.builder \
        .appName("FeaturePipeline") \
        .getOrCreate()


def compute_user_transaction_features(transactions_df, reference_date):
    """
    Computes user-level aggregate features from raw transaction data.

    Features computed:
    - user_total_spend_30d      : total spend in last 30 days
    - user_txn_count_30d        : transaction count in last 30 days
    - user_avg_txn_amount_30d   : average transaction amount
    - user_max_txn_amount_30d   : maximum single transaction
    - user_unique_merchants_30d : number of unique merchants visited
    - user_txn_count_7d         : transaction count in last 7 days
    - user_spend_velocity       : ratio of 7-day to 30-day spend
    """

    ref_date = reference_date
    date_30d = ref_date - timedelta(days=30)
    date_7d  = ref_date - timedelta(days=7)

    # Filter to relevant time window
    df_30d = transactions_df.filter(
        (F.col("transaction_ts") >= date_30d) &
        (F.col("transaction_ts") <  ref_date)
    )

    df_7d = transactions_df.filter(
        (F.col("transaction_ts") >= date_7d) &
        (F.col("transaction_ts") <  ref_date)
    )

    # Compute 30-day aggregates
    features_30d = df_30d.groupBy("user_id").agg(
        F.sum("amount").alias("user_total_spend_30d"),
        F.count("*").alias("user_txn_count_30d"),
        F.avg("amount").alias("user_avg_txn_amount_30d"),
        F.max("amount").alias("user_max_txn_amount_30d"),
        F.countDistinct("merchant_id").alias("user_unique_merchants_30d"),
    )

    # Compute 7-day aggregates
    features_7d = df_7d.groupBy("user_id").agg(
        F.sum("amount").alias("user_spend_7d"),
        F.count("*").alias("user_txn_count_7d"),
    )

    # Join 30d and 7d features
    features = features_30d.join(features_7d, on="user_id", how="left")

    # Compute derived feature: spend velocity
    features = features.withColumn(
        "user_spend_velocity",
        F.when(
            F.col("user_total_spend_30d") > 0,
            F.col("user_spend_7d") / (F.col("user_total_spend_30d") / 4.0)
        ).otherwise(0.0)
    )

    # Add event_timestamp — critical for point-in-time joins!
    features = features.withColumn(
        "event_timestamp",
        F.lit(ref_date.strftime("%Y-%m-%d %H:%M:%S")).cast("timestamp")
    )

    # Round numerical columns to 4 decimal places
    numerical_cols = [
        "user_total_spend_30d",
        "user_avg_txn_amount_30d",
        "user_max_txn_amount_30d",
        "user_spend_7d",
        "user_spend_velocity"
    ]

    for col in numerical_cols:
        features = features.withColumn(col, F.round(F.col(col), 4))

    return features


def run_batch_pipeline(spark, transactions_path, output_path, reference_date):
    """
    Main pipeline: read raw data → compute features → save to parquet.
    """

    print(f"[{datetime.now()}] Starting batch feature pipeline...")
    print(f"  Reference date: {reference_date}")

    # Step 1: Load raw transactions
    print(f"  Loading transactions from: {transactions_path}")
    transactions_df = spark.read.parquet(transactions_path)

    raw_count = transactions_df.count()
    print(f"  Raw transactions loaded: {raw_count:,}")

    # Step 2: Compute features
    print("  Computing user features...")
    features_df = compute_user_transaction_features(
        transactions_df, reference_date
    )

    feature_count = features_df.count()
    print(f"  Features computed for {feature_count:,} users")

    # Step 3: Write features to offline store (parquet on S3/GCS)
    output_partition = f"{output_path}/date={reference_date.strftime('%Y-%m-%d')}"
    features_df.write.mode("overwrite").parquet(output_partition)
    print(f"  Features written to: {output_partition}")

    # Show sample output
    print("\n  Sample feature values:")
    features_df.show(5, truncate=False)

    print(f"[{datetime.now()}] Pipeline complete! ✅")
    return features_df


# Entry point
if __name__ == "__main__":
    spark = create_spark_session()

    run_batch_pipeline(
        spark             = spark,
        transactions_path = "s3://my-data-lake/transactions/",
        output_path       = "s3://my-feature-store/user_transaction_features/",
        reference_date    = datetime(2025, 3, 10)
    )

📋 Section 6: Feature Registry — The Master Catalog

The Feature Registry is the brain and memory of the Feature Store. It's a centralized catalog that stores metadata about every feature — not the actual data values, but information about those features.

📚 The Library Card Catalog Analogy

Before digital search existed, libraries had a physical card catalog — a giant cabinet of index cards, each describing a book: title, author, location, subject, date added.

You couldn't read the book from the catalog card. But the card told you everything you needed to find and understand any book in the entire library.

The Feature Registry is exactly this — but for ML features instead of books. 📇

🗂️ What Does a Feature Registry Store?

Feature Name & Description: user_total_spend_30d — "Total USD amount spent by user in last 30 days"
Data Type: Float64, Int32, Boolean, String, etc.
Feature Group / Entity: Which entity does this feature belong to? (user, product, merchant)
Source: What pipeline produces this feature? What raw table does it come from?
Transformation Logic: The exact SQL/code used to compute this feature — so it can be reproduced.
Owner: Which team owns this feature? Who to contact?
Tags: fraud, recommendation, user, real-time, daily
Version History: When was this feature changed? What was version 1 vs version 2?
Statistics: Mean, min, max, null rate, distribution — auto-computed from the data.
Model Usage: Which models currently use this feature? (critical for impact analysis)

✅ Why the Registry is the Most Underrated Component:

Without a Registry, features are invisible to other teams. A data scientist can't discover that the feature they need has already been built — so they rebuild it.

With a Registry, anyone can search "user spending last 30 days" and find 3 existing features that might already fit their need. Zero rework. Instant reuse. 🔍

📌 Code Purpose — Feature Registry Definition (Feast)

What this code does: Defines the official "source of truth" for all features in the system using Feast's Python SDK. It creates two Entities (user and product), two FeatureViews (transaction features and behavior features — each with full metadata, owners, tags, TTL, and descriptions), and one FeatureService that bundles the exact features needed by the Fraud Detection v3 model.

Why it matters: Once this is registered (feast apply), every team in the organization can discover these features, see how they're computed, and reuse them in their own models — without duplicating any engineering work.

Key concept introduced: A FeatureService acts as a named contract between a specific ML model and the features it needs. The model requests the service by name — not individual features.

"""
feature_registry.py
Define features in the Feature Registry using Feast.
This is the 'source of truth' for what features exist and how they're defined.
"""

from datetime import timedelta
from feast import (
    Entity,
    FeatureView,
    Feature,
    FileSource,
    ValueType,
    FeatureService,
)
from feast.types import Float64, Int64, Bool, String


# ─── STEP 1: Define Entities ──────────────────────────────────────
# An Entity is the primary key your features are associated with.
# Most features are about a specific user, product, merchant, etc.

user_entity = Entity(
    name        = "user_id",
    description = "Unique user identifier across all platforms",
    value_type  = ValueType.INT64,
    tags        = {"team": "platform", "domain": "identity"},
)

product_entity = Entity(
    name        = "product_id",
    description = "Unique product identifier in the catalog",
    value_type  = ValueType.STRING,
    tags        = {"team": "catalog", "domain": "product"},
)


# ─── STEP 2: Define Data Sources ──────────────────────────────────
# Where the feature data physically lives (offline store location)

user_transaction_source = FileSource(
    name              = "user_transaction_features_source",
    path              = "s3://my-feature-store/user_transaction_features/",
    timestamp_field   = "event_timestamp",  # Critical for point-in-time joins
    created_timestamp_column = "created_ts",
    description       = "Batch-computed user transaction aggregate features. "
                        "Updated daily at 03:00 UTC.",
)

user_behavior_source = FileSource(
    name            = "user_behavior_features_source",
    path            = "s3://my-feature-store/user_behavior_features/",
    timestamp_field = "event_timestamp",
)


# ─── STEP 3: Define Feature Views ─────────────────────────────────
# A FeatureView is a NAMED GROUP of related features for a given entity.
# This is what gets registered in the Feature Registry.

user_transaction_feature_view = FeatureView(
    name        = "user_transaction_features",
    description = "Aggregated transaction behavior features for each user. "
                  "Computed from raw transactions table. "
                  "Used by: fraud_model_v3, credit_risk_model_v1",
    entities    = [user_entity],
    ttl         = timedelta(days=2),    # Features expire after 2 days

    schema = [
        Feature(
            name        = "user_total_spend_30d",
            dtype       = Float64,
            description = "Total USD amount spent by user in last 30 days",
            tags        = {"category": "spend", "window": "30d"},
        ),
        Feature(
            name        = "user_txn_count_30d",
            dtype       = Int64,
            description = "Number of transactions by user in last 30 days",
            tags        = {"category": "activity", "window": "30d"},
        ),
        Feature(
            name        = "user_avg_txn_amount_30d",
            dtype       = Float64,
            description = "Average transaction amount (USD) over last 30 days",
            tags        = {"category": "spend", "window": "30d"},
        ),
        Feature(
            name        = "user_max_txn_amount_30d",
            dtype       = Float64,
            description = "Largest single transaction amount in last 30 days",
            tags        = {"category": "spend", "window": "30d"},
        ),
        Feature(
            name        = "user_unique_merchants_30d",
            dtype       = Int64,
            description = "Number of unique merchants visited in last 30 days",
            tags        = {"category": "diversity", "window": "30d"},
        ),
        Feature(
            name        = "user_txn_count_7d",
            dtype       = Int64,
            description = "Transaction count in last 7 days (velocity signal)",
            tags        = {"category": "activity", "window": "7d"},
        ),
        Feature(
            name        = "user_spend_velocity",
            dtype       = Float64,
            description = "Ratio of 7d/30d spend rate. "
                          ">1.5 indicates unusual spending increase. "
                          "Key fraud signal.",
            tags        = {"category": "velocity", "risk": "true"},
        ),
    ],

    source = user_transaction_source,
    tags   = {
        "team":       "data-engineering",
        "owner":      "alice@company.com",
        "domain":     "fraud",
        "status":     "production",
        "sla":        "daily-3am",
    },
)


user_behavior_feature_view = FeatureView(
    name        = "user_behavior_features",
    description = "User app/web behavior signals for personalization and churn",
    entities    = [user_entity],
    ttl         = timedelta(hours=48),

    schema = [
        Feature(
            name        = "days_since_last_login",
            dtype       = Int64,
            description = "Number of days since user last logged in",
        ),
        Feature(
            name        = "app_session_count_7d",
            dtype       = Int64,
            description = "Number of app sessions in last 7 days",
        ),
        Feature(
            name        = "is_premium_subscriber",
            dtype       = Bool,
            description = "Whether the user has an active premium subscription",
        ),
        Feature(
            name        = "preferred_category",
            dtype       = String,
            description = "User's most-browsed product category in last 30 days",
        ),
    ],

    source = user_behavior_source,
    tags   = {"team": "product", "owner": "bob@company.com"},
)


# ─── STEP 4: Define Feature Services ──────────────────────────────
# A Feature Service bundles features for a specific ML model.
# This is what a model asks for — it doesn't request individual features.

fraud_detection_feature_service = FeatureService(
    name        = "fraud_detection_v3_features",
    description = "Feature set for Fraud Detection Model v3. "
                  "Requires both transaction and behavior features.",
    features    = [
        # Grab specific features from each view
        user_transaction_feature_view[[
            "user_total_spend_30d",
            "user_txn_count_30d",
            "user_max_txn_amount_30d",
            "user_txn_count_7d",
            "user_spend_velocity",
        ]],
        user_behavior_feature_view[[
            "days_since_last_login",
            "is_premium_subscriber",
        ]],
    ],
    tags = {
        "model":     "fraud_detection_v3",
        "owner":     "ml-team@company.com",
        "deployed":  "2025-01-15",
    },
)

print("✅ Feature Registry definitions created!")
print(f"   Entities:         user_entity, product_entity")
print(f"   Feature Views:    user_transaction_features, user_behavior_features")
print(f"   Feature Services: fraud_detection_v3_features")

⚡ Section 7: Online Store (Feature Server) — Real-Time Serving

The Online Store — also called the Feature Server — is the component that serves feature values at inference time with ultra-low latency.

🏎️ The Fast Food Counter Analogy

Think of a fast-food restaurant. The kitchen (offline store) preps large batches of ingredients and stores them in warming trays. When you place an order (inference request), the counter staff (Feature Server) grabs your items from the warming tray instantly — they don't cook them from scratch on the spot!

The Feature Server does exactly this for ML models. Features are pre-computed by batch/streaming pipelines and stored in a fast key-value store (Redis, DynamoDB, Cassandra). At inference time, the server fetches them in microseconds. 🚀

🔑 Key Requirements of the Online Store

Ultra-low latency: <10ms (often <1ms) per lookup
High throughput: Handle 100,000+ requests per second
Latest values only: No history needed — just the current feature value
Point-key lookup: "Give me all features for user_id=12345"
High availability: 99.99% uptime — models can't wait

🗄️ What Powers the Online Store

Store	Latency	Best For	Used By
Redis	<1ms	Highest speed, moderate size	Uber, Lyft, DoorDash
DynamoDB	<5ms	AWS-native, massive scale	Amazon, Robinhood
Cassandra	<5ms	Massive write throughput	Netflix, Apple
Bigtable	<10ms	GCP-native, petabyte scale	Google, Spotify
SQLite (dev)	<5ms local	Local development & testing	Feast default (local mode)

📌 Code Purpose — Feature Server: Materialization & Real-Time Inference

What this code does: Demonstrates the complete online store workflow in 5 steps: (1) initialize the Feature Store from config, (2) apply the registry definitions, (3) materialize (copy) features from the offline store (S3) into the online store (Redis), (4) fetch feature values in real-time for a given user, (5) run a full end-to-end fraud inference pipeline that combines stored features with live transaction data to make a BLOCK/ALLOW decision in under 3ms.

Why it matters: This shows how a fraud model running in production gets its features instantly — without recomputing them on every request. The features were pre-computed and are simply looked up by user_id, making the entire feature retrieval step take just 2–3ms regardless of complexity.

Key concept introduced: materialize_incremental() is the bridge that keeps the online store fresh. Run this on a schedule (e.g., every hour via Airflow).

"""
feature_server.py
Complete workflow for materializing features to the online store
and serving them at inference time.
"""

import subprocess
from feast import FeatureStore
from datetime import datetime
import pandas as pd


# ─── STEP 1: Initialize the Feature Store ─────────────────────────
# feature_store.yaml specifies online/offline store backends
store = FeatureStore(repo_path="./feature_repo")


# ─── STEP 2: Apply Registry (register all Feature Views) ──────────
# This pushes your feature definitions to the central registry
# Run once when definitions change:
# $ feast apply

print("Applying feature definitions to registry...")
# In code: store.apply([user_transaction_feature_view, ...])
# In CLI:  feast apply


# ─── STEP 3: Materialize Features to Online Store ─────────────────
# This copies the LATEST feature values from offline → online store.
# Run this on a schedule (e.g., every hour via Airflow/cron).

def materialize_to_online_store(store: FeatureStore):
    """
    Pushes latest feature values from offline store to online store.
    Only copies data within the feature view's TTL window.
    """
    print("\n📦 Materializing features to online store...")

    # This reads from S3/BigQuery (offline) and writes to Redis (online)
    store.materialize_incremental(
        end_date = datetime.utcnow()
    )

    print("✅ Materialization complete! Online store is up-to-date.")


# ─── STEP 4: Serve Features at Inference Time ─────────────────────

def get_features_for_inference(store: FeatureStore, user_ids: list) -> pd.DataFrame:
    """
    Fetches latest feature values from the ONLINE store for real-time inference.
    This runs in production — must be fast (sub-10ms)!

    Returns a DataFrame with one row per user and all requested features.
    """

    entity_rows = [{"user_id": uid} for uid in user_ids]

    feature_vector = store.get_online_features(
        feature_refs = [
            "user_transaction_features:user_total_spend_30d",
            "user_transaction_features:user_txn_count_30d",
            "user_transaction_features:user_max_txn_amount_30d",
            "user_transaction_features:user_txn_count_7d",
            "user_transaction_features:user_spend_velocity",
            "user_behavior_features:days_since_last_login",
            "user_behavior_features:is_premium_subscriber",
        ],
        entity_rows = entity_rows,
    ).to_df()

    return feature_vector


# ─── STEP 5: Full Inference Pipeline ──────────────────────────────

def run_fraud_inference(transaction_event: dict) -> dict:
    """
    End-to-end inference pipeline:
    1. Receive transaction event
    2. Fetch features from online store
    3. Run fraud model
    4. Return fraud score + decision
    """
    import numpy as np

    user_id    = transaction_event["user_id"]
    txn_amount = transaction_event["amount"]
    merchant   = transaction_event["merchant_country"]

    t0 = datetime.utcnow()
    features_df = get_features_for_inference(store, [user_id])
    fetch_ms = (datetime.utcnow() - t0).total_seconds() * 1000
    print(f"  Feature fetch time: {fetch_ms:.2f}ms")

    features = features_df.iloc[0].to_dict()
    features["txn_amount"]           = txn_amount
    features["merchant_country"]     = merchant
    features["is_high_risk_country"] = merchant in ["NG", "RU", "VN"]

    fraud_score = _mock_fraud_/model(features)

    return {
        "user_id":       user_id,
        "fraud_score":   round(fraud_score, 4),
        "decision":      "BLOCK" if fraud_score > 0.85 else "ALLOW",
        "features_used": list(features.keys()),
        "latency_ms":    fetch_ms,
    }


def _mock_fraud_model(features: dict) -> float:
    """Simple rule-based mock model for demonstration."""
    score = 0.1

    if features.get("user_spend_velocity", 0) > 2.0:
        score += 0.35

    if features.get("user_txn_count_7d", 0) > 20:
        score += 0.20

    if features.get("txn_amount", 0) > features.get("user_max_txn_amount_30d", 0) * 2:
        score += 0.25

    if features.get("is_high_risk_country", False):
        score += 0.15

    if features.get("days_since_last_login", 0) < 1:
        score += 0.05

    return min(score, 1.0)


# ─── DEMO ──────────────────────────────────────────────────────────
print("\n" + "="*55)
print("  🔍 FRAUD DETECTION — REAL-TIME INFERENCE DEMO")
print("="*55)

transaction = {
    "transaction_id":   "TXN-20250310-99182",
    "user_id":          12345,
    "amount":           890.00,
    "merchant_country": "NG",
    "timestamp":        datetime.utcnow().isoformat(),
}

print(f"\nIncoming transaction: {transaction}")
print("\nFetching features from online store (Redis)...")

result = run_fraud_inference(transaction)

print(f"\n📊 Fraud Decision:")
print(f"  Score:    {result['fraud_score']}")
print(f"  Decision: {result['decision']}")
print(f"  Latency:  {result['latency_ms']:.2f}ms total")

Sample Output:

========================================================
  🔍 FRAUD DETECTION — REAL-TIME INFERENCE DEMO
========================================================

Incoming transaction: {'transaction_id': 'TXN-20250310-99182',
                        'user_id': 12345, 'amount': 890.0, ...}

Fetching features from online store (Redis)...
  Feature fetch time: 2.31ms

📊 Fraud Decision:
  Score:    0.9100
  Decision: BLOCK
  Latency:  2.31ms total

🗄️ Section 8: Offline Store — Historical Training Data

The Offline Store is the historical archive of all feature values — every value that was ever computed, with timestamps. It's designed for training ML models, not serving them.

📜 The Archives Analogy

Think of the Online Store as today's newspaper on your doorstep — current, instant, always fresh. The Offline Store is the newspaper archive going back 5 years — every edition, perfectly organized by date, so you can reconstruct exactly what the world looked like on any given day in the past. 📰

🔧 What Powers the Offline Store

Amazon S3 + Parquet — most common setup
Google BigQuery — for SQL-based feature queries
Snowflake — enterprise data warehouse
Apache Hive / Delta Lake — open-source options
Databricks / Apache Iceberg — time-travel capable tables

📌 Code Purpose — Training Data Retrieval from Offline Store

What this code does: Retrieves historical feature values from the offline store using Feast's get_historical_features() function, which automatically performs Point-in-Time correct joins. It then trains a GradientBoosting fraud classifier on the resulting dataset and prints feature importances — showing which features are most predictive.

Why it matters: This is the complete model training workflow. The PIT join ensures the training data has no data leakage — each labeled example only sees feature values that existed before that moment. This is the difference between a trustworthy model and a "cheating" one.

Key concept introduced: get_historical_features(entity_df, features) is the offline store's key API. The entity_df must always include event_timestamp so the PIT join knows which historical snapshot to use.

"""
training_data_retrieval.py
Retrieve historical feature values for training an ML model.
This uses Point-in-Time Correct joins — the most critical concept!
"""

import pandas as pd
from feast import FeatureStore
from datetime import datetime


store = FeatureStore(repo_path="./feature_repo")


def create_training_dataset(store: FeatureStore) -> pd.DataFrame:
    """
    Creates a training dataset using historical feature values.
    Uses Point-in-Time (PIT) correct joins to prevent data leakage.
    """

    entity_df = pd.DataFrame({
        "user_id": [101, 102, 103, 101, 104, 102, 105],
        "event_timestamp": [
            datetime(2025, 1, 10, 14, 22, 0),
            datetime(2025, 1, 10, 15, 45, 0),
            datetime(2025, 1, 11,  9, 10, 0),
            datetime(2025, 1, 12, 18, 30, 0),
            datetime(2025, 1, 13, 11, 00, 0),
            datetime(2025, 1, 14,  8, 15, 0),
            datetime(2025, 1, 15, 20, 45, 0),
        ],
        "label_is_fraud": [1, 0, 0, 1, 0, 0, 1],
    })

    print(f"Training events: {len(entity_df)} rows")

    # PIT join: each row uses features computed BEFORE its event_timestamp
    training_df = store.get_historical_features(
        entity_df = entity_df,
        features  = [
            "user_transaction_features:user_total_spend_30d",
            "user_transaction_features:user_txn_count_30d",
            "user_transaction_features:user_max_txn_amount_30d",
            "user_transaction_features:user_txn_count_7d",
            "user_transaction_features:user_spend_velocity",
            "user_behavior_features:days_since_last_login",
            "user_behavior_features:is_premium_subscriber",
        ],
    ).to_df()

    print(f"Training dataset shape: {training_df.shape}")
    return training_df


def train_fraud_model(training_df: pd.DataFrame):
    """
    Train a GradientBoosting classifier on the retrieved feature dataset.
    """
    from sklearn.ensemble import GradientBoostingClassifier
    from sklearn.model_selection import train_test_split
    from sklearn.metrics import roc_auc_score

    feature_cols = [
        "user_total_spend_30d",
        "user_txn_count_30d",
        "user_max_txn_amount_30d",
        "user_txn_count_7d",
        "user_spend_velocity",
        "days_since_last_login",
        "is_premium_subscriber",
    ]

    X = training_df[feature_cols].fillna(0)
    y = training_df["label_is_fraud"]

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )

    model = GradientBoostingClassifier(
        n_estimators  = 200,
        learning_rate = 0.05,
        max_depth     = 4,
        random_state  = 42,
    )
    model.fit(X_train, y_train)

    y_pred_proba = model.predict_proba(X_test)[:, 1]
    auc          = roc_auc_score(y_test, y_pred_proba)

    print(f"\n🎯 Model Training Complete! ROC-AUC: {auc:.4f}")
    print(f"\n   Feature Importances:")
    for feat, imp in sorted(
        zip(feature_cols, model.feature_importances_),
        key=lambda x: -x[1]
    ):
        bar = "█" * int(imp * 50)
        print(f"   {feat:40s} {bar} {imp:.4f}")

    return model


# Run the full training workflow
training_data = create_training_dataset(store)
model         = train_fraud_model(training_data)

⚠️ Section 9: Training-Serving Skew — And How Feature Stores Fix It

Training-Serving Skew is when a model experiences different feature values during training vs. during production inference. It's the #1 cause of "why does my model work in training but fail in production?"

🔪 A Real-World Skew Example

A data scientist builds a churn prediction model. During training, they compute days_since_last_purchase in Python:

⚠️ Code Purpose — Training-Side Feature Computation (WRONG WAY)

What this code shows: A data scientist computing days_since_last_purchase using Python's pd.Timestamp.today() during model training. This is the training-side implementation.

Why this is dangerous: today() returns whatever today's date was when training ran — which could be months in the past by now. When the production system computes the same feature differently (next snippet), the values won't match. Model performance silently degrades.

# Training code (Python/Pandas)
df["days_since_last_purchase"] = (
    pd.Timestamp.today() - df["last_purchase_date"]
).dt.days
# Uses today() — but "today" during training was 3 months ago!

The backend engineer deploys the model and computes the same feature in SQL:

⚠️ Code Purpose — Serving-Side Feature Computation (MISMATCHED)

What this code shows: The backend engineer computing the "same" feature in SQL using CURRENT_DATE. Looks correct on the surface — but hides a timezone mismatch with the Python version.

The hidden bug: Python used UTC; SQL used EST (UTC-5). For users who transacted between 8 PM and midnight, the feature days_since_last_purchase is 1 day off in production vs training. Model performance drops 8%. Nobody can find the bug for weeks. This is why Feature Stores exist.

-- Serving code (SQL in production)
SELECT
    DATEDIFF('day', last_purchase_date, CURRENT_DATE) AS days_since_last_purchase
FROM users
-- Uses CURRENT_DATE — correct, but different timezone handling!

Python used UTC. SQL used local server time (EST, UTC-5). Feature values differ by 5 hours. For users who purchased late at night, days_since_last_purchase is 1 day off. Model performance drops by 8%. Nobody can find the bug for weeks.

✅ How Feature Stores Eliminate Skew:

The feature transformation logic is defined ONCE in the Feature Registry. The exact same code is used by both:

• The offline pipeline (for training data)
• The online server (for inference)

One definition. No translation. No discrepancy. Training and serving always see identical features. 🎯

⏰ Section 10: Point-in-Time Correct Joins — The Most Critical Concept

This is arguably the most important concept in Feature Stores. If you don't understand this, your training data will have data leakage, and your model will be unreliable in production.

🕰️ The Time Machine Problem

Imagine you're training a model to predict if a user will churn within the next 30 days, as of January 15th, 2025.

For this training example, you need the user's features as they appeared on January 15th. But your feature table has been updated many times since then — January 20th, February 1st, etc.

If you naively join the feature table, you might accidentally use the February 1st feature values to predict what happened in January. Your model "knew the future" during training — that's data leakage!

  WRONG JOIN (data leakage!) ❌:

  Label event: user=101, date=Jan 15, label=churned

  Feature join: user=101, using LATEST features (Feb 5)

  Problem: On Feb 5, user_spend_30d=0 (they already churned!)

  Model learns: "0 spend = churn" — it's cheating with future data.

  ─────────────────────────────────────────────────────────────

  CORRECT POINT-IN-TIME JOIN (no leakage!) ✅:

  Label event: user=101, date=Jan 15, label=churned

  Feature join: user=101, using features computed BEFORE Jan 15

  Use the most recent feature snapshot before Jan 15

  Correct: user_spend_30d=450.00 (they were still active on Jan 15)

  Model learns actual signals that preceded the churn. ✅

📌 Code Purpose — Point-in-Time Join: Wrong vs Correct

What this code does: Builds a working side-by-side comparison of two join strategies on the same dataset. The naive join blindly uses the latest feature value for each user. The PIT-correct join finds the most recent feature snapshot that was computed strictly before each label event's timestamp.

Why it matters: For user_id=101 predicting churn on Jan 12, the naive join uses the Jan 22 value (spend=0 — they already churned by then!), giving the model future information it couldn't have had in reality. The PIT join correctly uses the Jan 8 snapshot (spend=620). This one difference can swing model accuracy from 95% (cheating) to 72% (realistic).

Key concept introduced: The inner pit_join() function is essentially what store.get_historical_features() does internally. Understanding this manual implementation makes the Feast API clicks into place.

"""
point_in_time_join.py
Demonstrates what a Point-in-Time correct join looks like
and why it's essential for avoiding data leakage.
"""

import pandas as pd
from datetime import datetime


def demonstrate_pit_join():
    """
    Shows the difference between a naive join and a PIT-correct join.
    """

    # Feature snapshots stored over time (as computed by batch pipeline)
    feature_history = pd.DataFrame({
        "user_id": [101, 101, 101, 101, 102, 102, 102],
        "feature_timestamp": [
            datetime(2025, 1,  1),
            datetime(2025, 1,  8),
            datetime(2025, 1, 15),
            datetime(2025, 1, 22),   # Jan 22 — user 101 already churned!
            datetime(2025, 1,  1),
            datetime(2025, 1,  8),
            datetime(2025, 1, 15),
        ],
        "user_total_spend_30d": [
            450.0, 620.0, 380.0, 0.0,   # user 101: drops to 0 after churn
            80.0,  110.0,  95.0,         # user 102: stable
        ],
        "user_txn_count_7d": [
            8, 12, 6, 0,
            3,  4, 3,
        ],
    })

    # Training events (labels)
    training_events = pd.DataFrame({
        "user_id": [101, 102, 101],
        "event_timestamp": [
            datetime(2025, 1, 12),   # user 101 prediction point: Jan 12
            datetime(2025, 1, 10),   # user 102 prediction point: Jan 10
            datetime(2025, 1, 20),   # user 101 again at Jan 20
        ],
        "label_churned_30d": [1, 0, 1],
    })

    print("📊 Feature History:")
    print(feature_history.to_string(index=False))
    print("\n📋 Training Events:")
    print(training_events.to_string(index=False))

    # ── NAIVE JOIN (WRONG) ────────────────────────────────────────
    latest_features = feature_history.sort_values("feature_timestamp") \
        .groupby("user_id").last().reset_index()

    naive_joined = training_events.merge(
        latest_features[["user_id", "user_total_spend_30d", "user_txn_count_7d"]],
        on="user_id",
        how="left"
    )

    print("\n\n❌ NAIVE JOIN (uses LATEST features — data leakage!):")
    print(naive_joined.to_string(index=False))
    print("⚠️  user_id=101 at Jan 12 gets spend=0.0 (from Jan 22!) — WRONG!")

    # ── POINT-IN-TIME CORRECT JOIN ────────────────────────────────
    def pit_join(events_df, features_df):
        """
        For each event row, find the most recent feature snapshot
        that occurred BEFORE or AT the event_timestamp.
        """
        result_rows = []

        for _, event_row in events_df.iterrows():
            uid        = event_row["user_id"]
            event_time = event_row["event_timestamp"]

            user_features = features_df[
                (features_df["user_id"] == uid) &
                (features_df["feature_timestamp"] <= event_time)
            ]

            if user_features.empty:
                row = event_row.to_dict()
                row.update({"user_total_spend_30d": None, "user_txn_count_7d": None})
            else:
                latest_valid = user_features.sort_values("feature_timestamp").iloc[-1]
                row = event_row.to_dict()
                row.update({
                    "user_total_spend_30d": latest_valid["user_total_spend_30d"],
                    "user_txn_count_7d":    latest_valid["user_txn_count_7d"],
                    "feature_used_date":    latest_valid["feature_timestamp"],
                })

            result_rows.append(row)

        return pd.DataFrame(result_rows)

    pit_result = pit_join(training_events, feature_history)

    print("\n\n✅ POINT-IN-TIME CORRECT JOIN (no data leakage):")
    print(pit_result.to_string(index=False))
    print("\n✅ user_id=101 at Jan 12 correctly uses Jan 8 snapshot (spend=620.0)")
    print("✅ user_id=101 at Jan 20 correctly uses Jan 15 snapshot (spend=380.0)")
    print("   (Not the Jan 22 value of 0.0 — that's in the future!)")


demonstrate_pit_join()

Output:

✅ POINT-IN-TIME CORRECT JOIN:
   user_id  event_timestamp  label  user_total_spend_30d  feature_used_date
      101   2025-01-12          1           620.0          2025-01-08
      102   2025-01-10          0            80.0          2025-01-08
      101   2025-01-20          1           380.0          2025-01-15

✅ user_id=101 at Jan 12 correctly uses Jan 8 snapshot (spend=620.0)
✅ user_id=101 at Jan 20 correctly uses Jan 15 snapshot (spend=380.0)

🛠️ Section 11: Popular Feature Store Platforms

Platform	Type	Best For	Online Store	Offline Store
Feast 🍽️	Open Source	Learning, small-medium teams	Redis, DynamoDB, SQLite	S3, BigQuery, Snowflake
Tecton	Managed SaaS	Enterprise, streaming features	DynamoDB, Redis	S3, Databricks
Hopsworks	Open Source + Managed	Full MLOps platform + features	RonDB (MySQL NDB)	Hudi on S3/HDFS
Vertex AI Feature Store	Managed (GCP)	GCP-native ML pipelines	Bigtable	BigQuery
SageMaker Feature Store	Managed (AWS)	AWS-native ML pipelines	DynamoDB	S3
Databricks Feature Store	Managed	Spark/Databricks users	DynamoDB, CosmosDB	Delta Lake
Fennel	Managed SaaS	Real-time streaming features	Built-in	Built-in

✅ Recommendation for 2025 Beginners:

Start with Feast (open source, free, great documentation).
It runs locally with SQLite/files — no cloud needed to learn.
When you grow: move to Hopsworks (community edition is free) or Vertex AI Feature Store if you're on GCP. 🎓

📡 Section 12: Feature Monitoring — Keeping Features Healthy

A Feature Store isn't just about storage and serving. A critical operational layer is feature monitoring — continuously watching your features for problems.

🩺 What to Monitor

Freshness: Has the batch pipeline run on schedule? Are feature values stale? If user_total_spend_30d hasn't updated in 36 hours, alert!
Distribution Drift: Is the distribution of feature values changing over time?
Null Rate: Are more features becoming NULL than expected? Sudden null increase = upstream data pipeline broke.
Range Violations: Is user_age suddenly showing values of -5 or 200? Data quality issues surface here first.
Online/Offline Skew: Are online store values matching offline store values? Materialization bugs show up as skew between the two stores.

📌 Code Purpose — Feature Health Monitoring System

What this code does: Builds a reusable FeatureHealthMonitor class that runs three types of automated checks on live production feature data: (1) null rate spike detection — alerts when missing values exceed 2× baseline, (2) distribution drift detection — alerts when the mean shifts more than 2 standard deviations, (3) range violation detection — alerts when values fall outside historical min/max bounds. A structured FeatureAlert dataclass captures each issue with a severity level (INFO / WARNING / CRITICAL) and a human-readable message.

Why it matters: Silent feature degradation is more dangerous than visible errors. A broken upstream pipeline may quietly produce NULL values for days — your model keeps running, but its predictions are based on zeroed-out features. This monitoring system catches that within minutes, not days.

Demo scenario injected: The demo deliberately injects a 25% null rate (simulating a pipeline failure) on user_txn_count_30d and shifts the mean of user_total_spend_30d from 420 to 850 (simulating user behavior drift) — both are correctly caught as CRITICAL alerts.

"""
feature_monitor.py
Simple feature health monitoring system.
"""

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Dict, List


@dataclass
class FeatureAlert:
    feature_name: str
    alert_type:   str
    message:      str
    severity:     str   # "INFO", "WARNING", "CRITICAL"
    timestamp:    datetime = None

    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = datetime.utcnow()


class FeatureHealthMonitor:
    """
    Monitors feature quality across three dimensions:
    - Null rates (missing value spikes)
    - Distribution drift (statistical shift from baseline)
    - Range violations (values outside expected bounds)
    """

    def __init__(self, baseline_stats: Dict):
        """
        baseline_stats: {feature_name: {mean, std, min, max, null_rate}}
        Computed from your training dataset — the "healthy" reference.
        """
        self.baseline = baseline_stats
        self.alerts   = []

    def check_null_rate(self, feature_name: str, current_df: pd.Series) -> None:
        """Alert if null rate exceeds 2x baseline."""
        baseline_null = self.baseline.get(feature_name, {}).get("null_rate", 0)
        current_null  = current_df.isnull().mean()

        if current_null > baseline_null * 2 and current_null > 0.05:
            self.alerts.append(FeatureAlert(
                feature_name = feature_name,
                alert_type   = "NULL_RATE_SPIKE",
                message      = (f"Null rate jumped from {baseline_null:.1%} "
                                f"to {current_null:.1%}. "
                                f"Possible upstream data pipeline failure."),
                severity     = "CRITICAL" if current_null > 0.20 else "WARNING",
            ))

    def check_distribution_drift(
        self,
        feature_name: str,
        current_df: pd.Series
    ) -> None:
        """Alert if mean shifts by more than 2 standard deviations from baseline."""
        baseline      = self.baseline.get(feature_name, {})
        baseline_mean = baseline.get("mean", 0)
        baseline_std  = baseline.get("std", 1)
        current_mean  = current_df.dropna().mean()

        if baseline_std > 0:
            z_score = abs(current_mean - baseline_mean) / baseline_std
            if z_score > 2.0:
                self.alerts.append(FeatureAlert(
                    feature_name = feature_name,
                    alert_type   = "DISTRIBUTION_DRIFT",
                    message      = (f"Mean drifted from {baseline_mean:.2f} "
                                    f"to {current_mean:.2f} "
                                    f"(z-score={z_score:.2f}). "
                                    f"Possible data drift."),
                    severity     = "CRITICAL" if z_score > 3.0 else "WARNING",
                ))

    def check_range_violations(
        self,
        feature_name: str,
        current_df: pd.Series
    ) -> None:
        """Alert if values fall outside expected min/max range."""
        baseline     = self.baseline.get(feature_name, {})
        expected_min = baseline.get("min", None)

        if expected_min is not None:
            buffer      = abs(expected_min) * 0.1
            below_count = (current_df < expected_min - buffer).sum()
            if below_count > 0:
                self.alerts.append(FeatureAlert(
                    feature_name = feature_name,
                    alert_type   = "RANGE_VIOLATION_LOW",
                    message      = (f"{below_count} values below min "
                                    f"({expected_min:.2f}). "
                                    f"Possible data quality issue."),
                    severity     = "WARNING",
                ))

    def run_all_checks(
        self,
        feature_data: Dict[str, pd.Series]
    ) -> List[FeatureAlert]:
        """Run all health checks on a dict of feature series."""
        self.alerts = []

        for feature_name, series in feature_data.items():
            self.check_null_rate(feature_name, series)
            self.check_distribution_drift(feature_name, series)
            self.check_range_violations(feature_name, series)

        return self.alerts

    def print_report(self) -> None:
        """Print a formatted monitoring report."""
        print("\n" + "="*60)
        print("  📊 FEATURE HEALTH MONITORING REPORT")
        print(f"  Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}")
        print("="*60)

        if not self.alerts:
            print("\n  ✅ All features healthy! No issues detected.")
            return

        for severity in ["CRITICAL", "WARNING", "INFO"]:
            level_alerts = [a for a in self.alerts if a.severity == severity]
            if not level_alerts:
                continue

            icon = "🚨" if severity == "CRITICAL" else "⚠️"
            print(f"\n  {icon} {severity} ({len(level_alerts)} alerts):")
            for alert in level_alerts:
                print(f"    [{alert.feature_name}] {alert.alert_type}")
                print(f"      → {alert.message}")

        print("="*60)


# ─── DEMO ──────────────────────────────────────────────────────────

# Baseline statistics from training data
baseline = {
    "user_total_spend_30d": {
        "mean": 420.0, "std": 180.0, "min": 0.0, "max": 5000.0, "null_rate": 0.02
    },
    "user_txn_count_30d": {
        "mean": 8.5,   "std": 4.0,  "min": 0.0, "max": 120.0,  "null_rate": 0.01
    },
    "user_spend_velocity": {
        "mean": 1.0,   "std": 0.3,  "min": 0.0, "max": 5.0,    "null_rate": 0.03
    },
}

# Simulated current production data (with injected issues!)
np.random.seed(42)
n = 1000

current_data = {
    "user_total_spend_30d": pd.Series(
        # Mean shifted from 420 to 850 — drift detected!
        np.random.normal(850, 180, n)
    ),
    "user_txn_count_30d": pd.Series(
        # Injected 25% null rate (pipeline failure!)
        [None if np.random.random() < 0.25 else v
         for v in np.random.normal(8.5, 4, n)]
    ),
    "user_spend_velocity": pd.Series(
        # Normal distribution — should be fine
        np.random.normal(1.05, 0.3, n).clip(0, 5)
    ),
}

monitor = FeatureHealthMonitor(baseline_stats=baseline)
alerts  = monitor.run_all_checks(current_data)
monitor.print_report()