Online-SSA

pysatl_cpd/core/algorithms/bayesian_linear_heuristic.py

@@ -89,7 +89,7 @@ def detect(self, observation: np.float64 | npt.NDArray[np.float64]) -> bool:
         :param observation: a new observation from a time series. Note: only univariate data is supported for now.
         :return: whether a change point was detected by a main algorithm.
         """
-        if observation is npt.NDArray[np.float64]:
+        if isinstance(observation, np.ndarray):
             raise TypeError("Multivariate observations are not supported")
         assert self.__main_algorithm is not None, "Main algorithm must be initialized"
 
@@ -111,7 +111,7 @@ def localize(self, observation: np.float64 | npt.NDArray[np.float64]) -> Optiona
         :param observation: a new observation from a time series. Note: only univariate data is supported for now.
         :return: a change point, if it was localized, None otherwise.
         """
-        if observation is npt.NDArray[np.float64]:
+        if isinstance(observation, np.ndarray):
             raise TypeError("Multivariate observations are not supported")
         assert self.__main_algorithm is not None, "Main algorithm must be initialized"
 

pysatl_cpd/core/algorithms/bayesian_online_algorithm.py

@@ -188,7 +188,7 @@ def detect(self, observation: np.float64 | npt.NDArray[np.float64]) -> bool:
         :param observation: new observation of a time series. Note: multivariate time series aren't supported for now.
         :return: whether a change point was detected after processing the new observation.
         """
-        if observation is npt.NDArray[np.float64]:
+        if isinstance(observation, np.ndarray):
             raise TypeError("Multivariate observations are not supported")
 
         self.__process_point(np.float64(observation), False)
@@ -203,7 +203,7 @@ def localize(self, observation: np.float64 | npt.NDArray[np.float64]) -> Optiona
         :return: absolute location of a change point, acquired after processing the new observation,
         or None if there wasn't any.
         """
-        if observation is npt.NDArray[np.float64]:
+        if isinstance(observation, np.ndarray):
             raise TypeError("Multivariate observations are not supported")
 
         self.__process_point(np.float64(observation), True)

pysatl_cpd/core/algorithms/ssa/__init__.py

@@ -0,0 +1,33 @@
+"""
+Module for SSA CPD algorithm's customization blocks.
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+from pysatl_cpd.core.algorithms.ssa.abstracts import (
+    SVD,
+    IDecomposition,
+    IDetectorSSA,
+    IEmbedding,
+    IGrouping,
+)
+from pysatl_cpd.core.algorithms.ssa.decomposition import BasicSVD
+from pysatl_cpd.core.algorithms.ssa.detectors import DistanceThreshold
+from pysatl_cpd.core.algorithms.ssa.embedding import BasicEmbedding
+from pysatl_cpd.core.algorithms.ssa.grouping import ConstantGrouping
+from pysatl_cpd.core.algorithms.ssa.ssa import SSA
+
+__all__ = [
+    "SSA",
+    "SVD",
+    "BasicEmbedding",
+    "BasicSVD",
+    "ConstantGrouping",
+    "DistanceThreshold",
+    "IDecomposition",
+    "IDetectorSSA",
+    "IEmbedding",
+    "IGrouping",
+]

pysatl_cpd/core/algorithms/ssa/abstracts/__init__.py

@@ -0,0 +1,14 @@
+"""
+Module for abstract base classes for SSA CPD algorithm.
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+from pysatl_cpd.core.algorithms.ssa.abstracts.idecomposition import SVD, IDecomposition
+from pysatl_cpd.core.algorithms.ssa.abstracts.idetector import IDetectorSSA
+from pysatl_cpd.core.algorithms.ssa.abstracts.iembedding import IEmbedding
+from pysatl_cpd.core.algorithms.ssa.abstracts.igrouping import IGrouping
+
+__all__ = ["SVD", "IDecomposition", "IDetectorSSA", "IEmbedding", "IGrouping"]

pysatl_cpd/core/algorithms/ssa/abstracts/idecomposition.py

