tseda.anomaly

tseda.anomaly

Point and contextual anomaly detection for time series.

Public API

AnomalyReport

Frozen dataclass with mask, indices, timestamps, values, scores, method, and n_anomalies.

AnomalyDetector

Stateless detector: rolling IQR, rolling Z-score, STL-residual, and GESD methods.

class tseda.anomaly.AnomalyReport(mask, indices, timestamps, values, scores, method, n_anomalies)[source]

Bases: object

Immutable anomaly detection result.

Parameters:
mask

Boolean array of shape (n,); True where an anomaly was detected.

Type:

numpy.ndarray

indices

Integer positions (0-based) of detected anomalies.

Type:

numpy.ndarray

timestamps

Timestamps of detected anomalies.

Type:

pandas.DatetimeIndex

values

Observed values at anomaly positions.

Type:

numpy.ndarray

scores

Continuous anomaly score in [0, 1] for each observation. Higher values indicate stronger evidence of anomaly. 0 for normal observations.

Type:

numpy.ndarray

method

Name of the detection method.

Type:

str

n_anomalies

Number of anomalies detected.

Type:

int

mask: ndarray
indices: ndarray
timestamps: DatetimeIndex
values: ndarray
scores: ndarray
method: str
n_anomalies: int
__repr__()[source]

Return repr(self).

Return type:

str

__init__(mask, indices, timestamps, values, scores, method, n_anomalies)
Parameters:
Return type:

None

class tseda.anomaly.AnomalyDetector[source]

Bases: object

Detect point and contextual anomalies in a TimeSeries.

This class is stateless — one instance, many series.

rolling_iqr(ts, window, k)[source]

Rolling Tukey IQR fence.

Parameters:
Return type:

AnomalyReport

rolling_z(ts, window, threshold)[source]

Rolling mean ± k × std.

Parameters:
Return type:

AnomalyReport

stl_residual(ts, period, method, threshold)[source]

STL decompose then flag large residuals.

Parameters:
Return type:

AnomalyReport

gesd(ts, alpha, max_outliers)[source]

Global Generalized ESD test (re-uses quality module).

Parameters:
Return type:

AnomalyReport

remove(ts, report)[source]

Replace anomaly positions with NaN.

Parameters:
Return type:

TimeSeries

label(ts, report)[source]

Return a 0/1 TimeSeries of anomaly labels.

Parameters:
Return type:

TimeSeries

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector

>>> rng  = np.random.default_rng(1)
>>> idx  = pd.date_range("2020", periods=100, freq="D")
>>> vals = rng.standard_normal(100)
>>> vals[20] = 12.0
>>> ts   = TimeSeries(vals, index=idx)
>>> det  = AnomalyDetector()
>>> r    = det.rolling_z(ts, window=30)
>>> 20 in r.indices
True
rolling_iqr(ts, window=30, *, k=2.5, center=True, min_periods=None)[source]

Detect anomalies using a rolling IQR fence.

An observation y[t] is flagged when it falls outside [Q1(t) − k×IQR(t),  Q3(t) + k×IQR(t)] where the quartiles are computed over a rolling window centred at t.

Parameters:

  • ts (TimeSeries) – Input series.

  • window (int, optional) – Rolling window width in observations. Default 30.

  • k (float, optional) – Fence multiplier. Default 2.5 (tighter than the global 1.5 because the window is local and thus more sensitive).

  • center (bool, optional) – Centre the window on each observation. Default True.

  • min_periods (int, optional) – Minimum non-NaN observations required in each window. Defaults to window // 2.

Return type:

AnomalyReport

Raises:

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector
>>> rng  = np.random.default_rng(0)
>>> idx  = pd.date_range("2020", periods=100, freq="D")
>>> vals = rng.standard_normal(100)
>>> vals[40] = 10.0
>>> ts   = TimeSeries(vals, index=idx)
>>> r    = AnomalyDetector().rolling_iqr(ts)
>>> 40 in r.indices
True
rolling_z(ts, window=30, *, threshold=3.0, center=True, min_periods=None)[source]

Detect anomalies using a rolling Z-score.

An observation is flagged when |y[t] − μ(t)| / σ(t) > threshold, where μ and σ are the rolling mean and standard deviation computed over a window.

