auto model evalulator added for forecasting tasks

Author: prasankh

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

@@ -97,7 +97,8 @@ def _set_series_id_column(self, df):
                 for value in merged_values:
                     self._target_category_columns_map[value] = df[df[DataColumns.Series] == value][self.target_category_columns].drop_duplicates().iloc[0].to_dict()
 
-            df = df.drop(self.target_category_columns, axis=1)
+            if self.target_category_columns != [DataColumns.Series]:
+                df = df.drop(self.target_category_columns, axis=1)
         return df
 
     def _format_datetime_col(self, df):

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

@@ -12,7 +12,6 @@
 from .base_model import ForecastOperatorBaseModel
 from .neuralprophet import NeuralProphetOperatorModel
 from .prophet import ProphetOperatorModel
-from ..utils import select_auto_model
 from .forecast_datasets import ForecastDatasets
 
 class UnSupportedModelError(Exception):
@@ -62,7 +61,33 @@ def get_model(
         """
         model_type = operator_config.spec.model
         if model_type == "auto":
-            model_type = select_auto_model(datasets, operator_config)
+            model_type = cls.auto_select_model(datasets, operator_config)
         if model_type not in cls._MAP:
             raise UnSupportedModelError(model_type)
         return cls._MAP[model_type](config=operator_config, datasets=datasets)
+
+    @classmethod
+    def auto_select_model(
+            cls, datasets: ForecastDatasets, operator_config: ForecastOperatorConfig
+    ) -> str:
+        """
+        Selects AutoMLX or Arima model based on column count.
+
+        If the number of columns is less than or equal to the maximum allowed for AutoMLX,
+        returns 'AutoMLX'. Otherwise, returns 'Arima'.
+
+        Parameters
+        ------------
+        datasets:  ForecastDatasets
+                Datasets for predictions
+
+        Returns
+        --------
+        str
+            The type of the 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]

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

@@ -9,37 +9,42 @@
 from ads.opctl.operator.lowcode.common.utils import (
     find_output_dirname,
 )
-from .const import ForecastOutputColumns
+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):
         self.models = models
         self.k = k
         self.subsample_ratio = subsample_ratio
+        self.minimum_sample_count = 5
+
+    def generate_cutoffs(self, unique_dates, horizon):
+        sorted_dates = np.sort(unique_dates)
+        train_window_size = [len(sorted_dates) - (i + 1) * horizon for i in range(self.k)]
+        valid_train_window_size = [ws for ws in train_window_size if ws >= horizon * 3]
+        if len(valid_train_window_size) < self.k:
+            logger.warn(f"Only {valid_train_window_size} backtests can be created")
+        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):
         historical_data = datasets.historical_data.data.reset_index()
-        series_col = ForecastOutputColumns.SERIES
+        series_col = DataColumns.Series
         group_counts = historical_data[series_col].value_counts()
 
-        sample_count = max(5, int(len(group_counts) * self.subsample_ratio))
+        sample_count = max(self.minimum_sample_count, int(len(group_counts) * self.subsample_ratio))
         sampled_groups = group_counts.head(sample_count)
         sampled_historical_data = historical_data[historical_data[series_col].isin(sampled_groups.index)]
 
         min_group = group_counts.idxmin()
         min_series_data = historical_data[historical_data[series_col] == min_group]
         unique_dates = min_series_data[date_col].unique()
 
-        sorted_dates = np.sort(unique_dates)
-        train_window_size = [len(sorted_dates) - (i + 1) * horizon for i in range(self.k)]
-        valid_train_window_size = [ws for ws in train_window_size if ws >= horizon * 3]
-        if len(valid_train_window_size) < self.k:
-            logger.warn(f"Only ${valid_train_window_size} backtests can be created")
-
-        cut_offs = sorted_dates[-horizon - 1:-horizon * (self.k + 1):-horizon][:len(valid_train_window_size)]
+        cut_offs = self.generate_cutoffs(unique_dates, horizon)
         training_datasets = [sampled_historical_data[sampled_historical_data[date_col] <= cut_off_date] for cut_off_date
                              in cut_offs]
         test_datasets = [sampled_historical_data[sampled_historical_data[date_col] > cut_offs[0]]]
