Ad/add support for valid data (#563)

Author: ahosler

ads/opctl/operator/common/operator_config.py

@@ -8,13 +8,32 @@
 import json
 from abc import abstractmethod
 from dataclasses import dataclass
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 from ads.common.serializer import DataClassSerializable
 
 from ads.opctl.operator.common.utils import OperatorValidator
 from ads.opctl.operator.common.errors import InvalidParameterError
 
+@dataclass(repr=True)
+class InputData(DataClassSerializable):
+    """Class representing operator specification input data details."""
+
+    connect_args: Dict = None
+    format: str = None
+    columns: List[str] = None
+    url: str = None
+    filters: List[str] = None
+    options: Dict = None
+    limit: int = None
+    sql: str = None
+    table_name: str = None
+
+
+@dataclass(repr=True)
+class OutputDirectory(InputData):
+    """Class representing operator specification output directory details."""
+
 
 @dataclass(repr=True)
 class OperatorConfig(DataClassSerializable):

ads/opctl/operator/lowcode/anomaly/__main__.py

@@ -24,8 +24,6 @@ def operate(operator_config: AnomalyOperatorConfig) -> None:
     from .model.factory import AnomalyOperatorModelFactory
 
     datasets = AnomalyDatasets(operator_config.spec)
-    datasets2 = AnomalyData(operator_config.spec)
-    print(f"d1: {datasets.data}\n\n d2: {datasets2.data}")
     AnomalyOperatorModelFactory.get_model(operator_config, datasets).generate_report()
 
 

ads/opctl/operator/lowcode/anomaly/environment.yaml

@@ -8,5 +8,4 @@ dependencies:
       - datapane
       - cerberus
       - oracle-automlx==23.2.3
-      - "git+https://github.com/datamllab/tods.git"
-      - "git+https://github.com/oracle/accelerated-data-science.git@feature/anomaly#egg=oracle-ads"
+      - "oracle-ads[anomaly]"

ads/opctl/operator/lowcode/anomaly/model/anomaly_dataset.py

@@ -28,6 +28,18 @@ def __init__(self, spec: AnomalyOperatorSpec):
         super().__init__(spec=spec, name="test_data")
 
 
+class ValidationData(AbstractData):
+    def __init__(self, spec: AnomalyOperatorSpec):
+        super().__init__(spec=spec, name="validation_data")
+
+    def _ingest_data(self, spec):
+        self.X_valid_dict = dict()
+        self.y_valid_dict = dict()
+        for s_id, df in self.get_dict_by_series().items():
+            self.X_valid_dict[s_id] = df.drop([OutputColumns.ANOMALY_COL], axis=1)
+            self.y_valid_dict[s_id] = df[OutputColumns.ANOMALY_COL]
+
+
 class AnomalyDatasets:
     def __init__(self, spec: AnomalyOperatorSpec):
         """Instantiates the DataIO instance.
@@ -39,63 +51,23 @@ def __init__(self, spec: AnomalyOperatorSpec):
         """
         self._data = AnomalyData(spec)
         self.data = self._data.get_data_long()
-        # self.test_data = None
-        # self.target_columns = None
         self.full_data_dict = self._data.get_dict_by_series()
-        # self._load_data(spec)
-
-    # def _load_data(self, spec):
-    #     """Loads anomaly input data."""
-    #     try:
-    #         self.data = load_data(
-    #             filename=spec.input_data.url,
-    #             format=spec.input_data.format,
-    #             columns=spec.input_data.columns,
-    #         )
-    #     except InvalidParameterError as e:
-    #         e.args = e.args + ("Invalid Parameter: input_data",)
-    #         raise e
-    #     date_col = spec.datetime_column.name
-    #     self.data[date_col] = pd.to_datetime(self.data[date_col])
-    #     try:
-    #         spec.freq = get_frequency_of_datetime(self.data, spec)
-    #     except TypeError as e:
-    #         logger.warn(
-    #             f"Error determining frequency: {e.args}. Setting Frequency to None"
-    #         )
-    #         logger.debug(f"Full traceback: {e}")
-    #         spec.freq = None
-
-    #     if spec.target_category_columns is None:
-    #         if spec.target_column is None:
-    #             target_col = [
-    #                 col
-    #                 for col in self.data.columns
-    #                 if col not in [spec.datetime_column.name]
-    #             ]
-    #             spec.target_column = target_col[0]
-    #         self.full_data_dict = {spec.target_column: self.data}
-    #     else:
-    #         # Merge target category columns
-
-    #         self.data[OutputColumns.Series] = merge_category_columns(
-    #             self.data, spec.target_category_columns
-    #         )
-    #         unique_categories = self.data[OutputColumns.Series].unique()
-    #         self.full_data_dict = dict()
-
-    #         for cat in unique_categories:
-    #             data_by_cat = self.data[self.data[OutputColumns.Series] == cat].drop(
-    #                 spec.target_category_columns + [OutputColumns.Series], axis=1
-    #             )
-    #             self.full_data_dict[cat] = data_by_cat
+        if spec.validation_data is not None:
+            self.valid_data = ValidationData(spec)
+            self.X_valid_dict = self.valid_data.X_valid_dict
+            self.y_valid_dict = self.valid_data.y_valid_dict
 
 
 class AnomalyOutput:
     def __init__(self, date_column):
         self.category_map = dict()
         self.date_column = date_column
 
+    def list_categories(self):
+        categories = list(self.category_map.keys())
+        categories.sort()
+        return categories
+
     def add_output(self, category: str, anomalies: pd.DataFrame, scores: pd.DataFrame):
         self.category_map[category] = (anomalies, scores)
 
@@ -126,7 +98,7 @@ def get_outliers_by_cat(self, category: str, data: pd.DataFrame):
     def get_inliers(self, data):
         inliers = pd.DataFrame()
 
-        for category in self.category_map.keys():
+        for category in self.list_categories():
             inliers = pd.concat(
                 [
                     inliers,
@@ -145,7 +117,7 @@ def get_inliers(self, data):
     def get_outliers(self, data):
         outliers = pd.DataFrame()
 
-        for category in self.category_map.keys():
+        for category in self.list_categories():
             outliers = pd.concat(
                 [
                     outliers,
@@ -163,10 +135,10 @@ def get_outliers(self, data):
 
     def get_scores(self, target_category_columns):
         if target_category_columns is None:
-            return self.get_scores_by_cat(list(self.category_map.keys())[0])
+            return self.get_scores_by_cat(self.list_categories()[0])
 
         scores = pd.DataFrame()
-        for category in self.category_map.keys():
+        for category in self.list_categories():
             score = self.get_scores_by_cat(category)
             score[target_category_columns[0]] = category
             scores = pd.concat([scores, score], axis=0, ignore_index=True)

ads/opctl/operator/lowcode/anomaly/model/automlx.py

@@ -27,10 +27,23 @@ def _build_model(self) -> pd.DataFrame:
         date_column = self.spec.datetime_column.name
         anomaly_output = AnomalyOutput(date_column=date_column)
 
+        time_budget = self.spec.model_kwargs.pop("time_budget", -1)
         # Iterate over the full_data_dict items
         for target, df in self.datasets.full_data_dict.items():
-            est = automl.Pipeline(task="anomaly_detection")
-            est.fit(df, y=None)
+            est = automl.Pipeline(task="anomaly_detection", **self.spec.model_kwargs)
+            est.fit(
+                X=df,
+                X_valid=self.X_valid_dict[target]
+                if self.X_valid_dict is not None
+                else None,
+                y_valid=self.y_valid_dict[target]
+                if self.y_valid_dict is not None
+                else None,
+                time_budget=time_budget,
+                contamination=self.spec.contamination
+                if self.y_valid_dict is not None
+                else None,
+            )
             y_pred = est.predict(df)
             scores = est.predict_proba(df)
 

ads/opctl/operator/lowcode/anomaly/model/autots.py

@@ -14,23 +14,24 @@
 
 
 class AutoTSOperatorModel(AnomalyOperatorBaseModel):
-    """Class representing TODS Anomaly Detection operator model."""
+    """Class representing AutoTS Anomaly Detection operator model."""
 
     @runtime_dependency(
         module="autots",
         err_msg=(
             "Please run `pip3 install autots` to "
-            "install the required dependencies for TODS."
+            "install the required dependencies for AutoTS."
         ),
     )
     def _build_model(self) -> AnomalyOutput:
         from autots.evaluator.anomaly_detector import AnomalyDetector
 
         method = self.spec.model_kwargs.get("method")
+        transform_dict = self.spec.model_kwargs.get("transform_dict", {})
 
         if method == "random" or method == "deep" or method == "fast":
             new_params = AnomalyDetector.get_new_params(method=method)
-            new_params.pop("transform_dict")
+            transform_dict = new_params.pop("transform_dict")
 
             for key, value in new_params.items():
                 self.spec.model_kwargs[key] = value
@@ -39,7 +40,12 @@ def _build_model(self) -> AnomalyOutput:
             self.spec.model_kwargs["output"] = "univariate"
 
         if "transform_dict" not in self.spec.model_kwargs:
-            self.spec.model_kwargs["transform_dict"] = {}
+            self.spec.model_kwargs["transform_dict"] = transform_dict
+
+        if self.spec.contamination != 0.1:  # TODO: remove hard-coding
+            self.spec.model_kwargs.get("method_params", {})[
+                "contamination"
+            ] = self.spec.contamination
 
         model = AnomalyDetector(**self.spec.model_kwargs)