Parameters:

  • ts (TimeSeries) – Input series.

  • window (int, optional) – Rolling window width. Default 30.

  • threshold (float, optional) – Z-score cut-off. Default 3.0.

  • center (bool, optional) – Centre the window. Default True.

  • min_periods (int, optional) – Minimum non-NaN observations per window. Defaults to window // 2.

Return type:

AnomalyReport

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector
>>> rng  = np.random.default_rng(2)
>>> idx  = pd.date_range("2020", periods=100, freq="D")
>>> vals = rng.standard_normal(100)
>>> vals[60] = -9.0
>>> ts   = TimeSeries(vals, index=idx)
>>> r    = AnomalyDetector().rolling_z(ts)
>>> 60 in r.indices
True
stl_residual(ts, period=None, *, residual_method='iqr', k=3.0, robust=True)[source]

Detect anomalies in the STL residual component.

Decomposes ts into trend + seasonal + residual using STL, then flags residual values that are extreme under the chosen criterion.

Parameters:

  • ts (TimeSeries) – Input series. Must be long enough for STL decomposition (at least 2 × period observations).

  • period (int, optional) – Seasonal period. Inferred from ts.freq when omitted.

  • residual_method (str, optional) –

    Criterion to flag residuals:

    • "iqr" — Tukey IQR fence with multiplier k (default).

    • "mad" — Median Absolute Deviation threshold k.

    • "z" — Z-score threshold k.

  • k (float, optional) – Threshold multiplier / fence factor. Default 3.0.

  • robust (bool, optional) – Use robust LOESS in STL. Default True.

Return type:

AnomalyReport

Raises:

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector
>>> rng  = np.random.default_rng(3)
>>> n    = 60
>>> seas = np.tile(np.sin(2*np.pi*np.arange(12)/12)*5, 5)
>>> vals = seas + rng.standard_normal(n) * 0.3
>>> vals[25] = 20.0
>>> idx  = pd.date_range("2018-01", periods=n, freq="MS")
>>> ts   = TimeSeries(vals, index=idx)
>>> r    = AnomalyDetector().stl_residual(ts, period=12)
>>> 25 in r.indices
True
gesd(ts, *, alpha=0.05, max_outliers=10)[source]

Global GESD anomaly detection.

Delegates to tseda.quality.OutlierDetector.gesd() and wraps the result as an AnomalyReport. Best suited for stationary series where the number of anomalies is small relative to n.

Parameters:

  • ts (TimeSeries) – Input series.

  • alpha (float, optional) – Significance level. Default 0.05.

  • max_outliers (int, optional) – Maximum number of anomalies to test for. Default 10.

Return type:

AnomalyReport

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector
>>> rng  = np.random.default_rng(0)
>>> idx  = pd.date_range("2020", periods=50, freq="D")
>>> vals = rng.standard_normal(50)
>>> vals[10] = 12.0
>>> ts   = TimeSeries(vals, index=idx)
>>> r    = AnomalyDetector().gesd(ts)
>>> 10 in r.indices
True
remove(ts, report)[source]

Replace detected anomaly values with NaN.

Parameters:
Returns:

A new series with anomaly positions set to NaN.

Return type:

TimeSeries

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector
>>> idx  = pd.date_range("2020", periods=50, freq="D")
>>> vals = np.zeros(50)
>>> vals[5] = 100.0
>>> ts   = TimeSeries(vals, index=idx)
>>> det  = AnomalyDetector()
>>> cleaned = det.remove(ts, det.rolling_iqr(ts))
>>> cleaned.has_nan
True
label(ts, report)[source]

Return a 0/1 TimeSeries of anomaly labels.

Parameters:

  • ts (TimeSeries) – The original series (used for index and metadata).

  • report (AnomalyReport) – Output of any detection method.

Returns:

Values are 1 at anomaly positions, 0 elsewhere. Name is "{ts.name}_anomaly_label".

Return type:

TimeSeries

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector
>>> idx  = pd.date_range("2020", periods=50, freq="D")
>>> vals = np.zeros(50); vals[5] = 100.0
>>> ts   = TimeSeries(vals, index=idx)
>>> det  = AnomalyDetector()
>>> lbl  = det.label(ts, det.rolling_iqr(ts))
>>> lbl.values[5]
1.0
>>> lbl.values[0]
0.0

