[ENH] Implemented COPOD for anomaly detection

aeon/anomaly_detection/__init__.py

@@ -9,8 +9,10 @@
     "STOMP",
     "LeftSTAMPi",
     "IsolationForest",
+    "COPOD",
 ]
 
+from aeon.anomaly_detection._copod import COPOD
 from aeon.anomaly_detection._dwt_mlead import DWT_MLEAD
 from aeon.anomaly_detection._iforest import IsolationForest
 from aeon.anomaly_detection._kmeans import KMeansAD

aeon/anomaly_detection/_copod.py

@@ -0,0 +1,87 @@
+"""COPOD for anomaly detection."""
+
+__maintainer__ = []
+__all__ = ["COPOD"]
+
+from typing import Union
+
+import numpy as np
+
+from aeon.anomaly_detection._pyodadapter import PyODAdapter
+from aeon.utils.validation._dependencies import _check_soft_dependencies
+
+
+class COPOD(PyODAdapter):
+    """COPOD for anomaly detection.
+
+    This class implements the COPOD using PyODAdadpter to be used in the aeon framework.
+    The parameter `n_jobs` is passed to COPOD model from PyOD, `window_size` and
+    `stride` are used to construct the sliding windows.
+
+    .. list-table:: Capabilities
+       :stub-columns: 1
+       * - Input data format
+         - univariate and multivariate
+       * - Output data format
+         - anomaly scores
+       * - Learning Type
+         - unsupervised or semi-supervised
+
+    Parameters
+    ----------
+    n_jobs : int, default=1
+        The number of jobs to run in parallel for the COPOD model.
+
+    window_size : int, default=10
+        Size of the sliding window.
+
+    stride : int, default=1
+        Stride of the sliding window.
+    """
+
+    _tags = {
+        "capability:multivariate": True,
+        "capability:univariate": True,
+        "capability:missing_values": False,
+        "fit_is_empty": False,
+        "python_dependencies": ["pyod"],
+    }
+
+    def __init__(self, n_jobs: int = 1, window_size: int = 10, stride: int = 1):
+        _check_soft_dependencies(*self._tags["python_dependencies"])
+        from pyod.models.copod import COPOD
+
+        model = COPOD(n_jobs=n_jobs)
+        self.n_jobs = n_jobs
+        super().__init__(model, window_size=window_size, stride=stride)
+
+    def _fit(self, X: np.ndarray, y: Union[np.ndarray, None] = None) -> None:
+        super()._fit(X, y)
+
+    def _predict(self, X: np.ndarray) -> np.ndarray:
+        return super()._predict(X)
+
+    def _fit_predict(
+        self, X: np.ndarray, y: Union[np.ndarray, None] = None
+    ) -> np.ndarray:
+        return super()._fit_predict(X, y)
+
+    @classmethod
+    def get_test_params(cls, parameter_set="default") -> dict:
+        """Return testing parameter settings for the estimator.
+
+        Parameters
+        ----------
+        parameter_set : str, default="default"
+            Name of the set of test parameters to return, for use in tests. If no
+            special parameters are defined for a value, will return `"default"` set.
+
+        Returns
+        -------
+        params : dict or list of dict, default={}
+            Parameters to create testing instances of the class.
+            Each dict are parameters to construct an "interesting" test instance, i.e.,
+            `MyClass(**params)` or `MyClass(**params[i])` creates a valid test instance.
+            `create_test_instance` uses the first (or only) dictionary in `params`.
+        """
+        return {}

aeon/anomaly_detection/tests/test_copod.py

@@ -0,0 +1,61 @@
+"""Tests for the COPOD class."""
+
+import numpy as np
+import pytest
+
+from aeon.anomaly_detection import COPOD
+from aeon.testing.data_generation import make_example_1d_numpy
+from aeon.utils.validation._dependencies import _check_soft_dependencies
+
+
+@pytest.mark.skipif(
+    not _check_soft_dependencies("pyod", severity="none"),
+    reason="required soft dependency PyOD not available",
+)
+def test_copod_default():
+    """Test COPOD."""
+    series = make_example_1d_numpy(n_timepoints=80, random_state=0)
+    series[50:58] -= 2
+
+    copod = COPOD(window_size=10, stride=1)
+    pred = copod.fit_predict(series, axis=0)
+
+    assert pred.shape == (80,)
+    assert pred.dtype == np.float_
+    assert 50 <= np.argmax(pred) <= 60
+
+
+@pytest.mark.skipif(
+    not _check_soft_dependencies("pyod", severity="none"),
+    reason="required soft dependency PyOD not available",
+)
+def test_copod_pyod_parameters():
+    """Test parameters are correctly passed to the PyOD model."""
+    params = {"n_jobs": 2}
+    copod = COPOD(**params)
+
+    assert copod.pyod_model.n_jobs == params["n_jobs"]
+
+
+@pytest.mark.skipif(
+    not _check_soft_dependencies("pyod", severity="none"),
+    reason="required soft dependency PyOD not available",
+)
+def test_aeon_copod_with_pyod_copod():
+    """Test COPOD with PyOD COPOD."""
+    from pyod.models.copod import COPOD as PyODCOPOD
+
+    series = make_example_1d_numpy(n_timepoints=100, random_state=0)
+    series[20:30] -= 2
+
+    # fit and predict with aeon COPOD
+    copod = COPOD(window_size=1, stride=1)
+    copod_preds = copod.fit_predict(series)
+
+    # fit and predict with PyOD COPOD
+    _series = series.reshape(-1, 1)
+    pyod_copod = PyODCOPOD()
+    pyod_copod.fit(_series)
+    pyod_copod_preds = pyod_copod.decision_function(_series)
+
+    assert np.allclose(copod_preds, pyod_copod_preds)

docs/api_reference/anomaly_detection.rst

@@ -69,6 +69,7 @@ Detectors
     :toctree: auto_generated/
     :template: class.rst
 
+    COPOD
     DWT_MLEAD
     IsolationForest
     KMeansAD