fraudtwin.ml.drift#

Deterministic data, domain, and concept drift reports.

Status: Experimental

Classes#

fraudtwin.ml.drift.DriftConfig

Immutable policy controlling a drift comparison.

fraudtwin.ml.drift.DriftMetric

One measured drift value and its decision threshold.

fraudtwin.ml.drift.DriftReport

Reproducible comparison of two row windows.

Functions#

fraudtwin.ml.drift.compare_performance

Compare matching evaluation metrics under one threshold policy.

fraudtwin.ml.drift.compare_windows

Compare two row windows using deterministic distribution metrics.

Detailed API#

Deterministic data, domain, concept, and performance drift reports.

The drift helpers operate on ordinary row mappings so they can compare PIT datasets, replay windows, or production extracts without requiring SciPy or a monitoring vendor. Reports contain the comparison policy and input fingerprints, making an alert reproducible instead of an opaque dashboard number.

class fraudtwin.ml.drift.DriftConfig(**data)[source][source]

Bases: BaseModel

Immutable policy controlling a drift comparison.

reference_name and comparison_name identify the windows in the report. Numeric PSI bins are fitted from the reference values only. Thresholds are absolute metric values; an alert is raised when a metric is greater than or equal to its threshold.

Parameters:

  • reference_name (Annotated[str, MinLen(min_length=1)])

  • comparison_name (Annotated[str, MinLen(min_length=1)])

  • numeric_bins (Annotated[int, Ge(ge=2), Le(le=100)])

  • psi_threshold (Annotated[float, Ge(ge=0)])

  • wasserstein_threshold (Annotated[float, Ge(ge=0)])

  • js_threshold (Annotated[float, Ge(ge=0)])

  • prevalence_threshold (Annotated[float, Ge(ge=0)])

  • performance_threshold (Annotated[float, Ge(ge=0)])

  • minimum_samples (Annotated[int, Ge(ge=1)])

  • smoothing (Annotated[float, Gt(gt=0), Lt(lt=0.5)])

  • fields (tuple[str, ...] | None)

  • label_policy (Annotated[str, MinLen(min_length=1)])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class fraudtwin.ml.drift.DriftMetric(**data)[source][source]

Bases: BaseModel

One measured drift value and its decision threshold.

Parameters:

  • field (Annotated[str, MinLen(min_length=1)])

  • field_type (Literal['numeric', 'categorical', 'prevalence', 'quality', 'performance'])

  • method (Annotated[str, MinLen(min_length=1)])

  • reference_value (float | None)

  • comparison_value (float | None)

  • threshold (Annotated[float | None, Ge(ge=0)])

  • alerted (bool)

  • reference_count (Annotated[int, Ge(ge=0)])

  • comparison_count (Annotated[int, Ge(ge=0)])

  • metadata (dict[str, Any])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class fraudtwin.ml.drift.DriftReport(**data)[source][source]

Bases: BaseModel

Reproducible comparison of two row windows.

Parameters:

  • report_version (str)

  • reference_window (str)

  • comparison_window (str)

  • reference_count (Annotated[int, Ge(ge=0)])

  • comparison_count (Annotated[int, Ge(ge=0)])

  • reference_fingerprint (str)

  • comparison_fingerprint (str)

  • label_policy (str)

  • metrics (tuple[DriftMetric, ...])

  • performance_metrics (tuple[DriftMetric, ...])

  • manifest (dict[str, Any])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property alerts: tuple[DriftMetric, ...]

Return metrics that exceeded their configured thresholds.

property fingerprint: str

Return a stable fingerprint for policy, inputs, and measurements.

fraudtwin.ml.drift.compare_performance(reference, comparison, *, config=None, label_policy=None)[source][source]

Compare matching evaluation metrics under one threshold policy.

Parameters:

  • reference (Mapping[str, float]) – Metric name to value mapping for the baseline window.

  • comparison (Mapping[str, float]) – Metric name to value mapping for the new window.

  • config (DriftConfig | None) – Provides performance_threshold and sample policy.

  • label_policy (str | None) – Optional explicit policy; it must match the config when both are supplied.

Return type:

tuple[DriftMetric, ...]

Returns:

One relative absolute-change metric for every common metric.

Raises:

ValueError – If metric keys differ or label policies conflict.

fraudtwin.ml.drift.compare_windows(reference, comparison, config=None)[source][source]

Compare two row windows using deterministic distribution metrics.

Parameters:

  • reference (Iterable[Mapping[str, Any]]) – Baseline rows. Numeric bins are fitted from this window.

  • comparison (Iterable[Mapping[str, Any]]) – New rows to evaluate against the baseline.

  • config (DriftConfig | None) – Thresholds, window names, and fields. Defaults to DriftConfig.

Return type:

DriftReport

Returns:

A report containing feature, missingness, prevalence, and duplicate metrics plus input fingerprints.

Raises:

ValueError – If either input is empty or configured fields are absent.