Core Models API¶

NullModel ¶

Python

NullModel()

Bases: BaseModel

Null model that serves as a baseline for anomaly detection.

This model never learns from data and always returns a score of 0.0, indicating no anomalies are detected. It's useful as a baseline for comparing other anomaly detection models.

Examples:

Python

from aberrant.model import NullModel

model = NullModel()
score = model.score_one({"feature1": 1.0, "feature2": 2.0})
model.learn_one({"feature1": 1.0, "feature2": 2.0})
assert score == 0.0

Initialize the null model.

learn_one ¶

Python

learn_one(x: dict[str, float]) -> None

Update the model with a single data point.

This method does nothing as the null model never learns.

Parameters:

Name	Type	Description	Default
`x`	`dict[str, float]`	Feature dictionary with string keys and float values.	required

score_one ¶

Python

score_one(x: dict[str, float]) -> float

Compute anomaly score for a single data point.

Parameters:

Name	Type	Description	Default
`x`	`dict[str, float]`	Feature dictionary with string keys and float values.	required

Returns:

Type	Description
`float`	Always returns 0.0 (no anomaly detected).

RandomModel ¶

Python

RandomModel(seed: int = 1)

Bases: BaseModel

Random model that generates random anomaly scores.

This model never learns from data and returns uniformly distributed random scores between 0 and 1. It serves as a random baseline for comparing other anomaly detection models.

Parameters:

Name	Type	Description	Default
`seed`	`int`	Random seed for reproducibility.	`1`

Examples:

Python

from aberrant.model import RandomModel

model = RandomModel(seed=42)
score = model.score_one({"feature1": 1.0, "feature2": 2.0})
model.learn_one({"feature1": 1.0, "feature2": 2.0})
assert 0.0 <= score < 1.0

Initialize the random model with a specified seed.

learn_one ¶

Python

learn_one(x: dict[str, float]) -> None

Update the model with a single data point.

This method does nothing as the random model never learns.

Parameters:

Name	Type	Description	Default
`x`	`dict[str, float]`	Feature dictionary with string keys and float values.	required

score_one ¶

Python

score_one(x: dict[str, float]) -> float

Compute a random anomaly score for a single data point.

Parameters:

Name	Type	Description	Default
`x`	`dict[str, float]`	Feature dictionary with string keys and float values.	required

Returns:

Type	Description
`float`	Random float between 0.0 and 1.0.

ThresholdModel ¶

Python

ThresholdModel(ceiling: float | dict[str, float] | None = None, floor: float | dict[str, float] | None = None)

Bases: BaseModel

Threshold model that detects anomalies based on boundary violations.

This model never learns from data and returns a binary anomaly score based on whether feature values exceed specified thresholds. It supports both one-sided (ceiling or floor only) and two-sided (corridor) detection.

Parameters:

Name	Type	Description	Default
`ceiling`	`float \| dict[str, float] \| None`	Upper threshold(s). Can be a scalar (applies to all features) or a dict mapping feature names to their upper thresholds.	`None`
`floor`	`float \| dict[str, float] \| None`	Lower threshold(s). Can be a scalar (applies to all features) or a dict mapping feature names to their lower thresholds.	`None`

Note

At least one of ceiling or floor must be provided.

Examples:

Python

from aberrant.model import ThresholdModel

ceiling_only = ThresholdModel(ceiling=10.0)
assert ceiling_only.score_one({"temp": 15.0}) == 1.0
assert ceiling_only.score_one({"temp": 5.0}) == 0.0

corridor = ThresholdModel(ceiling=100.0, floor=0.0)
assert corridor.score_one({"temp": 50.0}) == 0.0
assert corridor.score_one({"temp": -5.0}) == 1.0

per_feature = ThresholdModel(
    ceiling={"temp": 100.0, "pressure": 50.0},
    floor={"temp": 0.0, "pressure": 10.0},
)
assert per_feature.score_one({"temp": 50.0, "pressure": 30.0}) == 0.0
assert per_feature.score_one({"temp": 150.0, "pressure": 30.0}) == 1.0

Initialize the threshold model.

learn_one ¶

Python

learn_one(x: dict[str, float]) -> None

Update the model with a single data point.

This method does nothing as the threshold model never learns.

Parameters:

Name	Type	Description	Default
`x`	`dict[str, float]`	Feature dictionary with string keys and float values.	required

score_one ¶

Python

score_one(x: dict[str, float]) -> float

Compute anomaly score for a single data point.

Parameters:

Name	Type	Description	Default
`x`	`dict[str, float]`	Feature dictionary with string keys and float values.	required

Returns:

Type	Description
`float`	1.0 if any feature violates its threshold(s), 0.0 otherwise.

QuantileThreshold ¶

Python

QuantileThreshold(quantile: float = 0.95, window_size: int = 1000, score_key: str = 'score')

Bases: BaseModel

Adaptive threshold model based on score distribution quantiles.

This model maintains a sliding window of anomaly scores and computes an adaptive threshold based on a specified quantile. Points scoring at or above the threshold are classified as anomalies.

Unlike static thresholds, QuantileThreshold adapts to the actual distribution of scores observed during streaming.

Parameters:

Name	Type	Description	Default
`quantile`	`float`	Quantile level for the threshold. Values closer to 1.0 result in higher thresholds (fewer detections). Default is 0.95.	`0.95`
`window_size`	`int`	Number of scores to keep for quantile computation. Default is 1000.	`1000`
`score_key`	`str`	Name of the score feature in the input dictionary. Default is "score".	`'score'`

Examples:

Python

from aberrant.model import QuantileThreshold

threshold = QuantileThreshold(quantile=0.8, window_size=10)
for score in range(10):
    threshold.learn_one({"score": float(score)})
assert threshold.score_one({"score": 10.0}) == 1.0

Note

The model expects input dictionaries with a score key (default "score"). score_one returns: - 1.0 if the score is greater than or equal to the threshold (anomaly) - max(score/threshold, 0.0) if below a positive threshold (in [0, 1)) - 0.0 if below a zero or negative threshold - 0.0 during warmup (insufficient data for threshold)

threshold `property` ¶

Python

threshold: float | None

Current adaptive threshold.

Returns None if insufficient data has been collected.

n_scores `property` ¶

Python

n_scores: int

Number of scores currently in the window.

learn_one ¶

Python

learn_one(x: dict[str, float]) -> None

Update the threshold estimate with a new score.

Parameters:

Name	Type	Description	Default
`x`	`dict[str, float]`	Dictionary containing the score. Must have the score_key.	required

score_one ¶

Python

score_one(x: dict[str, float]) -> float

Evaluate a score against the adaptive threshold.

Parameters:

Name	Type	Description	Default
`x`	`dict[str, float]`	Dictionary containing the score. Must have the score_key.	required

Returns:

Type	Description
`float`	1.0 if score >= threshold (anomaly)
`float`	max(score/threshold, 0.0) if below a positive threshold
`float`	0.0 if below a zero or negative threshold
`float`	0.0 if threshold not yet computed (warmup period)

reset ¶

Python

reset() -> None

Reset the model to its initial state.

Core Models API¶

NullModel ¶

learn_one ¶

score_one ¶

RandomModel ¶

learn_one ¶

score_one ¶

ThresholdModel ¶

learn_one ¶

score_one ¶

QuantileThreshold ¶

threshold property ¶

n_scores property ¶

learn_one ¶

score_one ¶

reset ¶

threshold `property` ¶

n_scores `property` ¶