Feature/anomaly detect (#546)

Author: ahosler

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

@@ -59,13 +59,18 @@ def _load_data(self, spec):
                 spec.target_column = target_col[0]
             self.full_data_dict = {spec.target_column: self.data}
         else:
-            # Group the data by target column
-            self.full_data_dict = dict(
-                tuple(
-                    (group, df.reset_index(drop=True))
-                    for group, df in self.data.groupby(spec.target_category_columns[0])
+            # Merge target category columns
+
+            self.data["__Series__"] = utils._merge_category_columns(self.data, spec.target_category_columns)
+            unique_categories = self.data["__Series__"].unique()
+            self.full_data_dict = dict()
+
+            for cat in unique_categories:
+                data_by_cat = (
+                    self.data[self.data["__Series__"] == cat].drop(spec.target_category_columns + ["__Series__"],
+                                                                   axis=1)
                 )
-            )
+                self.full_data_dict[cat] = data_by_cat
 
 
 class AnomalyOutput:
@@ -108,25 +113,32 @@ def get_outliers_by_cat(self, category: str, data: pd.DataFrame):
                 how='inner')
         return outliers
 
-    def get_inliers(self, full_data_dict):
+    def get_inliers(self, data):
         inliers = pd.DataFrame()
 
         for category in self.category_map.keys():
             inliers = pd.concat(
-                [inliers, self.get_inliers_by_cat(category, full_data_dict[category])],
+                [
+                    inliers,
+                     self.get_inliers_by_cat(
+                        category, data[data['__Series__'] == category].reset_index(drop=True).drop('__Series__', axis=1)
+                     )
+                ],
                 axis=0,
                 ignore_index=True,
             )
         return inliers
 
-    def get_outliers(self, full_data_dict):
+    def get_outliers(self, data):
         outliers = pd.DataFrame()
 
         for category in self.category_map.keys():
             outliers = pd.concat(
                 [
                     outliers,
-                    self.get_outliers_by_cat(category, full_data_dict[category]),
+                    self.get_outliers_by_cat(
+                        category, data[data['__Series__'] == category].reset_index(drop=True).drop('__Series__', axis=1)
+                    )
                 ],
                 axis=0,
                 ignore_index=True,

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

@@ -42,11 +42,11 @@ def _build_model(self) -> pd.DataFrame:
             anomaly = pd.DataFrame({
                 date_column: df[date_column],
                 OutputColumns.ANOMALY_COL: y_pred
-            })
+            }).reset_index(drop=True)
             score = pd.DataFrame({
                 date_column: df[date_column],
                 OutputColumns.SCORE_COL: [item[1] for item in scores]
-            })
+            }).reset_index(drop=True)
             anomaly_output.add_output(target, anomaly, score)
 
         return anomaly_output

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

@@ -54,17 +54,14 @@ def _build_model(self) -> AnomalyOutput:
         for target, df in full_data_dict.items():
             data = df.set_index(date_column)
 
-            if self.spec.target_category_columns is not None:
-                data = data.drop(self.spec.target_category_columns[0], axis=1)
-
             (anomaly, score) = model.detect(data)
 
             if len(anomaly.columns) == 1:
                 score.rename(
                     columns={score.columns.values[0]: OutputColumns.SCORE_COL},
                     inplace=True,
                 )
-                score = 1-score
+                score = 1 - score
                 score = score.reset_index(drop=False)
 
                 col = anomaly.columns.values[0]

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

@@ -69,8 +69,6 @@ def generate_report(self):
 
         blocks = []
         for target, df in self.datasets.full_data_dict.items():
