Anomaly Detection Toolkit (ADTK)

0.6.2 · active · verified Wed Apr 15

ADTK (Anomaly Detection Toolkit) is a Python package designed for unsupervised and rule-based time series anomaly detection. It provides a modular and intuitive API, offering a collection of common detectors, transformers, and aggregators that can be combined into custom anomaly detection pipelines. Currently at version 0.6.2, ADTK aims to help users build effective models even with limited labeled historical anomaly data, with regular updates and maintenance.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to use `adtk` to detect seasonal anomalies in a synthetic time series. It involves validating the input series, initializing a `SeasonalAD` detector, fitting it to the data, detecting anomalies, and visualizing the results. Note that for real-world scenarios, you would load your data (e.g., `pd.read_csv`), ensure it has a `DatetimeIndex`, and choose appropriate detector parameters based on your data's characteristics and anomaly types.

import pandas as pd
from adtk.data import validate_series
from adtk.detector import SeasonalAD
from adtk.visualization import plot

# Create a dummy time series with a seasonal pattern and an anomaly
index = pd.date_range(start='2023-01-01', periods=100, freq='H')
data = [i % 24 for i in range(100)] # Daily seasonality
data[50:55] = [50, 51, 52, 53, 54] # Introduce an anomaly
s_train = pd.Series(data, index=index)

# Validate the series (important for ADTK compatibility)
s_train = validate_series(s_train)

# Initialize and train a SeasonalAD detector
# freq can be 'H' for hourly, 'D' for daily, etc., based on data frequency.
# c is a sensitivity parameter for anomaly detection.
seasonal_ad = SeasonalAD(freq=24, c=3.0)

# Fit the detector to the training data and detect anomalies
anomalies = seasonal_ad.fit_detect(s_train)

# Plot the time series with detected anomalies
plot(s_train, anomaly=anomalies, ts_linewidth=1, anomaly_markersize=5, anomaly_color='red', anomaly_tag='marker')

print("Detected anomalies:")
print(anomalies[anomalies].index)

view raw JSON →