@@ -0,0 +1,37 @@
+"""
+Module for the SSA decomposition step base class.
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+import numpy as np
+import numpy.typing as npt
+
+
+@dataclass
+class SVD:
+    """
+    Dataclass of the singular value decomposition.
+    """
+
+    U: npt.NDArray[np.float64]
+    sigma: npt.NDArray[np.float64]
+
+
+class IDecomposition(ABC):
+    """
+    Abstract class of the second step of SSA (Decomposition step).
+    """
+
+    @abstractmethod
+    def decompose(self, X: npt.NDArray[np.float64]) -> SVD:
+        """
+        Decomposes the trajectory matrix into elementary matrices.
+        :param X: trajectory matrix from embedding step.
+        :return: matrix decomposition for grouping step.
+        """

pysatl_cpd/core/algorithms/ssa/abstracts/idetector.py

@@ -0,0 +1,35 @@
+"""
+Module for SSA CPD algorithm detector's abstract base class.
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+from abc import ABC, abstractmethod
+
+import numpy as np
+import numpy.typing as npt
+
+
+class IDetectorSSA(ABC):
+    """
+    Abstract class for detectors that detect a change point.
+    """
+
+    @abstractmethod
+    def detect(
+        self, subspace: npt.NDArray[np.float64], test_data: list[np.float64]
+    ) -> bool:
+        """
+        Checks whether a changepoint has occurred at the start of the test data.
+        :param subspace: vectors of the training set subspace.
+        :param test_data: test sample for breakpoint detection.
+        :return: boolean indicating whether a changepoint occurred.
+        """
+
+    @abstractmethod
+    def clean(self) -> None:
+        """
+        Clears the detector's state.
+        """

pysatl_cpd/core/algorithms/ssa/abstracts/iembedding.py

@@ -0,0 +1,29 @@
+"""
+Module for the SSA embedding step base class.
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+from abc import ABC, abstractmethod
+
+import numpy as np
+import numpy.typing as npt
+
+
+class IEmbedding(ABC):
+    """
+    Abstract class of the first step of SSA (Embedding step).
+    """
+
+    @abstractmethod
+    def transform(
+        self, segment: npt.NDArray[np.float64], L: int
+    ) -> npt.NDArray[np.float64]:
+        """
+        Converts a segment of the time series into a trajectory matrix.
+        :param segment: segment of the time series.
+        :param L: window width.
+        :return: trajectory matrix for decomposition step.
+        """

pysatl_cpd/core/algorithms/ssa/abstracts/igrouping.py

@@ -0,0 +1,28 @@
+"""
+Module for the SSA grouping step base class.
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+from abc import ABC, abstractmethod
+
+import numpy as np
+import numpy.typing as npt
+
+from pysatl_cpd.core.algorithms.ssa.abstracts.idecomposition import SVD
+
+
+class IGrouping(ABC):
+    """
+    Abstract class of the third step of SSA (Grouping step).
+    """
+
+    @abstractmethod
+    def group(self, svd: SVD) -> npt.NDArray[np.float64]:
+        """
+        Groups vectors to define a subspace of the time series.
+        :param svd: decomposition of trajectory matrix.
+        :return: vector group characterizing a subspace.
+        """

pysatl_cpd/core/algorithms/ssa/decomposition/__init__.py

@@ -0,0 +1,11 @@
+"""
+Module for implementations of SSA decomposition step
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+from pysatl_cpd.core.algorithms.ssa.decomposition.standard_svd import BasicSVD
+
+__all__ = ["BasicSVD"]

pysatl_cpd/core/algorithms/ssa/decomposition/standard_svd.py

@@ -0,0 +1,27 @@
+"""
+Module for implementation of basic SSA decomposition step.
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+import numpy as np
+import numpy.typing as npt
+
+from pysatl_cpd.core.algorithms.ssa.abstracts import SVD, IDecomposition
+
+
+class BasicSVD(IDecomposition):
+    """
+    Class of basic SSA decomposition step based on SVD.
+    """
+
+    def decompose(self, X: npt.NDArray[np.float64]) -> SVD:
+        """
+        Decomposes the trajectory matrix using SVD.
+        :param X: trajectory matrix from embedding step.
+        :return: matrix decomposition for grouping step.
+        """
+        U, sigma, _ = np.linalg.svd(X, full_matrices=False)
+        return SVD(U=U, sigma=sigma)

