How to detect feature drift in Machine Learning: step by step

João Barros 20 de August de 2026 5 min read

Detecting feature drift in Machine Learning is useful to understand when production data changes relative to training data, avoiding performance loss. This tutorial shows, step by step, how to calculate simple drift measures per feature and flag relevant changes with Python.

Prerequisites

Python 3.8+ installed
Pandas and scipy installed (pip install pandas scipy)
Training and production datasets as CSV files
Basic knowledge of Python and descriptive statistics

Step 1: Why detect feature drift?

Feature drift happens when the distribution of a column changes over time. Identifying drift helps decide whether to retrain, adjust preprocessing, or raise alerts. We will use statistical tests and practical measures for each feature.

Step 2: Prepare data and load libraries

Load the CSVs for the training set and the production set and select the columns you want to monitor. We keep only numeric columns in this example; for categoricals we use frequency comparisons.

import pandas as pd
from scipy.stats import ks_2samp, chi2_contingency

train = pd.read_csv('train.csv')
prod = pd.read_csv('prod.csv')

# Lista de colunas a monitorizar (numéricas e categóricas)
num_cols = ['age','salary']
cat_cols = ['region','product_type']

Step 3: Compute drift for numeric features (Kolmogorov-Smirnov)

The Kolmogorov-Smirnov test compares two samples to see if they come from the same distribution. It does not assume normality. We use the p-value and an effect measure (D) to flag drift.

def detect_numeric_drift(train_ser, prod_ser, alpha=0.05):
    # Remove NaNs
    a = train_ser.dropna()
    b = prod_ser.dropna()
    if len(a) < 20 or len(b) < 20:
        return {'p_value': None, 'D': None, 'drift': 'insufficient_data'}
    stat, p = ks_2samp(a, b)
    drift = 'drift' if p < alpha else 'no_drift'
    return {'p_value': float(p), 'D': float(stat), 'drift': drift}

results_num = {}
for col in num_cols:
    results_num[col] = detect_numeric_drift(train[col], prod[col])

print(results_num)

Step 4: Compute drift for categorical features (chi-squared test)

For categoricals with counts, we compare frequency tables. If there are new or removed categories, this indicates qualitative drift.

def detect_categorical_drift(train_ser, prod_ser, alpha=0.05):
    a = train_ser.fillna('NULL')
    b = prod_ser.fillna('NULL')
    # Frequências alinhadas por categoria
    freq = pd.concat([a.value_counts(), b.value_counts()], axis=1, sort=False).fillna(0)
    freq.columns = ['train', 'prod']
    chi2, p, _, _ = chi2_contingency(freq.values)
    different_categories = set(b.unique()) - set(a.unique())
    drift = 'drift' if p < alpha or len(different_categories) > 0 else 'no_drift'
    return {'p_value': float(p), 'different_categories': list(different_categories), 'drift': drift}

results_cat = {}
for col in cat_cols:
    results_cat[col] = detect_categorical_drift(train[col], prod[col])

print(results_cat)

Step 5: Complementary measures and practical thresholds

Besides statistical tests, it is useful to compute differences in mean, median and proportions to prioritize features. Define practical thresholds (e.g.: mean difference > 10% or D > 0.1).

def numeric_statistics(train_ser, prod_ser):
    t = train_ser.dropna()
    p = prod_ser.dropna()
    stats = {
        'train_mean': float(t.mean()),
        'prod_mean': float(p.mean()),
        'mean_rel_diff': abs(t.mean() - p.mean()) / (abs(t.mean()) + 1e-9)
    }
    return stats

summary = {}
for col in num_cols:
    summary[col] = {**results_num[col], **numeric_statistics(train[col], prod[col])}

print(summary)

Step 6: Automate and generate report

Combine everything into a function that iterates over columns, applies tests and writes an alert CSV with columns flagged by priority.

def drift_report(train_df, prod_df, num_cols, cat_cols):
    rows = []
    for col in num_cols:
        r = detect_numeric_drift(train_df[col], prod_df[col])
        stats = numeric_statistics(train_df[col], prod_df[col])
        priority = 'high' if r['drift']=='drift' and stats['mean_rel_diff'] > 0.1 else 'low'
        rows.append({'feature': col, 'type': 'numeric', **r, **stats, 'priority': priority})
    for col in cat_cols:
        r = detect_categorical_drift(train_df[col], prod_df[col])
        priority = 'high' if r['drift']=='drift' else 'low'
        rows.append({'feature': col, 'type': 'categorical', **r, 'priority': priority})
    return pd.DataFrame(rows)

report = drift_report(train, prod, num_cols, cat_cols)
report.to_csv('drift_report.csv', index=False)
print(report)

Check the result

Open drift_report.csv: it should list each feature with p_value, drift indicator and priority. Inspect columns marked as high: examine histograms or boxplots to visually confirm. If several features are high, consider retraining, recalibrating thresholds, or adjusting ETL.

Conclusion

This simple method combines statistical tests (KS and chi-squared) with practical measures to detect feature drift in Machine Learning. Next steps: integrate this into a production monitoring pipeline, add tests by time window and use explainability techniques to understand causes. Tip: start monitoring gradually and adjust thresholds according to the cost of false positives/negatives in your scenario.