TSPulseAnomalyDetector

class TSPulseAnomalyDetector(model_path: str = 'ibm-granite/granite-timeseries-tspulse-r1', revision: str = 'main', mask_type: str = 'user', prediction_mode=None, aggregation_length: int = 64, aggr_function: str = 'max', smoothing_length: int = 8, least_significant_scale: float = 0.01, least_significant_score: float = 0.1, batch_size: int = 128, predictive_score_smoothing: bool = False, anomaly_threshold: float | None = 0.6, anomaly_percentile: float | None = None, config: dict | None = None)[source]

Anomaly detector wrapping IBM TSPulse via granite-tsfm.

Loads a pretrained TSPulseForReconstruction checkpoint and scores each time point with the Hugging Face TimeSeriesAnomalyDetectionPipeline for zero-shot detection.

DatetimeIndex on X is required: it is copied into an internal timestamp column. If X has no DatetimeIndex, a synthetic daily timestamp column is added.

Implementation adapted from [1].

Parameters:

model_pathstr, default=”ibm-granite/granite-timeseries-tspulse-r1”

Hugging Face model id or local path.

revisionstr, default=”main”

Model revision on the Hugging Face Hub.

mask_typestr, default=”user”

Patch-masking strategy during reconstruction. "user" applies masking from an optional past_observed_mask (the default for anomaly scoring). Other supported values include "block", "hybrid", "var_hybrid", and "random".

prediction_modestr or list of str, optional

Score type(s) computed on each context window before they are merged. Supported values:

"time": mean squared error between the input and the model’s time-domain reconstruction over the last aggregation_length points of the context window.
"fft": same error using the reconstruction from the frequency (FFT) branch.
"forecast": mean squared error between the next observed value and the model’s one-step forecast.

When None (default), uses ["time", "fft"]. A single string is wrapped in a one-element list. With several types, each produces its own score sequence; see smoothing_length and aggr_function.

aggregation_lengthint, default=64

Length of the context suffix used for patchwise stitched reconstruction when scoring with "time" or "fft". Also controls boundary padding when scores are aligned back to the full series length.

aggr_functionstr, default=”max”

How to merge score sequences from different entries in prediction_mode at each time point. One of "max", "min", or "mean".

smoothing_lengthint, default=8

Moving-average window applied to each score sequence listed in prediction_mode after boundary alignment, before aggr_function merges them. One-step "forecast" scores skip this smoothing unless predictive_score_smoothing is True.

least_significant_scalefloat, default=0.01

Value in (0, 1). Sets a variance-based floor on raw errors before scores are rescaled: deviations smaller than this fraction of the squared differences in the (standardized) input are treated as insignificant.

least_significant_scorefloat, default=0.1

Minimum scale factor applied when normalizing significant errors to the [0, 1] score range returned by the pipeline.

batch_sizeint, default=128

Batch size for pipeline inference.

predictive_score_smoothingbool, default=False

If True, apply smoothing_length smoothing to "forecast" scores as well; if False, forecast scores are left unsmoothed.

anomaly_thresholdfloat, optional, default=0.6

Scores strictly above this value are flagged as anomalies in predict. If None, anomaly_percentile is used instead.

anomaly_percentilefloat, optional, default=None

Percentile of the score vector on X used as the detection threshold when anomaly_threshold is None. Exactly one of anomaly_threshold and anomaly_percentile must be set.

configdict, optional, default=None

Extra keyword arguments forwarded to TSPulseForReconstruction.from_pretrained.

Any key you supply overrides the built-in default for that key, except for num_input_channels, which is always set from the number of columns in X at fit time.

If config is None, the following default override is applied:

ignore_mismatched_sizes=True: allows loading when channel or head shapes differ from the checkpoint

Other commonly useful overrides (see TSPulseConfig in granite-tsfm):

