added options for reading and model saving

Author: prasankh

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

@@ -19,6 +19,14 @@
 from .const import SpeedAccuracyMode, SupportedMetrics, SupportedModels
 
 
+@dataclass(repr=True)
+class WhatIfAnalysis(DataClassSerializable):
+    """Class representing operator specification for whatif-analysis."""
+    model_name: str = None
+    compartment_id: str = None
+    project_id: str = None
+
+
 @dataclass(repr=True)
 class TestData(InputData):
     """Class representing operator specification test data details."""
@@ -90,7 +98,7 @@ class ForecastOperatorSpec(DataClassSerializable):
     confidence_interval_width: float = None
     metric: str = None
     tuning: Tuning = field(default_factory=Tuning)
-    what_if_analysis: bool = False
+    what_if_analysis: WhatIfAnalysis = field(default_factory=WhatIfAnalysis)
 
     def __post_init__(self):
         """Adjusts the specification details."""

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

@@ -341,9 +341,20 @@ spec:
         description: "Report file generation can be enabled using this flag. Defaults to true."
 
     what_if_analysis:
-      type: boolean
+      type: dict
       required: false
-      default: false
+      schema:
+        model_name:
+          type: string
+          required: true
+        project_id:
+          type: string
+          required: false
+          meta: "If not provided, The project OCID from config.PROJECT_OCID is used"
+        compartment_id:
+          type: string
+          required: false
+          meta: "If not provided, The compartment OCID from config.NB_SESSION_COMPARTMENT_OCID is used."
       meta:
         description: "When enabled, the models are saved to the model catalog. Defaults to false."
 

ads/opctl/operator/lowcode/forecast/whatifserve/deployment_manager.py

