Real-Time What-If Analysis with Model Catalog and Model Deployment Integration (#1041)

Author: prasankh

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

@@ -12,6 +12,7 @@
 
 import fsspec
 import oracledb
+import json
 import pandas as pd
 
 from ads.common.object_storage_details import ObjectStorageDetails
@@ -125,7 +126,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):
     disable_print()
     if not format:
         _, format = os.path.splitext(filename)
@@ -141,6 +142,15 @@ def write_data(data, filename, format, storage_options, index=False, **kwargs):
     )
 
 
+def write_simple_json(data, path):
+    if ObjectStorageDetails.is_oci_path(path):
+        storage_options = default_signer()
+    else:
+        storage_options = {}
+    with fsspec.open(path, mode="w", **storage_options) as f:
+        json.dump(data, f, indent=4)
+
+
 def merge_category_columns(data, target_category_columns):
     result = data.apply(
         lambda x: "__".join([str(x[col]) for col in target_category_columns]), axis=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,15 @@ 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()
+        if spec.what_if_analysis.model_deployment:
+            mdm.create_deployment()
+        mdm.save_deployment_info()
+
 
 def verify(spec: Dict, **kwargs: Dict) -> bool:
     """Verifies the forecasting operator config."""

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

@@ -167,7 +167,7 @@ def get_data_multi_indexed(self):
                 self.historical_data.data,
                 self.additional_data.data,
             ],
-            axis=1,
+            axis=1
         )
 
     def get_data_by_series(self, include_horizon=True):

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

@@ -18,6 +18,35 @@
 
 from .const import SpeedAccuracyMode, SupportedMetrics, SupportedModels
 
+@dataclass
+class AutoScaling(DataClassSerializable):
+    """Class representing simple autoscaling policy"""
+    minimum_instance: int = 1
+    maximum_instance: int = None
+    cool_down_in_seconds: int = 600
+    scale_in_threshold: int = 10
+    scale_out_threshold: int = 80
+    scaling_metric: str = "CPU_UTILIZATION"
+
+@dataclass(repr=True)
+class ModelDeploymentServer(DataClassSerializable):
+    """Class representing model deployment server specification for whatif-analysis."""
+    display_name: str = None
+    initial_shape: str = None
+    description: str = None
+    log_group: str = None
+    log_id: str = None
+    auto_scaling: AutoScaling = field(default_factory=AutoScaling)
+
+
+@dataclass(repr=True)
+class WhatIfAnalysis(DataClassSerializable):
+    """Class representing operator specification for whatif-analysis."""
+    model_display_name: str = None
+    compartment_id: str = None
+    project_id: str = None
+    model_deployment: ModelDeploymentServer = field(default_factory=ModelDeploymentServer)
+
 
 @dataclass(repr=True)
 class TestData(InputData):
@@ -90,12 +119,14 @@ class ForecastOperatorSpec(DataClassSerializable):
     confidence_interval_width: float = None
     metric: str = None
     tuning: Tuning = field(default_factory=Tuning)
+    what_if_analysis: WhatIfAnalysis = field(default_factory=WhatIfAnalysis)
 
     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

@@ -353,6 +353,69 @@ spec:
       meta:
         description: "Report file generation can be enabled using this flag. Defaults to true."
 
+    what_if_analysis:
+      type: dict
+      required: false
+      schema:
+        model_deployment:
+          type: dict
+          required: false
+          meta: "If model_deployment id is not specified, a new model deployment is created; otherwise, the model is linked to the specified model deployment."
+          schema:
+            id:
+              type: string
+              required: false
+            display_name:
+              type: string
+              required: false
+            initial_shape:
+              type: string
+              required: false
+            description:
+              type: string
+              required: false
+            log_group:
+              type: string
+              required: true
+            log_id:
+              type: string
+              required: true
+            auto_scaling:
+              type: dict
+              required: false
+              schema:
+                minimum_instance:
+                  type: integer
+                  required: true
+                maximum_instance:
+                  type: integer
+                  required: true
+                scale_in_threshold:
+                  type: integer
+                  required: true
+                scale_out_threshold:
+                  type: integer
+                  required: true
+                scaling_metric:
+                  type: string
+                  required: true
+                cool_down_in_seconds:
+                  type: integer
+                  required: true
+        model_display_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."
+
     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