auto model evalulator added for forecasting operator

Author: prasankh

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

@@ -4,33 +4,29 @@
 # Copyright (c) 2023, 2024 Oracle and/or its affiliates.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
 
+import fsspec
+import numpy as np
 import os
+import pandas as pd
 import tempfile
 import time
 from abc import ABC, abstractmethod
-from typing import Tuple
-
-import fsspec
-import pandas as pd
-import numpy as np
 from sklearn import linear_model
+from typing import Tuple
 
+from ads.common.object_storage_details import ObjectStorageDetails
 from ads.opctl import logger
-
-from ..operator_config import AnomalyOperatorConfig, AnomalyOperatorSpec
-from .anomaly_dataset import AnomalyDatasets, AnomalyOutput, TestData
 from ads.opctl.operator.lowcode.anomaly.const import OutputColumns, SupportedMetrics
-from ..const import SupportedModels
+from ads.opctl.operator.lowcode.anomaly.utils import _build_metrics_df, default_signer
 from ads.opctl.operator.lowcode.common.utils import (
     human_time_friendly,
     enable_print,
     disable_print,
     write_data,
-    merge_category_columns,
-    find_output_dirname,
 )
-from ads.opctl.operator.lowcode.anomaly.utils import _build_metrics_df, default_signer
-from ads.common.object_storage_details import ObjectStorageDetails
+from .anomaly_dataset import AnomalyDatasets, AnomalyOutput, TestData
+from ..const import SupportedModels
+from ..operator_config import AnomalyOperatorConfig, AnomalyOperatorSpec
 
 
 class AnomalyOperatorBaseModel(ABC):
