new versions

airquality/util.py

@@ -340,13 +340,71 @@ def check_file_path(file_path):
         print(f"File successfully found at the path: {file_path}")
 
 def backfill_predictions_for_monitoring(weather_fg, air_quality_df, monitor_fg, model):
-    features_df = weather_fg.read()
-    features_df = features_df.sort_values(by=['date'], ascending=True)
-    features_df = features_df.tail(10)
-    features_df['predicted_pm25'] = model.predict(features_df[['temperature_2m_mean', 'precipitation_sum', 'wind_speed_10m_max', 'wind_direction_10m_dominant']])
-    df = pd.merge(features_df, air_quality_df[['date','pm25','street','country']], on="date")
-    df['days_before_forecast_day'] = 1
-    hindcast_df = df
-    df = df.drop('pm25', axis=1)
-    monitor_fg.insert(df, write_options={"wait_for_job": True})
-    return hindcast_df
+    # 1. Load some past weather
+    features_df = weather_fg.read().sort_values("date")
+    features_df = features_df.tail(10).reset_index(drop=True)
+
+    # 2. Load past PM25 (for building lag features)
+    air_quality_df = air_quality_df.sort_values("date").reset_index(drop=True)
+    pm25_series = air_quality_df["pm25"].values
+
+    # Need last 3 pm25 values to start autoregressive cycle
+    pm25_history = list(pm25_series[-3:])
+
+    predictions = []
+
+    for i in range(len(features_df)):
+        # Weather for this day
+        w = features_df.iloc[i]
+
+        # Build lag features
+        lag1 = pm25_history[-1]
+        lag2 = pm25_history[-2]
+        lag3 = pm25_history[-3]
+
+        roll_mean = np.mean(pm25_history[-3:])
+        roll_std = np.std(pm25_history[-3:])
+
+        X = pd.DataFrame({
+            "temperature_2m_mean": [w["temperature_2m_mean"]],
+            "precipitation_sum": [w["precipitation_sum"]],
+            "wind_speed_10m_max": [w["wind_speed_10m_max"]],
+            "wind_direction_10m_dominant": [w["wind_direction_10m_dominant"]],
+            "pm25_lag1": [lag1],
+            "pm25_lag2": [lag2],
+            "pm25_lag3": [lag3],
+            "pm25_roll3_mean": [roll_mean],
+            "pm25_roll3_std": [roll_std]
+        })
+
+        pred = float(model.predict(X)[0])
+        predictions.append(pred)
+
+        pm25_history.append(pred)
+
+    # Combine predictions with weather for monitoring insert
+    out = features_df.copy()
+    out["predicted_pm25"] = predictions
+    out["days_before_forecast_day"] = 1
+
+    # Pick metadata from AQ DF
+    out["street"] = air_quality_df.iloc[-1]["street"]
+    out["city"] = air_quality_df.iloc[-1]["city"]
+    out["country"] = air_quality_df.iloc[-1]["country"]
+
+    # Fix types
+    out["predicted_pm25"] = out["predicted_pm25"].astype(float)
+    out["days_before_forecast_day"] = out["days_before_forecast_day"].astype("int64")
+
+    # Insert monitoring data
+    monitor_fg.insert(out, write_options={"wait_for_job": True})
+
+    # Merge with pm25 ground truth for return
+    merged = pd.merge(
+        out,
+        air_quality_df[["date", "pm25"]],
+        on="date",
+        how="left"
+    )
+
+    return merged

app.py

@@ -19,7 +19,7 @@ def load_resources():
     # Load Feature View
     fv = fs.get_feature_view(
         name="air_quality_fv",
-        version=1
+        version=2
     )
     fv.init_batch_scoring(1)
 
@@ -28,7 +28,7 @@ def load_resources():
 
     # Load Model from Registry
     mr = project.get_model_registry()
-    model_obj = mr.get_model("air_quality_xgboost_model", version=1)
+    model_obj = mr.get_model("air_quality_xgboost_model", version=2)
     model_dir = model_obj.download()
 
     model = XGBRegressor()