Merge pull request #104 from amosproj/feature/iqr-anomaly-detection

Add IQR anomaly detection RTDIP component

Author: hpfref

.gitignore

@@ -140,3 +140,20 @@ config.share
 
 # JetBrains
 .idea/
+
+
+
+# Notebooks & outputs
+*.ipynb
+outputs/
+
+# Local data
+amos_team_resources/shell/data/
+
+# Python cache
+__pycache__/
+
+
+# Backup folders
+src/sdk/python/rtdip_sdk/pipelines/anomaly_detection/spark/iqr-backup/
+

amos_team_resources/anomaly_detection/iqr/ad_shell_iqr.ipynb

@@ -0,0 +1,10 @@
+from .iqr_anomaly_detection import IQRAnomalyDetectionComponent
+from .decomposition_iqr_anomaly_detection import (
+    DecompositionIQRAnomalyDetectionComponent
+)
+
+__all__ = [
+    "IQRAnomalyDetectionComponent",
+    "DecompositionIQRAnomalyDetectionComponent",
+]
+

amos_team_resources/anomaly_detection/iqr/outputs/iqr_global_vs_rolling.png

@@ -0,0 +1,37 @@
+import pandas as pd
+
+from .iqr_anomaly_detection import IQRAnomalyDetectionComponent
+from .interfaces import IQRAnomalyDetectionConfig
+
+
+class DecompositionIQRAnomalyDetectionComponent(
+    IQRAnomalyDetectionComponent
+):
+    """
+    IQR anomaly detection on decomposed time series.
+
+    Expected input columns:
+    - residual (default)
+    - trend
+    - seasonal
+    """
+
+    def __init__(self, config: IQRAnomalyDetectionConfig):
+        super().__init__(config)
+        self.input_component: str = config.get("input_component", "residual")
+
+    def run(self, df: pd.DataFrame) -> pd.DataFrame:
+        """
+        Run anomaly detection on a selected decomposition component.
+        """
+
+        if self.input_component not in df.columns:
+            raise ValueError(
+                f"Column '{self.input_component}' not found in input DataFrame"
+            )
+
+        df = df.copy()
+        df[self.value_column] = df[self.input_component]
+
+        return super().run(df)
+

amos_team_resources/anomaly_detection/iqr/outputs/iqr_shell_raw.png

@@ -0,0 +1,21 @@
+from typing import TypedDict, Optional
+
+
+class IQRAnomalyDetectionConfig(TypedDict, total=False):
+    """
+    Configuration schema for IQR anomaly detection components.
+    """
+
+    # IQR sensitivity factor
+    k: float
+
+    # Rolling window size (None = global IQR)
+    window: Optional[int]
+
+    # Column names
+    value_column: str
+    time_column: str
+
+    # Used only for decomposition-based component
+    input_component: str
+

src/sdk/python/rtdip_sdk/pipelines/anomaly_detection/spark/iqr/__init__.py

@@ -0,0 +1,69 @@
+import pandas as pd
+from typing import Optional
+
+from rtdip_sdk.pipelines.interfaces import PipelineComponent
+from .interfaces import IQRAnomalyDetectionConfig
+
+
+class IQRAnomalyDetectionComponent(PipelineComponent):
+    """
+    RTDIP component implementing IQR-based anomaly detection.
+
+    Supports:
+    - Global IQR (window=None)
+    - Rolling IQR (window=int)
+    """
+
+    def __init__(self, config: IQRAnomalyDetectionConfig):
+        self.k: float = config.get("k", 1.5)
+        self.window: Optional[int] = config.get("window", None)
+
+        self.value_column: str = config.get("value_column", "value")
+        self.time_column: str = config.get("time_column", "timestamp")
+
+    def run(self, df: pd.DataFrame) -> pd.DataFrame:
+        """
+        Run IQR anomaly detection on a time series DataFrame.
+
+        Input:
+            df with columns [time_column, value_column]
+
+        Output:
+            df with additional column:
+            - is_anomaly (bool)
+        """
+
+        if self.value_column not in df.columns:
+            raise ValueError(
+                f"Column '{self.value_column}' not found in input DataFrame"
+            )
+
+        values = df[self.value_column]
+
+        # -----------------------
+        # Global IQR
+        # -----------------------
+        if self.window is None:
+            q1 = values.quantile(0.25)
+            q3 = values.quantile(0.75)
+            iqr = q3 - q1
+
+            lower = q1 - self.k * iqr
+            upper = q3 + self.k * iqr
+
+        # -----------------------
+        # Rolling IQR
+        # -----------------------
+        else:
+            q1 = values.rolling(self.window).quantile(0.25)
+            q3 = values.rolling(self.window).quantile(0.75)
+            iqr = q3 - q1
+
+            lower = q1 - self.k * iqr
+            upper = q3 + self.k * iqr
+
+        df = df.copy()
+        df["is_anomaly"] = (values < lower) | (values > upper)
+
+        return df
+