Types

Bucket

Bucket dataclass

Bucket(bucket_type: BucketType, index: int, lower: float = float('-inf'), upper: float = float('inf'), categories: Optional[frozenset] = None, mean_shap: Optional[float] = None)

A single bucket for a feature.

For numeric features: defined by [lower, upper) interval. For null bucket: lower=upper=NaN. For categorical: categories is a frozenset of category values.

contains

contains(value: float | str | None) -> bool

Check if a value falls into this bucket.

Source code in src/swift/types.py

def contains(self, value: float | str | 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

    if self.bucket_type == BucketType.CATEGORICAL:
        return self.categories is not None and value in self.categories

    # 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

BucketSet dataclass

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

Collection of buckets for a single feature.

Attributes: feature_name: Name of the feature. buckets: List of Bucket objects (including null bucket). decision_points: Sorted array of decision points (split thresholds).

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: 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: 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

BucketType

Bases: Enum

Type of bucket.

WassersteinOrder

WassersteinOrder

Bases: Enum

Order of Wasserstein distance.

CorrectionMethod

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

FeatureSWIFTResult dataclass

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

SWIFT result for a single feature.

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

SWIFTResult

SWIFTResult dataclass

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

Aggregate SWIFT result across all features.

Attributes: feature_results: Per-feature results. swift_max: Maximum SWIFT score across features. swift_mean: Mean SWIFT score across features. swift_weighted: Importance-weighted SWIFT score (optional). alpha: Significance level used for testing. correction_method: Multiple testing correction method used.

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.