API reference

Full auto-generated API documentation from source code docstrings.

SWIFTMonitor

The main entry point — the scikit-learn compatible monitor orchestrating all five stages.

SWIFTMonitor

SWIFTMonitor(model: object = None, order: int = 1, n_permutations: int = 1000, alpha: float = 0.05, correction: CorrectionMethod | str = BH, n_synthetic: int = 10, max_samples: int | None = None, random_state: int = 42)

Bases: MonitorPlottingMixin, BaseEstimator, TransformerMixin

SHAP-Weighted Impact Feature Testing monitor.

Orchestrates the 5-stage SWIFT pipeline:

1. Extract decision points from trained model
2. Build buckets from decision points
3. Compute bucket-level mean SHAP (SHAP normalization σ_j)
4. Compute Wasserstein distance on SHAP-transformed distributions
5. Permutation-based significance testing with MTC

The transformation σ_j is computed ONCE during fit() and applied identically to all monitoring samples via transform().

PARAMETER	DESCRIPTION
model	Trained tree-ensemble model (LightGBM Booster or XGBoost Booster). Required — passed as a constructor dependency (analogous to sklearn.feature_selection.SelectFromModel). TYPE: object DEFAULT: None
order	Wasserstein order (1 → W₁, 2 → W₂). TYPE: int DEFAULT: 1
n_permutations	Number of permutations for p-value estimation in test(). TYPE: int DEFAULT: 1000
alpha	Significance level for multiple testing correction. TYPE: float DEFAULT: 0.05
correction	Multiple testing correction method. Accepts enum members or strings ("bonferroni", "benjamini-hochberg", "bh"). TYPE: CorrectionMethod or str DEFAULT: "benjamini-hochberg"
n_synthetic	Number of synthetic observations to create for empty buckets during fit(). TYPE: int DEFAULT: 10
max_samples	Maximum total pool size (n_ref + n_mon) for the permutation test. If exceeded, subsample proportionally. None = no limit. TYPE: int or None DEFAULT: None
random_state	Seed for the random number generator, ensuring reproducibility. TYPE: int DEFAULT: 42

ATTRIBUTE	DESCRIPTION
bucket_sets_	Per-feature bucket sets with mean_shap populated (set by fit). TYPE: dict[str, BucketSet]
X_ref_	Copy of the reference DataFrame (stored for permutation testing). TYPE: DataFrame
shap_values_	SHAP values computed on the reference data. TYPE: ndarray
importance_weights_	Normalized mean |SHAP| importance weight per feature (sums to 1); used for the swift_weighted aggregate (set by fit). TYPE: dict[str, float]
feature_names_in_	Feature names inferred from X.columns during fit. TYPE: ndarray
n_features_in_	Number of features seen during fit. TYPE: int

Examples:

>>> monitor = SWIFTMonitor(model=lgb_model, n_permutations=200)
>>> monitor.fit(X_ref)
SWIFTMonitor(...)
>>> result = monitor.test(X_mon)
>>> result.drifted_features
('feature_3',)

Source code in src/swift/pipeline.py

def __init__(
    self,
    model: object = None,
    order: int = 1,
    n_permutations: int = 1000,
    alpha: float = 0.05,
    correction: CorrectionMethod | str = CorrectionMethod.BH,
    n_synthetic: int = 10,
    max_samples: int | None = None,
    random_state: int = 42,
) -> None:
    self.model = model
    self.order = order
    self.n_permutations = n_permutations
    self.alpha = alpha
    self.correction = correction
    self.n_synthetic = n_synthetic
    self.max_samples = max_samples
    self.random_state = random_state

fit

fit(X: DataFrame, y: None = None) -> SWIFTMonitor

Fit the SWIFT monitor on reference data.

Executes stages 1–3: extraction → bucketing → SHAP normalization.

PARAMETER	DESCRIPTION
X	Reference DataFrame. Feature names are inferred from X.columns. TYPE: pd.DataFrame of shape (n_ref, n_features)
y	Not used; present for API compatibility. TYPE: ignored DEFAULT: None

RETURNS	DESCRIPTION
self

Source code in src/swift/pipeline.py

def fit(
    self,
    X: pd.DataFrame,
    y: None = None,
) -> SWIFTMonitor:
    """Fit the SWIFT monitor on reference data.

    Executes stages 1–3: extraction → bucketing → SHAP normalization.

    Parameters
    ----------
    X : pd.DataFrame of shape (n_ref, n_features)
        Reference DataFrame.  Feature names are inferred from
        ``X.columns``.
    y : ignored
        Not used; present for API compatibility.

    Returns
    -------
    self
    """
    if self.model is None:
        raise ValueError(
            "SWIFTMonitor requires a trained model.  "
            "Pass it via the constructor: SWIFTMonitor(model=my_model)."
        )

    rng = np.random.default_rng(self.random_state)

    # Infer feature names (sklearn convention).
    self.feature_names_in_ = np.asarray(X.columns)
    self.n_features_in_ = len(self.feature_names_in_)
    feature_names = list(self.feature_names_in_)

    self.X_ref_ = X.copy()

    # Stage 1: Extract decision points
    logger.info("Stage 1: Extracting decision points...")
    decision_points = extract_decision_points(self.model, feature_names)

    # Stage 2: Build buckets
    logger.info("Stage 2: Building buckets...")
    bucket_sets = build_all_buckets(decision_points)

    # Compute SHAP values on reference data
    logger.info("Computing SHAP values on reference data...")
    explainer = shap.TreeExplainer(self.model)
    shap_values = explainer.shap_values(X)
    shap_values = np.asarray(shap_values)
    self.shap_values_ = shap_values
    self.importance_weights_ = compute_importance_weights(
        shap_values, feature_names
    )

    # Stage 3: SHAP normalization
    logger.info("Stage 3: Computing bucket-level mean SHAP...")
    self.bucket_sets_ = compute_bucket_shap(
        bucket_sets,
        X,
        shap_values,
        model=self.model,
        n_synthetic=self.n_synthetic,
        rng=rng,
    )

    n_buckets_total = sum(
        bs.num_buckets for bs in self.bucket_sets_.values()
    )
    logger.info(
        "SWIFT monitor fitted: %d features, %d total buckets.",
        len(feature_names),
        n_buckets_total,
    )

    return self

transform

transform(X: DataFrame) -> DataFrame

Apply the SHAP transformation σ_j to every feature.

Each value x_ij is mapped to the mean SHAP of its bucket: σ_j(x_ij) = mean_shap_j^{bucket(x_ij)}.

PARAMETER	DESCRIPTION
X	Input data. TYPE: pd.DataFrame of shape (n_samples, n_features)

RETURNS	DESCRIPTION
DataFrame	SHAP-transformed DataFrame (same shape and column names).

Source code in src/swift/pipeline.py

def transform(self, X: pd.DataFrame) -> pd.DataFrame:
    """Apply the SHAP transformation σ_j to every feature.

    Each value ``x_ij`` is mapped to the mean SHAP of its bucket:
    ``σ_j(x_ij) = mean_shap_j^{bucket(x_ij)}``.

    Parameters
    ----------
    X : pd.DataFrame of shape (n_samples, n_features)
        Input data.

    Returns
    -------
    pd.DataFrame
        SHAP-transformed DataFrame (same shape and column names).
    """
    check_is_fitted(self)
    result = pd.DataFrame(index=X.index)
    for fname in self.feature_names_in_:
        result[fname] = transform_feature(
            X[fname].values, self.bucket_sets_[fname]
        )
    return result

score

score(X: DataFrame, X_compare: DataFrame | None = None) -> dict[str, float]

Compute per-feature SWIFT scores (stage 4 only, no testing).

PARAMETER	DESCRIPTION
X	Monitoring DataFrame. When X_compare is None, this is compared against the fitted reference X_ref_. TYPE: pd.DataFrame of shape (n_samples, n_features)
X_compare	Optional second sample. When provided, SWIFT scores are computed between X and X_compare instead of between X_ref_ and X. The SHAP transformation σ_j is always the one fitted on X_ref_. TYPE: DataFrame or None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, float]	Feature name → SWIFT score (Wasserstein distance on SHAP-transformed distributions).

Source code in src/swift/pipeline.py

def score(
    self,
    X: pd.DataFrame,
    X_compare: pd.DataFrame | None = None,
) -> dict[str, float]:
    """Compute per-feature SWIFT scores (stage 4 only, no testing).

    Parameters
    ----------
    X : pd.DataFrame of shape (n_samples, n_features)
        Monitoring DataFrame.  When *X_compare* is ``None``, this is
        compared against the fitted reference ``X_ref_``.
    X_compare : pd.DataFrame or None
        Optional second sample.  When provided, SWIFT scores are
        computed between *X* and *X_compare* instead of between
        ``X_ref_`` and *X*.  The SHAP transformation σ_j is always
        the one fitted on ``X_ref_``.

    Returns
    -------
    dict[str, float]
        Feature name → SWIFT score (Wasserstein distance on
        SHAP-transformed distributions).
    """
    check_is_fitted(self)

    if X_compare is not None:
        return compute_swift_scores(
            X, X_compare, self.bucket_sets_, order=self.order
        )

    return compute_swift_scores(
        self.X_ref_, X, self.bucket_sets_, order=self.order
    )

test

test(X: DataFrame, X_compare: DataFrame | None = None) -> SWIFTResult

Run the full SWIFT pipeline: score + test + aggregate.

Stages 4–5 + aggregation:

• Compute per-feature SWIFT scores (Wasserstein on SHAP-transformed).
• Permutation test for p-values.
• Multiple testing correction.
• Model-level aggregation.

All hyperparameters (order, n_permutations, alpha, correction, max_samples) are taken from instance attributes set in the constructor. Use set_params() to override for individual calls (e.g. different random_state per experiment repetition).

PARAMETER	DESCRIPTION
X	Monitoring DataFrame. When X_compare is None, this is compared against the fitted reference X_ref_. TYPE: pd.DataFrame of shape (n_samples, n_features)
X_compare	Optional second sample. When provided, the test compares X against X_compare instead of X_ref_ against X. The SHAP transformation σ_j is always the one fitted on X_ref_. TYPE: DataFrame or None DEFAULT: None

RETURNS	DESCRIPTION
SWIFTResult	Per-feature and model-level results.

Source code in src/swift/pipeline.py