@@ -246,7 +242,7 @@ def _save_report(
         """Saves resulting reports to the given folder."""
         import report_creator as rc
 
-        unique_output_dir = find_output_dirname(self.spec.output_directory)
+        unique_output_dir = self.spec.output_directory.url
 
         if ObjectStorageDetails.is_oci_path(unique_output_dir):
             storage_options = default_signer()

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

@@ -215,7 +215,7 @@ def human_time_friendly(seconds):
 
 
 def find_output_dirname(output_dir: OutputDirectory):
-    if output_dir:
+    if output_dir and output_dir.url:
         return output_dir.url
     output_dir = "results"
 

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

@@ -24,24 +24,9 @@ def operate(operator_config: ForecastOperatorConfig) -> None:
     from .model.factory import ForecastOperatorModelFactory
 
     datasets = ForecastDatasets(operator_config)
-    try:
-        ForecastOperatorModelFactory.get_model(
-            operator_config, datasets
-        ).generate_report()
-    except Exception as e:
-        if operator_config.spec.model == "auto":
-            logger.debug(
-                f"Failed to forecast with error {e.args}. Trying again with model `prophet`."
-            )
-            operator_config.spec.model = "prophet"
-            operator_config.spec.model_kwargs = dict()
-            datasets = ForecastDatasets(operator_config)
-            ForecastOperatorModelFactory.get_model(
-                operator_config, datasets
-            ).generate_report()
-        else:
-            raise
-
+    ForecastOperatorModelFactory.get_model(
+        operator_config, datasets
+    ).generate_report()
 
 def verify(spec: Dict, **kwargs: Dict) -> bool:
     """Verifies the forecasting operator config."""

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

@@ -4,31 +4,19 @@
 # Copyright (c) 2023, 2024 Oracle and/or its affiliates.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
 
-import json
+import fsspec
+import numpy as np
 import os
+import pandas as pd
 import tempfile
 import time
+import traceback
 from abc import ABC, abstractmethod
 from typing import Tuple
-import traceback
-
-import fsspec
-import numpy as np
-import pandas as pd
 
-from ads.opctl.operator.lowcode.forecast.utils import (
-    default_signer,
-    evaluate_train_metrics,
-    get_forecast_plots,
-    _build_metrics_df,
-    _build_metrics_per_horizon,
-    load_pkl,
-    write_pkl,
-    _label_encode_dataframe,
-)
+from ads.common.decorator.runtime_dependency import runtime_dependency
 from ads.common.object_storage_details import ObjectStorageDetails
 from ads.opctl import logger
-
 from ads.opctl.operator.lowcode.common.utils import (
     human_time_friendly,
     enable_print,
@@ -37,18 +25,27 @@
     merged_category_column_name,
     datetime_to_seconds,
     seconds_to_datetime,
-    find_output_dirname,
 )
+from ads.opctl.operator.lowcode.forecast.model.forecast_datasets import TestData
+from ads.opctl.operator.lowcode.forecast.utils import (
+    default_signer,
+    evaluate_train_metrics,
+    get_forecast_plots,
+    get_auto_select_plot,
+    _build_metrics_df,
+    _build_metrics_per_horizon,
+    load_pkl,
+    write_pkl,
+    _label_encode_dataframe,
+)
+from .forecast_datasets import ForecastDatasets
 from ..const import (
     SUMMARY_METRICS_HORIZON_LIMIT,
     SupportedMetrics,
     SupportedModels,
     SpeedAccuracyMode,
 )
 from ..operator_config import ForecastOperatorConfig, ForecastOperatorSpec
-from ads.common.decorator.runtime_dependency import runtime_dependency
-from .forecast_datasets import ForecastDatasets, ForecastOutput
-from ads.opctl.operator.lowcode.forecast.model.forecast_datasets import TestData
 
 
 class ForecastOperatorBaseModel(ABC):
@@ -250,6 +247,17 @@ def generate_report(self):
                     sec9 = rc.DataTable(self.eval_metrics, index=True)
                     train_metrics_sections = [sec9_text, sec9]
 
+                backtest_sections = []
+                if self.spec.model == "auto-select":
+                    output_dir = self.spec.output_directory.url
+                    backtest_report_name = "backtest_stats.csv"
+                    backtest_stats = pd.read_csv(f"{output_dir}/{backtest_report_name}")
+                    backtest_text = rc.Heading("Back Testing Metrics", level=2)
+                    backtest_table = rc.DataTable(backtest_stats, index=True)
+                    liner_plot = get_auto_select_plot(backtest_stats)
+                    backtest_sections = [backtest_text, backtest_table, liner_plot]
+
+
                 forecast_plots = []
                 if len(self.forecast_output.list_series_ids()) > 0:
                     forecast_text = rc.Heading(
@@ -276,6 +284,7 @@ def generate_report(self):
                 yaml_appendix = rc.Yaml(self.config.to_dict())
                 report_sections = (
                     [summary]
+                    + backtest_sections
                     + forecast_plots
                     + other_sections
                     + test_metrics_sections
@@ -409,7 +418,7 @@ def _save_report(
         """Saves resulting reports to the given folder."""
         import report_creator as rc
 
-        unique_output_dir = find_output_dirname(self.spec.output_directory)
+        unique_output_dir = self.spec.output_directory.url
 
         if ObjectStorageDetails.is_oci_path(unique_output_dir):
             storage_options = default_signer()

ads/opctl/operator/lowcode/forecast/model/factory.py

@@ -60,7 +60,7 @@ def get_model(
             In case of not supported model.
         """
         model_type = operator_config.spec.model
-        if model_type == "auto":
+        if model_type == "auto-select":
             model_type = cls.auto_select_model(datasets, operator_config)
         if model_type not in cls._MAP:
             raise UnSupportedModelError(model_type)
@@ -89,5 +89,4 @@ def auto_select_model(
         from ..model_evaluator import ModelEvaluator
         all_models = cls._MAP.keys()
         model_evaluator = ModelEvaluator(all_models)
-        best_model = model_evaluator.find_best_model(datasets, operator_config)
-        return cls._MAP[best_model]
+        return model_evaluator.find_best_model(datasets, operator_config)

ads/opctl/operator/lowcode/forecast/model_evaluator.py

@@ -5,15 +5,14 @@
 
 
 import numpy as np
+import pandas as pd
+from pathlib import Path
+
 from ads.opctl import logger
-from ads.opctl.operator.lowcode.common.utils import (
-    find_output_dirname,
-)
 from ads.opctl.operator.lowcode.common.const import DataColumns
 from .model.forecast_datasets import ForecastDatasets
 from .operator_config import ForecastOperatorConfig
-from pathlib import Path
-import pandas as pd
+
 
 class ModelEvaluator:
     def __init__(self, models, k=5, subsample_ratio=0.20):
@@ -31,7 +30,9 @@ def generate_cutoffs(self, unique_dates, horizon):
         cut_offs = sorted_dates[-horizon - 1:-horizon * (self.k + 1):-horizon][:len(valid_train_window_size)]
         return cut_offs
 
-    def generate_k_fold_data(self, datasets: ForecastDatasets, date_col: str, horizon: int):
+    def generate_k_fold_data(self, datasets: ForecastDatasets, operator_config: ForecastOperatorConfig):
+        date_col = operator_config.spec.datetime_column.name
+        horizon = operator_config.spec.horizon
         historical_data = datasets.historical_data.data.reset_index()
         series_col = DataColumns.Series
         group_counts = historical_data[series_col].value_counts()
@@ -51,63 +52,80 @@ def generate_k_fold_data(self, datasets: ForecastDatasets, date_col: str, horizo
         for i, current in enumerate(cut_offs[1:]):
             test_datasets.append(sampled_historical_data[(current < sampled_historical_data[date_col]) & (
                     sampled_historical_data[date_col] <= cut_offs[i])])
-        return cut_offs, training_datasets, test_datasets
+        all_additional = datasets.additional_data.data.reset_index()
+        sampled_additional_data = all_additional[all_additional[series_col].isin(sampled_groups.index)]
+        max_historical_date = sampled_historical_data[date_col].max()
+        additional_data = [sampled_additional_data[sampled_additional_data[date_col] <= max_historical_date]]
+        for cut_off in cut_offs[:-1]:
+            trimmed_additional_data = sampled_additional_data[sampled_additional_data[date_col] <= cut_off]
+            additional_data.append(trimmed_additional_data)
+        return cut_offs, training_datasets, additional_data, test_datasets
 
     def remove_none_values(self, obj):
         if isinstance(obj, dict):
             return {k: self.remove_none_values(v) for k, v in obj.items() if k is not None and v is not None}
         else:
             return obj
 
-    def create_operator_config(self, operator_config, backtest, model, historical_data, test_data):
-        output_dir = find_output_dirname(operator_config.spec.output_directory)
-        output_file_path = f'{output_dir}back_testing/{model}/{backtest}'
+    def create_operator_config(self, operator_config, backtest, model, historical_data, additional_data, test_data):
+        output_dir = operator_config.spec.output_directory.url
+        output_file_path = f'{output_dir}/back_testing/{model}/{backtest}'
         Path(output_file_path).mkdir(parents=True, exist_ok=True)
         historical_data_url = f'{output_file_path}/historical.csv'
+        additional_data_url = f'{output_file_path}/additional.csv'
         test_data_url = f'{output_file_path}/test.csv'
         historical_data.to_csv(historical_data_url, index=False)
+        additional_data.to_csv(additional_data_url, index=False)
         test_data.to_csv(test_data_url, index=False)
         backtest_op_config_draft = operator_config.to_dict()
         backtest_spec = backtest_op_config_draft["spec"]
         backtest_spec["historical_data"]["url"] = historical_data_url
+        backtest_spec["additional_data"]["url"] = additional_data_url
         backtest_spec["test_data"]["url"] = test_data_url
         backtest_spec["model"] = model
-        backtest_spec["output_directory"]["url"] = output_file_path
+        backtest_spec["output_directory"] = {"url": output_file_path}
         backtest_spec["target_category_columns"] = [DataColumns.Series]
-        backtest_spec.pop('additional_data', None)  # todo create additional data
+        backtest_spec['generate_explanations'] = False
         cleaned_config = self.remove_none_values(backtest_op_config_draft)
 
         backtest_op_config = ForecastOperatorConfig.from_dict(
             obj_dict=cleaned_config)
         return backtest_op_config
 
     def run_all_models(self, datasets: ForecastDatasets, operator_config: ForecastOperatorConfig):
-        date_col = operator_config.spec.datetime_column.name
-        horizon = operator_config.spec.horizon
-        cut_offs, train_sets, test_sets = self.generate_k_fold_data(datasets, date_col, horizon)
+        cut_offs, train_sets, additional_data, test_sets = self.generate_k_fold_data(datasets, operator_config)
         metrics = {}
         for model in self.models:
             from .model.factory import ForecastOperatorModelFactory
             metrics[model] = {}
             for i in range(len(cut_offs)):
                 backtest_historical_data = train_sets[i]
+                backtest_additional_data = additional_data[i]
                 backtest_test_data = test_sets[i]
                 backtest_operator_config = self.create_operator_config(operator_config, i, model,
                                                                        backtest_historical_data,
+                                                                       backtest_additional_data,
                                                                        backtest_test_data)
                 datasets = ForecastDatasets(backtest_operator_config)
                 ForecastOperatorModelFactory.get_model(
                     backtest_operator_config, datasets
                 ).generate_report()
-                metrics_df = pd.read_csv(f"{backtest_operator_config.spec.output_directory.url}/metrics.csv")
-                metrics_df["average_accross_series"] = metrics_df.drop('metrics', axis=1).mean(axis=1)
-                metrics_average_dict = dict(zip(metrics_df['metrics'].str.lower(), metrics_df['average_accross_series']))
+                test_metrics_filename = backtest_operator_config.spec.test_metrics_filename
+                metrics_df = pd.read_csv(
+                    f"{backtest_operator_config.spec.output_directory.url}/{test_metrics_filename}")
+                metrics_df["average_across_series"] = metrics_df.drop('metrics', axis=1).mean(axis=1)
+                metrics_average_dict = dict(zip(metrics_df['metrics'].str.lower(), metrics_df['average_across_series']))
                 metrics[model][i] = metrics_average_dict[operator_config.spec.metric]
         return metrics
 
     def find_best_model(self, datasets: ForecastDatasets, operator_config: ForecastOperatorConfig):
         metrics = self.run_all_models(datasets, operator_config)
-        avg_backtests_metrics = {key : sum(value.values()) / len(value.values()) for key, value in metrics.items()}
+        avg_backtests_metrics = {key: sum(value.values()) / len(value.values()) for key, value in metrics.items()}
         best_model = min(avg_backtests_metrics, key=avg_backtests_metrics.get)
         logger.info(f"Among models {self.models}, {best_model} model shows better performance during backtesting.")
+        backtest_stats = pd.DataFrame(metrics).rename_axis('backtest')
+        backtest_stats.reset_index(inplace=True)
+        output_dir = operator_config.spec.output_directory.url
+        backtest_report_name = "backtest_stats.csv"
+        backtest_stats.to_csv(f"{output_dir}/{backtest_report_name}", index=False)
         return best_model

ads/opctl/operator/lowcode/forecast/operator_config.py

@@ -14,7 +14,7 @@
 
 from .const import SupportedMetrics, SpeedAccuracyMode
 from .const import SupportedModels
-
+from ads.opctl.operator.lowcode.common.utils import find_output_dirname
 
 @dataclass(repr=True)
 class TestData(InputData):
@@ -90,6 +90,7 @@ class ForecastOperatorSpec(DataClassSerializable):
 
     def __post_init__(self):
         """Adjusts the specification details."""
+        self.output_directory = self.output_directory or OutputDirectory(url=find_output_dirname(self.output_directory))
         self.metric = (self.metric or "").lower() or SupportedMetrics.SMAPE.lower()
         self.model = self.model or SupportedModels.Auto
         self.confidence_interval_width = self.confidence_interval_width or 0.80

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

@@ -365,14 +365,14 @@ spec:
     model:
       type: string
       required: false
-      default: auto
+      default: auto-select
       allowed:
         - prophet
         - arima
         - neuralprophet
         - automlx
         - autots
-        - auto
+        - auto-select
 
     model_kwargs:
       type: dict

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

@@ -265,6 +265,24 @@ def _select_plot_list(fn, series_ids):
 def _add_unit(num, unit):
     return f"{num} {unit}"
 
+def get_auto_select_plot(backtest_results):
+    fig = go.Figure()
+    columns = backtest_results.columns.tolist()
+    back_test_column = "backtest"
+    columns.remove(back_test_column)
+    for i, column in enumerate(columns):
+        color = 0 #int(i * 255 / len(columns))
+        fig.add_trace(
+            go.Scatter(
+            x=backtest_results[back_test_column],
+            y=backtest_results[column],
+            mode="lines",
+            name=column,
+        ))
+
+    import report_creator as rc
+    return rc.Widget(fig)
+
 
 def get_forecast_plots(
     forecast_output,