Getting Started with SWIFT Monitoring¶

SWIFT (SHAP-Weighted Impact Feature Testing) is a model-aware distribution monitoring framework that detects feature drift by comparing SHAP-transformed distributions between reference and monitoring data.

Unlike traditional drift detection methods that treat features independently of the model, SWIFT weights distribution changes by their impact on model predictions — so it flags the shifts that actually matter.

This notebook walks through:

1. Fitting a SWIFT monitor on reference data
2. Testing for drift in monitoring data
3. Injecting synthetic drift and detecting it
4. Visualizing bucket profiles and SWIFT scores
5. Comparing drifted vs. clean samples side-by-side
6. Sample-vs-sample comparison (without reference)
7. Advanced usage (hyperparameters, W2, sklearn integration)

1. Setup¶

Install the package (if not already installed):

uv pip install -e ".[dev]"

In [1]:

import warnings

import lightgbm as lgb
import numpy as np
import pandas as pd
from sklearn.datasets import make_classification

from swift import SWIFTMonitor

warnings.filterwarnings("ignore", category=UserWarning)

%matplotlib inline

import warnings import lightgbm as lgb import numpy as np import pandas as pd from sklearn.datasets import make_classification from swift import SWIFTMonitor warnings.filterwarnings("ignore", category=UserWarning) %matplotlib inline

/home/danielwlazlo/Code/swift/.venv/lib/python3.11/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html
  from .autonotebook import tqdm as notebook_tqdm

Generate synthetic data and train a model¶

In [2]:

# Generate a binary classification dataset
X, y = make_classification(
    n_samples=5000,
    n_features=10,
    n_informative=6,
    n_redundant=2,
    n_clusters_per_class=2,
    random_state=42,
)
feature_names = [f"feature_{i}" for i in range(X.shape[1])]
X = pd.DataFrame(X, columns=feature_names)

# Split into train / reference / monitoring
X_train, y_train = X.iloc[:3000], y[:3000]
X_ref = X.iloc[3000:4000].reset_index(drop=True)
X_mon = X.iloc[4000:].reset_index(drop=True)

print(f"Train: {X_train.shape}, Reference: {X_ref.shape}, Monitor: {X_mon.shape}")

# Generate a binary classification dataset X, y = make_classification( n_samples=5000, n_features=10, n_informative=6, n_redundant=2, n_clusters_per_class=2, random_state=42, ) feature_names = [f"feature_{i}" for i in range(X.shape[1])] X = pd.DataFrame(X, columns=feature_names) # Split into train / reference / monitoring X_train, y_train = X.iloc[:3000], y[:3000] X_ref = X.iloc[3000:4000].reset_index(drop=True) X_mon = X.iloc[4000:].reset_index(drop=True) print(f"Train: {X_train.shape}, Reference: {X_ref.shape}, Monitor: {X_mon.shape}")

Train: (3000, 10), Reference: (1000, 10), Monitor: (1000, 10)

In [3]:

# Train a LightGBM model
dtrain = lgb.Dataset(X_train, label=y_train)
params = {"objective": "binary", "verbose": -1, "num_leaves": 31, "n_estimators": 100}
model = lgb.train(params, dtrain, num_boost_round=100)

print(f"Model trained with {model.num_trees()} trees")

# Train a LightGBM model dtrain = lgb.Dataset(X_train, label=y_train) params = {"objective": "binary", "verbose": -1, "num_leaves": 31, "n_estimators": 100} model = lgb.train(params, dtrain, num_boost_round=100) print(f"Model trained with {model.num_trees()} trees")

Model trained with 100 trees

2. Basic Usage¶

SWIFT follows a scikit-learn-style API: create a monitor, fit() on reference data, then test() on monitoring data.

In [4]:

# Create and fit the monitor
monitor = SWIFTMonitor(model=model, n_permutations=200, random_state=42)
monitor.fit(X_ref)

print(f"Fitted on {monitor.n_features_in_} features")
print(f"Feature names: {list(monitor.feature_names_in_)}")

# Create and fit the monitor monitor = SWIFTMonitor(model=model, n_permutations=200, random_state=42) monitor.fit(X_ref) print(f"Fitted on {monitor.n_features_in_} features") print(f"Feature names: {list(monitor.feature_names_in_)}")

