skip check for global explanations, exception handling

Author: codeloop

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

@@ -413,41 +413,45 @@ def explain_model(self):
         for s_id, data_i in self.datasets.get_data_by_series(
             include_horizon=False
         ).items():
-            if self.spec.explanations_accuracy_mode == SpeedAccuracyMode.AUTOMLX:
-                # Use the MLExplainer class from AutoMLx to generate explanations
-                explainer = automlx.MLExplainer(
-                    self.models[s_id],
-                    self.datasets.additional_data.get_data_for_series(series_id=s_id)
-                    .drop(self.spec.datetime_column.name, axis=1)
-                    .head(-self.spec.horizon)
-                    if self.spec.additional_data
-                    else None,
-                    pd.DataFrame(data_i[self.spec.target_column]),
-                    task="forecasting",
-                )
+            try:
+                if self.spec.explanations_accuracy_mode == SpeedAccuracyMode.AUTOMLX:
+                    # Use the MLExplainer class from AutoMLx to generate explanations
+                    explainer = automlx.MLExplainer(
+                        self.models[s_id],
+                        self.datasets.additional_data.get_data_for_series(series_id=s_id)
+                        .drop(self.spec.datetime_column.name, axis=1)
+                        .head(-self.spec.horizon)
+                        if self.spec.additional_data
+                        else None,
+                        pd.DataFrame(data_i[self.spec.target_column]),
+                        task="forecasting",
+                    )
 
-                # Generate explanations for the forecast
-                explanations = explainer.explain_prediction(
-                    X=self.datasets.additional_data.get_data_for_series(series_id=s_id)
-                    .drop(self.spec.datetime_column.name, axis=1)
-                    .tail(self.spec.horizon)
-                    if self.spec.additional_data
-                    else None,
-                    forecast_timepoints=list(range(self.spec.horizon + 1)),
-                )
+                    # Generate explanations for the forecast
+                    explanations = explainer.explain_prediction(
+                        X=self.datasets.additional_data.get_data_for_series(series_id=s_id)
+                        .drop(self.spec.datetime_column.name, axis=1)
+                        .tail(self.spec.horizon)
+                        if self.spec.additional_data
+                        else None,
+                        forecast_timepoints=list(range(self.spec.horizon + 1)),
+                    )
 
-                # Convert the explanations to a DataFrame
-                explanations_df = pd.concat(
-                    [exp.to_dataframe() for exp in explanations]
-                )
-                explanations_df["row"] = explanations_df.groupby("Feature").cumcount()
-                explanations_df = explanations_df.pivot(
-                    index="row", columns="Feature", values="Attribution"
-                )
-                explanations_df = explanations_df.reset_index(drop=True)
+                    # Convert the explanations to a DataFrame
+                    explanations_df = pd.concat(
+                        [exp.to_dataframe() for exp in explanations]
+                    )
+                    explanations_df["row"] = explanations_df.groupby("Feature").cumcount()
+                    explanations_df = explanations_df.pivot(
+                        index="row", columns="Feature", values="Attribution"
+                    )
+                    explanations_df = explanations_df.reset_index(drop=True)
 
-                # Store the explanations in the local_explanation dictionary
-                self.local_explanation[s_id] = explanations_df
-            else:
-                # Fall back to the default explanation generation method
-                super().explain_model()
+                    # Store the explanations in the local_explanation dictionary
+                    self.local_explanation[s_id] = explanations_df
+                else:
+                    # Fall back to the default explanation generation method
+                    super().explain_model()
+            except Exception as e:
+                logger.warning(f"Failed to generate explanations for series {s_id} with error: {e}.")
+                logger.debug(f"Full Traceback: {traceback.format_exc()}")

tests/operators/forecast/test_errors.py

@@ -591,6 +591,8 @@ def test_all_series_failure(model):
     yaml_i["spec"]["preprocessing"] = {"enabled": True, "steps": preprocessing_steps}
     if yaml_i["spec"].get("additional_data") is not None and model != "autots":
         yaml_i["spec"]["generate_explanations"] = True
+    else:
+        yaml_i["spec"]["generate_explanations"] = False
     if model == "autots":
         yaml_i["spec"]["model_kwargs"] = {"model_list": "superfast"}
     if model == "automlx":
@@ -700,21 +702,15 @@ def test_arima_automlx_errors(operator_setup, model):
                 in error_content["13"]["error"]
             ), "Error message mismatch"
 
-    if model not in ["autots", "automlx"]:  # , "lgbforecast"
-        global_fn = f"{tmpdirname}/results/global_explanation.csv"
-        assert os.path.exists(
-            global_fn
-        ), f"Global explanation file not found at {report_path}"
+    if model not in ["autots"]:  # , "lgbforecast"
+        if yaml_i["spec"].get("explanations_accuracy_mode") != "AUTOMLX":
+            global_fn = f"{tmpdirname}/results/global_explanation.csv"
+            assert os.path.exists(global_fn), f"Global explanation file not found at {report_path}"
+            assert not pd.read_csv(global_fn, index_col=0).empty
 
         local_fn = f"{tmpdirname}/results/local_explanation.csv"
-        assert os.path.exists(
-            local_fn
-        ), f"Local explanation file not found at {report_path}"
-
-        glb_expl = pd.read_csv(global_fn, index_col=0)
-        loc_expl = pd.read_csv(local_fn)
-        assert not glb_expl.empty
-        assert not loc_expl.empty
+        assert os.path.exists(local_fn), f"Local explanation file not found at {report_path}"
+        assert not pd.read_csv(local_fn).empty
 
 
 def test_smape_error():