-            if self.spec.target_category_columns is not None:
-                df = df.drop(columns=[self.spec.target_category_columns[0]])
             figure_blocks = []
             time_col = df[date_column].reset_index(drop=True)
             anomaly_col = anomaly_output.get_anomalies_by_cat(category=target)[
@@ -179,12 +177,12 @@ def _validation_data_evaluate_metrics(self, anomaly_output, filename, elapsed_ti
 
         if data.empty:
             return total_metrics, summary_metrics, None
-
+        data["__Series__"] = utils._merge_category_columns(data, self.spec.target_category_columns)
         for cat in anomaly_output.category_map:
             output = anomaly_output.category_map[cat][0]
             date_col = self.spec.datetime_column.name
 
-            val_data = data[data[self.spec.target_category_columns[0]] == cat]
+            val_data = data[data["__Series__"] == cat]
             val_data[date_col] = pd.to_datetime(val_data[date_col])
 
             dates = output[output[date_col].isin(val_data[date_col])][date_col]
@@ -291,15 +289,15 @@ def _save_report(
                     f2.write(f1.read())
 
         if self.spec.generate_inliers:
-            inliers = anomaly_output.get_inliers(self.datasets.full_data_dict)
+            inliers = anomaly_output.get_inliers(self.datasets.data)
             utils._write_data(
                 data=inliers,
                 filename=os.path.join(output_dir, self.spec.inliers_filename),
                 format="csv",
                 storage_options=storage_options,
             )
 
-        outliers=anomaly_output.get_outliers(self.datasets.full_data_dict)
+        outliers=anomaly_output.get_outliers(self.datasets.data)
         utils._write_data(
             data=outliers,
             filename=os.path.join(output_dir, self.spec.outliers_filename),

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

@@ -80,17 +80,6 @@ def _build_model(self) -> pd.DataFrame:
 
             # Store the model and predictions in dictionaries
             models[target] = model
-            if self.spec.target_category_columns is None:
-                dataset.data[OutputColumns.ANOMALY_COL] = predictions_train[target]
-            else:
-                dataset.data.loc[
-                    dataset.data[self.spec.target_category_columns[0]] == target,
-                    OutputColumns.ANOMALY_COL,
-                ] = predictions_train[target]
-
-            self.datasets.full_data_dict[target][
-                OutputColumns.ANOMALY_COL
-            ] = predictions_train[target]
 
             anomaly = pd.DataFrame({
                 date_column: df[date_column],

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

@@ -282,6 +282,13 @@ spec:
       meta:
         description: "The model to be used for anomaly detection"
 
+    contamination:
+      required: false
+      default: 0.1
+      type: float
+      meta:
+        description: "Fraction of training dataset corresponding to anomalies (between 0.0 and 0.5)"
+
     model_kwargs:
       type: dict
       required: false
@@ -293,33 +300,4 @@ spec:
       meta:
         description: "preprocessing and feature engineering can be disabled using this flag, Defaults to true"
 
-    time_budget:
-      type: float
-      required: false
-      default: 0
-      meta:
-        description: "Time budget for optimization in seconds, defaults to 0 which means no limit & optimization continues till convergence."
-
-    training_ratio:
-      type: float
-      required: false
-      default: 0.7
-      meta:
-        description: "Ratio of data to be used for training, rest of the data is used for validation. defaults to 0.7"
-
-    false_alarm_ratio:
-      type: float
-      required: false
-      default: 0.01
-      meta:
-        description: "False alarm ratio is the ratio of the expected anomaly in the train dataset, defaults to 0.01"
-
-    metric:
-      type: string
-      required: false
-      default: F1_MACRO
-      allowed:
-        - F1_MACRO
-        - unsupervised_unify95
-        - unsupervised_unify95_log_loss
   type: dict

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

@@ -1,3 +1,4 @@
+
 #!/usr/bin/env python
 # -*- coding: utf-8 -*--
 
@@ -9,7 +10,7 @@
 import fsspec
 from .operator_config import AnomalyOperatorSpec
 from .const import SupportedMetrics
-
+from ads.opctl import logger
 
 def _build_metrics_df(y_true, y_pred, column_name):
     from sklearn.metrics import recall_score, precision_score, accuracy_score, f1_score, confusion_matrix, \
@@ -19,12 +20,18 @@ def _build_metrics_df(y_true, y_pred, column_name):
     metrics[SupportedMetrics.PRECISION] = precision_score(y_true, y_pred)
     metrics[SupportedMetrics.ACCURACY] = accuracy_score(y_true, y_pred)
     metrics[SupportedMetrics.F1_SCORE] = f1_score(y_true, y_pred)
-    tn, fp, fn, tp = confusion_matrix(y_true, y_pred).ravel()
+    tn, *fn_fp_tp = confusion_matrix(y_true, y_pred).ravel()
+    fp, fn, tp = fn_fp_tp if fn_fp_tp else (0, 0, 0)
     metrics[SupportedMetrics.FP] = fp
     metrics[SupportedMetrics.FN] = fn
     metrics[SupportedMetrics.TP] = tp
     metrics[SupportedMetrics.TN] = tn
-    metrics[SupportedMetrics.ROC_AUC] = roc_auc_score(y_true, y_pred)
+    try:
+        # Throws exception if y_true has only one class
+        metrics[SupportedMetrics.ROC_AUC] = roc_auc_score(y_true, y_pred)
+    except Exception as e:
+        logger.warn(f"An exception occurred: {e}")
+        metrics[SupportedMetrics.ROC_AUC] = None
     precision, recall, thresholds = precision_recall_curve(y_true, y_pred)
     metrics[SupportedMetrics.PRC_AUC] = auc(recall, precision)
     metrics[SupportedMetrics.MCC] = matthews_corrcoef(y_true, y_pred)
@@ -64,6 +71,12 @@ def _write_data(data, filename, format, storage_options, index=False, **kwargs):
         )
     raise ValueError(f"Unrecognized format: {format}")
 
+def _merge_category_columns(data, target_category_columns):
+    result = data.apply(
+        lambda x: "__".join([str(x[col]) for col in target_category_columns]), axis=1
+    )
+    return result if not result.empty else pd.Series([], dtype=str)
+
 
 def get_frequency_of_datetime(data: pd.DataFrame, dataset_info: AnomalyOperatorSpec):
     """

tests/operators/anomaly/test_anomaly_simple.py

@@ -16,7 +16,7 @@
 
 MODELS = ["automlx", "autots"]  # , "auto", "tods",
 
-
+# Mandatory YAML parameters
 TEMPLATE_YAML = {
     "kind": "operator",
     "type": "anomaly",
@@ -126,6 +126,7 @@ def test_artificial_small(model):
         yaml_i["spec"]["model"] = model
         yaml_i["spec"]["input_data"]["url"] = input_data
         yaml_i["spec"]["output_directory"]["url"] = output_dirname
+        yaml_i["spec"]["contamination"] = 0.3
 
         with open(anomaly_yaml_filename, "w") as f:
             f.write(yaml.dump(yaml_i))