test pandas df directly

Author: ahosler

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

@@ -210,12 +210,13 @@ def _outlier_treatment(self, df):
         -------
             A new Pandas DataFrame with treated outliears.
         """
-        df["z_score"] = (
+        return df
+        df["__z_score"] = (
             df[self.target_column_name]
             .groupby(DataColumns.Series)
             .transform(lambda x: (x - x.mean()) / x.std())
         )
-        outliers_mask = df["z_score"].abs() > 3
+        outliers_mask = df["__z_score"].abs() > 3
 
         if df[self.target_column_name].dtype == np.int:
             df[self.target_column_name].astype(np.float)
@@ -225,7 +226,7 @@ def _outlier_treatment(self, df):
             .groupby(DataColumns.Series)
             .transform(lambda x: np.median(x))
         )
-        df_ret = df.drop("z_score", axis=1)
+        df_ret = df.drop("__z_score", axis=1)
         return df_ret
 
     def _check_historical_dataset(self, df):

tests/operators/forecast/test_datasets.py

@@ -3,6 +3,7 @@
 # Copyright (c) 2023, 2025 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 json
 import yaml
 import tempfile
 import subprocess
@@ -15,6 +16,8 @@
 import pathlib
 import datetime
 from ads.opctl.operator.cmd import run
+from ads.opctl.operator.lowcode.forecast.__main__ import operate as forecast_operate
+from ads.opctl.operator.lowcode.forecast.operator_config import ForecastOperatorConfig
 
 
 DATASET_PREFIX = f"{os.path.dirname(os.path.abspath(__file__))}/../data/timeseries/"
@@ -28,10 +31,10 @@
 
 MODELS = [
     "arima",
-    "automlx",
+    # "automlx",
     "prophet",
     "neuralprophet",
-    "autots",
+    # "autots",
     # "lgbforecast",
     "auto-select",
 ]
@@ -159,38 +162,76 @@ def test_load_datasets(model, data_details):
             print(train_metrics)
 
 
-@pytest.mark.parametrize("model", MODELS[:-1])
-def test_pandas_df_historical(model):
+@pytest.mark.parametrize("model", MODELS[:1])
+def test_pandas_to_historical(model):
     df = pd.read_csv(f"{DATASET_PREFIX}dataset1.csv")
 
-    yaml_i = deepcopy(TEMPLATE_YAML)
-    yaml_i["spec"]["model"] = model
-    yaml_i["spec"]["historical_data"].pop("url")
-    yaml_i["spec"]["historical_data"]["data"] = df
-    yaml_i["spec"]["target_column"] = "Y"
-    yaml_i["spec"]["datetime_column"]["name"] = DATETIME_COL
-    yaml_i["spec"]["horizon"] = 5
-    run(yaml_i, backend="operator.local", debug=False)
-    subprocess.run(f"ls -a {output_data_path}", shell=True)
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        output_data_path = f"{tmpdirname}/results"
+        yaml_i = deepcopy(TEMPLATE_YAML)
+        yaml_i["spec"]["model"] = model
+        yaml_i["spec"]["historical_data"].pop("url")
+        yaml_i["spec"]["historical_data"]["data"] = df
+        yaml_i["spec"]["target_column"] = "Y"
+        yaml_i["spec"]["datetime_column"]["name"] = DATETIME_COL
+        yaml_i["spec"]["horizon"] = PERIODS
+        yaml_i["spec"]["output_directory"]["url"] = output_data_path
+        operator_config = ForecastOperatorConfig.from_dict(yaml_i)
+        forecast_operate(operator_config)
+        check_output_for_errors(output_data_path)
 
 
-@pytest.mark.parametrize("model", MODELS[:-1])
-def test_pandas_historical_test(model):
+@pytest.mark.parametrize("model", ["neuralprophet"])
+def test_pandas_to_historical_test(model):
     df = pd.read_csv(f"{DATASET_PREFIX}dataset4.csv")
-    df_train = df[:-1]
-    df_test = df[-1:]
+    df_train = df[:-PERIODS]
+    df_test = df[-PERIODS:]
 
-    yaml_i = deepcopy(TEMPLATE_YAML)
-    yaml_i["spec"]["model"] = model
-    yaml_i["spec"]["historical_data"].pop("url")
-    yaml_i["spec"]["historical_data"]["data"] = df_train
-    yaml_i["spec"]["test_data"]["data"] = df_test
-    yaml_i["spec"]["target_column"] = "Y"
-    yaml_i["spec"]["datetime_column"]["name"] = DATETIME_COL
-    yaml_i["spec"]["horizon"] = 5
-    run(yaml_i, backend="operator.local", debug=False)
-    subprocess.run(f"ls -a {output_data_path}", shell=True)
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        output_data_path = f"{tmpdirname}/results"
+        yaml_i = deepcopy(TEMPLATE_YAML)
+        yaml_i["spec"]["model"] = model
+        yaml_i["spec"]["historical_data"].pop("url")
+        yaml_i["spec"]["historical_data"]["data"] = df_train
+        yaml_i["spec"]["test_data"] = {"data": df_test}
+        yaml_i["spec"]["target_column"] = "Y"
+        yaml_i["spec"]["datetime_column"]["name"] = DATETIME_COL
+        yaml_i["spec"]["horizon"] = PERIODS
+        yaml_i["spec"]["output_directory"]["url"] = output_data_path
+        operator_config = ForecastOperatorConfig.from_dict(yaml_i)
+        forecast_operate(operator_config)
+        check_output_for_errors(output_data_path)
+        test_metrics = pd.read_csv(f"{output_data_path}/metrics.csv")
+        print(test_metrics)
 
+def check_output_for_errors(output_data_path):
+    # try:
+    # List files in the directory
+    result = subprocess.run(f"ls -a {output_data_path}", shell=True, check=True, text=True, capture_output=True)
+    files = result.stdout.splitlines()
+
+    # Check if errors.json is in the directory
+    if "errors.json" in files:
+        errors_file_path = os.path.join(output_data_path, "errors.json")
+        
+        # Read the errors.json file
+        with open(errors_file_path, "r") as f:
+            errors_content = json.load(f)
+        
+        # Extract and raise the error message
+        # error_message = errors_content.get("message", "An error occurred.")
+        raise Exception(errors_content)
+
+    print("No errors.json file found. Directory is clear.")
+    
+    # except subprocess.CalledProcessError as e:
+    #     print(f"Error listing files in directory: {e}")
+    # except FileNotFoundError:
+    #     print("The directory does not exist.")
+    # except json.JSONDecodeError:
+    #     print("errors.json is not a valid JSON file.")
+    # except Exception as e:
+    #     print(f"Raised error: {e}")
 
 def run_operator(
     historical_data_path,
@@ -229,9 +270,9 @@ def run_operator(
         sleep(0.1)
         subprocess.run(f"ls -a {output_data_path}", shell=True)
 
-        test_metrics = pd.read_csv(f"{tmpdirname}/results/test_metrics.csv")
+        test_metrics = pd.read_csv(f"{tmpdirname}/results/metrics.csv")
         print(test_metrics)
-        train_metrics = pd.read_csv(f"{tmpdirname}/results/metrics.csv")
+        train_metrics = pd.read_csv(f"{tmpdirname}/results/train_metrics.csv")
         print(train_metrics)