Report

class tseda.anomaly.detector.AnomalyReport(mask, indices, timestamps, values, scores, method, n_anomalies)[source]

Bases: object

Immutable anomaly detection result.

Parameters:
mask

Boolean array of shape (n,); True where an anomaly was detected.

Type:

numpy.ndarray

indices

Integer positions (0-based) of detected anomalies.

Type:

numpy.ndarray

timestamps

Timestamps of detected anomalies.

Type:

pandas.DatetimeIndex

values

Observed values at anomaly positions.

Type:

numpy.ndarray

scores

Continuous anomaly score in [0, 1] for each observation. Higher values indicate stronger evidence of anomaly. 0 for normal observations.

Type:

numpy.ndarray

method

Name of the detection method.

Type:

str

n_anomalies

Number of anomalies detected.

Type:

int

mask: ndarray
indices: ndarray
timestamps: DatetimeIndex
values: ndarray
scores: ndarray
method: str
n_anomalies: int
__repr__()[source]

Return repr(self).

Return type:

str

__init__(mask, indices, timestamps, values, scores, method, n_anomalies)
Parameters:
Return type:

None

Detector

class tseda.anomaly.detector.AnomalyDetector[source]

Bases: object

Detect point and contextual anomalies in a TimeSeries.

This class is stateless — one instance, many series.

rolling_iqr(ts, window, k)[source]

Rolling Tukey IQR fence.

Parameters:
Return type:

AnomalyReport

rolling_z(ts, window, threshold)[source]

Rolling mean ± k × std.

Parameters:
Return type:

AnomalyReport

stl_residual(ts, period, method, threshold)[source]

STL decompose then flag large residuals.

Parameters:
Return type:

AnomalyReport

gesd(ts, alpha, max_outliers)[source]

Global Generalized ESD test (re-uses quality module).

Parameters:
Return type:

AnomalyReport

remove(ts, report)[source]

Replace anomaly positions with NaN.

Parameters:
Return type:

TimeSeries

label(ts, report)[source]

Return a 0/1 TimeSeries of anomaly labels.

Parameters:
Return type:

TimeSeries

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector

>>> rng  = np.random.default_rng(1)
>>> idx  = pd.date_range("2020", periods=100, freq="D")
>>> vals = rng.standard_normal(100)
>>> vals[20] = 12.0
>>> ts   = TimeSeries(vals, index=idx)
>>> det  = AnomalyDetector()
>>> r    = det.rolling_z(ts, window=30)
>>> 20 in r.indices
True
rolling_iqr(ts, window=30, *, k=2.5, center=True, min_periods=None)[source]

Detect anomalies using a rolling IQR fence.

An observation y[t] is flagged when it falls outside [Q1(t) − k×IQR(t),  Q3(t) + k×IQR(t)] where the quartiles are computed over a rolling window centred at t.

Parameters:

  • ts (TimeSeries) – Input series.

  • window (int, optional) – Rolling window width in observations. Default 30.

  • k (float, optional) – Fence multiplier. Default 2.5 (tighter than the global 1.5 because the window is local and thus more sensitive).

  • center (bool, optional) – Centre the window on each observation. Default True.

  • min_periods (int, optional) – Minimum non-NaN observations required in each window. Defaults to window // 2.

Return type:

AnomalyReport

Raises:

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector
>>> rng  = np.random.default_rng(0)
>>> idx  = pd.date_range("2020", periods=100, freq="D")
>>> vals = rng.standard_normal(100)
>>> vals[40] = 10.0
>>> ts   = TimeSeries(vals, index=idx)
>>> r    = AnomalyDetector().rolling_iqr(ts)
>>> 40 in r.indices
True
rolling_z(ts, window=30, *, threshold=3.0, center=True, min_periods=None)[source]

Detect anomalies using a rolling Z-score.

An observation is flagged when |y[t] − μ(t)| / σ(t) > threshold, where μ and σ are the rolling mean and standard deviation computed over a window.

Parameters:

  • ts (TimeSeries) – Input series.

  • window (int, optional) – Rolling window width. Default 30.

  • threshold (float, optional) – Z-score cut-off. Default 3.0.

  • center (bool, optional) – Centre the window. Default True.

  • min_periods (int, optional) – Minimum non-NaN observations per window. Defaults to window // 2.

