Add integration change detection test

Author: jaime-cespedes-sisniega

frouros/detectors/concept_drift/streaming/change_detection/base.py

@@ -0,0 +1,23 @@
+"""Base concept drift ChangeDetection based module."""
+
+import abc
+from typing import Union
+
+from frouros.detectors.concept_drift.streaming.base import (
+    BaseConceptDriftStreaming,
+    BaseConceptDriftStreamingConfig,
+)
+
+
+class BaseChangeDetectionConfig(BaseConceptDriftStreamingConfig):
+    """Class representing a ChangeDetection based configuration class."""
+
+
+class BaseChangeDetection(BaseConceptDriftStreaming):
+    """ChangeDetection based algorithm class."""
+
+    config_type = BaseChangeDetectionConfig
+
+    @abc.abstractmethod
+    def _update(self, value: Union[int, float], **kwargs) -> None:
+        pass

frouros/tests/conftest.py

@@ -50,6 +50,34 @@ def sea() -> SEA:
     return generator
 
 
+@pytest.fixture(scope="function")
+def stream_drift() -> np.ndarray:
+    """Stream with drift.
+
+    :return: stream with drift
+    :rtype: np.ndarray
+    """
+    np.random.seed(seed=31)
+
+    dist_normal = np.random.normal(
+        loc=0.0,
+        scale=0.5,
+        size=100,
+    )
+    dist_drift_1 = np.random.normal(
+        loc=4.0,
+        scale=0.2,
+        size=100,
+    )
+    dist_drift_2 = np.random.normal(
+        loc=6.0,
+        scale=0.5,
+        size=100,
+    )
+
+    return np.concatenate((dist_normal, dist_drift_1, dist_drift_2))
+
+
 @pytest.fixture(scope="module", name="clf_dataset")
 def classification_dataset() -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
     """Classification dataset using SEA generator.

frouros/tests/integration/test_change_detection.py

@@ -0,0 +1,57 @@
+"""Test change detection detectors."""
+
+from typing import Tuple, List
+
+import numpy as np  # type: ignore
+import pytest  # type: ignore
+
+from frouros.detectors.concept_drift import (
+    BOCD,
+    BOCDConfig,
+)
+from frouros.detectors.concept_drift.streaming.change_detection.base import (
+    BaseChangeDetection,
+)
+from frouros.detectors.concept_drift.streaming.change_detection.bocd import (
+    GaussianUnknownMean,
+)
+
+detectors = [
+    (
+        BOCD(
+            config=BOCDConfig(
+                min_num_instances=1,
+                model=GaussianUnknownMean(
+                    prior_mean=0,
+                    prior_var=1,
+                    data_var=0.5,
+                ),
+                hazard=0.01,
+            ),
+        ),
+        [100, 203],
+    ),
+]
+
+
+@pytest.mark.parametrize("detector_info", detectors)
+def test_change_detection_detector(
+    stream_drift: np.ndarray,
+    detector_info: Tuple[BaseChangeDetection, List[int]],
+) -> None:
+    """Test streaming detector.
+
+    :param stream_drift: stream with drift
+    :type stream_drift: numpy.ndarray
+    :param detector_info: concept drift detector and value function
+    :type detector_info: Tuple[BaseConceptDrift, Callable]
+    """
+    detector, idx_drifts = detector_info
+    idx_detected_drifts = []
+    for i, val in enumerate(stream_drift):
+        detector.update(value=val)
+        if detector.status["drift"]:
+            detector.reset()
+            idx_detected_drifts.append(i)
+
+    assert idx_detected_drifts == idx_drifts