def test(
    self,
    X: pd.DataFrame,
    X_compare: pd.DataFrame | None = None,
) -> SWIFTResult:
    """Run the full SWIFT pipeline: score + test + aggregate.

    Stages 4–5 + aggregation:

    - Compute per-feature SWIFT scores (Wasserstein on SHAP-transformed).
    - Permutation test for p-values.
    - Multiple testing correction.
    - Model-level aggregation.

    All hyperparameters (``order``, ``n_permutations``, ``alpha``,
    ``correction``, ``max_samples``) are taken from instance attributes
    set in the constructor.  Use ``set_params()`` to override for
    individual calls (e.g. different ``random_state`` per experiment
    repetition).

    Parameters
    ----------
    X : pd.DataFrame of shape (n_samples, n_features)
        Monitoring DataFrame.  When *X_compare* is ``None``, this is
        compared against the fitted reference ``X_ref_``.
    X_compare : pd.DataFrame or None
        Optional second sample.  When provided, the test compares
        *X* against *X_compare* instead of ``X_ref_`` against *X*.
        The SHAP transformation σ_j is always the one fitted on
        ``X_ref_``.

    Returns
    -------
    SWIFTResult
        Per-feature and model-level results.
    """
    check_is_fitted(self)

    rng = np.random.default_rng(self.random_state)
    correction = CorrectionMethod.resolve(self.correction)

    # Determine the two samples to compare
    if X_compare is not None:
        X_a, X_b = X, X_compare
    else:
        X_a, X_b = self.X_ref_, X

    # Stage 4: Compute SWIFT scores
    logger.info("Stage 4: Computing SWIFT scores...")
    scores = compute_swift_scores(
        X_a, X_b, self.bucket_sets_, order=self.order
    )

    # Stage 5: Permutation test + MTC
    logger.info(
        "Stage 5: Permutation test (B=%d)...", self.n_permutations
    )
    pvalues = permutation_test(
        X_a,
        X_b,
        self.bucket_sets_,
        order=self.order,
        n_permutations=self.n_permutations,
        max_samples=self.max_samples,
        rng=rng,
    )

    decisions = correct_pvalues(pvalues, correction, self.alpha)

    # Per-feature results + aggregation (importance-weighted when
    # fitted weights are available)
    result = _assemble_result(
        feature_names=list(self.feature_names_in_),
        scores=scores,
        pvalues=pvalues,
        decisions=decisions,
        bucket_sets=self.bucket_sets_,
        order=self.order,
        weights=getattr(self, "importance_weights_", None),
        alpha=self.alpha,
        correction=correction,
    )

    logger.info(
        "SWIFT test complete: %d/%d features drifted (α=%.3f, %s). "
        "SWIFT_max=%.6f, SWIFT_mean=%.6f, SWIFT_weighted=%s",
        result.num_drifted,
        result.num_features,
        self.alpha,
        correction.value,
        result.swift_max,
        result.swift_mean,
        f"{result.swift_weighted:.6f}" if result.swift_weighted is not None else "n/a",
    )

    return result

Types

Data types used across the pipeline: buckets, bucket sets, result records, and enums.

Bucket dataclass

Bucket(bucket_type: BucketType, index: int, lower: float = float('-inf'), upper: float = float('inf'), mean_shap: float | None = None)

A single bucket for a feature.

For numeric features: defined by [lower, upper) interval. For null bucket: lower=upper=NaN.

contains

contains(value: float | None) -> bool

Check if a value falls into this bucket.

Source code in src/swift/types.py

def contains(self, value: float | None) -> bool:
    """Check if a value falls into this bucket."""
    if value is None or (isinstance(value, float) and np.isnan(value)):
        return self.bucket_type == BucketType.NULL

    if self.bucket_type == BucketType.NULL:
        return False

    # Numeric: [lower, upper) except for the first bucket which is (-inf, upper)
    if np.isneginf(self.lower):
        return value < self.upper
    if np.isposinf(self.upper):
        return value >= self.lower
    return self.lower <= value < self.upper

BucketSet dataclass

BucketSet(feature_name: str, buckets: tuple[Bucket, ...] = tuple(), decision_points: ndarray = (lambda: array([]))())

Collection of buckets for a single feature.

ATTRIBUTE	DESCRIPTION
feature_name	Name of the feature. TYPE: str
buckets	List of Bucket objects (including null bucket). TYPE: tuple[Bucket, ...]
decision_points	Sorted array of decision points (split thresholds). TYPE: ndarray

num_buckets property

num_buckets: int

Total number of buckets including null.

assign_bucket

assign_bucket(value: float | str | None) -> int

Return the index of the bucket that contains the given value.

RAISES	DESCRIPTION
ValueError	If the value does not fall into any bucket.

Source code in src/swift/types.py

def assign_bucket(self, value: float | str | None) -> int:
    """Return the index of the bucket that contains the given value.

    Raises
    ------
    ValueError
        If the value does not fall into any bucket.
    """
    for bucket in self.buckets:
        if bucket.contains(value):
            return bucket.index
    raise ValueError(
        f"Value {value!r} does not fall into any bucket for feature '{self.feature_name}'"
    )

get_mean_shap

get_mean_shap(bucket_index: int) -> float

Return the mean SHAP value for the given bucket index.

RAISES	DESCRIPTION
KeyError	If the bucket index does not exist.

Source code in src/swift/types.py

def get_mean_shap(self, bucket_index: int) -> float:
    """Return the mean SHAP value for the given bucket index.

    Raises
    ------
    KeyError
        If the bucket index does not exist.
    """
    for bucket in self.buckets:
        if bucket.index == bucket_index:
            if bucket.mean_shap is None:
                logger.warning(
                    "Bucket %d of feature '%s' has no mean SHAP value assigned.",
                    bucket_index,
                    self.feature_name,
                )
                return 0.0
            return bucket.mean_shap
    raise KeyError(
        f"Bucket index {bucket_index} not found for feature '{self.feature_name}'"
    )

BucketType

Bases: Enum

Type of bucket.

WassersteinOrder

Bases: Enum

Order of Wasserstein distance.

CorrectionMethod

Bases: Enum

Multiple testing correction method.

Supports construction from strings via CorrectionMethod.resolve(), so users can write correction="benjamini-hochberg" instead of importing the enum.

resolve classmethod

resolve(value: CorrectionMethod | str) -> CorrectionMethod

Return a CorrectionMethod from a member, its value, or an alias.

PARAMETER	DESCRIPTION
value	Enum member, canonical string ("bonferroni", "benjamini-hochberg"), or alias ("bh", "fdr", "bonf"). Case-insensitive. TYPE: CorrectionMethod or str

RETURNS	DESCRIPTION
CorrectionMethod

RAISES	DESCRIPTION
ValueError	If value cannot be resolved.

Source code in src/swift/types.py

@classmethod
def resolve(cls, value: CorrectionMethod | str) -> CorrectionMethod:
    """Return a ``CorrectionMethod`` from a member, its value, or an alias.

    Parameters
    ----------
    value : CorrectionMethod or str
        Enum member, canonical string (``"bonferroni"``,
        ``"benjamini-hochberg"``), or alias (``"bh"``, ``"fdr"``,
        ``"bonf"``).  Case-insensitive.

    Returns
    -------
    CorrectionMethod

    Raises
    ------
    ValueError
        If *value* cannot be resolved.
    """
    if isinstance(value, cls):
        return value
    key = value.strip().lower()
    # Direct enum-value lookup.
    for member in cls:
        if member.value == key:
            return member
    # Alias lookup.
    canonical = _CORRECTION_ALIASES.get(key)
    if canonical is not None:
        for member in cls:
            if member.value == canonical:
                return member
    valid = sorted(
        {m.value for m in cls} | set(_CORRECTION_ALIASES.keys())
    )
    raise ValueError(
        f"Unknown correction method {value!r}. "
        f"Valid values: {valid}"
    )

FeatureSWIFTResult dataclass

FeatureSWIFTResult(feature_name: str, swift_score: float, wasserstein_order: WassersteinOrder = W1, p_value: float | None = None, is_drifted: bool | None = None, num_buckets: int = 0)

SWIFT result for a single feature.

ATTRIBUTE	DESCRIPTION
feature_name	Name of the feature. TYPE: str
swift_score	SWIFT score (Wasserstein distance on SHAP-transformed distributions). TYPE: float
wasserstein_order	Which Wasserstein order was used (W1 or W2). TYPE: WassersteinOrder
p_value	p-value from permutation or bootstrap test (None if not computed). TYPE: float or None
is_drifted	Whether the feature is flagged as drifted after correction. TYPE: bool or None
num_buckets	Number of buckets used. TYPE: int

SWIFTResult dataclass

SWIFTResult(feature_results: tuple[FeatureSWIFTResult, ...], swift_max: float, swift_mean: float, swift_weighted: float | None = None, alpha: float = 0.05, correction_method: CorrectionMethod | None = None, max_feature: str | None = None)

Aggregate SWIFT result across all features.

ATTRIBUTE	DESCRIPTION
feature_results	Per-feature results. TYPE: tuple[FeatureSWIFTResult, ...]
swift_max	Maximum SWIFT score across features. TYPE: float
swift_mean	Mean SWIFT score across features. TYPE: float
swift_weighted	Importance-weighted SWIFT score (optional). TYPE: float or None
alpha	Significance level used for testing. TYPE: float
correction_method	Multiple testing correction method used. TYPE: CorrectionMethod or None
max_feature	Name of the feature with the highest SWIFT score. TYPE: str or None

num_features property

num_features: int

Number of features monitored.

num_drifted property

num_drifted: int

Number of features flagged as drifted.

drifted_features property

drifted_features: tuple[str, ...]

Names of features flagged as drifted.

Extraction

Stage 1 — decision point extraction from trained tree models.

extraction

Stage 1: Decision point extraction from trained models.

Extracts the set of unique split thresholds (decision points) per feature from a trained model. These decision points define the bucket boundaries for the SWIFT pipeline.

Supported model types: - LightGBM Booster (extract_decision_points_lgb) - XGBoost Booster (extract_decision_points_xgb)

extract_decision_points

extract_decision_points(model: object, feature_names: Sequence[str]) -> dict[str, ndarray]

Auto-dispatch extraction based on model type.

