Detect Concept Drift in Python with a Simple Statistical Test
Detect concept drift by comparing the mean of recent data against a reference distribution using a z-score-like threshold.
Python code
27 linesimport random
import statistics
def detect_drift(recent, reference, threshold=1.5):
ref_mean = statistics.mean(reference)
ref_std = statistics.stdev(reference)
recent_mean = statistics.mean(recent)
drift_score = abs(recent_mean - ref_mean) / (ref_std if ref_std > 0 else 1)
drifted = drift_score > threshold
return drifted, drift_score
if __name__ == "__main__":
random.seed(42)
reference_data = [random.gauss(50, 5) for _ in range(100)]
normal_data = [random.gauss(50, 5) for _ in range(20)]
drift_data = [random.gauss(70, 5) for _ in range(20)]
print("Normal data drift:")
drifted, score = detect_drift(normal_data, reference_data)
print(f" Score: {score:.3f}, Drift detected: {drifted}")
print("Drifted data detection:")
drifted, score = detect_drift(drift_data, reference_data)
print(f" Score: {score:.3f}, Drift detected: {drifted}")
Output
Normal data drift:
Score: 0.327, Drift detected: False
Drifted data detection:
Score: 3.880, Drift detected: True
How it works
This function computes a drift score as the absolute difference between the recent and reference means divided by the reference standard deviation. A score above the threshold indicates that the recent data is statistically far from the reference baseline, suggesting concept drift. The threshold (default 1.5) is a typical z-score cutoff; you can adjust it based on sensitivity needs. Using statistics.mean and statistics.stdev keeps the implementation dependency-free and works with any iterable of numbers.
Common mistakes
- Using population stdev (`pstdev`) instead of sample stdev, which under/overestimates drift for small samples
- Not handling zero standard deviation, which causes division by zero
- Applying the test when the reference sample is too small, leading to unstable estimates
Variations
- Use a sliding window approach where reference data updates over time
- Use scipy.stats.ztest for a formal statistical test with p-values
Real-world use cases
- Monitoring ML model performance in production by comparing prediction distributions over time.
- Triggering retraining pipelines when input feature distributions shift beyond a threshold.
- Alerting on drift in user behavior data for recommendation or fraud detection systems.
Sponsored
More from ML engineering pipelines
- Bayesian Optimization in Python: A Simplified Mock Implementation medium
- Build a Data Helper Class in Python for ML Pipelines easy
- Build a Mock Random Forest Classifier in Python easy
- Champion Challenger Deployment Mock in Python easy
- Compare Model A vs Model B Metrics in Python easy
- Create a Minimal Great Expectations Suite Mock in Python easy
Keep learning
Related tutorials and quizzes for this topic.