relax error threshold

Author: ahosler

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

@@ -59,9 +59,15 @@ def run(self, data):
         clean_df = self._set_multi_index(clean_df)
 
         if self.name == "historical_data":
-            clean_df = self._missing_value_imputation_hist(clean_df)
+            try:
+                clean_df = self._missing_value_imputation_hist(clean_df)
+            except Exception as e:
+                logger.debug(f"Missing value imputation failed with {e.args}")
             if self.preprocessing:
-                clean_df = self._outlier_treatment(clean_df)
+                try:
+                    clean_df = self._outlier_treatment(clean_df)
+                except Exception as e:
+                    logger.debug(f"Outlier Treatment failed with {e.args}")
             else:
                 logger.debug("Skipping outlier treatment as preprocessing is disabled")
         elif self.name == "additional_data":
@@ -89,7 +95,7 @@ def _format_datetime_col(self, df):
             )
         except:
             raise InvalidParameterError(
-                f"Unable to determine the datetime type for column: {self.dt_column_name} in dataset: {self.name}. Please specify the format explicitly. (For example adding 'format: %d/%m/%Y' underneath 'name: {self.dt_column_name}' in the datetime_column section of the yaml file. For reference, here is the first datetime given: {df[self.dt_column_name].values[0]}"
+                f"Unable to determine the datetime type for column: {self.dt_column_name} in dataset: {self.name}. Please specify the format explicitly. (For example adding 'format: %d/%m/%Y' underneath 'name: {self.dt_column_name}' in the datetime_column section of the yaml file if you haven't already. For reference, here is the first datetime given: {df[self.dt_column_name].values[0]}"
             )
         return df
 

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

@@ -24,7 +24,18 @@ def operate(operator_config: ForecastOperatorConfig) -> None:
     from .model.factory import ForecastOperatorModelFactory
 
     datasets = ForecastDatasets(operator_config)
-    ForecastOperatorModelFactory.get_model(operator_config, datasets).generate_report()
+    try:
+        ForecastOperatorModelFactory.get_model(
+            operator_config, datasets
+        ).generate_report()
+    except Exception as e:
+        logger.debug(
+            f"Failed to forecast with error {e.args}. Trying again with model `prophet`."
+        )
+        operator_config.spec.model = "prophet"
+        ForecastOperatorModelFactory.get_model(
+            operator_config, datasets
+        ).generate_report()
 
 
 def verify(spec: Dict, **kwargs: Dict) -> bool:

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