@@ -27,6 +27,9 @@ def __init__(self, spec: ForecastOperatorSpec, additional_data: AdditionalData,
         self.horizon = spec.horizon
         self.additional_data = additional_data.get_dict_by_series()
         self.model_obj = {}
+        self.display_name = spec.what_if_analysis.model_name
+        self.project_id = spec.what_if_analysis.project_id
+        self.compartment_id = spec.what_if_analysis.compartment_id
         self.path_to_artifact = f"{self.spec.output_directory.url}/artifacts/"
         self.pickle_file_path = f"{self.spec.output_directory.url}/model.pkl"
         self.model_version = previous_model_version + 1 if previous_model_version else 1
@@ -48,8 +51,8 @@ def _satiny_test(self):
             date_col_format = self.spec.datetime_column.format
             sample_prediction_data[date_col_name] = sample_prediction_data[date_col_name].dt.strftime(date_col_format)
             sample_prediction_data.to_csv(temp_file.name, index=False)
-            additional_data_uri = "additional_data_uri"
-            input_data = {additional_data_uri: temp_file.name}
+            additional_data_uri = "additional_data"
+            input_data = {additional_data_uri: {"url": temp_file.name}}
             prediction_test = predict(input_data, _)
             logger.info(f"prediction test completed with result :{prediction_test}")
 
@@ -91,7 +94,9 @@ def save_to_catalog(self):
 
         catalog_id = "None"
         if not os.environ.get("TEST_MODE", False):
-            catalog_entry = artifact.save(display_name=f"{self.model_name}-v{self.model_version}",
+            catalog_entry = artifact.save(display_name=self.display_name,
+                                          compartment_id=self.compartment_id,
+                                          project_id=self.project_id,
                                           description=description)
             catalog_id = catalog_entry.id
 

ads/opctl/operator/lowcode/forecast/whatifserve/score.py

@@ -11,11 +11,8 @@
 from functools import lru_cache
 import logging
 import ads
-from prophet import Prophet
-from neuralprophet import NeuralProphet
-from pmdarima import ARIMA
-from autots import AutoTS
-from automlx._interface.forecaster import AutoForecaster
+from ads.opctl.operator.lowcode.common.utils import load_data
+from ads.opctl.operator.common.operator_config import InputData
 
 ads.set_auth("resource_principal")
 
@@ -29,6 +26,12 @@
    Inference script. This script is used for prediction by scoring server when schema is known.
 """
 
+AUTOTS = "autots"
+ARIMA = "arima"
+PROPHET = "prophet"
+NEURALPROPHET = "neuralprophet"
+AUTOMLX = "automlx"
+
 
 @lru_cache(maxsize=10)
 def load_model():
@@ -132,14 +135,14 @@ def post_inference(yhat):
     return yhat
 
 
-def get_forecast(future_df, series_id, model_object, date_col, target_column, target_cat_col, horizon):
+def get_forecast(future_df, model_name, series_id, model_object, date_col, target_column, target_cat_col, horizon):
     date_col_name = date_col["name"]
     date_col_format = date_col["format"]
     future_df[target_cat_col] = future_df[target_cat_col].astype("str")
     future_df[date_col_name] = pd.to_datetime(
         future_df[date_col_name], format=date_col_format
     )
-    if isinstance(model_object, AutoTS):
+    if model_name == AUTOTS:
         series_id_col = "Series"
         full_data_indexed = future_df.rename(columns={target_cat_col: series_id_col})
         additional_regressors = list(
@@ -152,12 +155,12 @@ def get_forecast(future_df, series_id, model_object, date_col, target_column, ta
         )
         pred_obj = model_object.predict(future_regressor=future_reg)
         return pred_obj.forecast[series_id].tolist()
-    elif series_id in model_object and isinstance(model_object[series_id], Prophet):
+    elif model_name == PROPHET and series_id in model_object:
         model = model_object[series_id]
         processed = future_df.rename(columns={date_col_name: 'ds', target_column: 'y'})
         forecast = model.predict(processed)
         return forecast['yhat'].tolist()
-    elif series_id in model_object and isinstance(model_object[series_id], NeuralProphet):
+    elif model_name == NEURALPROPHET and series_id in model_object:
         model = model_object[series_id]
         model.restore_trainer()
         accepted_regressors = list(model.config_regressors.keys())
@@ -166,7 +169,7 @@ def get_forecast(future_df, series_id, model_object, date_col, target_column, ta
         future["y"] = None
         forecast = model.predict(future)
         return forecast['yhat1'].tolist()
-    elif series_id in model_object and isinstance(model_object[series_id], ARIMA):
+    elif model_name == ARIMA and series_id in model_object:
         model = model_object[series_id]
         future_df = future_df.set_index(date_col_name)
         x_pred = future_df.drop(target_cat_col, axis=1)
@@ -177,7 +180,7 @@ def get_forecast(future_df, series_id, model_object, date_col, target_column, ta
         )
         yhat_clean = pd.DataFrame(yhat, index=yhat.index, columns=["yhat"])
         return yhat_clean['yhat'].tolist()
-    elif series_id in model_object and isinstance(model_object[series_id], AutoForecaster):
+    elif model_name == AUTOMLX and series_id in model_object:
         # automlx model
         model = model_object[series_id]
         x_pred = future_df.drop(target_cat_col, axis=1)
@@ -188,7 +191,7 @@ def get_forecast(future_df, series_id, model_object, date_col, target_column, ta
         )
         return forecast[target_column].tolist()
     else:
-        raise Exception( f"Invalid model object type: {type(model_object).__name__}.")
+        raise Exception(f"Invalid model object type: {type(model_object).__name__}.")
 
 
 def predict(data, model=load_model()) -> dict:
@@ -211,20 +214,26 @@ def predict(data, model=load_model()) -> dict:
     models = model["models"]
     specs = model["spec"]
     horizon = specs["horizon"]
+    model_name = specs["model"]
     date_col = specs["datetime_column"]
     target_column = specs["target_column"]
-    forecasts = {}
-    uri = f"{data['additional_data_uri']}"
     target_category_column = specs["target_category_columns"][0]
-    signer = ads.common.auth.default_signer() if uri.lower().startswith("oci://") else {}
-    additional_data = pd.read_csv(uri, storage_options=signer)
+
+    try:
+        input_data = InputData(**data["additional_data"])
+    except TypeError as e:
+        raise ValueError(f"Validation error: {e}")
+    additional_data = load_data(input_data)
+
     unique_values = additional_data[target_category_column].unique()
+    forecasts = {}
     for key in unique_values:
         try:
             s_id = str(key)
             filtered = additional_data[additional_data[target_category_column] == key]
             future = filtered.tail(horizon)
-            forecast = get_forecast(future, s_id, models, date_col, target_column, target_category_column, horizon)
+            forecast = get_forecast(future, model_name, s_id, models, date_col,
+                                    target_column, target_category_column, horizon)
             forecasts[s_id] = json.dumps(forecast)
         except Exception as e:
             raise RuntimeError(

tests/operators/forecast/test_errors.py

@@ -754,7 +754,7 @@ def test_what_if_analysis(operator_setup, model):
     )
     yaml_i["spec"]["horizon"] = 10
     yaml_i["spec"]["model"] = model
-    yaml_i["spec"]["what_if_analysis"] = True
+    yaml_i["spec"]["what_if_analysis"] = {"model_name": f"model_{model}"}
 
     run_yaml(
         tmpdirname=tmpdirname,