Skip to content

drift_detection¶

skforecast.drift_detection._range_drift.RangeDriftDetector ¶

RangeDriftDetector()

Detector of out-of-range values based on training feature ranges.

The detector is intentionally lightweight: it does not compute advanced drift statistics since it is used to check single observations during inference. Suitable for real-time applications.

Attributes:

Name Type Description
series_names_in_ list

Names of the series used during training.

series_values_range_ dict

Range of values of the target series used during training.

exog_names_in_ list

Names of the exogenous variables used during training.

exog_values_range_ dict

Range of values of the exogenous variables used during training.

series_specific_exog_ bool

Indicates whether exogenous variables have different values across target series during training (i.e., exogenous is series-specific rather than global).

is_fitted bool

Whether the detector has been fitted to the training data.

skforecast_version str

Version of skforecast library used to create the detector.

Methods:

Name Description
fit

Fit detector, storing training ranges.

predict

Check if there is any value outside the training range for last_window and exog.

Source code in skforecast/drift_detection/_range_drift.py
57
58
59
60
61
62
63
64
65
def __init__(self) -> None:

    self.series_names_in_      = None
    self.series_values_range_  = None
    self.exog_names_in_        = None
    self.exog_values_range_    = None
    self.series_specific_exog_ = False
    self.is_fitted             = False
    self.skforecast_version    = __version__

Attributes¶

series_names_in_ instance-attribute ¶

series_names_in_ = None

series_values_range_ instance-attribute ¶

series_values_range_ = None

exog_names_in_ instance-attribute ¶

exog_names_in_ = None

exog_values_range_ instance-attribute ¶

exog_values_range_ = None

series_specific_exog_ instance-attribute ¶

series_specific_exog_ = False

is_fitted instance-attribute ¶

is_fitted = False

skforecast_version instance-attribute ¶

skforecast_version = __version__

Methods:¶

fit ¶

fit(series=None, exog=None, **kwargs)

Fit detector, storing training ranges.

Parameters:

Name Type Description Default
series pandas Series, pandas DataFrame, dict, aliases: `y`

Input time series data to fit the detector, ideally the same ones used to fit the forecaster.

None
exog pandas Series, pandas DataFrame, dict

Exogenous variables to include in the forecaster.

None

Returns:

Type Description
None
Source code in skforecast/drift_detection/_range_drift.py
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
def fit(
    self,
    series: pd.DataFrame | pd.Series | dict[str, pd.Series | pd.DataFrame] | None = None,
    exog: pd.DataFrame | pd.Series | dict[str, pd.Series | pd.DataFrame] | None = None,
    **kwargs
) -> None:
    """
    Fit detector, storing training ranges.

    Parameters
    ----------
    series : pandas Series, pandas DataFrame, dict, aliases: `y`
        Input time series data to fit the detector, ideally the same ones
        used to fit the forecaster.
    exog : pandas Series, pandas DataFrame, dict, default None
        Exogenous variables to include in the forecaster.

    Returns
    -------
    None

    """

    if series is None and ('y' not in kwargs or kwargs['y'] is None):
        raise ValueError(
            "One of `series` or `y` must be provided."
        )
    if 'y' in kwargs:
        if series is not None:
            raise ValueError(
                "Cannot specify both `series` and `y`. Please provide only one of them."
            )
        series = kwargs.pop('y')

    if not isinstance(series, (pd.Series, pd.DataFrame, dict)):
        raise TypeError("Input must be a pandas Series, DataFrame or dict.")

    if not isinstance(exog, (pd.Series, pd.DataFrame, dict, type(None))):
        raise TypeError(
            "Exogenous variables must be a pandas Series, DataFrame or dict."
        )

    self.series_names_in_      = []
    self.series_values_range_  = {}
    self.exog_names_in_        = None
    self.exog_values_range_    = None
    self.series_specific_exog_ = False
    self.is_fitted             = False

    series = self._normalize_input(series, name="series")
    for key, value in series.items():
        self.series_names_in_.append(key)
        self.series_values_range_[key] = self._get_features_range(X=value)

    if exog is not None:

        exog = self._normalize_input(exog, name="exog")

        self.exog_names_in_ = []
        self.exog_values_range_ = {}
        for key, value in exog.items():
            if isinstance(value, pd.Series):
                self.exog_names_in_.append(key)
            else:
                self.exog_names_in_.extend(value.columns)
            self.exog_values_range_[key] = self._get_features_range(X=value)

        self.exog_names_in_ = list(dict.fromkeys(self.exog_names_in_))
        series_names_set = set(self.series_names_in_)
        self.series_specific_exog_ = any(key in series_names_set for key in exog.keys())

    self.is_fitted = True

