pdmlabs.thresholding.constant

pdmlabs.thresholding.constant#

Constant (fixed) thresholder for simple decision boundaries.

ConstantThresholder returns the same threshold value for all scores. Simple but effective for many use cases where a fixed decision boundary works.

Useful when: - Anomaly detection model outputs are well-calibrated (0-1) - Domain knowledge specifies a fixed decision boundary - Need a fast baseline thresholder - Want control over False Positive vs False Negative trade-off

Example: If detector outputs [0.1, 0.3, 0.7, 0.9] with threshold=0.5, anomaly labels would be [0, 0, 1, 1] (score > threshold = anomaly).

Classes

ConstantThresholder(event_preferences[, ...])

Apply a constant threshold to all anomaly scores.
class pdmlabs.thresholding.constant.ConstantThresholder(event_preferences: EventPreferences, threshold_value: float = 0.5)#

Bases: ThresholderInterface

Apply a constant threshold to all anomaly scores.

Simplest thresholder: returns same fixed threshold value for every score. Used when a single decision boundary works across all conditions.

threshold_value#

Fixed threshold. Any score > threshold_value is classified as anomalous. Default 0.5 (typical for normalized scores).

Type:

float

Examples

>>> from pdmlabs.thresholding.constant import ConstantThresholder
>>> thresholder = ConstantThresholder(
...     event_preferences={'failure': [], 'reset': []},
...     threshold_value=0.5
... )
>>> thresholder.fit([df_train], ['bearing_1'], events_df)  # No-op
>>>
>>> # Batch mode
>>> scores = [0.1, 0.3, 0.7, 0.9, 0.5]
>>> thresholds = thresholder.infer_threshold(scores, 'bearing_1', events_df, dates)
>>> print(thresholds)  # [0.5, 0.5, 0.5, 0.5, 0.5]
>>>
>>> # Online mode
>>> threshold = thresholder.infer_threshold_one(0.7, 'bearing_1', events_df)
>>> print(threshold)  # 0.5
>>> is_anomaly = 0.7 > threshold  # True
fit(historic_data: list, historic_sources: list[str], event_data: DataFrame, anomaly_ranges=None) → None#

No-op fit (thresholder is stateless).

Constant thresholder doesn’t learn from data. Threshold is fixed at init.

Parameters:

  • historic_data (list) – Ignored.

  • historic_sources (list[str]) – Ignored.

  • event_data (pd.DataFrame) – Ignored.

  • anomaly_ranges – Ignored.

get_params()#

Return thresholder parameters.

Returns:

{‘threshold_value’: the fixed threshold}

Return type:

dict

infer_threshold(scores: list[float], source: str, event_data: DataFrame, scores_dates: list[Timestamp]) → list[float]#

Return constant threshold for each score.

Parameters:

  • scores (list[float]) – Anomaly scores (unused).

  • source (str) – Source identifier (unused).

  • event_data (pd.DataFrame) – Event log (unused).

  • scores_dates (list[pd.Timestamp]) – Score timestamps (unused).

Returns:

List of constant threshold values repeated for each score.

Return type:

list[float]

Examples

>>> scores = [0.1, 0.3, 0.7, 0.9]
>>> thresholds = thresholder.infer_threshold(scores, 'bearing_1', events_df, dates)
>>> print(thresholds)  # [0.5, 0.5, 0.5, 0.5]
infer_threshold_one(score: float, source: str, event_data: DataFrame) → float#

Return constant threshold for single score.

Parameters:

  • score (float) – Single anomaly score (unused).

  • source (str) – Source identifier (unused).

  • event_data (pd.DataFrame) – Event log (unused).

Returns:

The constant threshold_value.

Return type:

float

Contents