BaseDetector

BaseDetector is the abstract base class all detectors in telemetry-anomdet inherit from. It enforces a consistent interface modeled on PyOD convention.

Abstract methods (subclasses must implement):

  • fit(X, y=None) — train on nominal telemetry windows; must call _set_post_fit(scores) before returning

  • decision_function(X) — return raw anomaly scores, shape (n_windows,); higher = more anomalous

Provided methods (inherited free by all subclasses):

  • predict(X) — binary labels (0 = normal, 1 = anomaly) using threshold_

  • is_anomaly(X, *, threshold=None, percentile=None) — boolean mask with optional runtime threshold override

  • score_samples(X) — alias for decision_function()

  • save(path) / load(path) — pickle serialization

Post-fit guarantee: after fit(), every detector always exposes:

  • decision_scores_ — training anomaly scores, shape (n_windows,)

  • threshold_ — score cutoff at self.percentile of training scores

  • labels_ — binary training labels derived from decision_scores_ and threshold_

BaseDetector: Shared interface for all anomaly detectors in telemetry_anomdet.

Design

All detectors in this library follow PyOD convention.

fit(x): Train the model on the provided data, sets post-fit attributes. predict(x): Return binary labels (0 for normal, 1 for anomaly). decision_function(x): Return anomaly scores (higher means more anomalous). is_anomaly(x): Boolean mask. Supports runtime thresholding override.

Input Convention

All detectors accept X of shape (n_windows, window_size, n_features). Classical detecrtors flatten X internally using feature_stat(). Sequence detectors (GDN, TranAD) consume X directly. The caller never maneges this distinction.

Post-fit Attributes

After fit(), every detectror exposes:

decision_scores_: np.ndarray (n_windows,) Training anomaly scores. threshold_: float Score cutoff from training. labels_: np.ndarray (n_windows,) Binary labels from training (0 normal, 1 anomaly).

class telemetry_anomdet.models.base.BaseDetector(percentile: float = 95.0)[source]

Bases: ABC

Abstract base class for all anomaly detectors in telemetry-anomdet.

Parameters:
  • percentile (float, default = 95.0) – Percentile of training anomaly scores used to set threshold_. For example, 95.0 means the top 5% most anomalous training windows are labelled as anomalies by default. Must be in (0, 100).

  • fit) (Attributes (set after)

  • --------------------------

  • decision_scores (np.ndarray, shape (n_windows,)) – Anomaly scores on the training data. Higher values indicate greater anomaly likelihood. Set by _set_post_fit() at the end of fit().

  • threshold (float) – Score cutoff derived from decision_scores_ at self.percentile. Used as the default decision boundary in predict() and is_anomaly().

  • labels (np.ndarray, shape (n_windows,)) – Binary anomaly labels on the training data. 0 = normal, 1 = anomaly. Derived from decision_scores_ and threshold_.

Notes

Subclasses must implement fit() and decision_function(). All other methods are provided and should not need to be overridden unless the detector has non-standard scoring behavior.

abstractmethod fit(X: ndarray, y: ndarray | None = None) BaseDetector[source]

Train the detector on nominal telemetry windows.

Subclasses must call self._set_post_fit(scores) at the end of this method, where scores are the training anomaly scores. This sets decision_scores_, threshold_, and labels_ automatically.

Parameters:
  • X (np.ndarray, shape (n_windows, window_size, n_features)) – Windowed telemetry tensor from windowify().

  • y (np.ndarray, optional) – Ignored for unsupervised detectors. Present for API consistency.

Returns:

self – Fitted detector instance, for method chaining.

Return type:

BaseDetector

abstractmethod decision_function(X: ndarray) ndarray[source]

Compute raw anomaly scores for each input window.

Higher scores indicate greater anomaly likelihood. This is the core scoring method. All other scoring methods call this one.

Parameters:

X (np.ndarray, shape (n_windows, window_size, n_features)) – Windowed telemetry tensor from windowify().

Returns:

scores – Anomaly score per window.

Return type:

np.ndarray, shape (n_windows,)

predict(X: ndarray)[source]

Classify each window as normal (0) or anomalous (1).

Uses threshold_ from training by default. For runtime threshold adjustment, use is_anomaly() with a threshold or percentile override instead.

Parameters:

X (np.ndarray, shape (n_windows, window_size, n_features)) – Windowed telemetry tensor from windowify().

Returns:

labels – Binary labels. 0 = normal, 1 = anomaly.

Return type:

np.ndarray, shape (n_windows,)

is_anomaly(X: ndarray, *, threshold: float | None = None, percentile: float | None = None) ndarray[source]

Boolean anomaly mask with optional runtime threshold override.

This is the human-in-the-loop hook. Operators can adjust sensitivity at inference time without retraining by passing a custom threshold or percentile. If neither is provided, the training derived threshold_ is used.

Priority: threshold > percentile > threshold_.

Parameters:
  • X (np.ndarray, shape (n_windows, window_size, n_features)) – Windowed telemetry tensor from windowify().

  • threshold (float, optional) – Fixed score cutoff. Windows with scores above this are anomalies.

  • percentile (float, optional) – Compute the cutoff as this percentile of the current batch’s scores. Useful when the operator wants to flag only the top N% of a given pass’s windows rather than using training statistics. Must be in (0, 100).

Returns:

mask – True where the window is anomalous.

Return type:

np.ndarray of bool, shape (n_windows,)

Examples

Default — use training threshold:

flags = detector.is_anomaly(X3d)

Tighten sensitivity (only top 2% flagged):

flags = detector.is_anomaly(X3d, percentile=98.0)

Fixed cutoff from operator feedback:

flags = detector.is_anomaly(X3d, threshold=0.72)

score_samples(X: ndarray) ndarray[source]

Alias for decision_function().

Provided for compatibility with code that uses the sklearn/PyOD score_samples convention. Prefer decision_function() for new code.

Parameters:

X (np.ndarray, shape (n_windows, window_size, n_features))

Returns:

scores

Return type:

np.ndarray, shape (n_windows,)

save(path: str | Path) None[source]

Serialize the fitted detector to disk using pickle.

Parameters:

path (str or Path) – Destination file path. Conventionally ends in .pkl.

classmethod load(path: str | Path) BaseDetector[source]

Deserialize a detector from disk.

Parameters:

path (str or Path) – Path to a pickled detector file created by save().

Returns:

detector – The deserialized, fitted detector instance.

Return type:

BaseDetector

Notes

Only load files from trusted sources.