predict ¶

predict(
    last_window=None,
    exog=None,
    verbose=True,
    suppress_warnings=False,
)

Check if there is any value outside the training range for last_window and exog.

Parameters:

Name Type Description Default
last_window pandas Series, pandas DataFrame, dict

Series values used to create the predictors (lags) needed in the first iteration of the prediction (t + 1).

None
exog pandas Series, pandas DataFrame, dict

Exogenous variable/s included as predictor/s.

None
verbose bool

Whether to print a summary of the check.

True
suppress_warnings bool

Whether to suppress warnings.

False

Returns:

Name Type Description
flag_out_of_range bool

True if there is any value outside the training range, False otherwise.

out_of_range_series list

List of series names that are out of range.

out_of_range_exog (list, dict)

Exogenous variables that are out of range.

  • If self.series_specific_exog_ is False: returns a list with the names of exogenous variables that are out of range (global exogenous).
  • If self.series_specific_exog_ is True: returns a dictionary where keys are series names and values are lists of out-of-range exogenous variables for each series.
Source code in skforecast/drift_detection/_range_drift.py
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
@manage_warnings
def predict(
    self,
    last_window: pd.Series | pd.DataFrame | dict[str, pd.Series | pd.DataFrame] | None = None,
    exog: pd.Series | pd.DataFrame | dict[str, pd.Series | pd.DataFrame] | None = None,
    verbose: bool = True,
    suppress_warnings: bool = False
) -> tuple[bool, list[str], list[str] | dict[str, list[str]]]:
    """
    Check if there is any value outside the training range for last_window and exog.

    Parameters
    ----------
    last_window : pandas Series, pandas DataFrame, dict, default None
        Series values used to create the predictors (lags) needed in the
        first iteration of the prediction (t + 1).
    exog : pandas Series, pandas DataFrame, dict, default None
        Exogenous variable/s included as predictor/s.
    verbose : bool, default True
        Whether to print a summary of the check.
    suppress_warnings : bool, default False
        Whether to suppress warnings.

    Returns
    -------
    flag_out_of_range : bool
        True if there is any value outside the training range, False otherwise.
    out_of_range_series : list
        List of series names that are out of range.
    out_of_range_exog : list, dict
        Exogenous variables that are out of range.

        - If `self.series_specific_exog_` is False: returns a list with the names
        of exogenous variables that are out of range (global exogenous).
        - If `self.series_specific_exog_` is True: returns a dictionary where
        keys are series names and values are lists of out-of-range exogenous
        variables for each series.

    """

    if not self.is_fitted:
        raise RuntimeError("Model is not fitted yet.")

    if not isinstance(last_window, (pd.Series, pd.DataFrame, dict, type(None))):
        raise TypeError(
            "`last_window` must be a pandas Series, DataFrame, dict or None."
        )

    if not isinstance(exog, (pd.Series, pd.DataFrame, dict, type(None))):
        raise TypeError(
            "`exog` must be a pandas Series, DataFrame, dict or None."
        )

    flag_out_of_range = False

    out_of_range_series = []
    out_of_range_series_ranges = []
    if last_window is not None:
        last_window = self._normalize_input(last_window, name="last_window")
        series_names_set = set(self.series_names_in_)
        for key, value in last_window.items():
            if isinstance(value, pd.Series):
                value = value.to_frame()
            for col in value.columns:
                if key not in series_names_set:
                    warnings.warn(
                        f"'{key}' was not seen during training. Its range is unknown.",
                        UnknownLevelWarning
                    )
                    continue
                is_out_of_range = self._check_feature_range(
                    feature_range=self.series_values_range_[col], X=value[col]
                )
                if is_out_of_range:
                    flag_out_of_range = True
                    out_of_range_series.append(col)
                    out_of_range_series_ranges.append(self.series_values_range_[col])
                    self._display_warnings(
                        not_compliant_feature = col,
                        feature_range         = self.series_values_range_[col],
                        series_name           = None
                    )

    out_of_range_exog = {} if self.series_specific_exog_ else []
    out_of_range_exog_ranges = {} if self.series_specific_exog_ else []
    if exog is not None:
        series_ids = list(last_window.keys()) if last_window is not None else self.series_names_in_
        exog = self._normalize_input(exog, name="exog", series_ids=series_ids)
        exog_names_set = set(self.exog_names_in_) if self.exog_names_in_ else set()
        for key, value in exog.items():

            if isinstance(value, pd.Series):
                value = value.to_frame()
            features_ranges = self.exog_values_range_.get(key, None)

            if self.series_specific_exog_:
                out_of_range_exog[key] = []
                out_of_range_exog_ranges[key] = []

            for col in value.columns:

                if not isinstance(features_ranges, dict):
                    features_ranges = {key: features_ranges}

                if col not in exog_names_set:
                    warnings.warn(
                        f"'{col}' was not seen during training. Its range is unknown.",
                        MissingExogWarning,
                    )
                    continue

                is_out_of_range = self._check_feature_range(
                    feature_range=features_ranges[col], X=value[col]
                )

                if is_out_of_range:

                    flag_out_of_range = True
                    if self.series_specific_exog_:
                        out_of_range_exog[key].append(col)
                        out_of_range_exog_ranges[key].append(features_ranges[col])
                    else:
                        out_of_range_exog.append(col)
                        out_of_range_exog_ranges.append(features_ranges[col])

                    self._display_warnings(
                        not_compliant_feature = col,
                        feature_range         = features_ranges[col],
                        series_name           = key if self.series_specific_exog_ else None,
                    )

            if self.series_specific_exog_ and not out_of_range_exog[key]:
                out_of_range_exog.pop(key)
                out_of_range_exog_ranges.pop(key)

    if verbose:
        self._summary(
            out_of_range_series        = out_of_range_series,
            out_of_range_series_ranges = out_of_range_series_ranges,
            out_of_range_exog          = out_of_range_exog,
            out_of_range_exog_ranges   = out_of_range_exog_ranges
        )

    return flag_out_of_range, out_of_range_series, out_of_range_exog

