From 119016d6a5778343f627f3ea0ae4088f3098d2e8 Mon Sep 17 00:00:00 2001 From: Tony Bagnall Date: Tue, 30 Jun 2026 18:39:41 -0400 Subject: [PATCH] left stampi --- .../series/distance_based/_left_stampi.py | 36 +++----- .../distance_based/tests/test_left_stampi.py | 88 +++++++++++-------- aeon/testing/testing_config.py | 2 - 3 files changed, 64 insertions(+), 62 deletions(-) diff --git a/aeon/anomaly_detection/series/distance_based/_left_stampi.py b/aeon/anomaly_detection/series/distance_based/_left_stampi.py index fa4ceed226..6772e5d8db 100644 --- a/aeon/anomaly_detection/series/distance_based/_left_stampi.py +++ b/aeon/anomaly_detection/series/distance_based/_left_stampi.py @@ -22,6 +22,11 @@ class LeftSTAMPi(BaseSeriesAnomalyDetector): LeftSTAMPi supports univariate time series only. + This estimator is unsupervised and ``fit`` is a no-op. Calls to ``predict(X)`` + and ``fit_predict(X)`` treat ``X`` as an independent stream: the initial + matrix profile is seeded from the first ``n_init_train`` points in ``X`` and + then updated with the remaining points in ``X``. State is not carried between + separate series passed to ``predict``. Parameters ---------- @@ -66,7 +71,7 @@ class LeftSTAMPi(BaseSeriesAnomalyDetector): "capability:univariate": True, "capability:multivariate": False, "capability:missing_values": False, - "fit_is_empty": False, + "fit_is_empty": True, "cant_pickle": True, "python_dependencies": ["stumpy"], "anomaly_output_type": "anomaly_scores", @@ -81,7 +86,6 @@ def __init__( p: float = 2.0, k: int = 1, ): - self.mp_: np.ndarray | None = None self.window_size = window_size self.n_init_train = n_init_train self.normalize = normalize @@ -109,41 +113,29 @@ def _check_params(self, X): "the time series minus the window size." ) - def _fit(self, X: np.ndarray, y=None) -> "LeftSTAMPi": + def _predict(self, X: np.ndarray) -> np.ndarray: if X.ndim > 1: X = X.squeeze() self._check_params(X) - self._call_stumpi(X) - - return self + init = X[: self.n_init_train] + stream = X[self.n_init_train :] - def _predict(self, X: np.ndarray) -> np.ndarray: - if X.ndim > 1: - X = X.squeeze() - self._check_params(X) + mp = self._call_stumpi(init) - for x in X: - self.mp_.update(x) + for x in stream: + mp.update(x) - lmp = self.mp_._left_P + lmp = mp._left_P lmp[: self.n_init_train] = 0 point_anomaly_scores = reverse_windowing(lmp, self.window_size) return point_anomaly_scores - def _fit_predict(self, X: np.ndarray, y=None) -> np.ndarray: - if X.ndim > 1: - X = X.squeeze() - - self.fit(X[: self.n_init_train]) - - return self.predict(X[self.n_init_train :]) - def _call_stumpi(self, X: np.ndarray): import stumpy - self.mp_ = stumpy.stumpi( + return stumpy.stumpi( X, m=self.window_size, egress=False, diff --git a/aeon/anomaly_detection/series/distance_based/tests/test_left_stampi.py b/aeon/anomaly_detection/series/distance_based/tests/test_left_stampi.py index 2e14928625..4bbd17fe64 100644 --- a/aeon/anomaly_detection/series/distance_based/tests/test_left_stampi.py +++ b/aeon/anomaly_detection/series/distance_based/tests/test_left_stampi.py @@ -25,39 +25,20 @@ def __init__( p, k, ): - self.X = X + self.X = np.asarray(X) self.m = m self.egress = egress self.normalize = normalize self.p = p self.k = k - self._left_P = np.array( - [ - 5.92, - 4.43, - 3.07, - 1.25, - 3.07, - 2.82, - 1.19, - 0.81, - 0.58, - 0.75, - 0.70, - 0.10, - 0.35, - 0.47, - 0.77, - 0.82, - 0.62, - ] - ) - + n_windows = self.X.shape[0] - self.m + 1 + self._left_P = np.arange(n_windows, dtype=np.float64) self._update_count = 0 def update(self, X): """Fake update method.""" + self._left_P = np.append(self._left_P, self._left_P.shape[0]) self._update_count += 1 @@ -65,9 +46,12 @@ def update(self, X): def mock_stumpy_pkg(): """Mock stumpy package.""" mock_pkg = mock.MagicMock() + mock_pkg.stumpi_instances = [] def fake_stumpi(X, m, egress, normalize, p, k): - return FakeStumpi(X, m, egress, normalize, p, k) + stumpi = FakeStumpi(X, m, egress, normalize, p, k) + mock_pkg.stumpi_instances.append(stumpi) + return stumpi mock_pkg.stumpi.side_effect = fake_stumpi with mock.patch.dict("sys.modules", {"stumpy": mock_pkg}): @@ -81,22 +65,21 @@ class TestLeftSTAMPi: not _check_soft_dependencies("stumpy", severity="none"), reason="required soft dependency stumpy not available", ) - def test_functional_it_allows_batch_processing_two_step(self): - """Functional testing the batch mode without mocking stumpy.""" + def test_functional_predict_after_fit_is_independent_series(self): + """Functional testing predict returns scores for the predict series only.""" # given - series = make_example_1d_numpy(n_timepoints=20, random_state=42) - series[7:10] += 3 + train = make_example_1d_numpy(n_timepoints=20, random_state=42) + test = make_example_1d_numpy(n_timepoints=10, random_state=43) - model = LeftSTAMPi(window_size=4, n_init_train=5) + model = LeftSTAMPi(window_size=3, n_init_train=3) # when - model = model.fit(series[:5]) - pred = model.predict(series[5:]) + model = model.fit(train) + pred = model.predict(test) # then - assert pred.shape == (20,) + assert pred.shape == (10,) assert pred.dtype == np.float64 - assert np.argmax(pred) == 8 @pytest.mark.skipif( not _check_soft_dependencies("stumpy", severity="none"), @@ -130,13 +113,42 @@ def test_it_allows_batch_processing(self, mock_stumpy_pkg): pred = ad.fit_predict(series) # then - mock_stumpy_pkg.stumpi.has_been_called_once_with( - series[:5], m=4, egress=False, normalize=True, p=2 - ) - assert ad.mp_._update_count == 15 + mock_stumpy_pkg.stumpi.assert_called_once() + call_args, call_kwargs = mock_stumpy_pkg.stumpi.call_args + np.testing.assert_array_equal(call_args[0], series[:5]) + assert call_kwargs == { + "m": 4, + "egress": False, + "normalize": True, + "p": 2.0, + "k": 1, + } + assert mock_stumpy_pkg.stumpi_instances[-1]._update_count == 15 assert pred.shape == (20,) assert pred.dtype == np.float64 - assert np.argmax(pred) == 8 + + def test_predict_after_fit_returns_predict_series_length(self, mock_stumpy_pkg): + """Unit testing predict does not append to the fit series.""" + train = make_example_1d_numpy(n_timepoints=20, random_state=42) + test = make_example_1d_numpy(n_timepoints=10, random_state=43) + ad = LeftSTAMPi(window_size=3, n_init_train=3) + + ad.fit(train) + pred = ad.predict(test) + + mock_stumpy_pkg.stumpi.assert_called_once() + call_args, call_kwargs = mock_stumpy_pkg.stumpi.call_args + np.testing.assert_array_equal(call_args[0], test[:3]) + assert call_kwargs == { + "m": 3, + "egress": False, + "normalize": True, + "p": 2.0, + "k": 1, + } + assert mock_stumpy_pkg.stumpi_instances[-1]._update_count == 7 + assert pred.shape == (10,) + assert pred.dtype == np.float64 def test_window_size_defaults_to_3(self, mock_stumpy_pkg): """Unit testing the default window size.""" diff --git a/aeon/testing/testing_config.py b/aeon/testing/testing_config.py index d1a009e745..d6b9fef17f 100644 --- a/aeon/testing/testing_config.py +++ b/aeon/testing/testing_config.py @@ -46,8 +46,6 @@ "check_persistence_via_pickle", "check_save_estimators_to_file", ], - # needs investigation - "LeftSTAMPi": ["check_series_anomaly_detector_output"], "SeriesToCollectionBroadcaster": ["check_transform_inverse_transform_equivalent"], "CollectionToSeriesWrapper": ["check_transform_inverse_transform_equivalent"], # missed in legacy testing, changes state in predict/transform