PR review

Author: htahir1

floracast/README.md

@@ -97,7 +97,7 @@ python run.py --config configs/inference.yaml --pipeline inference
 ```
 
 3. **View results**:
-- Check `outputs/forecast_inference.csv` for predictions
+- Check the predictions artifact for predictions
 - Use ZenML dashboard to view artifacts and metrics
 
 ## ⚙️ Configuration Files
@@ -118,35 +118,33 @@ Edit the appropriate config file to customize:
 - **Evaluation**: Forecasting horizon, metrics
 - **Output**: File paths and formats
 
-
-
 ```
 floracast/
 ├── README.md
 ├── requirements.txt
 ├── .env.example
 ├── configs/
-│   ├── training.yaml       # Training pipeline config
-│   ├── inference.yaml      # Inference pipeline config  
+│   ├── training.yaml            # Training pipeline config
+│   └── inference.yaml           # Inference pipeline config
 ├── data/
-│   └── ecommerce_daily.csv # Generated sample data
+│   └── ecommerce_daily.csv      # Example input data
 ├── pipelines/
-│   ├── train_forecast_pipeline.py
-│   └── batch_inference_pipeline.py
+│   ├── train_forecast_pipeline.py      # Training pipeline definition
+│   └── batch_inference_pipeline.py     # Batch inference pipeline definition
 ├── steps/
-│   ├── ingest.py          # Data loading
-│   ├── preprocess.py      # Time series preprocessing  
-│   ├── train.py           # Model training
-│   ├── evaluate.py        # Model evaluation
-│   ├── promote.py         # Model registration
-│   ├── batch_infer.py     # Batch predictions
-│   └── load_model.py      # Model loading utilities
+│   ├── ingest.py                # Data ingestion step
+│   ├── preprocess.py            # Preprocessing step (train/val split, scaling)
+│   ├── train.py                 # Model training step
+│   ├── evaluate.py              # Model evaluation step
+│   ├── promote.py               # Model registration/promotion step
+│   ├── batch_infer.py           # Batch inference step
+│   └── load_model.py            # Model loading utilities
 ├── materializers/
-│   └── tft_materializer.py         # Custom TFTModel materializer
-|   └── timeseries_materializer.py  # Custom TimeSeries materializer
+│   ├── tft_materializer.py              # Custom TFTModel materializer
+│   └── timeseries_materializer.py       # Custom TimeSeries materializer
 ├── utils/