skforecast.drift_detection._population_drift.PopulationDriftDetector ¶

PopulationDriftDetector(
    chunk_size=None,
    threshold=3,
    threshold_method="std",
    max_out_of_range_proportion=0.1,
)

A class to detect population drift between reference and new datasets.

This implementation computes Kolmogorov-Smirnov (KS) test for numeric features, Chi-Square test for categorical features, and Jensen-Shannon (JS) distance for all features. It calculates empirical distributions of these statistics from the reference data and uses quantile thresholds to determine drift in new data.

Unlike fixed statistical cutoffs, all drift thresholds are calibrated from the reference data itself, allowing the detector to adapt to the natural variability of each feature.

This implementation is inspired by NannyML's DriftDetector. See Notes for details.

For an in-depth explanation of the underlying calculations, see https://skforecast.org/latest/user_guides/drift-detection.html#deep-dive-into-temporal-drift-detection-in-time-series

Parameters:

Name Type Description Default
chunk_size (int, str)

Size of chunks for sequential drift analysis.

  • If int, it represents the number of observations per chunk.
  • If str (e.g., 'D', 'W', 'MS'), it defines time-based chunks assuming a datetime index.
  • If None, the entire dataset is analyzed as a single chunk.
None
threshold (int, float)

Threshold for KS, Chi2, and JS statistics. Interpretation depends on threshold_method:

  • If threshold_method='std', threshold is interpreted as a multiplier of the standard deviation, and thresholds are computed as: mean + threshold * std.
  • If threshold_method='quantile', threshold represents the quantile level (between 0 and 1) used to compute the empirical threshold.
