Merge pull request #106 from andrewm4894/make-list-of-models

Make list of models

Author: andrewm4894

anomstack/jobs/score.py

@@ -58,7 +58,8 @@ def noop():
 
     metric_batch = spec["metric_batch"]
     model_path = spec["model_path"]
-    model_tag = spec["model_config"].get("model_tag", "")
+    model_configs = spec["model_configs"]
+    model_combination_method = spec.get("model_combination_method", "mean")
     table_key = spec["table_key"]
     db = spec["db"]
     preprocess_params = spec["preprocess_params"]
@@ -110,17 +111,6 @@ def score(df) -> pd.DataFrame:
                 logger.debug(f"preprocess {metric_name} in {metric_batch} score job.")
                 logger.debug(f"df_metric:\n{df_metric.head()}")
 
-                # try load model and catch google.api_core.exceptions.NotFound
-                try:
-                    model = load_model(metric_name, model_path, metric_batch, model_tag)
-                except NotFound as e:
-                    logger.warning(e)
-                    logger.warning(
-                        f"model not found for {metric_name} in "
-                        f"{metric_batch} score job."
-                    )
-                    continue
-
                 X = preprocess(df_metric, **preprocess_params)
 
                 if len(X) == 0:
@@ -131,11 +121,40 @@ def score(df) -> pd.DataFrame:
 
                 logger.debug(f"X:\n{X.head()}")
 
-                scores = model.predict_proba(X)
+                scores = {}
+                for model_config in model_configs:
+
+                    model_tag = model_config.get("model_tag", "")
+
+                    try:
+                        model = load_model(
+                            metric_name, model_path, metric_batch, model_tag
+                        )
+                        scores_tmp = model.predict_proba(X)
+                        scores_tmp = scores_tmp[:, 1]  # probability of anomaly
+                        scores[f'{metric_name}_{model_tag}'] = scores_tmp
+                    except NotFound as e:
+                        logger.warning(e)
+                        logger.warning(
+                            f"model not found for {metric_name} in "
+                            f"{metric_batch} score job."
+                        )
+                        continue
+
+                if model_combination_method == "mean":
+                    scores = pd.DataFrame(scores).mean(axis=1).values
+                elif model_combination_method == "max":
+                    scores = pd.DataFrame(scores).max(axis=1).values
+                elif model_combination_method == "min":
+                    scores = pd.DataFrame(scores).min(axis=1).values
+                else:
+                    raise ValueError(
+                        f"model_combination_method {model_combination_method} not supported."
+                    )
 
                 # create initial df_score
                 df_score = pd.DataFrame(
-                    data=scores[:, 1],  # probability of anomaly
+                    data=scores,
                     index=X.index,
                     columns=["metric_value"],
                 ).round(3)

anomstack/jobs/train.py

@@ -57,9 +57,7 @@ def noop():
     db = spec["db"]
     model_path = spec["model_path"]
     preprocess_params = spec["preprocess_params"]
-    model_name = spec["model_config"]["model_name"]
-    model_tag = spec["model_config"].get("model_tag", "")
-    model_params = spec["model_config"]["model_params"]
+    model_configs = spec["model_configs"]
 
     @job(
         name=f"{metric_batch}_train",
@@ -130,10 +128,18 @@ def train(df) -> List[Tuple[str, BaseDetector, str]]:
                                 f"len(X)={len(X)}"
                             )
                         )
-                        model = train_model(
-                            X, metric_name, model_name, model_params, model_tag
-                        )
-                        models.append((metric_name, model, model_tag))
+                        for model_config in model_configs:
+                            model_name = model_config["model_name"]
+                            model_params = model_config["model_params"]
+                            model_tag = model_config.get("model_tag", "")
+                            model = train_model(
+                                X,
+                                metric_name,
+                                model_name,
+                                model_params,
+                                model_tag
+                            )
+                            models.append((metric_name, model, model_tag))
                     else:
                         logger.info(
                             f"no data for {metric_name} in {metric_batch} train job."

metrics/defaults/defaults.yaml

@@ -5,12 +5,16 @@ model_path: "local://./models" # path to where models are to be stored.
 # model_path: "gs://your-bucket/models" # gcs path to where models are to be stored.
 # model_path: "s3://your-bucket/models" # s3 path to where models are to be stored.
 # model configs to pass to PyOD, model_params are passed to the model constructor.
-model_config:
-  model_name: 'PCA'
-  model_tag: 'pca_default'
-  model_params:
-    contamination: 0.01
-# metric_tags is a map of metric key value tags to metric names
+model_configs:
+  - model_name: 'PCA'
+    model_tag: 'pca_default'
+    model_params:
+      contamination: 0.01
+  - model_name: 'KNN'
+    model_tag: 'knn_default'
+    model_params:
+      contamination: 0.01
+model_combination_method: 'mean' # method to combine model scores, 'mean', 'min' or 'max'.
 # metric_tags:
 #   metric_name:
 #     key1: value1

metrics/examples/netdata/netdata.yaml

@@ -1,7 +1,6 @@
 metric_batch: "netdata"
-db: "bigquery"
-table_key: "andrewm4894.metrics.metrics"
-model_path: "gs://andrewm4894-tmp/models"
+db: "sqlite"
+table_key: "metrics"
 ingest_cron_schedule: "*/10 * * * *"
 train_cron_schedule: "*/60 * * * *"
 score_cron_schedule: "*/15 * * * *"
@@ -11,6 +10,6 @@ llmalert_cron_schedule: "*/20 * * * *"
 plot_cron_schedule: "*/25 * * * *"
 alert_always: False
 disable_llmalert: False
-alert_methods: "email"
+alert_methods: "email,slack"
 ingest_fn: >
   {% include "./examples/netdata/netdata.py" %}