hindcast

airquality/util.py

@@ -14,6 +14,7 @@ from retry_requests import retry
 import hopsworks
 import hsfs
 from pathlib import Path
+import numpy as np
 
 def get_historical_weather(city, start_date,  end_date, latitude, longitude):
     # latitude, longitude = get_city_coordinates(city)

app.py

@@ -142,21 +142,35 @@ def generate_forecast_v2(days):
 def generate_hindcast_v1(days):
     model, fv, weather_fg, project = resources_v1
 
+    start_date = datetime.utcnow().date() - timedelta(days=days)
+    end_date = datetime.utcnow().date()
+
+    # Load training data from Feature View
     features_df, labels_df = fv.training_data(
-        start_time=datetime.utcnow().date() - timedelta(days=days),
-        end_time=datetime.utcnow().date(),
+        start_time=start_date,
+        end_time=end_date,
         statistics_config=False
     )
 
     features_df["date"] = pd.to_datetime(features_df["date"]).dt.date
 
-    X = features_df[["temperature_2m_mean", "precipitation_sum",
-                     "wind_speed_10m_max", "wind_direction_10m_dominant"]]
+    # The model for v1 ONLY uses weather features
+    X = features_df[[
+        "temperature_2m_mean", 
+        "precipitation_sum",
+        "wind_speed_10m_max", 
+        "wind_direction_10m_dominant"
+    ]]
 
     features_df["predicted_pm25"] = model.predict(X)
 
     tmp_path = tempfile.NamedTemporaryFile(suffix=".png", delete=False).name
-    plot_air_quality_forecast("linkoping", "hamngatan-10", features_df, tmp_path, hindcast=True)
+    plot_air_quality_forecast(
+        "linkoping", "hamngatan-10",
+        features_df,
+        tmp_path,
+        hindcast=True
+    )
 
     return tmp_path
 
@@ -164,21 +178,60 @@ def generate_hindcast_v1(days):
 def generate_hindcast_v2(days):
     model, fv, weather_fg, project = resources_v2
 
-    features_df, labels_df = fv.training_data(
-        start_time=datetime.utcnow().date() - timedelta(days=days),
-        end_time=datetime.utcnow().date(),
-        statistics_config=False
-    )
-
-    features_df["date"] = pd.to_datetime(features_df["date"]).dt.date
+    # Time window
+    start_date = datetime.utcnow().date() - timedelta(days=days + 3)
+    end_date = datetime.utcnow().date()
 
-    # Full lag-feature dataset already inside feature view v2
-    X = features_df.drop(columns=["date"])
+    # Load weather history
+    weather_df = weather_fg.read()
+    weather_df["date"] = pd.to_datetime(weather_df["date"]).dt.date
+    weather_df = weather_df[(weather_df["date"] >= start_date) &
+                            (weather_df["date"] <= end_date)]
+    weather_df = weather_df.sort_values("date")
 
-    features_df["predicted_pm25"] = model.predict(X)
+    # Load PM2.5 history (actual values, not predictions)
+    aq_fg = project.get_feature_store().get_feature_group("air_quality", version=1)
+    aq_df = aq_fg.read()
+    aq_df["date"] = pd.to_datetime(aq_df["date"]).dt.date
+    aq_df = aq_df[(aq_df["date"] >= start_date) &
+                  (aq_df["date"] <= end_date)]
+    aq_df = aq_df.sort_values("date")
+
+    # Merge actual historical PM2.5 + weather
+    df = pd.merge(weather_df, aq_df, on="date")
+
+    # Build lag features
+    df["pm25_lag1"] = df["pm25"].shift(1)
+    df["pm25_lag2"] = df["pm25"].shift(2)
+    df["pm25_lag3"] = df["pm25"].shift(3)
+    df["pm25_roll3_mean"] = df["pm25"].rolling(3).mean()
+    df["pm25_roll3_std"] = df["pm25"].rolling(3).std()
+
+    # Only keep the last N days (where all lags exist)
+    df = df.dropna().tail(days)
+
+    # Features for model v2
+    X = df[[
+        "temperature_2m_mean",
+        "precipitation_sum",
+        "wind_speed_10m_max",
+        "wind_direction_10m_dominant",
+        "pm25_lag1",
+        "pm25_lag2",
+        "pm25_lag3",
+        "pm25_roll3_mean",
+        "pm25_roll3_std",
+    ]]
+
+    df["predicted_pm25"] = model.predict(X)
 
     tmp_path = tempfile.NamedTemporaryFile(suffix=".png", delete=False).name
-    plot_air_quality_forecast("linkoping", "hamngatan-10", features_df, tmp_path, hindcast=True)
+    plot_air_quality_forecast(
+        "linkoping", "hamngatan-10",
+        df,
+        tmp_path,
+        hindcast=True
+    )
 
     return tmp_path