3
threshold_method str

Strategy used to estimate threshold from empirical distributions computed on the reference data:

  • 'std': Thresholds are estimated as a function of the mean and standard deviation of the empirical distribution (mean + threshold * std). This approach is computationally efficient, as it does not rely on leave-one-chunk-out procedures.
  • 'quantile': Thresholds are derived from a specified quantile of the empirical distribution using leave-one-chunk-out cross-validation to avoid self-comparison bias. This is statistically more correct for quantile-based thresholds but computationally more expensive.
'std'
max_out_of_range_proportion float

Maximum allowed proportion of observations outside the reference value range for numeric features. If the proportion of out-of-range observations in a new data chunk exceeds this value, drift is flagged for the corresponding feature. This parameter must be between 0 and 1.

0.1

Attributes:

Name Type Description
chunk_size (int, str)

Size of chunks for sequential drift analysis.

threshold float

Threshold for KS, Chi2, and JS statistics. Interpretation depends on threshold_method.

threshold_method str

Method for calculating threshold ('quantile' or 'std').

max_out_of_range_proportion float

Proportion threshold for out-of-range observations (numeric features).

is_fitted bool

Indicates if the detector has been fitted with reference data.

ref_features_ list

List of features in the reference data.

empirical_dist_ks_ dict

Empirical distributions of KS test statistics for each numeric feature in reference data.

empirical_dist_chi2_ dict

Empirical distributions of Chi-Square test statistics for each categorical feature in reference data.

empirical_dist_js_ dict

Empirical distributions of Jensen-Shannon distance for each feature in reference data (numeric and categorical).

empirical_threshold_ks_ dict

Computed thresholds for KS statistics based on empirical distributions for each numeric feature in reference data.

empirical_threshold_chi2_ dict

Computed thresholds for Chi-Square statistics based on empirical distributions for each categorical feature in reference data.

empirical_threshold_js_ dict

Computed thresholds for Jensen-Shannon distance based on empirical distributions for each feature in reference data (numeric and categorical).

n_chunks_reference_data_ int

Number of chunks in reference data used during fitting to compute empirical distributions.

ref_ecdf_ dict

Precomputed ECDFs for numeric features in the reference data.

ref_bins_edges_ dict

Precomputed histogram bin edges for numeric features in the reference data.

ref_hist_ dict

Precomputed histograms for numeric features in the reference data.

ref_probs_ dict

Precomputed normalized value counts (probabilities) for each category of categorical features in the reference data.

ref_ranges_ dict

Min and max values for numeric features in the reference data.

ref_categories_ dict

Unique categories for categorical features in the reference data.

detectors_ dict

Dictionary of PopulationDriftDetector instances for each group when fitting/predicting on MultiIndex DataFrames.

series_names_in_ list

List of series IDs present during fitting when using MultiIndex DataFrames.

skforecast_version str

Version of skforecast library used to create the detector.

Notes

This implementation is inspired by NannyML's DriftDetector [1]_.

It is a lightweight version adapted for skforecast's needs: - It does not store the raw reference data, only the necessary precomputed information to calculate the statistics efficiently during prediction. - All empirical thresholds are calculated using the specified quantile from the empirical distributions obtained from the reference data chunks. - It includes checks for out of range values in numeric features and new categories in categorical features. - It supports multiple time series by fitting separate detectors for each series ID when provided with a MultiIndex DataFrame.

For advanced features (multivariate drift, data quality checks), consider using NannyML directly: https://nannyml.readthedocs.io/en/stable/

References

.. [1] NannyML API Reference. https://nannyml.readthedocs.io/en/stable/tutorials/detecting_data_drift/univariate_drift_detection.html

Methods:

Name Description
fit

Fit the drift detector by calculating empirical distributions and thresholds

predict

Predict drift in new data by comparing the estimated statistics to

get_thresholds

Return a DataFrame with all computed thresholds per feature.