pysatl_cpd/core/algorithms/ssa/detectors/__init__.py

@@ -0,0 +1,13 @@
+"""
+Module for implementations of SSA CPD algorithm detectors.
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+from pysatl_cpd.core.algorithms.ssa.detectors.distance_threshold import (
+    DistanceThreshold,
+)
+
+__all__ = ["DistanceThreshold"]

pysatl_cpd/core/algorithms/ssa/detectors/distance_threshold.py

@@ -0,0 +1,59 @@
+"""
+Module for implementations of SSA CPD algorithm distance detector using a threshold.
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+import numpy as np
+import numpy.typing as npt
+
+from pysatl_cpd.core.algorithms.ssa.abstracts.idetector import IDetectorSSA
+
+
+class DistanceThreshold(IDetectorSSA):
+    """
+    Class of implementations of SSA CPD algorithm detector using
+    the normalized Euclidean distance and a threshold.
+    """
+
+    def __init__(self, threshold: float) -> None:
+        """
+        Initializes SSA CPD algorithm distance detector with given threshold.
+        :param threshold: threshold for distance calculation.
+        """
+        if not(0 <= threshold <= 1):
+            raise ValueError("Threshold must be in [0.0, 1.0]")
+        self._threshold = threshold
+
+    def detect(
+        self, subspace: npt.NDArray[np.float64], test_data: list[np.float64]
+    ) -> bool:
+        """
+        Checks whether a changepoint has occurred at the start of the test data using
+        the normalized Euclidean distance and comparing it to a threshold.
+        :param subspace: vectors of the training set subspace.
+        :param test_data: test sample for breakpoint detection.
+        :return: boolean indicating whether a changepoint occurred.
+        """
+        L = subspace.shape[0]
+        n_test = len(test_data) - L + 1
+        X_test = np.zeros((L, n_test))
+
+        for i in range(n_test):
+            X_test[:, i] = test_data[i : i + L]
+
+        proj_sum = np.sum((subspace.T @ X_test) ** 2)
+        total_sum = np.sum(X_test**2)
+
+        if total_sum == 0:
+            return False
+
+        return bool(1 - (proj_sum / total_sum) > self._threshold)
+
+    def clean(self) -> None:
+        """
+        Clears the detector's state.
+        """
+        return

pysatl_cpd/core/algorithms/ssa/embedding/__init__.py

@@ -0,0 +1,11 @@
+"""
+Module for implementations of SSA embedding step
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+from pysatl_cpd.core.algorithms.ssa.embedding.basic_embedding import BasicEmbedding
+
+__all__ = ["BasicEmbedding"]

pysatl_cpd/core/algorithms/ssa/embedding/basic_embedding.py

@@ -0,0 +1,34 @@
+"""
+Module for implementation of basic SSA embedding step.
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+import numpy as np
+import numpy.typing as npt
+
+from pysatl_cpd.core.algorithms.ssa.abstracts.iembedding import IEmbedding
+
+
+class BasicEmbedding(IEmbedding):
+    """
+    Class of basic SSA embedding step based on a sliding window.
+    """
+
+    def transform(
+        self, segment: npt.NDArray[np.float64], L: int
+    ) -> npt.NDArray[np.float64]:
+        """
+        Converts a segment of the time series into a trajectory matrix based on a sliding window.
+        :param segment: segment of the time series.
+        :param L: window width.
+        :return: trajectory matrix for decomposition step.
+        """
+        K = len(segment) - L + 1
+        X = np.zeros((L, K))
+        for i in range(K):
+            X[:, i] = segment[i : i + L]
+
+        return X

pysatl_cpd/core/algorithms/ssa/grouping/__init__.py

@@ -0,0 +1,11 @@
+"""
+Module for implementations of SSA grouping step
+"""
+
+__author__ = "Mark Dubrovchenko"
+__copyright__ = "Copyright (c) 2026 PySATL project"
+__license__ = "SPDX-License-Identifier: MIT"
+
+from pysatl_cpd.core.algorithms.ssa.grouping.constant_grouping import ConstantGrouping
+
+__all__ = ["ConstantGrouping"]