Forecasting Operator

README.md

@@ -27,6 +27,14 @@ You have various options when installing ADS.
  python3 -m pip install oracle-ads
 ```
 
+### Installing OCI AI Operators
+
+To use the AI Forecast Operator, install the "forecast" dependencies using the following command:
+
+```bash
+ python3 -m pip install 'oracle_ads[forecast]==2.9.0rc1'
+```
+
 ### Installing extras libraries
 
 To work with gradient boosting models, install the `boosted` module. This module includes XGBoost and LightGBM model classes.

ads/opctl/config/merger.py

@@ -65,7 +65,6 @@ def process(self, **kwargs) -> None:
 
  self._config_flex_shape_details()
 
- logger.debug(f"Config: {self.config}")
  return self
 
  def _merge_config_with_cmd_args(self, cmd_args: Dict) -> None:
@@ -135,7 +134,6 @@ def _fill_config_with_defaults(self, ads_config_path: str) -> None:
  )
  self.config["execution"]["oci_profile"] = profile
  # loading config for corresponding profile
- logger.debug(f"Loading service config for profile {profile}.")
  infra_config = self._get_service_config(profile, ads_config_path)
  if infra_config.get(
  "conda_pack_os_prefix"

ads/opctl/config/resolver.py

@@ -111,8 +111,6 @@ def process(self):
  self._resolve_mounted_volumes()
  self._resolve_job_name()
 
- logger.debug(f"Config: {self.config}")
-
  return self
 
  def _is_ads_operator(self) -> bool:

ads/opctl/config/versioner.py

@@ -64,6 +64,5 @@ def process(self):
  # - user runs ADS operator -> name/YAML
 
  # TODO: build this out
- logger.debug(f"Config: {self.config}")
 
  return self

ads/opctl/config/yaml_parsers/base.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8; -*-
 
-# Copyright (c) 2022 Oracle and/or its affiliates.
+# Copyright (c) 2022, 2023 Oracle and/or its affiliates.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
 
 from logging import getLogger
@@ -42,7 +42,6 @@ def parse(self):
 
  @classmethod
  def parse_content(cls, file):
- logger.debug(f"Config filename is {file}")
  yaml_spec = {}
  if isinstance(file, dict):
  yaml_spec = file
@@ -53,10 +52,7 @@ def parse_content(cls, file):
  parsed_output = None
  if kind:
  className = f"{kind[0].upper()}{kind[1:]}SpecParser"
- logger.debug(f"kind: {kind}")
- logger.debug(f"Parser Class name: {className}")
  m = importlib.import_module("ads.opctl.config.yaml_parsers")
  parser = getattr(m, className)
  parsed_output = parser(yaml_spec).parse()
- logger.debug(f"Parsed Output: {parsed_output}")
  return parsed_output

ads/opctl/operator/cli.py

@@ -9,8 +9,7 @@
 import click
 import fsspec
 import yaml
-
-from ads.common import auth as authutil
+from ads.opctl.operator.common.utils import default_signer
 from ads.common.auth import AuthType
 from ads.common.object_storage_details import ObjectStorageDetails
 from ads.opctl.constants import BACKEND_NAME, RUNTIME_TYPE
@@ -212,9 +211,7 @@ def verify(debug: bool, **kwargs: Dict[str, Any]) -> None:
  kwargs["file"],
  "r",
  **(
- authutil.default_signer()
- if ObjectStorageDetails.is_oci_path(kwargs["file"])
- else {}
+ default_signer() if ObjectStorageDetails.is_oci_path(kwargs["file"]) else {}
  ),
  ) as f:
  operator_spec = suppress_traceback(debug)(yaml.safe_load)(f.read())
@@ -318,9 +315,7 @@ def run(debug: bool, **kwargs: Dict[str, Any]) -> None:
  kwargs["file"],
  "r",
  **(
- authutil.default_signer()
- if ObjectStorageDetails.is_oci_path(kwargs["file"])
- else {}
+ default_signer() if ObjectStorageDetails.is_oci_path(kwargs["file"]) else {}
  ),
  ) as f:
  operator_spec = suppress_traceback(debug)(yaml.safe_load)(f.read())
@@ -329,11 +324,7 @@ def run(debug: bool, **kwargs: Dict[str, Any]) -> None:
  with fsspec.open(
  backend,
  "r",
- **(
- authutil.default_signer()
- if ObjectStorageDetails.is_oci_path(backend)
- else {}
- ),
+ **(default_signer() if ObjectStorageDetails.is_oci_path(backend) else {}),
  ) as f:
  backend = suppress_traceback(debug)(yaml.safe_load)(f.read())
 

ads/opctl/operator/common/operator_loader.py

@@ -19,7 +19,7 @@
 
 from yaml import SafeLoader as loader
 
-from ads.common import auth as authutil
+from ads.opctl.operator.common.utils import default_signer
 from ads.common.decorator.runtime_dependency import runtime_dependency
 from ads.common.serializer import DataClassSerializable
 from ads.common.utils import copy_from_uri
@@ -461,9 +461,7 @@ def __init__(self, uri: str, uri_dst: str = None, auth: Dict = None) -> None:
  auth (Dict, optional)
  Default authentication settings.
  """
- super().__init__(
- uri=uri, uri_dst=uri_dst, auth=auth or authutil.default_signer()
- )
+ super().__init__(uri=uri, uri_dst=uri_dst, auth=auth or default_signer())
 
  def _load(self, **kwargs: Dict) -> OperatorInfo:
  """Downloads the operator's source code to the local folder.

ads/opctl/operator/common/utils.py

@@ -149,3 +149,10 @@ def _load_yaml_from_uri(uri: str, **kwargs) -> str:
  """Loads YAML from the URI path. Can be Object Storage path."""
  with fsspec.open(uri) as f:
  return _load_yaml_from_string(str(f.read(), "UTF-8"), **kwargs)
+
+
+def default_signer(**kwargs):
+ os.environ["EXTRA_USER_AGENT_INFO"] = "Operator"
+ from ads.common.auth import default_signer
+
+ return default_signer(**kwargs)

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

@@ -18,9 +18,11 @@
 from .operator_config import ForecastOperatorConfig
 from .model.forecast_datasets import ForecastDatasets
 
+
 def operate(operator_config: ForecastOperatorConfig) -> None:
  """Runs the forecasting operator."""
  from .model.factory import ForecastOperatorModelFactory
+
  datasets = ForecastDatasets(operator_config)
  ForecastOperatorModelFactory.get_model(operator_config, datasets).generate_report()
 
@@ -65,8 +67,6 @@ def main(raw_args: List[str]):
  yaml_string=yaml_string,
  )
 
- logger.info(operator_config.to_yaml())
-
  # run operator
  if args.verify:
  verify(operator_config)

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

@@ -45,6 +45,18 @@ class SupportedMetrics(str, metaclass=ExtendedEnumMeta):
  ELAPSED_TIME = "Elapsed Time"
 
 
+class ForecastOutputColumns(str, metaclass=ExtendedEnumMeta):
+ """The column names for the forecast.csv output file"""
+
+ DATE = "Date"
+ SERIES = "Series"
+ INPUT_VALUE = "input_value"
+ FITTED_VALUE = "fitted_value"
+ FORECAST_VALUE = "forecast_value"
+ UPPER_BOUND = "upper_bound"
+ LOWER_BOUND = "lower_bound"
+
+
 AUTOMLX_METRIC_MAP = {
  "smape": "neg_sym_mean_abs_percent_error",
  "mape": "neg_sym_mean_abs_percent_error",

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

@@ -14,7 +14,8 @@
 from .base_model import ForecastOperatorBaseModel
 from ..operator_config import ForecastOperatorConfig
 import traceback
-from .forecast_datasets import ForecastDatasets
+from .forecast_datasets import ForecastDatasets, ForecastOutput
+from ..const import ForecastOutputColumns
 
 
 class ArimaOperatorModel(ForecastOperatorBaseModel):
@@ -24,11 +25,12 @@ def __init__(self, config: ForecastOperatorConfig, datasets: ForecastDatasets):
  super().__init__(config, datasets=datasets)
  self.global_explanation = {}
  self.local_explanation = {}
+ self.train_metrics = True
  self.formatted_global_explanation = None
  self.formatted_local_explanation = None
 
  def _build_model(self) -> pd.DataFrame:
- full_data_dict = self.full_data_dict
+ full_data_dict = self.datasets.full_data_dict
 
  # Extract the Confidence Interval Width and convert to arima's equivalent - alpha
  if self.spec.confidence_interval_width is None:
@@ -37,10 +39,20 @@ def _build_model(self) -> pd.DataFrame:
  )
  model_kwargs = self.spec.model_kwargs
  model_kwargs["alpha"] = 1 - self.spec.confidence_interval_width
+ if "error_action" not in model_kwargs.keys():
+ model_kwargs["error_action"] = "ignore"
 
  models = []
+ self.datasets.datetime_col = self.spec.datetime_column.name
+ self.forecast_output = ForecastOutput(
+ confidence_interval_width=self.spec.confidence_interval_width
+ )
+
  outputs = dict()
  outputs_legacy = []
+ fitted_values = dict()
+ actual_values = dict()
+ dt_columns = dict()
 
  for i, (target, df) in enumerate(full_data_dict.items()):
  # format the dataframe for this target. Dropping NA on target[df] will remove all future data
@@ -60,7 +72,9 @@ def _build_model(self) -> pd.DataFrame:
  target,
  self.spec.datetime_column.name,
  }
- logger.info(f"Additional Regressors Detected {list(additional_regressors)}")
+ logger.debug(
+ f"Additional Regressors Detected {list(additional_regressors)}"
+ )
 
  # Split data into X and y for arima tune method
  y = data_i[target]
@@ -71,6 +85,9 @@ def _build_model(self) -> pd.DataFrame:
  # Build and fit model
  model = pm.auto_arima(y=y, X=X_in, **self.spec.model_kwargs)
 
+ fitted_values[target] = model.predict_in_sample(X=X_in)
+ actual_values[target] = y
+
  # Build future dataframe
  start_date = y.index.values[-1]
  n_periods = self.spec.horizon
@@ -89,12 +106,19 @@ def _build_model(self) -> pd.DataFrame:
  alpha=model_kwargs["alpha"],
  )
  yhat_clean = pd.DataFrame(yhat, index=yhat.index, columns=["yhat"])
+
+ dt_columns[target] = pd.concat(
+ [
+ df_encoded[self.spec.datetime_column.name],
+ pd.Series(yhat_clean.index),
+ ]
+ )
  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.info(f"-----------------Model {i}----------------------")
- logger.info(forecast[["yhat", "yhat_lower", "yhat_upper"]].tail())
+ logger.debug(f"-----------------Model {i}----------------------")
+ logger.debug(forecast[["yhat", "yhat_lower", "yhat_upper"]].tail())
 
  # Collect all outputs
  models.append(model)
@@ -104,44 +128,35 @@ def _build_model(self) -> pd.DataFrame:
  outputs[target] = forecast
 
  self.models = models
- self.outputs = outputs_legacy
 
- logger.info("===========Done===========")
- outputs_merged = pd.DataFrame()
+ logger.debug("===========Done===========")
 
  # Merge the outputs from each model into 1 df with all outputs by target and category
  col = self.original_target_column
  output_col = pd.DataFrame()
- yhat_upper_percentage = int(100 - model_kwargs["alpha"] * 100 / 2)
- yhat_lower_name = "p" + str(int(100 - yhat_upper_percentage))
- yhat_upper_name = "p" + str(yhat_upper_percentage)
+ yhat_upper_name = ForecastOutputColumns.UPPER_BOUND
+ yhat_lower_name = ForecastOutputColumns.LOWER_BOUND
  for cat in self.categories:
  output_i = pd.DataFrame()
-
- output_i["Date"] = outputs[f"{col}_{cat}"].index
+ output_i["Date"] = dt_columns[f"{col}_{cat}"]
+ output_i = output_i.set_index("Date")
  output_i["Series"] = cat
- output_i["input_value"] = float("nan")
- output_i[f"fitted_value"] = float("nan")
- output_i[f"forecast_value"] = outputs[f"{col}_{cat}"]["yhat"].values
- output_i[yhat_upper_name] = outputs[f"{col}_{cat}"]["yhat_upper"].values
- output_i[yhat_lower_name] = outputs[f"{col}_{cat}"]["yhat_lower"].values
+ output_i["input_value"] = actual_values[f"{col}_{cat}"]
+
+ output_i["fitted_value"] = fitted_values[f"{col}_{cat}"]
+ output_i["forecast_value"] = outputs[f"{col}_{cat}"]["yhat"]
+ output_i[yhat_upper_name] = outputs[f"{col}_{cat}"]["yhat_upper"]
+ output_i[yhat_lower_name] = outputs[f"{col}_{cat}"]["yhat_lower"]
+
+ output_i = output_i.reset_index(drop=False)
  output_col = pd.concat([output_col, output_i])
+ self.forecast_output.add_category(
+ category=cat, target_category_column=f"{col}_{cat}", forecast=output_i
+ )
 
- # output_col = output_col.sort_values(operator.ds_column).reset_index(drop=True)
  output_col = output_col.reset_index(drop=True)
- outputs_merged = pd.concat([outputs_merged, output_col], axis=1)
-
- # Re-merge historical datas for processing
- data_merged = pd.concat(
- [
- v[v[k].notna()].set_index(self.spec.datetime_column.name)
- for k, v in full_data_dict.items()
- ],
- axis=1,
- ).reset_index()
 
- self.data = data_merged
- return outputs_merged
+ return output_col
 
  def _generate_report(self):
  """The method that needs to be implemented on the particular model level."""
@@ -221,16 +236,10 @@ def _generate_report(self):
  "it predicts future values based on past values."
  )
  other_sections = all_sections
- ds_column_series = self.data[self.spec.datetime_column.name]
- ds_forecast_col = self.outputs[0].index
- ci_col_names = ["yhat_lower", "yhat_upper"]
 
  return (
  model_description,
  other_sections,
- ds_column_series,
- ds_forecast_col,
- ci_col_names,
  )
 
  def _custom_predict_arima(self, data):