Source code in skforecast/drift_detection/_population_drift.py
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def __init__(
    self, 
    chunk_size: int | str | None = None,
    threshold: int | float = 3,
    threshold_method: str = 'std',
    max_out_of_range_proportion: float = 0.1
) -> None:

    self.ref_features_             = []
    self.is_fitted                 = False
    self.ref_ecdf_                 = {}
    self.ref_bins_edges_           = {}
    self.ref_hist_                 = {}
    self.ref_probs_                = {}
    self.ref_counts_               = {}
    self.empirical_dist_ks_        = {}
    self.empirical_dist_chi2_      = {}
    self.empirical_dist_js_        = {}
    self.empirical_threshold_ks_   = {}
    self.empirical_threshold_chi2_ = {}
    self.empirical_threshold_js_   = {}
    self.ref_ranges_               = {}
    self.ref_categories_           = {}
    self.n_chunks_reference_data_  = None
    self.detectors_                = {}    # NOTE: Only used for multiseries
    self.series_names_in_          = None  # NOTE: Only used for multiseries
    self.skforecast_version        = __version__

    error_msg = (
        "`chunk_size` must be a positive integer, a string compatible with "
        "pandas frequencies (e.g., 'D', 'W', 'MS'), or None."
    )
    if not (isinstance(chunk_size, (int, str, pd.DateOffset, type(None)))):
        raise TypeError(f"{error_msg} Got {type(chunk_size)}.")

    if isinstance(chunk_size, str):
        try:
            chunk_size = pd.tseries.frequencies.to_offset(chunk_size)
        except ValueError:
            raise ValueError(f"{error_msg} Got {type(chunk_size)}.")

    if isinstance(chunk_size, int) and chunk_size <= 0:
        raise ValueError(f"{error_msg} Got {chunk_size}.")

    self.chunk_size = chunk_size

    valid_threshold_methods = ['quantile', 'std']
    if threshold_method not in valid_threshold_methods:
        raise ValueError(
            f"`threshold_method` must be one of {valid_threshold_methods}. "
            f"Got '{threshold_method}'."
        )
    self.threshold_method = threshold_method

    if threshold_method == 'quantile':
        if not (0 < threshold < 1):
            raise ValueError(
                f"When `threshold_method='quantile'`, `threshold` must be between "
                f"0 and 1. Got {threshold}."
            )
    else:
        if threshold < 0:
            raise ValueError(
                f"When `threshold_method='std'`, `threshold` must be >= 0. "
                f"Got {threshold}."
            )

    self.threshold = threshold

    if not (0 <= max_out_of_range_proportion <= 1):
        raise ValueError(
            f"`max_out_of_range_proportion` must be between 0 and 1. "
            f"Got {max_out_of_range_proportion}."
        )
    self.max_out_of_range_proportion = max_out_of_range_proportion

Attributes¶

ref_features_ instance-attribute ¶

ref_features_ = []

is_fitted instance-attribute ¶

is_fitted = False

ref_ecdf_ instance-attribute ¶

ref_ecdf_ = {}

ref_bins_edges_ instance-attribute ¶

ref_bins_edges_ = {}

ref_hist_ instance-attribute ¶

ref_hist_ = {}

ref_probs_ instance-attribute ¶

ref_probs_ = {}

ref_counts_ instance-attribute ¶

ref_counts_ = {}

empirical_dist_ks_ instance-attribute ¶

empirical_dist_ks_ = {}

empirical_dist_chi2_ instance-attribute ¶

empirical_dist_chi2_ = {}

empirical_dist_js_ instance-attribute ¶

empirical_dist_js_ = {}

empirical_threshold_ks_ instance-attribute ¶

empirical_threshold_ks_ = {}

empirical_threshold_chi2_ instance-attribute ¶

empirical_threshold_chi2_ = {}

empirical_threshold_js_ instance-attribute ¶

empirical_threshold_js_ = {}

ref_ranges_ instance-attribute ¶

ref_ranges_ = {}

ref_categories_ instance-attribute ¶

ref_categories_ = {}