Fitted on 10 features
Feature names: ['feature_0', 'feature_1', 'feature_2', 'feature_3', 'feature_4', 'feature_5', 'feature_6', 'feature_7', 'feature_8', 'feature_9']

Transform — SHAP-weighted representation¶

In [5]:

# transform() maps each value to its bucket's mean SHAP
X_transformed = monitor.transform(X_mon)

print("Original monitoring data (first 3 rows):")
display(X_mon.head(3))

print("\nSHAP-transformed (first 3 rows):")
display(X_transformed.head(3))

# transform() maps each value to its bucket's mean SHAP X_transformed = monitor.transform(X_mon) print("Original monitoring data (first 3 rows):") display(X_mon.head(3)) print("\nSHAP-transformed (first 3 rows):") display(X_transformed.head(3))

Original monitoring data (first 3 rows):

	feature_0	feature_1	feature_2	feature_3	feature_4	feature_5	feature_6	feature_7	feature_8	feature_9
0	2.901141	-1.276328	0.037419	-3.454975	1.060145	-1.273333	-4.887935	1.057181	0.220411	-1.579664
1	4.403352	-2.958958	0.702262	0.177995	0.599032	-1.198069	-3.145705	-0.567785	-0.470583	-1.583910
2	0.737262	0.449872	-0.722989	0.822152	-2.640224	-0.375800	-2.859257	-1.873294	-1.843126	-2.841833

SHAP-transformed (first 3 rows):

	feature_0	feature_1	feature_2	feature_3	feature_4	feature_5	feature_6	feature_7	feature_8	feature_9
0	-0.364344	-0.213708	-0.010290	-0.766155	0.427617	-0.955129	-0.287524	-0.329060	0.027231	-2.372301
1	-0.715311	0.046607	-0.081627	-0.722812	0.358162	-1.519299	-0.327770	-0.138953	-0.157886	-2.372301
2	-0.236510	0.288767	-0.020792	-1.312289	-0.782878	-0.554468	-0.347392	0.056228	0.052909	-3.796613

Score — per-feature SWIFT distances¶

In [6]:

# score() computes Wasserstein distances without significance testing
scores = monitor.score(X_mon)

print("Per-feature SWIFT scores (no drift expected):")
for name, score in sorted(scores.items(), key=lambda x: x[1], reverse=True):
    print(f"  {name}: {score:.6f}")

# score() computes Wasserstein distances without significance testing scores = monitor.score(X_mon) print("Per-feature SWIFT scores (no drift expected):") for name, score in sorted(scores.items(), key=lambda x: x[1], reverse=True): print(f" {name}: {score:.6f}")

Per-feature SWIFT scores (no drift expected):
  feature_9: 0.115740
  feature_0: 0.045936
  feature_5: 0.042422
  feature_4: 0.035010
  feature_7: 0.029517
  feature_3: 0.028742
  feature_6: 0.022366
  feature_1: 0.022051
  feature_2: 0.004249
  feature_8: 0.002420

Test — full pipeline with significance testing¶

In [7]:

# test() runs the full pipeline: score + permutation test + MTC
result_clean = monitor.test(X_mon)

print(f"SWIFT max:  {result_clean.swift_max:.6f}")
print(f"SWIFT mean: {result_clean.swift_mean:.6f}")
print(f"Drifted:    {result_clean.num_drifted}/{result_clean.num_features} features")
print(f"Drifted features: {result_clean.drifted_features}")

# test() runs the full pipeline: score + permutation test + MTC result_clean = monitor.test(X_mon) print(f"SWIFT max: {result_clean.swift_max:.6f}") print(f"SWIFT mean: {result_clean.swift_mean:.6f}") print(f"Drifted: {result_clean.num_drifted}/{result_clean.num_features} features") print(f"Drifted features: {result_clean.drifted_features}")

SWIFT max:  0.115740
SWIFT mean: 0.034845
Drifted:    0/10 features
Drifted features: ()

3. Introducing Drift¶

Let's inject synthetic drift into a few features and see if SWIFT picks it up.