-│   └── metrics.py         # Forecasting metrics
-└── run.py                 # Main entry point
+│   └── metrics.py               # Forecasting metrics (e.g., SMAPE)
+└── run.py                      # CLI entry point for running pipelines
 ```
 
 ### Key Components
@@ -199,7 +197,7 @@ Read more:
 
 - **Set up an MLOps stack on Azure**: [ZenML Azure guide](https://docs.zenml.io/stacks/popular-stacks/azure-guide)
 - **Kubernetes Orchestrator (AKS)**: [Docs](https://docs.zenml.io/stacks/stack-components/orchestrators/kubernetes)
-- **Azure Blob Artifact Store**: [Docs](https://docs.zenml.io/stacks/stack-components/artifact-stores/azuree)
+- **Azure Blob Artifact Store**: [Docs](https://docs.zenml.io/stacks/stack-components/artifact-stores/azure)
 - **Azure Container Registry**: [Docs](https://docs.zenml.io/stacks/stack-components/container-registries/azure)
 - **AzureML Step Operator**: [Docs](https://docs.zenml.io/stacks/stack-components/step-operators/azureml)
 - **Terraform stack recipe for Azure**: [Hashicorp Registry](https://registry.terraform.io/modules/zenml-io/zenml-stack/azure/latest)

floracast/configs/inference.yaml

@@ -18,12 +18,9 @@ steps:
       datetime_col: "ds"
       target_col: "y"
 
-  preprocess_for_inference:
+  batch_inference_predict:
     parameters:
       datetime_col: "ds"
       target_col: "y"
       freq: "D"
-  
-  batch_inference_predict:
-    parameters:
       horizon: 14

floracast/configs/training.yaml

@@ -32,7 +32,7 @@ steps:
       datetime_col: "ds"
       target_col: "y"
 
-  preprocess_for_training:
+  preprocess_data:
     parameters:
       datetime_col: "ds"
       target_col: "y"
@@ -41,7 +41,6 @@ steps:
 
   train_model:
     parameters:
-      model_name: "TFTModel"
       input_chunk_length: 90
       output_chunk_length: 14
       hidden_size: 256
@@ -57,5 +56,5 @@ steps:
 
   evaluate:
     parameters:
-      horizon: 7
+      horizon: 14
       metric: "smape"

floracast/materializers/tft_materializer.py

@@ -168,12 +168,12 @@ def _load_with_pytorch_state(self) -> Any:
             )
 
             dates = pd.date_range("2020-01-01", periods=dummy_length, freq="D")
-            values = np.random.randn(dummy_length)
+            values = np.random.randn(dummy_length).astype(np.float32)
             dummy_series = TimeSeries.from_dataframe(
                 pd.DataFrame({"ds": dates, "y": values}),
                 time_col="ds",
                 value_cols="y",
-            )
+            ).astype(np.float32)
 
             # Partially fit to create the internal model structure
             temp_model.fit(dummy_series, epochs=1, verbose=False)

floracast/pipelines/batch_inference_pipeline.py

@@ -6,7 +6,6 @@
 from zenml.logger import get_logger
 
 from steps.ingest import ingest_data
-from steps.preprocess import preprocess_for_inference
 from steps.batch_infer import batch_inference_predict
 
 logger = get_logger(__name__)
@@ -22,10 +21,7 @@ def batch_inference_pipeline() -> None:
     # Step 1: Ingest data (simulate real-time data sources)
     raw_data = ingest_data(infer=True)
 
-    # Step 2: Preprocess data (use full series for inference context)
-    inference_series = preprocess_for_inference(df=raw_data)
-
-    # Step 3: Generate predictions using model from MCP
-    batch_inference_predict(series=inference_series)
+    # Step 2: Generate predictions using model from MCP (with scaling handled internally)
+    batch_inference_predict(df=raw_data)
 
     logger.info("Batch inference completed. Returning predictions DataFrame.")

floracast/pipelines/train_forecast_pipeline.py

@@ -6,7 +6,7 @@
 from zenml.logger import get_logger
 
 from steps.ingest import ingest_data
-from steps.preprocess import preprocess_for_training
+from steps.preprocess import preprocess_data
 from steps.train import train_model
 from steps.evaluate import evaluate
 from steps.promote import promote_model
@@ -24,7 +24,7 @@ def train_forecast_pipeline() -> None:
     raw_data = ingest_data()
 
     # Step 2: Preprocess data into Darts TimeSeries with train/val split
-    train_series, val_series = preprocess_for_training(df=raw_data)
+    train_series, val_series, _ = preprocess_data(df=raw_data)
 
     # Step 3: Train the forecasting model
     trained_model = train_model(train_series=train_series)

floracast/requirements.txt

@@ -19,5 +19,4 @@ azure-identity>=1.16.0,<2.0.0
 azure-storage-blob>=12.20.0,<13.0.0
 
 # Utilities
-matplotlib>=3.7.0,<4.0.0
-seaborn>=0.12.0,<0.14.0
+matplotlib>=3.7.0,<4.0.0

floracast/run.py

@@ -3,7 +3,6 @@
 """
 
 import click
-from datetime import datetime
 from pathlib import Path
 from pipelines import batch_inference_pipeline, train_forecast_pipeline
 from zenml.logger import get_logger

floracast/steps/__init__.py

@@ -1,16 +1,15 @@
 """ZenML pipeline steps for FloraCast."""
 
 from .ingest import ingest_data
-from .preprocess import preprocess_for_training, preprocess_for_inference
+from .preprocess import preprocess_data
 from .train import train_model
 from .evaluate import evaluate
 from .promote import promote_model
 from .batch_infer import batch_inference_predict
 
 __all__ = [
     "ingest_data",
-    "preprocess_for_training",
-    "preprocess_for_inference", 
+    "preprocess_data",
     "train_model",
     "evaluate",
     "promote_model",

floracast/steps/batch_infer.py

@@ -4,24 +4,32 @@
 
 from typing import Annotated
 import pandas as pd
+import numpy as np
 from darts import TimeSeries
 from zenml import step, get_step_context, log_metadata
 from zenml.logger import get_logger
 from zenml.client import Client
+from utils.prediction import iterative_predict
 
 logger = get_logger(__name__)
 
 
 @step
 def batch_inference_predict(
-    series: TimeSeries,
+    df: pd.DataFrame,
+    datetime_col: str = "ds",
+    target_col: str = "y",
+    freq: str = "D",
     horizon: int = 14,
 ) -> Annotated[pd.DataFrame, "predictions"]:
     """
     Perform batch inference using the trained model from Model Control Plane.
 
     Args:
-        series: Time series data for forecasting
+        df: Raw DataFrame with datetime and target columns
+        datetime_col: Name of datetime column
+        target_col: Name of target column
+        freq: Frequency string for time series
         horizon: Number of time steps to forecast
 
     Returns:
@@ -30,6 +38,21 @@ def batch_inference_predict(
     logger.info(f"Performing batch inference with horizon: {horizon}")
 
     try:
+        # Convert DataFrame to TimeSeries
+        logger.info("Converting DataFrame to TimeSeries")
+        series = TimeSeries.from_dataframe(
+            df, time_col=datetime_col, value_cols=target_col, freq=freq
+        )
+
+        # Cast to float32 for consistency with training data
+        logger.info("Converting TimeSeries to float32 for consistency")
+        series = series.astype(np.float32)
+
+        logger.info(f"Created TimeSeries with {len(series)} points")
+        logger.info(
+            f"Series range: {series.start_time()} to {series.end_time()}"
+        )
+
         # Get the model from Model Control Plane
         context = get_step_context()
         if not context.model:
@@ -72,41 +95,48 @@ def batch_inference_predict(
             f"Loaded model from Model Control Plane: {type(trained_model).__name__}"
         )
 
-        # Generate predictions using improved multi-step approach (same as evaluation)
-        logger.info(
-            f"Using iterative multi-step prediction for horizon={horizon}"
-        )
-
-        # Use multiple prediction steps for better long-term accuracy
-        predictions_list = []
-        context_series = series
-
-        # Predict in chunks of output_chunk_length (14 days)
-        remaining_steps = horizon
-        while remaining_steps > 0:
-            chunk_size = min(
-                14, remaining_steps
-            )  # Model's output_chunk_length
-            chunk_pred = trained_model.predict(
-                n=chunk_size, series=context_series
+        # Load the fitted scaler artifact
+        fitted_scaler = None
+        try:
+            scaler_artifact = context.model.get_artifact("fitted_scaler")
+            if scaler_artifact is None:
+                raise ValueError(
+                    "fitted_scaler artifact not found in model version"
+                )
+            fitted_scaler = scaler_artifact.load()
+            logger.info("Loaded fitted scaler artifact from model version")
+
+            # Apply scaling to the input series
+            logger.info("Applying scaling to input series for inference")
+            series = fitted_scaler.transform(series)
+            logger.info("Scaling applied successfully")
+        except Exception as scaler_error:
+            logger.error(f"Failed to load or apply scaler: {scaler_error}")
+            logger.warning(
+                "Proceeding without scaling - predictions may be incorrect!"
             )
-            predictions_list.append(chunk_pred)
+            # Continue without scaling for backward compatibility
 
-            # Extend context with the prediction for next iteration
-            context_series = context_series.concatenate(chunk_pred)
-            remaining_steps -= chunk_size
+        # Generate predictions using improved multi-step approach
+        predictions = iterative_predict(trained_model, series, horizon)
 
-        # Combine all predictions
-        if len(predictions_list) == 1:
-            predictions = predictions_list[0]
+        # Inverse transform predictions back to original scale
+        if fitted_scaler is not None:
+            try:
+                logger.info(
+                    "Inverse transforming predictions back to original scale"
+                )
+                predictions = fitted_scaler.inverse_transform(predictions)
+                logger.info("Inverse transformation applied successfully")
+            except Exception as inverse_error:
+                logger.error(
+                    f"Failed to inverse transform predictions: {inverse_error}"
+                )
+                logger.warning("Predictions remain in scaled format!")
         else:
-            predictions = predictions_list[0]
-            for pred_chunk in predictions_list[1:]:
-                predictions = predictions.concatenate(pred_chunk)
-
-        logger.info(
-            f"Generated {len(predictions)} predictions using multi-step approach"
-        )
+            logger.warning(
+                "No scaler available - predictions remain in original format"
+            )
 
         # Convert to DataFrame
         pred_df = predictions.pd_dataframe().reset_index()