PARAMETER	DESCRIPTION
model	A trained LightGBM Booster or XGBoost Booster. TYPE: object
feature_names	List of feature names (must match model's feature order). TYPE: Sequence[str]

RETURNS	DESCRIPTION
dict[str, ndarray]	Dict mapping feature name -> sorted 1-D array of unique split thresholds.

RAISES	DESCRIPTION
TypeError	If the model type is not supported.

Source code in src/swift/extraction.py

def extract_decision_points(
    model: object,
    feature_names: Sequence[str],
) -> dict[str, np.ndarray]:
    """Auto-dispatch extraction based on model type.

    Parameters
    ----------
    model : object
        A trained LightGBM Booster or XGBoost Booster.
    feature_names : Sequence[str]
        List of feature names (must match model's feature order).

    Returns
    -------
    dict[str, np.ndarray]
        Dict mapping feature name -> sorted 1-D array of unique split thresholds.

    Raises
    ------
    TypeError
        If the model type is not supported.
    """
    if isinstance(model, lgb.Booster):
        return extract_decision_points_lgb(model, feature_names)

    # XGBoost Booster: avoid importing xgboost at module level
    try:
        import xgboost as xgb

        if isinstance(model, xgb.Booster):
            return extract_decision_points_xgb(model, feature_names)
    except ImportError:
        pass

    raise TypeError(
        f"Unsupported model type: {type(model).__name__}. "
        "SWIFT supports LightGBM Booster and XGBoost Booster."
    )

extract_decision_points_lgb

extract_decision_points_lgb(model: Booster, feature_names: Sequence[str]) -> dict[str, ndarray]

Extract unique, sorted split thresholds per feature from a LightGBM Booster.

For each feature, collects every split threshold used across all trees in the ensemble, deduplicates, and returns them in ascending order.

PARAMETER	DESCRIPTION
model	A trained LightGBM Booster. TYPE: Booster
feature_names	List of feature names (must match model's feature order). TYPE: Sequence[str]

RETURNS	DESCRIPTION
dict[str, ndarray]	Dict mapping feature name -> sorted 1-D array of unique split thresholds. Features never used in any split get an empty array.

Source code in src/swift/extraction.py

def extract_decision_points_lgb(
    model: lgb.Booster,
    feature_names: Sequence[str],
) -> dict[str, np.ndarray]:
    """Extract unique, sorted split thresholds per feature from a LightGBM Booster.

    For each feature, collects every split threshold used across all trees
    in the ensemble, deduplicates, and returns them in ascending order.

    Parameters
    ----------
    model : lgb.Booster
        A trained LightGBM Booster.
    feature_names : Sequence[str]
        List of feature names (must match model's feature order).

    Returns
    -------
    dict[str, np.ndarray]
        Dict mapping feature name -> sorted 1-D array of unique split thresholds.
        Features never used in any split get an empty array.
    """
    model_dump = model.dump_model()
    trees = model_dump["tree_info"]

    # Build feature index -> name mapping
    n_features = len(feature_names)
    splits_per_feature: dict[str, list[float]] = {name: [] for name in feature_names}

    for tree_info in trees:
        tree = tree_info["tree_structure"]
        _collect_splits_recursive(tree, feature_names, n_features, splits_per_feature)

    # Sort and deduplicate
    result: dict[str, np.ndarray] = {}
    for fname in feature_names:
        raw = splits_per_feature[fname]
        if raw:
            unique_sorted = np.unique(np.array(raw, dtype=np.float64))
            result[fname] = unique_sorted
        else:
            result[fname] = np.array([], dtype=np.float64)

    n_total = sum(len(v) for v in result.values())
    logger.info(
        "Extracted %d unique split points across %d features from %d trees.",
        n_total,
        n_features,
        len(trees),
    )

    return result

extract_decision_points_xgb

extract_decision_points_xgb(model: object, feature_names: Sequence[str]) -> dict[str, ndarray]

Extract unique, sorted split thresholds per feature from an XGBoost Booster.

Uses get_dump(dump_format='json') to obtain JSON-serialised trees, then recursively collects every numeric split threshold per feature.

PARAMETER	DESCRIPTION
model	A trained xgboost.Booster. TYPE: object
feature_names	List of feature names (must match model's feature order). TYPE: Sequence[str]

RETURNS	DESCRIPTION
dict[str, ndarray]	Dict mapping feature name -> sorted 1-D array of unique split thresholds. Features never used in any split get an empty array.

Source code in src/swift/extraction.py

def extract_decision_points_xgb(
    model: object,
    feature_names: Sequence[str],
) -> dict[str, np.ndarray]:
    """Extract unique, sorted split thresholds per feature from an XGBoost Booster.

    Uses ``get_dump(dump_format='json')`` to obtain JSON-serialised trees,
    then recursively collects every numeric split threshold per feature.

    Parameters
    ----------
    model : object
        A trained ``xgboost.Booster``.
    feature_names : Sequence[str]
        List of feature names (must match model's feature order).

    Returns
    -------
    dict[str, np.ndarray]
        Dict mapping feature name -> sorted 1-D array of unique split thresholds.
        Features never used in any split get an empty array.
    """
    import xgboost as xgb  # local import — XGBoost is optional

    if not isinstance(model, xgb.Booster):
        raise TypeError(
            f"Expected xgboost.Booster, got {type(model).__name__}."
        )

    # Build feature-index-to-name mapping.
    # XGBoost uses "f0", "f1", ... by default, or user-set feature names.
    feat_name_set = set(feature_names)
    idx_to_name: dict[str, str] = {}
    for i, name in enumerate(feature_names):
        idx_to_name[f"f{i}"] = name
        idx_to_name[name] = name  # model may already use real names

    tree_dumps = model.get_dump(dump_format="json")

    splits_per_feature: dict[str, list[float]] = {n: [] for n in feature_names}

    for tree_json_str in tree_dumps:
        tree_node = json.loads(tree_json_str)
        _collect_xgb_splits_recursive(tree_node, idx_to_name, feat_name_set, splits_per_feature)

    # Sort and deduplicate
    result: dict[str, np.ndarray] = {}
    for fname in feature_names:
        raw = splits_per_feature[fname]
        if raw:
            result[fname] = np.unique(np.array(raw, dtype=np.float64))
        else:
            result[fname] = np.array([], dtype=np.float64)

    n_total = sum(len(v) for v in result.values())
    logger.info(
        "Extracted %d unique split points across %d features from %d XGBoost trees.",
        n_total,
        len(feature_names),
        len(tree_dumps),
    )

    return result

Bucketing

Stage 2 — bucket construction from decision points.

bucketing

Stage 2: Bucket construction from decision points.

Given sorted, unique decision points (split thresholds) per feature, constructs a BucketSet with: - A null bucket for missing values (index 0) - Numeric buckets: (-inf, t1), [t1, t2), ..., [tm, inf)

build_buckets

build_buckets(decision_points: ndarray, feature_name: str) -> BucketSet

Construct a BucketSet from sorted decision points for a single feature.

PARAMETER	DESCRIPTION
decision_points	Sorted 1-D array of unique split thresholds. TYPE: ndarray
feature_name	Name of the feature. TYPE: str

RETURNS	DESCRIPTION
BucketSet	A BucketSet containing: Bucket 0: null bucket (for missing values) Buckets 1..m+1: numeric interval buckets Total: len(decision_points) + 2 buckets.

Source code in src/swift/bucketing.py

def build_buckets(
    decision_points: np.ndarray,
    feature_name: str,
) -> BucketSet:
    """Construct a BucketSet from sorted decision points for a single feature.

    Parameters
    ----------
    decision_points : np.ndarray
        Sorted 1-D array of unique split thresholds.
    feature_name : str
        Name of the feature.

    Returns
    -------
    BucketSet
        A BucketSet containing:

        - Bucket 0: null bucket (for missing values)
        - Buckets 1..m+1: numeric interval buckets

        Total: len(decision_points) + 2 buckets.
    """
    dp = np.asarray(decision_points, dtype=np.float64)
    m = len(dp)
    buckets: list[Bucket] = []

    # Index 0: null bucket
    buckets.append(
        Bucket(
            bucket_type=BucketType.NULL,
            index=0,
            lower=float("nan"),
            upper=float("nan"),
        )
    )

    if m == 0:
        # No decision points: single catch-all bucket (-inf, inf)
        buckets.append(
            Bucket(
                bucket_type=BucketType.NUMERIC,
                index=1,
                lower=float("-inf"),
                upper=float("inf"),
            )
        )
    else:
        # First bucket: (-inf, t1)
        buckets.append(
            Bucket(
                bucket_type=BucketType.NUMERIC,
                index=1,
                lower=float("-inf"),
                upper=float(dp[0]),
            )
        )

        # Middle buckets: [t_k, t_{k+1})  for k = 0..m-2
        for k in range(m - 1):
            buckets.append(
                Bucket(
                    bucket_type=BucketType.NUMERIC,
                    index=k + 2,
                    lower=float(dp[k]),
                    upper=float(dp[k + 1]),
                )
            )

        # Last bucket: [t_m, inf)
        buckets.append(
            Bucket(
                bucket_type=BucketType.NUMERIC,
                index=m + 1,
                lower=float(dp[-1]),
                upper=float("inf"),
            )
        )

    bucket_set = BucketSet(
        feature_name=feature_name,
        buckets=tuple(buckets),
        decision_points=dp,
    )

    logger.debug(
        "Feature '%s': %d decision points -> %d buckets.",
        feature_name,
        m,
        bucket_set.num_buckets,
    )

    return bucket_set

build_all_buckets

build_all_buckets(decision_points_dict: dict[str, ndarray]) -> dict[str, BucketSet]

Construct BucketSets for all features.

PARAMETER	DESCRIPTION
decision_points_dict	Dict mapping feature name -> sorted decision points. TYPE: dict[str, ndarray]

RETURNS	DESCRIPTION
dict[str, BucketSet]	Dict mapping feature name -> BucketSet.

Source code in src/swift/bucketing.py

def build_all_buckets(
    decision_points_dict: dict[str, np.ndarray],
) -> dict[str, BucketSet]:
    """Construct BucketSets for all features.

    Parameters
    ----------
    decision_points_dict : dict[str, np.ndarray]
        Dict mapping feature name -> sorted decision points.

    Returns
    -------
    dict[str, BucketSet]
        Dict mapping feature name -> BucketSet.
    """
    result: dict[str, BucketSet] = {}
    for fname, dp in decision_points_dict.items():
        result[fname] = build_buckets(dp, fname)
    return result

Normalization

Stage 3 — bucket-level mean SHAP and the feature transformation.

normalization

Stage 3: SHAP normalization — bucket-level mean SHAP and feature transformation.

Computes the mean SHAP value per bucket on the reference sample, then provides a transformation function that maps any feature value to its bucket's mean SHAP (the "SHAP normalization" step).

For empty buckets (no reference observations): creates synthetic observations by sampling real rows and placing the feature value inside the empty bucket, then computes SHAP on those synthetic observations.

compute_bucket_shap

compute_bucket_shap(bucket_sets: dict[str, BucketSet], X_ref: DataFrame, shap_values: ndarray, model: object | None = None, n_synthetic: int = 10, rng: Generator | None = None) -> dict[str, BucketSet]

Compute mean SHAP per bucket for all features.

For each feature j and bucket k, computes: mean_shap_j^k = mean(shap_j(x_i) for all i where x_ij in bucket k)

If a bucket has zero observations in X_ref: - If model is provided: create n_synthetic observations by sampling real rows and setting the feature value to fall in the empty bucket, then compute SHAP on those synthetic observations. - If model is None: assign mean_shap = 0.0 with a warning.

PARAMETER	DESCRIPTION
bucket_sets	Dict of feature_name -> BucketSet (from build_all_buckets). TYPE: dict[str, BucketSet]
X_ref	Reference DataFrame (n_ref x p). TYPE: DataFrame
shap_values	SHAP values array of shape (n_ref, p). TYPE: ndarray
model	Trained model for computing SHAP on synthetic observations. If None, empty buckets get mean_shap = 0.0. TYPE: object or None DEFAULT: None
n_synthetic	Number of synthetic observations to create for empty buckets. TYPE: int DEFAULT: 10
rng	Random number generator for reproducibility. TYPE: Generator or None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, BucketSet]	Dict of feature_name -> BucketSet with mean_shap populated on each Bucket.

Source code in src/swift/normalization.py

def compute_bucket_shap(
    bucket_sets: dict[str, BucketSet],
    X_ref: pd.DataFrame,
    shap_values: np.ndarray,
    model: object | None = None,
    n_synthetic: int = 10,
    rng: np.random.Generator | None = None,
) -> dict[str, BucketSet]:
    """Compute mean SHAP per bucket for all features.

    For each feature j and bucket k, computes:
        mean_shap_j^k = mean(shap_j(x_i) for all i where x_ij in bucket k)

    If a bucket has zero observations in X_ref:
        - If model is provided: create n_synthetic observations by sampling
          real rows and setting the feature value to fall in the empty bucket,
          then compute SHAP on those synthetic observations.
        - If model is None: assign mean_shap = 0.0 with a warning.

    Parameters
    ----------
    bucket_sets : dict[str, BucketSet]
        Dict of feature_name -> BucketSet (from build_all_buckets).
    X_ref : pd.DataFrame
        Reference DataFrame (n_ref x p).
    shap_values : np.ndarray
        SHAP values array of shape (n_ref, p).
    model : object or None, default=None
        Trained model for computing SHAP on synthetic observations.
        If None, empty buckets get mean_shap = 0.0.
    n_synthetic : int, default=10
        Number of synthetic observations to create for empty buckets.
    rng : np.random.Generator or None, default=None
        Random number generator for reproducibility.

    Returns
    -------
    dict[str, BucketSet]
        Dict of feature_name -> BucketSet with mean_shap populated on each Bucket.
    """
    if rng is None:
        rng = np.random.default_rng(42)

    feature_names = list(bucket_sets.keys())
    shap_values = np.asarray(shap_values)

    result: dict[str, BucketSet] = {}

    for j, fname in enumerate(feature_names):
        bs = bucket_sets[fname]
        feature_col = X_ref[fname].values
        shap_col = shap_values[:, j]

        new_buckets = []
        for bucket in bs.buckets:
            # Find observations that fall into this bucket
            mask = _make_bucket_mask(feature_col, bucket)
            count = int(np.sum(mask))

            if count > 0:
                mean_shap = float(np.mean(shap_col[mask]))
                new_buckets.append(replace(bucket, mean_shap=mean_shap))
            else:
                # Empty bucket — try synthetic fill
                mean_shap = _fill_empty_bucket(
                    bucket, bs, fname, j, X_ref, model, n_synthetic, rng
                )
                new_buckets.append(replace(bucket, mean_shap=mean_shap))

        result[fname] = BucketSet(
            feature_name=bs.feature_name,
            buckets=tuple(new_buckets),
            decision_points=bs.decision_points,
        )

    return result

assign_bucket_indices

assign_bucket_indices(values: ndarray | Series, bucket_set: BucketSet) -> ndarray

Assign a bucket index to every value, vectorized.

Bucket layout: null bucket (index 0), then numeric buckets in interval order: bucket 1: (-inf, dp[0]) bucket 2: [dp[0], dp[1]) ... bucket k+1: [dp[k-1], inf)

np.searchsorted(dp, val, side='right') gives the number of decision points <= val, which maps to bucket index offset by 1 (since bucket 0 is null). NaN values map to the null bucket in O(n log k) time.

PARAMETER	DESCRIPTION
values	1-D array or Series of feature values (may contain NaN). TYPE: ndarray or Series
bucket_set	BucketSet with sorted decision_points. TYPE: BucketSet

RETURNS	DESCRIPTION
ndarray	1-D integer array of bucket indices (same length as input).

Source code in src/swift/normalization.py

def assign_bucket_indices(
    values: np.ndarray | pd.Series,
    bucket_set: BucketSet,
) -> np.ndarray:
    """Assign a bucket index to every value, vectorized.

    Bucket layout: null bucket (index 0), then numeric buckets in
    interval order:
        bucket 1: (-inf, dp[0])
        bucket 2: [dp[0], dp[1])
        ...
        bucket k+1: [dp[k-1], inf)

    np.searchsorted(dp, val, side='right') gives the number of decision
    points <= val, which maps to bucket index offset by 1 (since bucket
    0 is null). NaN values map to the null bucket in O(n log k) time.

    Parameters
    ----------
    values : np.ndarray or pd.Series
        1-D array or Series of feature values (may contain NaN).
    bucket_set : BucketSet
        BucketSet with sorted decision_points.

    Returns
    -------
    np.ndarray
        1-D integer array of bucket indices (same length as input).
    """
    values_arr = np.asarray(values, dtype=np.float64)
    n = len(values_arr)
    max_idx = max(b.index for b in bucket_set.buckets)

    # Identify null bucket
    null_bucket = None
    for b in bucket_set.buckets:
        if b.bucket_type == BucketType.NULL:
            null_bucket = b
            break

    nan_mask = np.isnan(values_arr)

    dp = bucket_set.decision_points
    if len(dp) > 0:
        # searchsorted with side='right' gives index i such that
        # dp[i-1] <= val < dp[i], which is bucket (i + 1) if null=0
        bucket_indices = np.searchsorted(dp, values_arr, side="right")
        # Offset by 1 because bucket 0 is the null bucket
        bucket_indices = bucket_indices + 1
    else:
        # No decision points: everything goes to bucket 1
        bucket_indices = np.ones(n, dtype=np.intp)

    # Handle NaN → null bucket
    if null_bucket is not None:
        bucket_indices[nan_mask] = null_bucket.index

    # Clip to valid range (safety)
    np.clip(bucket_indices, 0, max_idx, out=bucket_indices)

    return bucket_indices

transform_feature

transform_feature(values: ndarray | Series, bucket_set: BucketSet) -> ndarray

Map feature values to their bucket's mean SHAP value.

This is the SHAP transformation sigma_j defined in the paper: sigma_j(x_ij) = mean_shap_j^{bucket(x_ij)}

PARAMETER	DESCRIPTION
values	1-D array or Series of feature values (may contain NaN). TYPE: ndarray or Series
bucket_set	BucketSet with mean_shap populated on each bucket. TYPE: BucketSet

RETURNS	DESCRIPTION
ndarray	1-D numpy array of transformed values (same length as input).

Source code in src/swift/normalization.py

def transform_feature(
    values: np.ndarray | pd.Series,
    bucket_set: BucketSet,
) -> np.ndarray:
    """Map feature values to their bucket's mean SHAP value.

    This is the SHAP transformation sigma_j defined in the paper:
        sigma_j(x_ij) = mean_shap_j^{bucket(x_ij)}

    Parameters
    ----------
    values : np.ndarray or pd.Series
        1-D array or Series of feature values (may contain NaN).
    bucket_set : BucketSet
        BucketSet with mean_shap populated on each bucket.

    Returns
    -------
    np.ndarray
        1-D numpy array of transformed values (same length as input).
    """
    # Build a lookup array: bucket_index -> mean_shap
    max_idx = max(b.index for b in bucket_set.buckets)
    shap_lookup = np.zeros(max_idx + 1, dtype=np.float64)
    for b in bucket_set.buckets:
        shap_lookup[b.index] = b.mean_shap if b.mean_shap is not None else 0.0

    return shap_lookup[assign_bucket_indices(values, bucket_set)]

transform_frame

transform_frame(X: DataFrame, bucket_sets: dict[str, BucketSet]) -> dict[str, ndarray]

Apply transform_feature to every feature in bucket_sets.

PARAMETER	DESCRIPTION
X	DataFrame containing at least the features in bucket_sets. TYPE: DataFrame
bucket_sets	Dict of feature_name -> fitted BucketSet. TYPE: dict[str, BucketSet]

RETURNS	DESCRIPTION
dict[str, ndarray]	Dict of feature_name -> 1-D array of transformed values.

Source code in src/swift/normalization.py

def transform_frame(
    X: pd.DataFrame,
    bucket_sets: dict[str, BucketSet],
) -> dict[str, np.ndarray]:
    """Apply transform_feature to every feature in *bucket_sets*.

    Parameters
    ----------
    X : pd.DataFrame
        DataFrame containing at least the features in bucket_sets.
    bucket_sets : dict[str, BucketSet]
        Dict of feature_name -> fitted BucketSet.

    Returns
    -------
    dict[str, np.ndarray]
        Dict of feature_name -> 1-D array of transformed values.
    """
    return {
        fname: transform_feature(X[fname].values, bs)
        for fname, bs in bucket_sets.items()
    }

Distance

Stage 4 — Wasserstein distance on SHAP-transformed distributions.

distance

Stage 4: Wasserstein distance on SHAP-transformed distributions.

Computes the Wasserstein distance (W₁ or W₂) between two 1-D empirical distributions of SHAP-transformed feature values. The transformation σ_j is computed ONCE from D_ref (Stage 3) and applied identically to both reference and monitoring samples.

Functions: wasserstein_1d: W_p distance between two 1-D arrays. compute_swift_scores: Per-feature SWIFT scores for ref vs mon.

wasserstein_1d

wasserstein_1d(u: ndarray, v: ndarray, order: int = 1) -> float

Compute the p-th Wasserstein distance between two 1-D empirical distributions.

For order=1 this is the Earth Mover's Distance (L₁ area between CDFs). For order=2 this is the root of the integral of squared CDF differences.

Both are computed via the sorted-quantile formula: W_p^p = (1/N) Σ |F⁻¹_u(i/N) - F⁻¹_v(i/N)|^p with the step empirical quantile functions evaluated on the merged grid of both samples' CDF levels (grid spacings as weights), so that unequal sample sizes are handled correctly.

PARAMETER	DESCRIPTION
u	1-D array of samples from distribution P. TYPE: ndarray
v	1-D array of samples from distribution Q. TYPE: ndarray
order	Wasserstein order (1 or 2). TYPE: int DEFAULT: 1

RETURNS	DESCRIPTION
float	Non-negative float W_p(P, Q).

RAISES	DESCRIPTION
ValueError	If order is not 1 or 2, or arrays are empty.

Source code in src/swift/distance.py

def wasserstein_1d(
    u: np.ndarray,
    v: np.ndarray,
    order: int = 1,
) -> float:
    """Compute the p-th Wasserstein distance between two 1-D empirical distributions.

    For order=1 this is the Earth Mover's Distance (L₁ area between CDFs).
    For order=2 this is the root of the integral of squared CDF differences.

    Both are computed via the sorted-quantile formula:
        W_p^p = (1/N) Σ |F⁻¹_u(i/N) - F⁻¹_v(i/N)|^p
    with the step empirical quantile functions evaluated on the merged
    grid of both samples' CDF levels (grid spacings as weights), so that
    unequal sample sizes are handled correctly.

    Parameters
    ----------
    u : np.ndarray
        1-D array of samples from distribution P.
    v : np.ndarray
        1-D array of samples from distribution Q.
    order : int, default=1
        Wasserstein order (1 or 2).

    Returns
    -------
    float
        Non-negative float W_p(P, Q).

    Raises
    ------
    ValueError
        If order is not 1 or 2, or arrays are empty.
    """
    if order not in (1, 2):
        raise ValueError(f"order must be 1 or 2, got {order}")

    u = np.asarray(u, dtype=np.float64)
    v = np.asarray(v, dtype=np.float64)

    if u.size == 0 or v.size == 0:
        raise ValueError("Input arrays must not be empty.")

    u_sorted = np.sort(u)
    v_sorted = np.sort(v)

    if order == 1 and u.size == v.size:
        # Fast path: equal-size W₁ is just mean absolute difference of
        # sorted values (equivalent to scipy's implementation).
        return float(np.mean(np.abs(u_sorted - v_sorted)))

    # General case: evaluate step quantile functions on the merged CDF grid.
    # Build the combined set of quantile evaluation points.
    n_u = u.size
    n_v = v.size

    # CDF values at each sorted observation
    cdf_u = np.linspace(1.0 / n_u, 1.0, n_u)
    cdf_v = np.linspace(1.0 / n_v, 1.0, n_v)

    # Merge CDF grids and evaluate quantile functions at all grid points
    all_cdf = np.sort(np.concatenate([cdf_u, cdf_v]))
    all_cdf = np.unique(all_cdf)

    # Quantile function (inverse CDF): for a given probability p, the
    # quantile is the smallest x such that CDF(x) >= p.  With sorted
    # samples, np.searchsorted gives the correct index.
    q_u = u_sorted[np.clip(np.searchsorted(cdf_u, all_cdf, side="left"), 0, n_u - 1)]
    q_v = v_sorted[np.clip(np.searchsorted(cdf_v, all_cdf, side="left"), 0, n_v - 1)]

    # Compute spacing weights so we integrate over the [0,1] interval
    diffs = np.diff(np.concatenate([[0.0], all_cdf]))

    if order == 1:
        result = float(np.sum(np.abs(q_u - q_v) * diffs))
    else:  # order == 2
        result = float(np.sqrt(np.sum((q_u - q_v) ** 2 * diffs)))

    return result

compute_swift_scores

compute_swift_scores(X_ref: DataFrame, X_mon: DataFrame, bucket_sets: dict[str, BucketSet], order: int = 1) -> dict[str, float]

Compute per-feature SWIFT scores: W_p on SHAP-transformed distributions.

For each feature j: 1. Apply σ_j (from bucket_sets) to both ref and mon columns. 2. Compute W_p between the two transformed arrays.

The transformation σ_j was fitted on D_ref (Stage 3) and is not recomputed here — it is applied identically to both samples.

PARAMETER	DESCRIPTION
X_ref	Reference DataFrame (n_ref × p). TYPE: DataFrame
X_mon	Monitoring DataFrame (n_mon × p). TYPE: DataFrame
bucket_sets	Dict of feature_name → BucketSet with mean_shap already computed (output of compute_bucket_shap). TYPE: dict[str, BucketSet]
order	Wasserstein order (1 or 2). TYPE: int DEFAULT: 1

RETURNS	DESCRIPTION
dict[str, float]	Dict of feature_name → SWIFT score (non-negative float).

Source code in src/swift/distance.py

def compute_swift_scores(
    X_ref: pd.DataFrame,
    X_mon: pd.DataFrame,
    bucket_sets: dict[str, BucketSet],
    order: int = 1,
) -> dict[str, float]:
    """Compute per-feature SWIFT scores: W_p on SHAP-transformed distributions.

    For each feature j:
        1. Apply σ_j (from bucket_sets) to both ref and mon columns.
        2. Compute W_p between the two transformed arrays.

    The transformation σ_j was fitted on D_ref (Stage 3) and is *not*
    recomputed here — it is applied identically to both samples.

    Parameters
    ----------
    X_ref : pd.DataFrame
        Reference DataFrame (n_ref × p).
    X_mon : pd.DataFrame
        Monitoring DataFrame (n_mon × p).
    bucket_sets : dict[str, BucketSet]
        Dict of feature_name → BucketSet with mean_shap
        already computed (output of compute_bucket_shap).
    order : int, default=1
        Wasserstein order (1 or 2).

    Returns
    -------
    dict[str, float]
        Dict of feature_name → SWIFT score (non-negative float).
    """
    scores: dict[str, float] = {}

    # Apply the SAME σ_j (fitted on D_ref) to both samples.
    ref_transformed_all = transform_frame(X_ref, bucket_sets)
    mon_transformed_all = transform_frame(X_mon, bucket_sets)

    for fname in bucket_sets:
        ref_transformed = ref_transformed_all[fname]
        mon_transformed = mon_transformed_all[fname]

        # Compute Wasserstein distance
        score = wasserstein_1d(ref_transformed, mon_transformed, order=order)

        logger.debug(
            "Feature '%s': W%d = %.6f (ref_n=%d, mon_n=%d)",
            fname,
            order,
            score,
            len(ref_transformed),
            len(mon_transformed),
        )

        scores[fname] = score

    return scores

Threshold

Stage 5 — permutation test, bootstrap thresholds, and multiple testing correction.

threshold

Stage 5: Threshold calibration — permutation test, bootstrap, multiple testing correction.

Provides three functions: permutation_test: Per-feature p-values via permutation test. bootstrap_threshold: Per-feature absolute thresholds via bootstrap. correct_pvalues: Multiple testing correction (Bonferroni / BH).

Key invariant: the SHAP transformation σ_j is computed ONCE from D_ref (Stage 3) and reused for every permutation / bootstrap draw. SHAP is never recomputed inside the permutation loop.

permutation_test

permutation_test(X_ref: DataFrame, X_mon: DataFrame, bucket_sets: dict[str, BucketSet], order: int = 1, n_permutations: int = 1000, max_samples: int | None = None, rng: Generator | None = None) -> dict[str, float]

Compute per-feature p-values via permutation test.

Under H₀ (no drift), reference and monitoring samples are exchangeable. We pool them, draw random splits, and compute SWIFT scores under the null.

The pre-computed SHAP transformation σ_j (stored in bucket_sets) is applied identically to every permutation — it is NOT recomputed.

The p-value uses the conservative formula: p_j = (1 + #{b : SWIFT_j^(b) ≥ SWIFT_j^obs}) / (1 + B)

When max_samples is set and the pooled data exceeds it, both reference and monitoring data are randomly subsampled (preserving the ref/mon ratio) before running the permutation loop. This provides a significant speedup on large datasets with negligible impact on statistical power.

PARAMETER	DESCRIPTION
X_ref	Reference DataFrame (n_ref × p). TYPE: DataFrame
X_mon	Monitoring DataFrame (n_mon × p). TYPE: DataFrame
bucket_sets	Dict of feature → BucketSet with mean_shap populated. TYPE: dict[str, BucketSet]
order	Wasserstein order (1 or 2). TYPE: int DEFAULT: 1
n_permutations	Number of permutations B. TYPE: int DEFAULT: 1000
max_samples	Maximum total pool size. If the pool (n_ref + n_mon) exceeds this, subsample proportionally. None = no limit. TYPE: int or None DEFAULT: None
rng	Random number generator for reproducibility. TYPE: Generator or None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, float]	Dict of feature_name → p-value ∈ (0, 1].

Source code in src/swift/threshold.py

def permutation_test(
    X_ref: pd.DataFrame,
    X_mon: pd.DataFrame,
    bucket_sets: dict[str, BucketSet],
    order: int = 1,
    n_permutations: int = 1000,
    max_samples: int | None = None,
    rng: np.random.Generator | None = None,
) -> dict[str, float]:
    """Compute per-feature p-values via permutation test.

    Under H₀ (no drift), reference and monitoring samples are
    exchangeable.  We pool them, draw random splits, and compute
    SWIFT scores under the null.

    The pre-computed SHAP transformation σ_j (stored in bucket_sets)
    is applied identically to every permutation — it is NOT recomputed.

    The p-value uses the conservative formula:
        p_j = (1 + #{b : SWIFT_j^(b) ≥ SWIFT_j^obs}) / (1 + B)

    When ``max_samples`` is set and the pooled data exceeds it, both
    reference and monitoring data are randomly subsampled (preserving
    the ref/mon ratio) before running the permutation loop.  This
    provides a significant speedup on large datasets with negligible
    impact on statistical power.

    Parameters
    ----------
    X_ref : pd.DataFrame
        Reference DataFrame (n_ref × p).
    X_mon : pd.DataFrame
        Monitoring DataFrame (n_mon × p).
    bucket_sets : dict[str, BucketSet]
        Dict of feature → BucketSet with mean_shap populated.
    order : int, default=1
        Wasserstein order (1 or 2).
    n_permutations : int, default=1000
        Number of permutations B.
    max_samples : int or None, default=None
        Maximum total pool size.  If the pool (n_ref + n_mon)
        exceeds this, subsample proportionally.  None = no limit.
    rng : np.random.Generator or None, default=None
        Random number generator for reproducibility.

    Returns
    -------
    dict[str, float]
        Dict of feature_name → p-value ∈ (0, 1].
    """
    if rng is None:
        rng = np.random.default_rng(42)

    n_ref = len(X_ref)
    n_mon = len(X_mon)
    n_pool = n_ref + n_mon

    # ── Optional subsampling for performance ─────────────────────────
    if max_samples is not None and n_pool > max_samples:
        ref_ratio = n_ref / n_pool
        n_ref_sub = max(1, int(max_samples * ref_ratio))
        n_mon_sub = max(1, max_samples - n_ref_sub)

        ref_idx = rng.choice(n_ref, size=n_ref_sub, replace=False)
        mon_idx = rng.choice(n_mon, size=n_mon_sub, replace=False)

        X_ref = X_ref.iloc[ref_idx].reset_index(drop=True)
        X_mon = X_mon.iloc[mon_idx].reset_index(drop=True)

        logger.info(
            "Permutation test: subsampled pool from %d to %d "
            "(ref: %d→%d, mon: %d→%d)",
            n_pool, n_ref_sub + n_mon_sub,
            n_ref, n_ref_sub, n_mon, n_mon_sub,
        )

        n_ref = n_ref_sub
        n_mon = n_mon_sub
        n_pool = n_ref + n_mon

    feature_names = list(bucket_sets.keys())

    # ── Observed SWIFT scores ──────────────────────────────────────────
    observed_scores = compute_swift_scores(X_ref, X_mon, bucket_sets, order=order)

    # ── Pre-transform the pooled data per feature ─────────────────────
    # This avoids redundant transform_feature calls inside the loop.
    X_pool = pd.concat([X_ref, X_mon], ignore_index=True)
    pool_transformed = transform_frame(X_pool, bucket_sets)

    # ── Permutation loop ──────────────────────────────────────────────
    count_ge: dict[str, int] = {fname: 0 for fname in feature_names}

    for _ in range(n_permutations):
        # Random split of pooled indices
        perm = rng.permutation(n_pool)
        idx_ref = perm[:n_ref]
        idx_mon = perm[n_ref:]

        for fname in feature_names:
            ref_t = pool_transformed[fname][idx_ref]
            mon_t = pool_transformed[fname][idx_mon]
            perm_score = wasserstein_1d(ref_t, mon_t, order=order)

            if perm_score >= observed_scores[fname]:
                count_ge[fname] += 1

    # ── p-values: (1 + count) / (1 + B) ──────────────────────────────
    pvalues: dict[str, float] = {}
    for fname in feature_names:
        pvalues[fname] = (1 + count_ge[fname]) / (1 + n_permutations)

    logger.info(
        "Permutation test (B=%d, order=%d): p-values = %s",
        n_permutations,
        order,
        {k: f"{v:.4f}" for k, v in pvalues.items()},
    )

    return pvalues

bootstrap_threshold

bootstrap_threshold(X_ref: DataFrame, bucket_sets: dict[str, BucketSet], n_mon: int, order: int = 1, alpha: float = 0.05, n_bootstrap: int = 1000, rng: Generator | None = None) -> dict[str, float]

Compute per-feature absolute thresholds via bootstrap.

Draws bootstrap samples of size n_mon from X_ref, computes SWIFT scores against X_ref, and returns the (1 − α) quantile as the threshold for each feature.

This provides a principled alternative to PSI's ad-hoc 0.10 / 0.25.

PARAMETER	DESCRIPTION
X_ref	Reference DataFrame (n_ref × p). TYPE: DataFrame
bucket_sets	Dict of feature → BucketSet with mean_shap populated. TYPE: dict[str, BucketSet]
n_mon	Size of monitoring sample (bootstrap sample size). TYPE: int
order	Wasserstein order (1 or 2). TYPE: int DEFAULT: 1
alpha	Significance level (threshold is (1-α) quantile). TYPE: float DEFAULT: 0.05
n_bootstrap	Number of bootstrap iterations B. TYPE: int DEFAULT: 1000
rng	Random number generator for reproducibility. TYPE: Generator or None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, float]	Dict of feature_name → threshold (non-negative float).

Source code in src/swift/threshold.py

def bootstrap_threshold(
    X_ref: pd.DataFrame,
    bucket_sets: dict[str, BucketSet],
    n_mon: int,
    order: int = 1,
    alpha: float = 0.05,
    n_bootstrap: int = 1000,
    rng: np.random.Generator | None = None,
) -> dict[str, float]:
    """Compute per-feature absolute thresholds via bootstrap.

    Draws bootstrap samples of size n_mon from X_ref, computes SWIFT
    scores against X_ref, and returns the (1 − α) quantile as the
    threshold for each feature.

    This provides a principled alternative to PSI's ad-hoc 0.10 / 0.25.

    Parameters
    ----------
    X_ref : pd.DataFrame
        Reference DataFrame (n_ref × p).
    bucket_sets : dict[str, BucketSet]
        Dict of feature → BucketSet with mean_shap populated.
    n_mon : int
        Size of monitoring sample (bootstrap sample size).
    order : int, default=1
        Wasserstein order (1 or 2).
    alpha : float, default=0.05
        Significance level (threshold is (1-α) quantile).
    n_bootstrap : int, default=1000
        Number of bootstrap iterations B.
    rng : np.random.Generator or None, default=None
        Random number generator for reproducibility.

    Returns
    -------
    dict[str, float]
        Dict of feature_name → threshold (non-negative float).
    """
    if rng is None:
        rng = np.random.default_rng(42)

    n_ref = len(X_ref)
    feature_names = list(bucket_sets.keys())

    # Pre-transform reference data
    ref_transformed = transform_frame(X_ref, bucket_sets)

    # Bootstrap loop
    boot_scores: dict[str, list[float]] = {fname: [] for fname in feature_names}

    for _ in range(n_bootstrap):
        # Draw bootstrap sample indices from reference
        boot_idx = rng.integers(0, n_ref, size=n_mon)

        for fname in feature_names:
            boot_t = ref_transformed[fname][boot_idx]
            score = wasserstein_1d(ref_transformed[fname], boot_t, order=order)
            boot_scores[fname].append(score)

    # Threshold at (1 - alpha) quantile
    quantile = 1.0 - alpha
    thresholds: dict[str, float] = {}
    for fname in feature_names:
        thresholds[fname] = float(np.quantile(boot_scores[fname], quantile))

    logger.info(
        "Bootstrap thresholds (B=%d, α=%.3f, order=%d): %s",
        n_bootstrap,
        alpha,
        order,
        {k: f"{v:.6f}" for k, v in thresholds.items()},
    )

    return thresholds

correct_pvalues

correct_pvalues(pvalues: dict[str, float], method: CorrectionMethod, alpha: float = 0.05) -> dict[str, bool]

Apply multiple testing correction and return rejection decisions.

Bonferroni: reject if p_j < α / p (controls FWER). Benjamini-Hochberg: sort p-values, find largest k s.t. p_(k) ≤ k·α / p, then reject all with rank ≤ k (controls FDR).

Bonferroni uses strict inequality (p < threshold); BH uses p ≤ threshold.

PARAMETER	DESCRIPTION
pvalues	Dict of feature_name → p-value. TYPE: dict[str, float]
method	CorrectionMethod.BONFERRONI or CorrectionMethod.BH. TYPE: CorrectionMethod
alpha	Significance level. TYPE: float DEFAULT: 0.05

RETURNS	DESCRIPTION
dict[str, bool]	Dict of feature_name → bool (True = reject H₀ / flagged as drifted).

Source code in src/swift/threshold.py

def correct_pvalues(
    pvalues: dict[str, float],
    method: CorrectionMethod,
    alpha: float = 0.05,
) -> dict[str, bool]:
    """Apply multiple testing correction and return rejection decisions.

    Bonferroni: reject if p_j < α / p  (controls FWER).
    Benjamini-Hochberg: sort p-values, find largest k s.t.
        p_(k) ≤ k·α / p, then reject all with rank ≤ k  (controls FDR).

    Bonferroni uses strict inequality (p < threshold); BH uses p ≤ threshold.

    Parameters
    ----------
    pvalues : dict[str, float]
        Dict of feature_name → p-value.
    method : CorrectionMethod
        CorrectionMethod.BONFERRONI or CorrectionMethod.BH.
    alpha : float, default=0.05
        Significance level.

    Returns
    -------
    dict[str, bool]
        Dict of feature_name → bool (True = reject H₀ / flagged as drifted).
    """
    p = len(pvalues)
    if p == 0:
        return {}

    if method == CorrectionMethod.BONFERRONI:
        threshold = alpha / p
        return {fname: pval < threshold for fname, pval in pvalues.items()}

    if method == CorrectionMethod.BH:
        # Sort p-values ascending
        sorted_items = sorted(pvalues.items(), key=lambda x: x[1])
        sorted_names = [name for name, _ in sorted_items]
        sorted_pvals = [pval for _, pval in sorted_items]

        # Find largest k such that p_(k) <= k * alpha / p
        max_k = 0
        for k_idx in range(p):
            rank = k_idx + 1  # 1-indexed
            bh_threshold = rank * alpha / p
            if sorted_pvals[k_idx] <= bh_threshold:
                max_k = rank

        # Reject all features with rank <= max_k
        rejected_names = set(sorted_names[:max_k])
        return {fname: fname in rejected_names for fname in pvalues}

    raise ValueError(f"Unknown correction method: {method}")

Aggregation

Model-level score aggregation (SWIFT_max, SWIFT_mean, SWIFT_weighted).

aggregation

Stage 6: Aggregation — feature-level to model-level SWIFT scores.

Provides three aggregation strategies from the paper (§3.6): - Maximum SWIFT: most-drifted feature (SWIFT_max) - Mean SWIFT: average feature drift (SWIFT_mean) - Weighted SWIFT: importance-weighted (SWIFT_weighted)

Functions: aggregate_scores: Compute model-level summary statistics. compute_importance_weights: Normalized mean |SHAP| weights w_j.

AggregatedScores dataclass

AggregatedScores(swift_max: float, swift_mean: float, max_feature: str, swift_weighted: float | None = None)

Model-level SWIFT aggregation.

ATTRIBUTE	DESCRIPTION
swift_max	Maximum per-feature SWIFT score. TYPE: float
swift_mean	Unweighted mean of per-feature scores. TYPE: float
swift_weighted	Importance-weighted score (None if no weights). TYPE: float or None
max_feature	Name of the feature with the highest score. TYPE: str

aggregate_scores

aggregate_scores(scores: dict[str, float], weights: dict[str, float] | None = None) -> AggregatedScores

Aggregate per-feature SWIFT scores to model level.

PARAMETER	DESCRIPTION
scores	Dict of feature_name → SWIFT score. TYPE: dict[str, float]
weights	Optional dict of feature_name → importance weight. If provided, weights should sum to 1 (or be normalizable). If None, swift_weighted is set to None. TYPE: dict[str, float] or None DEFAULT: None

RETURNS	DESCRIPTION
AggregatedScores	AggregatedScores with max, mean, and optionally weighted score.

Source code in src/swift/aggregation.py

def aggregate_scores(
    scores: dict[str, float],
    weights: dict[str, float] | None = None,
) -> AggregatedScores:
    """Aggregate per-feature SWIFT scores to model level.

    Parameters
    ----------
    scores : dict[str, float]
        Dict of feature_name → SWIFT score.
    weights : dict[str, float] or None, default=None
        Optional dict of feature_name → importance weight.
        If provided, weights should sum to 1 (or be normalizable).
        If None, swift_weighted is set to None.

    Returns
    -------
    AggregatedScores
        AggregatedScores with max, mean, and optionally weighted score.
    """
    if not scores:
        raise ValueError("scores dict must not be empty.")

    names = list(scores.keys())
    vals = np.array([scores[n] for n in names])

    swift_max = float(np.max(vals))
    swift_mean = float(np.mean(vals))
    max_feature = names[int(np.argmax(vals))]

    swift_weighted: float | None = None
    if weights is not None:
        weighted_sum = sum(weights[n] * scores[n] for n in names)
        swift_weighted = float(weighted_sum)

    return AggregatedScores(
        swift_max=swift_max,
        swift_mean=swift_mean,
        max_feature=max_feature,
        swift_weighted=swift_weighted,
    )

compute_importance_weights

compute_importance_weights(shap_values: ndarray, feature_names: list[str]) -> dict[str, float]

Compute normalized mean absolute SHAP importance weights.

w_j = |φ̄_j| / Σ_k |φ̄_k|

where φ̄_j = mean(|SHAP_j(x_i)|) across all reference observations.

This generalizes SSI's IV-based weights with model-aware SHAP-based feature importance.

PARAMETER	DESCRIPTION
shap_values	SHAP values array of shape (n, p). TYPE: ndarray
feature_names	List of p feature names. TYPE: list[str]

RETURNS	DESCRIPTION
dict[str, float]	Dict of feature_name → weight (non-negative, sums to 1).

Source code in src/swift/aggregation.py

def compute_importance_weights(
    shap_values: np.ndarray,
    feature_names: list[str],
) -> dict[str, float]:
    """Compute normalized mean absolute SHAP importance weights.

    w_j = |φ̄_j| / Σ_k |φ̄_k|

    where φ̄_j = mean(|SHAP_j(x_i)|) across all reference observations.

    This generalizes SSI's IV-based weights with model-aware
    SHAP-based feature importance.

    Parameters
    ----------
    shap_values : np.ndarray
        SHAP values array of shape (n, p).
    feature_names : list[str]
        List of p feature names.

    Returns
    -------
    dict[str, float]
        Dict of feature_name → weight (non-negative, sums to 1).
    """
    shap_values = np.asarray(shap_values)
    mean_abs_shap = np.mean(np.abs(shap_values), axis=0)

    total = np.sum(mean_abs_shap)
    if total == 0.0:
        # Uniform weights if all SHAP values are zero
        p = len(feature_names)
        return {name: 1.0 / p for name in feature_names}

    weights = mean_abs_shap / total
    return {name: float(weights[j]) for j, name in enumerate(feature_names)}

Plotting

Visualization functions behind plot_buckets and plot_swift_scores.

plotting

Visualization utilities for SWIFT monitoring results.

Provides two public plotting functions:

• plot_bucket_profile — SHAP response curve + density per bucket for a feature.
• plot_feature_swift_scores — Bar chart of SWIFT scores per feature.

Both return (Figure, Axes) tuples for further customization.

plot_bucket_profile

plot_bucket_profile(bucket_set: BucketSet, feature_values: ndarray, shap_values: ndarray, compare_values: ndarray | None = None, primary_values: ndarray | None = None, labels: tuple[str, str] = ('Reference', 'Comparison'), figsize: tuple[float, float] = (10, 5), title: str | None = None, max_label_buckets: int = 20, x_axis: str = 'bucket') -> tuple[Figure, Axes]

Plot the bucketing profile for a single feature.

Shows mean SHAP per bucket (line + error band) and observation density (filled line). Optionally overlays a comparison sample's density.

PARAMETER	DESCRIPTION
bucket_set	Fitted bucket set for the feature. TYPE: BucketSet
feature_values	Raw feature values from the reference sample (used for the SHAP curve and, when primary_values is None, for the primary density). TYPE: ndarray
shap_values	SHAP values for this feature on the reference sample. TYPE: ndarray
compare_values	Raw feature values from a comparison sample. If provided, shows a second density line. TYPE: ndarray or None DEFAULT: None
primary_values	If provided, these values are used for the primary density instead of feature_values. Useful for showing density of an arbitrary sample while the SHAP curve stays anchored to the reference. TYPE: ndarray or None DEFAULT: None
labels	Legend labels for (primary, comparison) densities. TYPE: tuple of str DEFAULT: ('Reference', 'Comparison')
figsize	Figure size. TYPE: tuple DEFAULT: (10, 5)
title	Custom title. Defaults to "Bucketing Profile: {feature_name}". TYPE: str or None DEFAULT: None
max_label_buckets	Switch from interval notation to bucket indices if exceeded. TYPE: int DEFAULT: 20
x_axis	"bucket" (default) uses integer bucket indices on the x-axis. "natural" uses actual feature-value positions (bucket midpoints, data min/max for edge buckets, NULL placed at the left with a gap). TYPE: ('bucket', 'natural') DEFAULT: "bucket"

RETURNS	DESCRIPTION
(Figure, Axes)

Source code in src/swift/plotting/bucket_profile.py

def plot_bucket_profile(
    bucket_set: BucketSet,
    feature_values: np.ndarray,
    shap_values: np.ndarray,
    compare_values: np.ndarray | None = None,
    primary_values: np.ndarray | None = None,
    labels: tuple[str, str] = ("Reference", "Comparison"),
    figsize: tuple[float, float] = (10, 5),
    title: str | None = None,
    max_label_buckets: int = 20,
    x_axis: str = "bucket",
) -> tuple[Figure, Axes]:
    """Plot the bucketing profile for a single feature.

    Shows mean SHAP per bucket (line + error band) and observation density
    (filled line).  Optionally overlays a comparison sample's density.

    Parameters
    ----------
    bucket_set : BucketSet
        Fitted bucket set for the feature.
    feature_values : np.ndarray
        Raw feature values from the reference sample (used for the SHAP
        curve and, when *primary_values* is ``None``, for the primary
        density).
    shap_values : np.ndarray
        SHAP values for this feature on the reference sample.
    compare_values : np.ndarray or None
        Raw feature values from a comparison sample.  If provided, shows
        a second density line.
    primary_values : np.ndarray or None
        If provided, these values are used for the primary density
        instead of *feature_values*.  Useful for showing density of an
        arbitrary sample while the SHAP curve stays anchored to the
        reference.
    labels : tuple of str
        Legend labels for (primary, comparison) densities.
    figsize : tuple, default (10, 5)
        Figure size.
    title : str or None
        Custom title.  Defaults to ``"Bucketing Profile: {feature_name}"``.
    max_label_buckets : int, default 20
        Switch from interval notation to bucket indices if exceeded.
    x_axis : {"bucket", "natural"}
        ``"bucket"`` (default) uses integer bucket indices on the x-axis.
        ``"natural"`` uses actual feature-value positions (bucket
        midpoints, data min/max for edge buckets, NULL placed at the
        left with a gap).

    Returns
    -------
    (Figure, Axes)
    """
    plt.style.use("seaborn-v0_8-whitegrid")
    palette = _PALETTE

    # Determine which values to use for primary density
    density_values = primary_values if primary_values is not None else feature_values

    stats = _compute_bucket_stats(bucket_set, feature_values, shap_values)
    primary_densities = _compute_sample_densities(bucket_set, density_values)
    bucket_labels = _format_bucket_labels(bucket_set, max_label_buckets)
    num_buckets = bucket_set.num_buckets

    # -- X positions --
    use_natural = x_axis == "natural"
    if use_natural:
        x_positions = _compute_natural_x_positions(bucket_set, feature_values)
    else:
        x_positions = np.arange(num_buckets, dtype=float)

    fig, ax_shap = plt.subplots(figsize=figsize)
    ax_density = ax_shap.twinx()

    # -- SHAP = 0 reference line (visible behind everything) --
    ax_shap.axhline(
        y=0,
        color="#888888",
        linewidth=1.4,
        linestyle="--",
        alpha=0.7,
        zorder=1.5,
        label="_nolegend_",
    )

    # -- Density lines + filled area (right y-axis) --
    comparison_mode = compare_values is not None

    # Primary density
    ax_density.plot(
        x_positions,
        primary_densities,
        color=palette[0],
        linewidth=1.5,
        alpha=0.8,
        zorder=1,
        label=labels[0] if comparison_mode else "Observation density",
    )
    ax_density.fill_between(
        x_positions,
        0,
        primary_densities,
        color=palette[0],
        alpha=0.15,
        zorder=0.9,
    )

    if comparison_mode:
        compare_densities = _compute_sample_densities(
            bucket_set, compare_values,
        )
        ax_density.plot(
            x_positions,
            compare_densities,
            color=palette[1],
            linewidth=1.5,
            alpha=0.8,
            zorder=1,
            label=labels[1],
        )
        ax_density.fill_between(
            x_positions,
            0,
            compare_densities,
            color=palette[1],
            alpha=0.15,
            zorder=0.9,
        )

    ax_density.set_ylabel("Density")

    # -- SHAP line (left y-axis) --
    shap_color = palette[2]
    mean_shaps = stats["mean_shaps"]
    shap_stds = stats["shap_stds"]

    ax_shap.plot(
        x_positions,
        mean_shaps,
        color=shap_color,
        marker="o",
        linewidth=2,
        markersize=6,
        zorder=3,
        label="Mean SHAP \u00b1 2 std",
    )

    # Error band (only where count > 0)
    upper = mean_shaps + 2 * shap_stds
    lower = mean_shaps - 2 * shap_stds
    mask_nonzero = stats["counts"] > 0
    ax_shap.fill_between(
        x_positions,
        lower,
        upper,
        where=mask_nonzero,
        alpha=0.2,
        color=shap_color,
        zorder=2,
    )

    ax_shap.set_ylabel("SHAP Value")

    # -- X-axis --
    if use_natural:
        # Natural axis: let matplotlib auto-format, but annotate NULL
        ax_shap.set_xlabel(bucket_set.feature_name)

        # Add a vertical dotted line to separate NULL from numeric range
        if num_buckets > 1:
            sep_x = (x_positions[0] + x_positions[1]) / 2
            ax_shap.axvline(
                sep_x,
                color="#aaaaaa",
                linewidth=1.0,
                linestyle=":",
                alpha=0.6,
                zorder=0.5,
            )
            # Label the NULL point
            ax_shap.annotate(
                "NULL",
                xy=(x_positions[0], 0),
                xytext=(x_positions[0], 0),
                fontsize=8,
                ha="center",
                va="bottom",
                color="#666666",
            )

        # Use auto-ticks for the numeric part, but add NULL tick
        # We set minor ticks off and let matplotlib handle major ticks
        ax_shap.xaxis.set_major_locator(mticker.AutoLocator())
    else:
        # Bucket-index mode: explicit tick labels
        ax_shap.set_xlabel("Bucket")
        ax_shap.set_xticks(x_positions)

        # Adaptive label formatting to avoid overlap
        if num_buckets > 10:
            fontsize = 7
            rotation = 90
        elif num_buckets > 3:
            fontsize = 8
            rotation = 60
        else:
            fontsize = 9
            rotation = 0

        ax_shap.set_xticklabels(
            bucket_labels,
            rotation=rotation,
            ha="right" if rotation > 0 else "center",
            fontsize=fontsize,
        )

    # -- Title --
    plot_title = title or f"Bucketing Profile: {bucket_set.feature_name}"
    ax_shap.set_title(plot_title)

    # -- Legend (combine both axes) --
    lines_shap, labels_shap = ax_shap.get_legend_handles_labels()
    lines_density, labels_density = ax_density.get_legend_handles_labels()
    ax_shap.legend(
        lines_shap + lines_density,
        labels_shap + labels_density,
        loc="upper right",
    )

    # Ensure SHAP line is drawn on top of density
    ax_shap.set_zorder(ax_density.get_zorder() + 1)
    ax_shap.patch.set_visible(False)

    fig.tight_layout()
    return fig, ax_shap

plot_feature_swift_scores

plot_feature_swift_scores(result: SWIFTResult, result_compare: SWIFTResult | None = None, labels: tuple[str, str] = ('Result A', 'Result B'), threshold: float | None = None, sort_by: str = 'score', feature_order: list[str] | None = None, figsize: tuple[float, float] = (12, 5), title: str | None = None) -> tuple[Figure, Axes]

Plot SWIFT scores per feature as a bar chart with reference lines.

Optionally compare two SWIFTResult objects side by side.

PARAMETER	DESCRIPTION
result	Primary result from SWIFTMonitor.test(). TYPE: SWIFTResult
result_compare	Optional second result for side-by-side comparison. TYPE: SWIFTResult or None DEFAULT: None
labels	Legend labels for the two results in comparison mode. TYPE: tuple of str DEFAULT: ('Result A', 'Result B')
threshold	Optional detection threshold horizontal line. TYPE: float or None DEFAULT: None
sort_by	Feature ordering on x-axis. "original" uses feature_order. TYPE: ('score', 'name', 'original') DEFAULT: "score"
feature_order	Original feature order (from monitor.feature_names_in_). Required when sort_by="original". TYPE: list[str] or None DEFAULT: None
figsize	Figure size. TYPE: tuple DEFAULT: (12, 5)
title	Custom title. TYPE: str or None DEFAULT: None

RETURNS	DESCRIPTION
(Figure, Axes)

Source code in src/swift/plotting/swift_scores.py

def plot_feature_swift_scores(
    result: SWIFTResult,
    result_compare: SWIFTResult | None = None,
    labels: tuple[str, str] = ("Result A", "Result B"),
    threshold: float | None = None,
    sort_by: str = "score",
    feature_order: list[str] | None = None,
    figsize: tuple[float, float] = (12, 5),
    title: str | None = None,
) -> tuple[Figure, Axes]:
    """Plot SWIFT scores per feature as a bar chart with reference lines.

    Optionally compare two ``SWIFTResult`` objects side by side.

    Parameters
    ----------
    result : SWIFTResult
        Primary result from ``SWIFTMonitor.test()``.
    result_compare : SWIFTResult or None
        Optional second result for side-by-side comparison.
    labels : tuple of str
        Legend labels for the two results in comparison mode.
    threshold : float or None
        Optional detection threshold horizontal line.
    sort_by : {"score", "name", "original"}
        Feature ordering on x-axis.  ``"original"`` uses *feature_order*.
    feature_order : list[str] or None
        Original feature order (from ``monitor.feature_names_in_``).
        Required when ``sort_by="original"``.
    figsize : tuple, default (12, 5)
        Figure size.
    title : str or None
        Custom title.

    Returns
    -------
    (Figure, Axes)
    """
    plt.style.use("seaborn-v0_8-whitegrid")

    comparison_mode = result_compare is not None

    # Build dicts: feature_name -> score / is_drifted
    scores_a = {fr.feature_name: fr.swift_score for fr in result.feature_results}
    drifted_a = {fr.feature_name: fr.is_drifted for fr in result.feature_results}

    if comparison_mode:
        scores_b = {
            fr.feature_name: fr.swift_score
            for fr in result_compare.feature_results
        }

    # Determine feature order
    feature_names = list(scores_a.keys())
    if sort_by == "score":
        feature_names = sorted(feature_names, key=lambda f: scores_a[f], reverse=True)
    elif sort_by == "name":
        feature_names = sorted(feature_names)
    elif sort_by == "original" and feature_order is not None:
        feature_names = [f for f in feature_order if f in scores_a]
    # else: keep dict order

    n_features = len(feature_names)
    x_positions = np.arange(n_features)

    fig, ax = plt.subplots(figsize=figsize)

    if comparison_mode:
        # -- Comparison mode: grouped bars --
        palette = _PALETTE
        bar_width = 0.35

        vals_a = [scores_a[f] for f in feature_names]
        vals_b = [scores_b.get(f, 0.0) for f in feature_names]

        ax.bar(
            x_positions - bar_width / 2,
            vals_a,
            width=bar_width,
            color=palette[0],
            edgecolor=_darken(palette[0]),
            label=labels[0],
        )
        ax.bar(
            x_positions + bar_width / 2,
            vals_b,
            width=bar_width,
            color=palette[1],
            edgecolor=_darken(palette[1]),
            label=labels[1],
        )

        # Only threshold line in comparison mode
        if threshold is not None:
            ax.axhline(
                threshold,
                linestyle=":",
                color="black",
                linewidth=1.5,
                label=f"Threshold = {threshold:.4f}",
            )

        plot_title = title or "SWIFT Scores Comparison"

    else:
        # -- Single result mode: drift-colored bars --
        color_drifted = "#e74c3c"
        color_ok = "#3498db"
        edge_drifted = _darken(color_drifted)
        edge_ok = _darken(color_ok)

        vals = [scores_a[f] for f in feature_names]
        colors = [
            color_drifted if drifted_a.get(f) else color_ok
            for f in feature_names
        ]
        edges = [
            edge_drifted if drifted_a.get(f) else edge_ok
            for f in feature_names
        ]

        ax.bar(
            x_positions,
            vals,
            color=colors,
            edgecolor=edges,
            width=0.6,
        )

        # Proxy artists for legend
        from matplotlib.patches import Patch

        legend_elements = [
            Patch(facecolor=color_drifted, edgecolor=edge_drifted, label="Drifted"),
            Patch(facecolor=color_ok, edgecolor=edge_ok, label="Not drifted"),
        ]

        # Horizontal lines
        ax.axhline(
            result.swift_max,
            linestyle="--",
            color=color_drifted,
            linewidth=1.2,
            label=f"SWIFT max = {result.swift_max:.4f}",
        )
        ax.axhline(
            result.swift_mean,
            linestyle="--",
            color=color_ok,
            linewidth=1.2,
            label=f"SWIFT mean = {result.swift_mean:.4f}",
        )

        if threshold is not None:
            ax.axhline(
                threshold,
                linestyle=":",
                color="black",
                linewidth=1.5,
                label=f"Threshold = {threshold:.4f}",
            )

        legend_elements.extend(ax.get_legend_handles_labels()[0])
        # Reset and rebuild legend
        ax.legend(handles=legend_elements, loc="upper right")

        plot_title = title or "SWIFT Scores per Feature"

    # -- Axes --
    ax.set_xticks(x_positions)
    ax.set_xticklabels(
        feature_names,
        rotation=45 if n_features > 5 else 0,
        ha="right" if n_features > 5 else "center",
    )
    ax.set_ylabel("SWIFT Score")
    ax.set_title(plot_title)

    if comparison_mode:
        ax.legend(loc="upper right")

    fig.tight_layout()
    return fig, ax