In [8]:

# Create a drifted copy of the monitoring data
X_drifted = X_mon.copy()

# Shift the mean of feature_0 and feature_1 by 2 standard deviations
rng = np.random.default_rng(123)
for feat in ["feature_0", "feature_1"]:
    shift = 2.0 * X_ref[feat].std()
    X_drifted[feat] = X_drifted[feat] + shift

# Add noise to feature_2 (increase variance by 3x)
X_drifted["feature_2"] = X_drifted["feature_2"] * 3.0

print("Drift injected into: feature_0 (mean shift), feature_1 (mean shift), feature_2 (variance increase)")

# Create a drifted copy of the monitoring data X_drifted = X_mon.copy() # Shift the mean of feature_0 and feature_1 by 2 standard deviations rng = np.random.default_rng(123) for feat in ["feature_0", "feature_1"]: shift = 2.0 * X_ref[feat].std() X_drifted[feat] = X_drifted[feat] + shift # Add noise to feature_2 (increase variance by 3x) X_drifted["feature_2"] = X_drifted["feature_2"] * 3.0 print("Drift injected into: feature_0 (mean shift), feature_1 (mean shift), feature_2 (variance increase)")

Drift injected into: feature_0 (mean shift), feature_1 (mean shift), feature_2 (variance increase)

In [9]:

# Test the drifted data
result_drifted = monitor.test(X_drifted)

print(f"SWIFT max:  {result_drifted.swift_max:.6f}")
print(f"SWIFT mean: {result_drifted.swift_mean:.6f}")
print(f"Drifted:    {result_drifted.num_drifted}/{result_drifted.num_features} features")
print(f"Drifted features: {result_drifted.drifted_features}")

print("\nPer-feature details:")
for fr in result_drifted.feature_results:
    flag = "DRIFT" if fr.is_drifted else "ok"
    print(f"  {fr.feature_name:12s}  score={fr.swift_score:.6f}  p={fr.p_value:.4f}  [{flag}]")

# Test the drifted data result_drifted = monitor.test(X_drifted) print(f"SWIFT max: {result_drifted.swift_max:.6f}") print(f"SWIFT mean: {result_drifted.swift_mean:.6f}") print(f"Drifted: {result_drifted.num_drifted}/{result_drifted.num_features} features") print(f"Drifted features: {result_drifted.drifted_features}") print("\nPer-feature details:") for fr in result_drifted.feature_results: flag = "DRIFT" if fr.is_drifted else "ok" print(f" {fr.feature_name:12s} score={fr.swift_score:.6f} p={fr.p_value:.4f} [{flag}]")

SWIFT max:  0.880430
SWIFT mean: 0.162945
Drifted:    3/10 features
Drifted features: ('feature_0', 'feature_1', 'feature_2')

Per-feature details:
  feature_0     score=0.880430  p=0.0050  [DRIFT]
  feature_1     score=0.427469  p=0.0050  [DRIFT]
  feature_2     score=0.045337  p=0.0050  [DRIFT]
  feature_3     score=0.028742  p=0.5025  [ok]
  feature_4     score=0.035010  p=0.3085  [ok]
  feature_5     score=0.042422  p=0.5274  [ok]
  feature_6     score=0.022366  p=0.1741  [ok]
  feature_7     score=0.029517  p=0.2687  [ok]
  feature_8     score=0.002420  p=0.9751  [ok]
  feature_9     score=0.115740  p=0.3134  [ok]

4. Visualization — Single Sample¶

SWIFT provides two built-in plots:

• plot_buckets() — Bucketing profile: SHAP response curve + observation density per bucket
• plot_swift_scores() — Bar chart of SWIFT scores per feature

Bucketing profile for a non-drifted feature¶

In [10]:

# Pick a feature that is not drifted
non_drifted = [f for f in monitor.feature_names_in_ if f not in result_drifted.drifted_features]
feat_ok = non_drifted[0] if non_drifted else "feature_5"

fig, ax = monitor.plot_buckets(feat_ok)
fig.suptitle(f"Non-drifted feature: {feat_ok}", y=1.02, fontsize=12)