n_chunks_reference_data_ instance-attribute ¶

n_chunks_reference_data_ = None

detectors_ instance-attribute ¶

detectors_ = {}

series_names_in_ instance-attribute ¶

series_names_in_ = None

skforecast_version instance-attribute ¶

skforecast_version = __version__

chunk_size instance-attribute ¶

chunk_size = chunk_size

threshold_method instance-attribute ¶

threshold_method = threshold_method

threshold instance-attribute ¶

threshold = threshold

max_out_of_range_proportion instance-attribute ¶

max_out_of_range_proportion = max_out_of_range_proportion

Methods:¶

fit ¶

fit(X)

Fit the drift detector by calculating empirical distributions and thresholds from reference data. The empirical distributions are computed by chunking the reference data according to the specified chunk_size and calculating the statistics for each chunk.

Parameters:

Name Type Description Default
X pandas DataFrame

Reference data used as the baseline for drift detection.

  • If X is a regular DataFrame, a single detector is fitted for all data. The index is assumed to be the temporal index and each column a feature.
  • If X has a MultiIndex, the first level is assumed to be the series ID and the second level the temporal index. A separate detector is fitted for each series.
required
Source code in skforecast/drift_detection/_population_drift.py
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
def fit(self, X) -> None:
    """
    Fit the drift detector by calculating empirical distributions and thresholds
    from reference data. The empirical distributions are computed by chunking
    the reference data according to the specified `chunk_size` and calculating
    the statistics for each chunk.

    Parameters
    ----------
    X : pandas DataFrame
        Reference data used as the baseline for drift detection.

        - If `X` is a regular DataFrame, a single detector is fitted for all data.
        The index is assumed to be the temporal index and each column a feature.
        - If `X` has a MultiIndex, the first level is assumed to be the series ID
        and the second level the temporal index. A separate detector is fitted for
        each series.        

    """

    self._reset_attributes()
    self.detectors_       = {}    # NOTE: Only used for multiseries
    self.series_names_in_ = None  # NOTE: Only used for multiseries

    if not isinstance(X, pd.DataFrame):
        raise ValueError(
            f"`X` must be a pandas DataFrame. Got {type(X)} instead."
        )

    if isinstance(X.index, pd.MultiIndex):
        X = X.groupby(level=0)

        for idx, group in X:
            group = group.droplevel(0)
            self.detectors_[idx] = PopulationDriftDetector(
                                       chunk_size                  = self.chunk_size,
                                       threshold                   = self.threshold,
                                       threshold_method            = self.threshold_method,
                                       max_out_of_range_proportion = self.max_out_of_range_proportion
                                   )
            self.detectors_[idx]._fit(group)
    else:
        self._fit(X)

    self.is_fitted = True
    self.series_names_in_ = list(self.detectors_.keys()) if self.detectors_ else None
    self._collect_attributes()

predict ¶

predict(X)

Predict drift in new data by comparing the estimated statistics to reference thresholds.

Parameters:

Name Type Description Default
X pandas DataFrame

New data to compare against the reference.

required

Returns:

Name Type Description
results pandas DataFrame

DataFrame with the drift detection results for each chunk.

summary pandas DataFrame

Summary DataFrame with the total number and percentage of chunks with detected drift per feature (or per series_id and feature if MultiIndex), and the list of chunk IDs where drift was detected.