Return type:

AnomalyReport

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector
>>> rng  = np.random.default_rng(2)
>>> idx  = pd.date_range("2020", periods=100, freq="D")
>>> vals = rng.standard_normal(100)
>>> vals[60] = -9.0
>>> ts   = TimeSeries(vals, index=idx)
>>> r    = AnomalyDetector().rolling_z(ts)
>>> 60 in r.indices
True
stl_residual(ts, period=None, *, residual_method='iqr', k=3.0, robust=True)[source]

Detect anomalies in the STL residual component.

Decomposes ts into trend + seasonal + residual using STL, then flags residual values that are extreme under the chosen criterion.

Parameters:

  • ts (TimeSeries) – Input series. Must be long enough for STL decomposition (at least 2 × period observations).

  • period (int, optional) – Seasonal period. Inferred from ts.freq when omitted.

  • residual_method (str, optional) –

    Criterion to flag residuals:

    • "iqr" — Tukey IQR fence with multiplier k (default).

    • "mad" — Median Absolute Deviation threshold k.

    • "z" — Z-score threshold k.

  • k (float, optional) – Threshold multiplier / fence factor. Default 3.0.

  • robust (bool, optional) – Use robust LOESS in STL. Default True.

Return type:

AnomalyReport

Raises:

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector
>>> rng  = np.random.default_rng(3)
>>> n    = 60
>>> seas = np.tile(np.sin(2*np.pi*np.arange(12)/12)*5, 5)
>>> vals = seas + rng.standard_normal(n) * 0.3
>>> vals[25] = 20.0
>>> idx  = pd.date_range("2018-01", periods=n, freq="MS")
>>> ts   = TimeSeries(vals, index=idx)
>>> r    = AnomalyDetector().stl_residual(ts, period=12)
>>> 25 in r.indices
True
gesd(ts, *, alpha=0.05, max_outliers=10)[source]

Global GESD anomaly detection.

Delegates to tseda.quality.OutlierDetector.gesd() and wraps the result as an AnomalyReport. Best suited for stationary series where the number of anomalies is small relative to n.

Parameters:

  • ts (TimeSeries) – Input series.

  • alpha (float, optional) – Significance level. Default 0.05.

  • max_outliers (int, optional) – Maximum number of anomalies to test for. Default 10.

Return type:

AnomalyReport

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector
>>> rng  = np.random.default_rng(0)
>>> idx  = pd.date_range("2020", periods=50, freq="D")
>>> vals = rng.standard_normal(50)
>>> vals[10] = 12.0
>>> ts   = TimeSeries(vals, index=idx)
>>> r    = AnomalyDetector().gesd(ts)
>>> 10 in r.indices
True
remove(ts, report)[source]

Replace detected anomaly values with NaN.

Parameters:
Returns:

A new series with anomaly positions set to NaN.

Return type:

TimeSeries

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector
>>> idx  = pd.date_range("2020", periods=50, freq="D")
>>> vals = np.zeros(50)
>>> vals[5] = 100.0
>>> ts   = TimeSeries(vals, index=idx)
>>> det  = AnomalyDetector()
>>> cleaned = det.remove(ts, det.rolling_iqr(ts))
>>> cleaned.has_nan
True
label(ts, report)[source]

Return a 0/1 TimeSeries of anomaly labels.

Parameters:

  • ts (TimeSeries) – The original series (used for index and metadata).

  • report (AnomalyReport) – Output of any detection method.

Returns:

Values are 1 at anomaly positions, 0 elsewhere. Name is "{ts.name}_anomaly_label".

Return type:

TimeSeries

Examples

>>> import numpy as np, pandas as pd
>>> from tseda import TimeSeries
>>> from tseda.anomaly.detector import AnomalyDetector
>>> idx  = pd.date_range("2020", periods=50, freq="D")
>>> vals = np.zeros(50); vals[5] = 100.0
>>> ts   = TimeSeries(vals, index=idx)
>>> det  = AnomalyDetector()
>>> lbl  = det.label(ts, det.rolling_iqr(ts))
>>> lbl.values[5]
1.0
>>> lbl.values[0]
0.0