added support for What-If Analysis

Author: prasankh

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

@@ -121,7 +121,7 @@ def load_data(data_spec, storage_options=None, **kwargs):
     return data
 
 
-def write_data(data, filename, format, storage_options, index=False, **kwargs):
+def write_data(data, filename, format, storage_options=None, index=False, **kwargs):
     if not format:
         _, format = os.path.splitext(filename)
         format = format[1:]

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

@@ -17,6 +17,7 @@
 
 from .operator_config import ForecastOperatorConfig
 from .model.forecast_datasets import ForecastDatasets
+from .whatifserve import ModelDeploymentManager
 
 
 def operate(operator_config: ForecastOperatorConfig) -> None:
@@ -27,6 +28,12 @@ def operate(operator_config: ForecastOperatorConfig) -> None:
     ForecastOperatorModelFactory.get_model(
         operator_config, datasets
     ).generate_report()
+    # saving to model catalog
+    spec = operator_config.spec
+    if spec.what_if_analysis and datasets.additional_data:
+        mdm = ModelDeploymentManager(spec, datasets.additional_data)
+        mdm.save_to_catalog()
+
 
 def verify(spec: Dict, **kwargs: Dict) -> bool:
     """Verifies the forecasting operator config."""

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

@@ -168,6 +168,7 @@ def get_data_multi_indexed(self):
                 self.additional_data.data,
             ],
             axis=1,
+            join='inner'
         )
 
     def get_data_by_series(self, include_horizon=True):

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

@@ -90,12 +90,14 @@ class ForecastOperatorSpec(DataClassSerializable):
     confidence_interval_width: float = None
     metric: str = None
     tuning: Tuning = field(default_factory=Tuning)
+    what_if_analysis: bool = False
 
     def __post_init__(self):
         """Adjusts the specification details."""
         self.output_directory = self.output_directory or OutputDirectory(
             url=find_output_dirname(self.output_directory)
         )
+        self.generate_model_pickle = True if self.generate_model_pickle or self.what_if_analysis else False
         self.metric = (self.metric or "").lower() or SupportedMetrics.SMAPE.lower()
         self.model = self.model or SupportedModels.Prophet
         self.confidence_interval_width = self.confidence_interval_width or 0.80

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

@@ -340,6 +340,13 @@ spec:
       meta:
         description: "Report file generation can be enabled using this flag. Defaults to true."
 
+    what_if_analysis:
+      type: boolean
+      required: false
+      default: false
+      meta:
+        description: "When enabled, the models are saved to the model catalog. Defaults to false."
+
     generate_metrics:
       type: boolean
       required: false

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

@@ -0,0 +1,7 @@
+#!/usr/bin/env python
+
+# 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/
+
+
+from .deployment_manager import ModelDeploymentManager

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

@@ -0,0 +1,115 @@
+#!/usr/bin/env python
+import json
+# 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 os
+import pickle
+import shutil
+import sys
+import tempfile
+
+import pandas as pd
+from joblib import dump
+
+from ads.common.model_export_util import prepare_generic_model
+from ads.opctl.operator.lowcode.common.utils import write_data, call_pandas_fsspec
+
+from ..model.forecast_datasets import AdditionalData
+from ..operator_config import ForecastOperatorSpec
+
+
+class ModelDeploymentManager:
+    def __init__(self, spec: ForecastOperatorSpec, additional_data: AdditionalData, previous_model_version=None):
+        self.spec = spec
+        # self.model_path = spec.output_directory.url
+        self.model_name = spec.model
+        self.horizon = spec.horizon
+        self.additional_data = additional_data.get_dict_by_series()
+        self.model_obj = {}
+        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
+
+    def _satiny_test(self):
+        """
+        Function perform sanity test for saved artifact
+        """
+        sys.path.insert(0, f"{self.path_to_artifact}")
+        from score import load_model, predict
+        _ = load_model()
+
+        # Write additional data to tmp file and perform sanity check
+        with tempfile.NamedTemporaryFile(suffix='.csv') as temp_file:
+            one_series = next(iter(self.additional_data))
+            sample_prediction_data = self.additional_data[one_series].tail(self.horizon)
+            sample_prediction_data[self.spec.target_category_columns[0]] = one_series
+            date_col_name = self.spec.datetime_column.name
+            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}
+            prediction_test = predict(input_data, _)
+            print(f"prediction test completed with result :{prediction_test}")
+
+    def _copy_score_file(self):
+        """
+        Copies the score.py to the artifact_path.
+        """
+        try:
+            current_dir = os.path.dirname(os.path.abspath(__file__))
+            score_file = os.path.join(current_dir, "score.py")
+            destination_file = os.path.join(self.path_to_artifact, os.path.basename(score_file))
+            shutil.copy2(score_file, destination_file)
+            print(f"score.py copied successfully to {self.path_to_artifact}")
+        except Exception as e:
+            print(f"Error copying file: {e}")
+            raise e
+
+    def save_to_catalog(self):
+        """Save the model to a model catalog"""
+        with open(self.pickle_file_path, 'rb') as file:
+            self.model_obj = pickle.load(file)
+
+        if not os.path.exists(self.path_to_artifact):
+            os.mkdir(self.path_to_artifact)
+
+        artifact_dict = {"spec": self.spec.to_dict(), "models": self.model_obj}
+        dump(artifact_dict, os.path.join(self.path_to_artifact, "model.joblib"))
+        artifact = prepare_generic_model(self.path_to_artifact, function_artifacts=False, force_overwrite=True,
+                                         data_science_env=True)
+
+        self._copy_score_file()
+        self._satiny_test()
+        
+        if isinstance(self.model_obj, dict):
+            series = self.model_obj.keys()
+        else:
+            series = self.additional_data.keys()
+        description = f"The object contains {len(series)} {self.model_name} models"
+
+        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}",
+                                          description=description)
+            catalog_id = catalog_entry.id
+
+
+        print(f"Saved {self.model_name} version-v{self.model_version} to model catalog"
+              f" with catalog id : {catalog_id}")
+
+        catalog_mapping = {"catalog_id": catalog_id, "series": list(series)}
+
+        write_data(
+            data=pd.DataFrame([catalog_mapping]),
+            filename=os.path.join(
+                self.spec.output_directory.url, "model_ids.csv"
+            ),
+            format="csv"
+        )
+        return catalog_id
+
+    def create_deployment(self, deployment_config):
+        """Create a model deployment serving"""
+        pass