Source code in skforecast/drift_detection/_population_drift.py
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
def predict(self, X) -> tuple[pd.DataFrame, pd.DataFrame]:
    """
    Predict drift in new data by comparing the estimated statistics to
    reference thresholds.

    Parameters
    ----------
    X : pandas DataFrame
        New data to compare against the reference.

    Returns
    -------
    results : pandas DataFrame
        DataFrame with the drift detection results for each chunk.
    summary : pandas DataFrame
        Summary DataFrame with the total number and percentage of chunks
        with detected drift per feature (or per series_id and feature if
        MultiIndex), and the list of chunk IDs where drift was detected.

    """

    if not self.is_fitted:
        raise NotFittedError(
            "This PopulationDriftDetector instance is not fitted yet. "
            "Call 'fit' with appropriate arguments before using this estimator."
        )

    if not isinstance(X, pd.DataFrame):
        raise ValueError(f"`X` must be a pandas DataFrame. Got {type(X)} instead.")

    if isinstance(X.index, pd.MultiIndex):
        results = []
        for idx, group in X.groupby(level=0):
            group = group.droplevel(0)
            if idx not in self.detectors_:
                warnings.warn(
                    f"Series '{idx}' was not present during fitting. Drift detection skipped.",
                    UnknownLevelWarning
                )
                continue

            detector = self.detectors_[idx]
            result = detector._predict(group)
            result.insert(0, 'series_id', idx)
            results.append(result)

        results = pd.concat(results, ignore_index=True)
    else:
        results = self._predict(X)

    if results.columns[0] == 'series_id':
        groupby_cols = ['series_id', 'feature']
    else:
        groupby_cols = ['feature']

    def _get_drift_chunk_ids(group):
        return group.loc[group['drift_detected'], 'chunk'].tolist()

    summary = (
        results.groupby(groupby_cols)
        .agg(
            n_chunks_with_drift=('drift_detected', 'sum'),
            pct_chunks_with_drift=('drift_detected', 'mean'),
            chunks_with_drift=('drift_detected', lambda x: _get_drift_chunk_ids(
                results.loc[x.index]
            ))
        )
        .reset_index()
    )

    summary['pct_chunks_with_drift'] = summary['pct_chunks_with_drift'] * 100

    return results, summary

get_thresholds ¶

get_thresholds()

Return a DataFrame with all computed thresholds per feature. For multi-series, returns thresholds per series_id and feature.

Returns:

Name Type Description
thresholds pandas DataFrame

DataFrame with the computed thresholds per feature (and per series_id if MultiIndex was used during fitting).

Source code in skforecast/drift_detection/_population_drift.py
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
def get_thresholds(
    self
) -> pd.DataFrame:
    """
    Return a DataFrame with all computed thresholds per feature.
    For multi-series, returns thresholds per series_id and feature.

    Returns
    -------
    thresholds : pandas DataFrame
        DataFrame with the computed thresholds per feature (and per series_id
        if MultiIndex was used during fitting).

    """

    if not self.is_fitted:
        raise NotFittedError(
            "This PopulationDriftDetector instance is not fitted yet. "
            "Call 'fit' with appropriate arguments before using this estimator."
        )

    # Multi-series case: ref_features_ is a dict keyed by series_id
    if self.detectors_:
        thresholds = {
            "series_id": [],
            "feature": [],
            "ks_threshold": [],
            "chi2_threshold": [],
            "js_threshold": [],
            "max_out_of_range_proportion": []
        }
        for series_id, detector in self.detectors_.items():
            for feature in detector.ref_features_:
                thresholds["series_id"].append(series_id)
                thresholds["feature"].append(feature)
                thresholds["ks_threshold"].append(
                    detector.empirical_threshold_ks_.get(feature)
                )
                thresholds["chi2_threshold"].append(
                    detector.empirical_threshold_chi2_.get(feature)
                )
                thresholds["js_threshold"].append(
                    detector.empirical_threshold_js_.get(feature)
                )
                thresholds["max_out_of_range_proportion"].append(
                    detector.max_out_of_range_proportion
                )
    else:
        # Single-series case
        thresholds = {
            "feature": [],
            "ks_threshold": [],
            "chi2_threshold": [],
            "js_threshold": [],
            "max_out_of_range_proportion": []
        }
        for feature in self.ref_features_:
            thresholds["feature"].append(feature)
            thresholds["ks_threshold"].append(
                self.empirical_threshold_ks_.get(feature)
            )
            thresholds["chi2_threshold"].append(
                self.empirical_threshold_chi2_.get(feature)
            )
            thresholds["js_threshold"].append(
                self.empirical_threshold_js_.get(feature)
            )
            thresholds["max_out_of_range_proportion"].append(
                self.max_out_of_range_proportion
            )

    return pd.DataFrame(thresholds)