context_length: history length per window (checkpoint default often 512 for r1 models)
patch_length / patch_stride: patch size and stride for the encoder
decoder_mode: channel mixing in the decoder ("mix_channel" or "common_channel")
scaling: input normalization ("revin", "mean", "std", or None)
mask_ratio: fraction of patches masked when mask_type is not "user"
fft_time_consistent_masking: if True, masked series is used for the FFT branch during training-style masking
reconstruction_type: "patchwise" or "full" reconstruction
prediction_length: horizon for the forecast head when using prediction_mode="forecast"

Attributes:

is_fitted: Whether fit has been called.

References

[1]

https://github.com/ibm-granite/granite-tsfm/blob/main/notebooks/hfdemo/tspulse_anomaly_detection.ipynb

[2]

https://arxiv.org/abs/2505.13033

Examples

>>> from sktime.detection.tspulse import TSPulseAnomalyDetector
>>> import pandas as pd
>>> import numpy as np
>>> idx = pd.date_range("2020-01-01", periods=200, freq="D")
>>> X = pd.DataFrame(np.random.randn(200, 1), index=idx, columns=["value"])
>>> detector = TSPulseAnomalyDetector()
>>> detector.fit(X)
>>> detector.predict(X)
>>> detector.predict_scores(X)

Methods

`change_points_to_segments`(y_sparse[, start, end])	Convert an series of change point indexes to segments.
`check_is_fitted`([method_name])	Check if the estimator has been fitted.
`clone`()	Obtain a clone of the object with same hyper-parameters and config.
`clone_tags`(estimator[, tag_names])	Clone tags from another object as dynamic override.
`create_test_instance`([parameter_set])	Construct an instance of the class, using first test parameter set.
`create_test_instances_and_names`([parameter_set])	Create list of all test instances and a list of names for them.
`dense_to_sparse`(y_dense)	Convert the dense output from an detector to a sparse format.
`fit`(X[, y])	Fit to training data.
`fit_predict`(X[, y])	Fit to data, then predict it.
`fit_transform`(X[, y])	Fit to data, then transform it.
`get_class_tag`(tag_name[, tag_value_default])	Get class tag value from class, with tag level inheritance from parents.
`get_class_tags`()	Get class tags from class, with tag level inheritance from parent classes.
`get_config`()	Get config flags for self.
`get_fitted_params`([deep])	Get fitted parameters.
`get_param_defaults`()	Get object's parameter defaults.
`get_param_names`([sort])	Get object's parameter names.
`get_params`([deep])	Get a dict of parameters values for this object.
`get_tag`(tag_name[, tag_value_default, ...])	Get tag value from instance, with tag level inheritance and overrides.
`get_tags`()	Get tags from instance, with tag level inheritance and overrides.
`get_test_params`([parameter_set])	Return testing parameter settings for the estimator.
`is_composite`()	Check if the object is composed of other BaseObjects.
`load_from_path`(serial)	Load object from file location.
`load_from_serial`(serial)	Load object from serialized memory container.
`predict`(X)	Create labels on test/deployment data.
`predict_points`(X)	Predict changepoints/anomalies on test/deployment data.
`predict_scores`(X)	Return scores for predicted labels on test/deployment data.
`predict_segments`(X)	Predict segments on test/deployment data.
`reset`()	Reset the object to a clean post-init state.
`save`([path, serialization_format])	Save serialized self to bytes-like object or to (.zip) file.
`segments_to_change_points`(y_sparse)	Convert segments to change points.
`set_config`(**config_dict)	Set config flags to given values.
`set_params`(**params)	Set the parameters of this object.
`set_random_state`([random_state, deep, ...])	Set random_state pseudo-random seed parameters for self.
`set_tags`(**tag_dict)	Set instance level tag overrides to given values.
`sparse_to_dense`(y_sparse, index)	Convert the sparse output from an detector to a dense format.
`transform`(X)	Create labels on test/deployment data.
`transform_scores`(X)	Return scores for predicted labels on test/deployment data.
`update`(X[, y])	Update model with new data and optional ground truth labels.
`update_predict`(X[, y])	Update model with new data and create labels for it.