@@ -54,35 +59,55 @@ def remove_none_values(self, obj):
         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}'
+        Path(output_file_path).mkdir(parents=True, exist_ok=True)
+        historical_data_url = f'{output_file_path}/historical.csv'
+        test_data_url = f'{output_file_path}/test.csv'
+        historical_data.to_csv(historical_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["test_data"]["url"] = test_data_url
+        backtest_spec["model"] = model
+        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
+        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)
-
+        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_test_data = test_sets[i]
-                output_dir = find_output_dirname(operator_config.spec.output_directory)
-                output_file_path = f'{output_dir}back_test/{i}'
-                from pathlib import Path
-                Path(output_file_path).mkdir(parents=True, exist_ok=True)
-                historical_data_url = f'{output_file_path}/historical.csv'
-                test_data_url = f'{output_file_path}/test.csv'
-                backtest_historical_data.to_csv(historical_data_url, index=False)
-                backtest_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["test_data"]["url"] = test_data_url
-                backtest_spec["model"] = model
-                backtest_spec["output_directory"]["url"] = output_dir
-                cleaned_config = self.remove_none_values(backtest_op_config_draft)
-                backtest_op_cofig = ForecastOperatorConfig.from_dict(
-                    obj_dict=cleaned_config)
-                datasets = ForecastDatasets(backtest_op_cofig)
-
+                backtest_operator_config = self.create_operator_config(operator_config, i, model,
+                                                                       backtest_historical_data,
+                                                                       backtest_test_data)
+                datasets = ForecastDatasets(backtest_operator_config)
                 ForecastOperatorModelFactory.get_model(
-                    operator_config, datasets
+                    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']))
+                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()}
+        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.")
+        return best_model

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

@@ -413,4 +413,8 @@ spec:
         - RMSE
         - MSE
         - SMAPE
+        - mape
+        - rmse
+        - mse
+        - smape
   type: dict

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

@@ -19,7 +19,6 @@
     mean_absolute_percentage_error,
     mean_squared_error,
 )
-
 try:
     from scipy.stats import linregress
 except:
@@ -34,7 +33,6 @@
 from .operator_config import ForecastOperatorSpec, ForecastOperatorConfig
 from ads.opctl.operator.lowcode.common.utils import merge_category_columns
 from ads.opctl.operator.lowcode.forecast.const import ForecastOutputColumns
-
 # from ads.opctl.operator.lowcode.forecast.model.forecast_datasets import TestData, ForecastOutput
 
 
@@ -371,45 +369,6 @@ def plot_forecast_plotly(s_id):
 
     return _select_plot_list(plot_forecast_plotly, forecast_output.list_series_ids())
 
-
-def select_auto_model(
-    datasets: "ForecastDatasets", operator_config: ForecastOperatorConfig
-) -> str:
-    """
-    Selects AutoMLX or Arima model based on column count.
-
-    If the number of columns is less than or equal to the maximum allowed for AutoMLX,
-    returns 'AutoMLX'. Otherwise, returns 'Arima'.
-
-    Parameters
-    ------------
-    datasets:  ForecastDatasets
-            Datasets for predictions
-
-    Returns
-    --------
-    str
-        The type of the model.
-    """
-    freq_in_secs = datasets.get_datetime_frequency_in_seconds()
-    num_of_additional_cols = len(datasets.get_additional_data_column_names())
-    row_count = datasets.get_num_rows()
-    number_of_series = len(datasets.list_series_ids())
-    if (
-        num_of_additional_cols < 15
-        and row_count < 10000
-        and number_of_series < 10
-        and freq_in_secs > 3600
-    ):
-        return SupportedModels.AutoMLX
-    elif row_count < 10000 and number_of_series > 10:
-        return SupportedModels.AutoTS
-    elif row_count > 20000:
-        return SupportedModels.NeuralProphet
-    else:
-        return SupportedModels.NeuralProphet
-
-
 def convert_target(target: str, target_col: str):
     """
     Removes the target_column that got appended to target.