Distribution Drift Detection

VigilCV detects statistical covariate shift between a reference distribution (your training set) and live inference data — without running a single neural network forward pass.

The Problem: Silent Drift

A model trained on summer daytime images may receive winter nighttime images at inference time. Accuracy degrades silently. VigilCV intercepts this by comparing the statistical fingerprint of incoming batches against a recorded baseline.

Feature Representation: 54D Spatial Color Moments

Each image is represented as a 54-dimensional vector of spatial color statistics:

Total: 54 dimensions — computed in ~85µs per image with pure NumPy.

Drift Metrics

Wasserstein-1 Distance (Earth Mover's Distance)

For 1D marginals of each feature dimension k:

W₁(P, Q) = ∫|F_P(x) - F_Q(x)| dx

where F_P and F_Q are the cumulative distribution functions of the reference and query batches. This is computed via scipy.stats.wasserstein_distance over the 54D feature vectors projected onto each dimension.

Interpretation: W₁ is the minimum "work" required to transform distribution P into Q. Units are in feature-space units. Higher values indicate more shift.

Maximum Mean Discrepancy (MMD) with RBF Kernel

The unbiased MMD² estimator with a Radial Basis Function (Gaussian) kernel:

MMD²(P, Q) = E[k(x,x')] - 2·E[k(x,y)] + E[k(y,y')]

where k(x,y) = exp(-||x-y||² / (2σ²))

The bandwidth σ is set to the median pairwise distance of the reference distribution (Silverman's rule). This makes MMD scale-free and robust to the absolute magnitude of the 54D features.

Interpretation: MMD² = 0 means the distributions are identical. Higher values indicate statistically significant shift.

Usage

Step 1: Record a Baseline

vigilcv baseline dataset/train/ --output baseline.pkl

Or programmatically:

from vigilcv.core.drift import DriftEngine

engine = DriftEngine()
engine.fit(reference_directory="dataset/train/")
engine.save("baseline.pkl")

Step 2: Detect Drift

vigilcv drift dataset/new/ --baseline baseline.pkl --threshold 0.15

Or in your pipeline:

from vigilcv.core.drift import DriftEngine

engine = DriftEngine.load("baseline.pkl")
report = engine.detect(query_directory="dataset/new/")

if report.is_drifted:
    print(f"DRIFT DETECTED")
    print(f"  Wasserstein-1: {report.wasserstein_distance:.4f}")
    print(f"  MMD²:          {report.mmd_score:.6f}")
    print(f"  Top features:  {report.top_drift_features}")

DriftReport Schema

@dataclass(frozen=True)
class DriftReport:
    is_drifted: bool
    wasserstein_distance: float   # W₁ over 54D moments
    mmd_score: float              # Unbiased MMD² (RBF kernel)
    top_drift_features: list[str] # Feature names with highest W₁
    reference_n: int              # Number of reference images
    query_n: int                  # Number of query images
    threshold: float              # Threshold used for decision

Choosing Thresholds