@@ -61,66 +61,65 @@ def _train_model(self, i, s_id, df, model_kwargs):
         df: pd.DataFrame
             The dataframe containing the target data
         """
-        # try:
-        target = self.original_target_column
-        self.forecast_output.init_series_output(series_id=s_id, data_at_series=df)
-
-        # format the dataframe for this target. Dropping NA on target[df] will remove all future data
-        data = self.preprocess(df, s_id)
-        data_i = self.drop_horizon(data)
-
-        # Split data into X and y for arima tune method
-        y = data_i[target]
-        X_in = data_i.drop(target, axis=1) if len(data_i.columns) > 1 else None
-        X_pred = self.get_horizon(data).drop(target, axis=1)
-
-        if self.loaded_models is not None:
-            model = self.loaded_models[s_id]
-        else:
-            # Build and fit model
-            model = pm.auto_arima(y=y, X=X_in, **model_kwargs)
-
-        fitted_values = model.predict_in_sample(X=X_in).values
-
-        # Predict and format forecast
-        yhat, conf_int = model.predict(
-            n_periods=self.spec.horizon,
-            X=X_pred,
-            return_conf_int=True,
-            alpha=model_kwargs["alpha"],
-        )
-        yhat_clean = pd.DataFrame(yhat, index=yhat.index, columns=["yhat"])
+        try:
+            target = self.original_target_column
+            self.forecast_output.init_series_output(series_id=s_id, data_at_series=df)
+
+            # format the dataframe for this target. Dropping NA on target[df] will remove all future data
+            data = self.preprocess(df, s_id)
+            data_i = self.drop_horizon(data)
+
+            # Split data into X and y for arima tune method
+            y = data_i[target]
+            X_in = data_i.drop(target, axis=1) if len(data_i.columns) > 1 else None
+            X_pred = self.get_horizon(data).drop(target, axis=1)
+
+            if self.loaded_models is not None:
+                model = self.loaded_models[s_id]
+            else:
+                # Build and fit model
+                model = pm.auto_arima(y=y, X=X_in, **model_kwargs)
+
+            fitted_values = model.predict_in_sample(X=X_in).values
+
+            # Predict and format forecast
+            yhat, conf_int = model.predict(
+                n_periods=self.spec.horizon,
+                X=X_pred,
+                return_conf_int=True,
+                alpha=model_kwargs["alpha"],
+            )
+            yhat_clean = pd.DataFrame(yhat, index=yhat.index, columns=["yhat"])
 
-        conf_int_clean = pd.DataFrame(
-            conf_int, index=yhat.index, columns=["yhat_lower", "yhat_upper"]
-        )
-        forecast = pd.concat([yhat_clean, conf_int_clean], axis=1)
-        logger.debug(f"-----------------Model {i}----------------------")
-        logger.debug(forecast[["yhat", "yhat_lower", "yhat_upper"]].tail())
-
-        self.forecast_output.populate_series_output(
-            series_id=s_id,
-            fit_val=fitted_values,
-            forecast_val=self.get_horizon(forecast["yhat"]).values,
-            upper_bound=self.get_horizon(forecast["yhat_upper"]).values,
-            lower_bound=self.get_horizon(forecast["yhat_lower"]).values,
-        )
+            conf_int_clean = pd.DataFrame(
+                conf_int, index=yhat.index, columns=["yhat_lower", "yhat_upper"]
+            )
+            forecast = pd.concat([yhat_clean, conf_int_clean], axis=1)
+            logger.debug(f"-----------------Model {i}----------------------")
+            logger.debug(forecast[["yhat", "yhat_lower", "yhat_upper"]].tail())
+
+            self.forecast_output.populate_series_output(
+                series_id=s_id,
+                fit_val=fitted_values,
+                forecast_val=self.get_horizon(forecast["yhat"]).values,
+                upper_bound=self.get_horizon(forecast["yhat_upper"]).values,
+                lower_bound=self.get_horizon(forecast["yhat_lower"]).values,
+            )
+
+            self.models[s_id] = model
+
+            params = vars(model).copy()
+            for param in ["arima_res_", "endog_index_"]:
+                if param in params:
+                    params.pop(param)
+            self.model_parameters[s_id] = {
+                "framework": SupportedModels.Arima,
+                **params,
+            }
 
-        self.models[s_id] = model
-
-        params = vars(model).copy()
-        for param in ["arima_res_", "endog_index_"]:
-            if param in params:
-                params.pop(param)
-        self.model_parameters[s_id] = {
-            "framework": SupportedModels.Arima,
-            **params,
-        }
-
-        logger.debug("===========Done===========")
-        # except Exception as e:
-        #     self.errors_dict[s_id] = {"model_name": self.spec.model, "error": str(e)}
-        #     raise
+            logger.debug("===========Done===========")
+        except Exception as e:
+            self.errors_dict[s_id] = {"model_name": self.spec.model, "error": str(e)}
 
     def _build_model(self) -> pd.DataFrame:
         full_data_dict = self.datasets.get_data_by_series()

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

@@ -45,11 +45,11 @@ def set_kwargs(self):
             model_kwargs_cleaned.get("score_metric", AUTOMLX_DEFAULT_SCORE_METRIC),
         )
         model_kwargs_cleaned.pop("task", None)
-        time_budget = model_kwargs_cleaned.pop("time_budget", 0)
+        time_budget = model_kwargs_cleaned.pop("time_budget", None)
         model_kwargs_cleaned[
             "preprocessing"
         ] = self.spec.preprocessing or model_kwargs_cleaned.get("preprocessing", True)
-        return model_kwargs_cleaned
+        return model_kwargs_cleaned, time_budget
 
     def preprocess(self, data, series_id=None):
         return data.set_index(self.spec.datetime_column.name)
@@ -91,8 +91,7 @@ def _build_model(self) -> pd.DataFrame:
         )
 
         # Clean up kwargs for pass through
-        model_kwargs_cleaned = self.set_kwargs()
-        time_budget = model_kwargs_cleaned.pop("time_budget", -1)
+        model_kwargs_cleaned, time_budget = self.set_kwargs()
 
         for i, (s_id, df) in enumerate(full_data_dict.items()):
             try:
@@ -170,7 +169,6 @@ def _build_model(self) -> pd.DataFrame:
                     "model_name": self.spec.model,
                     "error": str(e),
                 }
-                raise
 
         logger.debug("===========Forecast Generated===========")
 
@@ -325,13 +323,16 @@ def _custom_predict_fn(
                 if row.index[0] > last_train_date:
                     X_new = horizon_data.copy()
                     X_new.loc[row.index[0]] = row.iloc[0]
-                    row_i = model.forecast(X=X_new, periods=self.spec.horizon)[
-                        [target_col]
-                    ].loc[row.index[0]]
+                    row_i = (
+                        model.forecast(X=X_new, periods=self.spec.horizon)[[target_col]]
+                        .loc[row.index[0]]
+                        .values[0]
+                    )
                 else:
-                    row_i = model.predict(X=row)
+                    row_i = model.predict(X=row).values[0][0]
                 rows.append(row_i)
-            return pd.concat(rows)[target_col].reset_index(drop=True)
+            ret = np.asarray(rows).flatten()
+            return ret
 
         return _custom_predict_fn
 

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

@@ -131,7 +131,6 @@ def generate_report(self):
                     except Exception as e:
                         logger.warn("Unable to generate Test Metrics.")
                         logger.debug(f"Full Traceback: {traceback.format_exc()}")
-                        raise e  # TODO remove
             report_sections = []
 
             if self.spec.generate_report:
@@ -278,7 +277,10 @@ def generate_report(self):
                     test_data=test_data,
                     ci_interval_width=self.spec.confidence_interval_width,
                 )
-                forecast_plots = [forecast_text, series_subtext, forecast_sec]
+                if series_name is not None and len(self.datasets.list_series_ids()) > 1:
+                    forecast_plots = [forecast_text, series_subtext, forecast_sec]
+                else:
+                    forecast_plots = [forecast_text, forecast_sec]
 
                 yaml_appendix_title = dp.Text(f"## Reference: YAML File")
                 yaml_appendix = dp.Code(code=self.config.to_yaml(), language="yaml")
@@ -529,7 +531,7 @@ def _save_report(
                 logger.warn(
                     "Unable to generate explanations for this model type or for this dataset."
                 )
-                raise
+                logger.debug(f"Got error: {e.args}")
 
         if self.spec.generate_model_parameters:
             # model params
@@ -662,6 +664,14 @@ def explain_model(self):
             )
             kernel_explnr_vals = kernel_explnr.shap_values(data_trimmed)
 
+            exp_end_time = time.time()
+            global_ex_time = global_ex_time + exp_end_time - exp_start_time
+
+            self.local_explainer(
+                kernel_explnr, series_id=s_id, datetime_col_name=datetime_col_name
+            )
+            local_ex_time = local_ex_time + time.time() - exp_end_time
+
             if not len(kernel_explnr_vals):
                 logger.warn(
                     f"No explanations generated. Ensure that additional data has been provided."
@@ -673,13 +683,7 @@ def explain_model(self):
                         np.average(np.absolute(kernel_explnr_vals[:, 1:]), axis=0),
                     )
                 )
-            exp_end_time = time.time()
-            global_ex_time = global_ex_time + exp_end_time - exp_start_time
 
-            self.local_explainer(
-                kernel_explnr, series_id=s_id, datetime_col_name=datetime_col_name
-            )
-            local_ex_time = local_ex_time + time.time() - exp_end_time
         logger.info(
             "Global explanations generation completed in %s seconds", global_ex_time
         )

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

@@ -65,6 +65,16 @@ class AdditionalData(AbstractData):
     def __init__(self, spec, historical_data):
         if spec.additional_data is not None:
             super().__init__(spec=spec, name="additional_data")
+            add_dates = self.data.index.get_level_values(0).unique().tolist()
+            add_dates.sort()
+            if historical_data.get_max_time() > add_dates[-spec.horizon]:
+                raise DataMismatchError(
+                    f"The Historical Data ends on {historical_data.get_max_time()}. The additional data horizon starts on {add_dates[-spec.horizon]}. The horizon should have exactly {spec.horizon} dates after the Hisotrical at a frequency of {historical_data.freq}"
+                )
+            elif historical_data.get_max_time() != add_dates[-(spec.horizon + 1)]:
+                raise DataMismatchError(
+                    f"The Additional Data must be present for all historical data and the entire horizon. The Historical Data ends on {historical_data.get_max_time()}. The additonal data horizon starts after {add_dates[-(spec.horizon+1)]}. These should be the same date."
+                )
         else:
             self.name = "additional_data"
             self.data = None

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

@@ -209,7 +209,6 @@ def _train_model(self, i, s_id, df, model_kwargs):
             logger.debug("===========Done===========")
         except Exception as e:
             self.errors_dict[s_id] = {"model_name": self.spec.model, "error": str(e)}
-            raise e
 
     def _build_model(self) -> pd.DataFrame:
         full_data_dict = self.datasets.get_data_by_series()
@@ -402,7 +401,6 @@ def _generate_report(self):
                 # Do not fail the whole run due to explanations failure
                 logger.warn(f"Failed to generate Explanations with error: {e}.")
                 logger.debug(f"Full Traceback: {traceback.format_exc()}")
-                raise
 
         model_description = dp.Text(
             "NeuralProphet is an easy to learn framework for interpretable time "

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

@@ -131,7 +131,6 @@ def _train_model(self, i, series_id, df, model_kwargs):
                 "model_name": self.spec.model,
                 "error": str(e),
             }
-            raise
 
     def _build_model(self) -> pd.DataFrame:
         from prophet import Prophet

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

@@ -250,7 +250,6 @@ def evaluate_train_metrics(output, metrics_col_name=None):
                 f"Failed to generate training metrics for target_series: {s_id}"
             )
             logger.debug(f"Recieved Error Statement: {e}")
-            raise
     return total_metrics