# Pick a feature that is not drifted non_drifted = [f for f in monitor.feature_names_in_ if f not in result_drifted.drifted_features] feat_ok = non_drifted[0] if non_drifted else "feature_5" fig, ax = monitor.plot_buckets(feat_ok) fig.suptitle(f"Non-drifted feature: {feat_ok}", y=1.02, fontsize=12)

Out[10]:

Text(0.5, 1.02, 'Non-drifted feature: feature_3')

Bucketing profile for a drifted feature¶

In [11]:

fig, ax = monitor.plot_buckets("feature_0")

fig, ax = monitor.plot_buckets("feature_0")

Natural x-axis¶

By default, bucket profiles use integer indices on the x-axis (x_axis="bucket"). Set x_axis="natural" to place buckets at their midpoint on the actual feature-value scale. NULL values appear at the left edge with a dotted separator.

In [12]:

# Same feature, now with natural x-axis
fig, ax = monitor.plot_buckets("feature_0", x_axis="natural")

# Same feature, now with natural x-axis fig, ax = monitor.plot_buckets("feature_0", x_axis="natural")

Custom density source¶

By default plot_buckets() shows the reference density (from X_ref_). Pass X=... to show the density of a different sample. The SHAP response curve always comes from the fitted reference.

In [13]:

# Show the monitoring sample's density instead of the reference
fig, ax = monitor.plot_buckets("feature_0", X=X_mon)
fig.suptitle("Monitoring density (feature_0)", y=1.02, fontsize=12)

# Show the monitoring sample's density instead of the reference fig, ax = monitor.plot_buckets("feature_0", X=X_mon) fig.suptitle("Monitoring density (feature_0)", y=1.02, fontsize=12)

Out[13]:

Text(0.5, 1.02, 'Monitoring density (feature_0)')

SWIFT scores overview¶

In [14]:

# Bar chart of SWIFT scores with a user-defined threshold
fig, ax = monitor.plot_swift_scores(result_drifted, threshold=0.01)

# Bar chart of SWIFT scores with a user-defined threshold fig, ax = monitor.plot_swift_scores(result_drifted, threshold=0.01)

5. Visualization — Comparison¶

Both plots support a comparison mode to visualize differences between two samples or two test results.

Bucket density comparison: reference vs. drifted¶

In [15]:

# Compare reference density vs. drifted density for feature_0
fig, ax = monitor.plot_buckets(
    "feature_0",
    X_compare=X_drifted,
    labels=("Reference", "Drifted"),
)

# Compare reference density vs. drifted density for feature_0 fig, ax = monitor.plot_buckets( "feature_0", X_compare=X_drifted, labels=("Reference", "Drifted"), )

In [16]:

# Compare clean vs. drifted density for feature_2 (variance change)
fig, ax = monitor.plot_buckets(
    "feature_2",
    X_compare=X_drifted,
    labels=("Reference", "Drifted"),
)

# Compare clean vs. drifted density for feature_2 (variance change) fig, ax = monitor.plot_buckets( "feature_2", X_compare=X_drifted, labels=("Reference", "Drifted"), )

Comparison on the natural x-axis¶

In [17]:

# Natural x-axis with comparison overlay
fig, ax = monitor.plot_buckets(
    "feature_0",
    X_compare=X_drifted,
    labels=("Reference", "Drifted"),
    x_axis="natural",
)

# Natural x-axis with comparison overlay fig, ax = monitor.plot_buckets( "feature_0", X_compare=X_drifted, labels=("Reference", "Drifted"), x_axis="natural", )

Custom primary density in comparison mode¶

You can combine X and X_compare to compare two arbitrary samples — for example, monitoring vs. drifted, without showing reference density.

In [18]:

# Compare monitoring density vs. drifted density
fig, ax = monitor.plot_buckets(
    "feature_0",
    X=X_mon,
    X_compare=X_drifted,
    labels=("Monitoring", "Drifted"),
)

# Compare monitoring density vs. drifted density fig, ax = monitor.plot_buckets( "feature_0", X=X_mon, X_compare=X_drifted, labels=("Monitoring", "Drifted"), )

SWIFT scores comparison: clean vs. drifted¶

In [19]:

# Side-by-side grouped bars comparing two test results
fig, ax = monitor.plot_swift_scores(
    result_clean,
    result_compare=result_drifted,
    labels=("Clean", "Drifted"),
    threshold=0.01,
)

# Side-by-side grouped bars comparing two test results fig, ax = monitor.plot_swift_scores( result_clean, result_compare=result_drifted, labels=("Clean", "Drifted"), threshold=0.01, )

6. Sample vs. Sample Comparison¶

By default, score() and test() compare a monitoring sample against the fitted reference (X_ref_). You can instead compare two arbitrary samples by passing X_compare. The SHAP transformation (bucket definitions and mean SHAP per bucket) is always the one learned from the reference.

Scoring two samples¶

In [20]:

# Score monitoring vs. drifted (reference is NOT involved in the comparison)
scores_s2s = monitor.score(X_mon, X_compare=X_drifted)

print("Sample-vs-sample SWIFT scores (monitoring vs. drifted):")
for name, s in sorted(scores_s2s.items(), key=lambda x: x[1], reverse=True):
    print(f"  {name}: {s:.6f}")

# Score monitoring vs. drifted (reference is NOT involved in the comparison) scores_s2s = monitor.score(X_mon, X_compare=X_drifted) print("Sample-vs-sample SWIFT scores (monitoring vs. drifted):") for name, s in sorted(scores_s2s.items(), key=lambda x: x[1], reverse=True): print(f" {name}: {s:.6f}")

Sample-vs-sample SWIFT scores (monitoring vs. drifted):
  feature_0: 0.836672
  feature_1: 0.449174
  feature_2: 0.048190
  feature_3: 0.000000
  feature_4: 0.000000
  feature_5: 0.000000
  feature_6: 0.000000
  feature_7: 0.000000
  feature_8: 0.000000
  feature_9: 0.000000

Testing two samples¶

The permutation test pools X and X_compare (instead of X_ref_ and X), then permutes to estimate p-values.

In [21]:

# Full test: monitoring vs. drifted
result_s2s = monitor.test(X_mon, X_compare=X_drifted)

print(f"SWIFT max:  {result_s2s.swift_max:.6f}")
print(f"SWIFT mean: {result_s2s.swift_mean:.6f}")
print(f"Drifted:    {result_s2s.num_drifted}/{result_s2s.num_features} features")
print(f"Drifted features: {result_s2s.drifted_features}")

# Full test: monitoring vs. drifted result_s2s = monitor.test(X_mon, X_compare=X_drifted) print(f"SWIFT max: {result_s2s.swift_max:.6f}") print(f"SWIFT mean: {result_s2s.swift_mean:.6f}") print(f"Drifted: {result_s2s.num_drifted}/{result_s2s.num_features} features") print(f"Drifted features: {result_s2s.drifted_features}")

SWIFT max:  0.836672
SWIFT mean: 0.133404
Drifted:    3/10 features
Drifted features: ('feature_0', 'feature_1', 'feature_2')

In [22]:

# Compare the sample-vs-sample result against the ref-vs-drifted result
fig, ax = monitor.plot_swift_scores(
    result_drifted,
    result_compare=result_s2s,
    labels=("Ref vs Drifted", "Mon vs Drifted"),
)

# Compare the sample-vs-sample result against the ref-vs-drifted result fig, ax = monitor.plot_swift_scores( result_drifted, result_compare=result_s2s, labels=("Ref vs Drifted", "Mon vs Drifted"), )

7. Advanced Usage¶

Changing hyperparameters with set_params()¶

SWIFT inherits from scikit-learn's BaseEstimator, so you can use get_params() and set_params().

In [23]:

# Inspect current parameters
monitor.get_params()

# Inspect current parameters monitor.get_params()

Out[23]:

{'alpha': 0.05,
 'correction': <CorrectionMethod.BH: 'benjamini-hochberg'>,
 'max_samples': None,
 'model': <lightgbm.basic.Booster at 0x72bd3d584bd0>,
 'n_permutations': 200,
 'n_synthetic': 10,
 'order': 1,
 'random_state': 42}

In [24]:

