tdaad package

Subpackages

• tdaad.utils package
  • Submodules
  • tdaad.utils.local_elliptic_envelope module
    • pandas_mahalanobis()
    • pandas_score_samples()
  • tdaad.utils.remapping_functions module
    • score_flat_fast_remapping()
  • tdaad.utils.tda_functions module
    • transform_to_persistence_diagram()
  • tdaad.utils.window_functions module
    • hash_window()
    • sliding_window_3D_view()
    • sliding_window_ppl_pp()
  • Module contents

Submodules

tdaad.anomaly_detectors module

Topological Anomaly Detectors.

class tdaad.anomaly_detectors.TopologicalAnomalyDetector(window_size: int = 100, step: int = 5, tda_max_dim: int = 1, n_centers_by_dim: int = 5, support_fraction: float | None = None, contamination: float = 0.1, random_state: int | RandomState | None = 42)

Bases: EllipticEnvelope, TransformerMixin

Anomaly detection for multivariate time series using topological embeddings and robust covariance estimation.

This detector extracts topological features from sliding windows of time series data and uses a robust Mahalanobis distance (via EllipticEnvelope) to score anomalies.

Read more in the User Guide.

Parameters:

• window_size (int, default=100) – Sliding window size for extracting time series subsequences.

• step (int, default=5) – Step size between windows.

• tda_max_dim (int, default=1) – Maximum homology dimension used for topological feature extraction.

• n_centers_by_dim (int, default=5) – Number of k-means centers per topological dimension (for vectorization).

• support_fraction (float or None, default=None) – Proportion of data to use for robust covariance estimation. If None, computed automatically.

• contamination (float, default=0.1) – Proportion of anomalies in the data, used to compute decision threshold.

• random_state (int, RandomState instance, or None, default=42) – Controls randomness of the topological embedding and robust estimator.

topological_embedding_

TopologicalEmbedding transformer object that is fitted at fit.

Type:

object

Examples

>>> n_timestamps = 1000
>>> n_sensors = 20
>>> import pandas as pd
>>> timestamps = pd.to_datetime('2024-01-01', utc=True) + pd.Timedelta(1, 'h') * np.arange(n_timestamps)
>>> X = pd.DataFrame(np.random.random(size=(n_timestamps, n_sensors)), index=timestamps)
>>> X.iloc[n_timestamps//2:,:10] = -X.iloc[n_timestamps//2:,10:20]
>>> detector = TopologicalAnomalyDetector(n_centers_by_dim=2, tda_max_dim=1).fit(X)
>>> anomaly_scores = detector.score_samples(X)
>>> decision = detector.decision_function(X)
>>> anomalies = detector.predict(X)

decision_function(X)

Return the distance to the decision boundary.

fit(X, y=None)

Fit the TopologicalAnomalyDetector model.

Parameters:

• X ({array-like, sparse matrix} of shape (n_timestamps, n_sensors)) – Multiple time series to transform, where n_timestamps is the number of timestamps in the series X, and n_sensors is the number of sensors.

• y (Ignored) – Not used, present for API consistency by convention.

Returns:

self – Returns the instance itself.

Return type:

object

predict(X)

Predict inliers (1) and outliers (-1) using learned threshold.

required_properties: Sequence[str] = ['multiple_time_series']

score_samples(X, y=None)

Compute anomaly scores from topological features.

transform(X)

Alias for score_samples. Returns anomaly scores.

tdaad.topological_embedding module

Topological Embedding Transformers.

class tdaad.topological_embedding.TopologicalEmbedding(window_size: int = 40, step: int = 5, tda_max_dim: int = 2, n_centers_by_dim: int = 5)

Bases: BaseEstimator, TransformerMixin

Topological embedding for multiple time series.

Slices time series into smaller time series windows, forms an affinity matrix on each window and applies a Rips procedure to produce persistence diagrams for each affinity matrix. Then uses Atol [ref:Atol] on each dimension through the gudhi.representation.Archipelago representation to produce topological vectorization.

Read more in the User Guide.

Parameters:

• window_size (int, default=40) – Size of the sliding window algorithm to extract subsequences as input to named_pipeline.

• step (int, default=5) – Size of the sliding window steps between each window.

• n_centers_by_dim (int, default=5) – The number of centroids to generate by dimension for vectorizing topological features. The resulting embedding will have total dimension =< tda_max_dim * n_centers_by_dim. The resulting embedding dimension might be smaller because of the KMeans algorithm in the Archipelago step.

• tda_max_dim (int, default=2) – The maximum dimension of the topological feature extraction.

Examples

>>> n_timestamps = 100
>>> n_sensors = 5
>>> timestamps = pd.to_datetime('2024-01-01', utc=True) + pd.Timedelta(1, 'h') * np.arange(n_timestamps)
>>> X = pd.DataFrame(np.random.random(size=(n_timestamps, n_sensors)), index=timestamps)
>>> TopologicalEmbedding(n_centers_by_dim=2, tda_max_dim=1).fit_transform(X)

fit(X, y=None)

Fit the internal pipeline to the data.

Parameters:

• X (pandas.DataFrame) – Input feature matrix.

• y (array-like, optional) – Target values (not used here, but accepted for compatibility with sklearn).

Returns:

self – Fitted transformer.

Return type:

object

fit_transform(X, y=None, **fit_params)

Fit to data, then transform it.

Returns:

X_transformed

Return type:

array-like

transform(X)

Apply transformations to the input data using the fitted pipeline.

Parameters:

X (pandas.DataFrame) – Input data to transform.

Returns:

X_transformed – Transformed data.

Return type:

array-like or DataFrame

tdaad.topological_embedding.local_atol_fit(self, X, y=None, sample_weight=None)

local modification to prevent FutureWarning triggered by np.concatenate(X) when X is a pd.Series.

Module contents

Topological Data Analysis module for Anomaly Detection in Time Series

tdaad is a Python module integrating TDA tools from gudhi into learning algorithms designed to detect anomalies in Multiple/Multivariate Time Series.