skforecast/skforecast-ai/drift-detection

from skforecast.drift_detection import RangeDriftDetector
from skforecast.recursive import ForecasterRecursive
# 1. Train the forecaster
forecaster = ForecasterRecursive(estimator=estimator, lags=24)
forecaster.fit(y=y_train, exog=exog_train)
# 2. Fit the drift detector on training data
detector = RangeDriftDetector()
detector.fit(series=y_train, exog=exog_train)
# 3. Check new data before making predictions
flag_drift, out_of_range_series, out_of_range_exog = detector.predict(
    last_window=new_data,
    exog=new_exog,
    verbose=True,           # Print drift details
    suppress_warnings=False,
)
if flag_drift:
    print("WARNING: New data contains values outside training range!")
    print(f"Out-of-range series features: {out_of_range_series}")
    print(f"Out-of-range exog features: {out_of_range_exog}")

from skforecast.drift_detection import PopulationDriftDetector
# 1. Create detector
detector = PopulationDriftDetector(
    chunk_size=100,                     # Split data into chunks of 100 obs
    threshold=3,                        # Multiplier for std deviation
    threshold_method='std',             # 'std' or 'quantile'
    max_out_of_range_proportion=0.1,    # Max 10% out-of-range allowed
)
# 2. Fit on reference (training) data
detector.fit(X=X_train)
# 3. Detect drift in new data
results, summary = detector.predict(X=X_new)
# results: DataFrame with per-chunk drift statistics
# summary: DataFrame with per-feature drift summary
print(summary)

# Fixed number of observations per chunk
detector = PopulationDriftDetector(chunk_size=100)
# Time-based chunks
detector = PopulationDriftDetector(chunk_size='W')   # Weekly
detector = PopulationDriftDetector(chunk_size='M')   # Monthly
detector = PopulationDriftDetector(chunk_size='D')   # Daily
# No chunking — compare entire datasets
detector = PopulationDriftDetector(chunk_size=None)

# Standard deviation method: flag if statistic > mean + threshold * std
detector = PopulationDriftDetector(
    threshold=3,
    threshold_method='std',
)
# Quantile method: flag if statistic > empirical quantile
detector = PopulationDriftDetector(
    threshold=0.95,             # 95th percentile
    threshold_method='quantile',
)

from skforecast.recursive import ForecasterRecursive
from skforecast.drift_detection import RangeDriftDetector
# Train
forecaster = ForecasterRecursive(estimator=estimator, lags=24)
forecaster.fit(y=y_train, exog=exog_train)
# Set up monitoring
detector = RangeDriftDetector()
detector.fit(series=y_train, exog=exog_train)
# Production loop
def predict_with_monitoring(new_window, new_exog):
    flag, _, _ = detector.predict(
        last_window=new_window, exog=new_exog, verbose=False
    )
    if flag:
        print("Drift detected — consider retraining the model")
    return forecaster.predict(steps=10, exog=new_exog)