# Change correction method and re-test
monitor.set_params(correction="bonferroni", alpha=0.01)
result_strict = monitor.test(X_drifted)

print(f"With Bonferroni + alpha=0.01:")
print(f"  Drifted: {result_strict.num_drifted}/{result_strict.num_features}")
print(f"  Drifted features: {result_strict.drifted_features}")

# Restore defaults
monitor.set_params(correction="benjamini-hochberg", alpha=0.05)

# Change correction method and re-test monitor.set_params(correction="bonferroni", alpha=0.01) result_strict = monitor.test(X_drifted) print(f"With Bonferroni + alpha=0.01:") print(f" Drifted: {result_strict.num_drifted}/{result_strict.num_features}") print(f" Drifted features: {result_strict.drifted_features}") # Restore defaults monitor.set_params(correction="benjamini-hochberg", alpha=0.05)

With Bonferroni + alpha=0.01:
  Drifted: 0/10
  Drifted features: ()

Out[24]:

SWIFTMonitor(correction='benjamini-hochberg',
             model=<lightgbm.basic.Booster object at 0x72bd3d584bd0>,
             n_permutations=200)

In a Jupyter environment, please rerun this cell to show the HTML representation or trust the notebook.
On GitHub, the HTML representation is unable to render, please try loading this page with nbviewer.org.

Using W2 (Wasserstein order 2)¶

In [25]:

# W2 is more sensitive to variance changes
monitor_w2 = SWIFTMonitor(model=model, order=2, n_permutations=200, random_state=42)
monitor_w2.fit(X_ref)
result_w2 = monitor_w2.test(X_drifted)

print(f"W2 results:")
print(f"  Drifted: {result_w2.num_drifted}/{result_w2.num_features}")
print(f"  Drifted features: {result_w2.drifted_features}")

# W2 is more sensitive to variance changes monitor_w2 = SWIFTMonitor(model=model, order=2, n_permutations=200, random_state=42) monitor_w2.fit(X_ref) result_w2 = monitor_w2.test(X_drifted) print(f"W2 results:") print(f" Drifted: {result_w2.num_drifted}/{result_w2.num_features}") print(f" Drifted features: {result_w2.drifted_features}")

W2 results:
  Drifted: 3/10
  Drifted features: ('feature_0', 'feature_1', 'feature_2')

fit_transform() shorthand¶

In [26]:

# fit_transform() fits on the data and returns the SHAP-transformed version
monitor_ft = SWIFTMonitor(model=model)
X_ref_transformed = monitor_ft.fit_transform(X_ref)

print(f"Shape: {X_ref_transformed.shape}")
display(X_ref_transformed.head(3))

# fit_transform() fits on the data and returns the SHAP-transformed version monitor_ft = SWIFTMonitor(model=model) X_ref_transformed = monitor_ft.fit_transform(X_ref) print(f"Shape: {X_ref_transformed.shape}") display(X_ref_transformed.head(3))

Shape: (1000, 10)

	feature_0	feature_1	feature_2	feature_3	feature_4	feature_5	feature_6	feature_7	feature_8	feature_9
0	-0.172587	-0.166767	0.017300	0.460865	0.136649	0.015150	0.072554	0.615067	0.041145	0.913536
1	-0.264326	0.016085	0.453870	-0.636090	0.405499	-0.706170	-0.287524	-0.290771	0.030322	-3.496276
2	-0.611653	-0.323008	0.040233	0.387942	-0.463422	0.958026	-0.453722	-0.210279	-0.126483	-2.467320

Summary¶

Method	Purpose
fit(X)	Learn reference distribution (stages 1-3)
transform(X)	Map values to bucket-level mean SHAP
score(X)	Per-feature Wasserstein distances vs. reference
score(X, X_compare=Y)	Per-feature distances: X vs. Y
test(X)	Full drift test vs. reference (stages 4-5)
test(X, X_compare=Y)	Full drift test: X vs. Y
plot_buckets(feat, x_axis=...)	SHAP response + density ("bucket" or "natural" axis)
plot_buckets(feat, X=..., X_compare=...)	Custom density source + comparison overlay
plot_swift_scores(result)	Bar chart of SWIFT scores per feature

For more details, see the README.