go-weatherstation/db/init/002_rain_monitoring_views.sql

-- Rain-model monitoring views (feature/prediction/calibration drift + pipeline freshness).
-- Safe to re-run.

CREATE OR REPLACE VIEW rain_feature_drift_daily AS
WITH ws90_daily AS (
  SELECT
    time_bucket(INTERVAL '1 day', ts) AS day,
    site,
    count(*) AS ws90_rows,
    avg(temperature_c) AS temp_mean,
    avg(humidity) AS humidity_mean,
    avg(wind_avg_m_s) AS wind_avg_mean,
    avg(wind_max_m_s) AS wind_gust_mean
  FROM observations_ws90
  GROUP BY 1,2
),
baro_daily AS (
  SELECT
    time_bucket(INTERVAL '1 day', ts) AS day,
    site,
    count(*) AS baro_rows,
    avg(pressure_hpa) AS pressure_mean
  FROM observations_baro
  GROUP BY 1,2
),
joined AS (
  SELECT
    w.day,
    w.site,
    w.ws90_rows,
    b.baro_rows,
    w.temp_mean,
    w.humidity_mean,
    w.wind_avg_mean,
    w.wind_gust_mean,
    b.pressure_mean
  FROM ws90_daily w
  LEFT JOIN baro_daily b
    ON b.day = w.day
   AND b.site = w.site
),
baseline AS (
  SELECT
    j1.day,
    j1.site,
    j1.ws90_rows,
    j1.baro_rows,
    j1.temp_mean,
    j1.humidity_mean,
    j1.wind_avg_mean,
    j1.wind_gust_mean,
    j1.pressure_mean,
    avg(j2.temp_mean) AS temp_mean_30d,
    stddev_samp(j2.temp_mean) AS temp_std_30d,
    avg(j2.humidity_mean) AS humidity_mean_30d,
    stddev_samp(j2.humidity_mean) AS humidity_std_30d,
    avg(j2.wind_avg_mean) AS wind_avg_mean_30d,
    stddev_samp(j2.wind_avg_mean) AS wind_avg_std_30d,
    avg(j2.wind_gust_mean) AS wind_gust_mean_30d,
    stddev_samp(j2.wind_gust_mean) AS wind_gust_std_30d,
    avg(j2.pressure_mean) AS pressure_mean_30d,
    stddev_samp(j2.pressure_mean) AS pressure_std_30d
  FROM joined j1
  LEFT JOIN joined j2
    ON j2.site = j1.site
   AND j2.day < j1.day
   AND j2.day >= j1.day - INTERVAL '30 days'
  GROUP BY
    j1.day,
    j1.site,
    j1.ws90_rows,
    j1.baro_rows,
    j1.temp_mean,
    j1.humidity_mean,
    j1.wind_avg_mean,
    j1.wind_gust_mean,
    j1.pressure_mean
)
SELECT
  day,
  site,
  ws90_rows,
  baro_rows,
  temp_mean,
  humidity_mean,
  wind_avg_mean,
  wind_gust_mean,
  pressure_mean,
  temp_mean_30d,
  humidity_mean_30d,
  wind_avg_mean_30d,
  wind_gust_mean_30d,
  pressure_mean_30d,
  CASE WHEN temp_std_30d > 0 THEN (temp_mean - temp_mean_30d) / temp_std_30d END AS temp_zscore_30d,
  CASE WHEN humidity_std_30d > 0 THEN (humidity_mean - humidity_mean_30d) / humidity_std_30d END AS humidity_zscore_30d,
  CASE WHEN wind_avg_std_30d > 0 THEN (wind_avg_mean - wind_avg_mean_30d) / wind_avg_std_30d END AS wind_avg_zscore_30d,
  CASE WHEN wind_gust_std_30d > 0 THEN (wind_gust_mean - wind_gust_mean_30d) / wind_gust_std_30d END AS wind_gust_zscore_30d,
  CASE WHEN pressure_std_30d > 0 THEN (pressure_mean - pressure_mean_30d) / pressure_std_30d END AS pressure_zscore_30d
FROM baseline;


CREATE OR REPLACE VIEW rain_prediction_drift_daily AS
SELECT
  time_bucket(INTERVAL '1 day', ts) AS day,
  site,
  model_name,
  model_version,
  count(*) AS prediction_rows,
  avg(probability) AS probability_mean,
  stddev_samp(probability) AS probability_stddev,
  avg(CASE WHEN predict_rain THEN 1.0 ELSE 0.0 END) AS predicted_positive_rate,
  avg(threshold) AS threshold_mean
FROM predictions_rain_1h
GROUP BY 1,2,3,4;


CREATE OR REPLACE VIEW rain_calibration_drift_daily AS
SELECT
  time_bucket(INTERVAL '1 day', ts) AS day,
  site,
  model_name,
  model_version,
  count(*) FILTER (WHERE rain_next_1h_actual IS NOT NULL) AS evaluated_rows,
  avg(probability) FILTER (WHERE rain_next_1h_actual IS NOT NULL) AS mean_probability,
  avg(CASE WHEN rain_next_1h_actual THEN 1.0 ELSE 0.0 END) FILTER (WHERE rain_next_1h_actual IS NOT NULL) AS observed_positive_rate,
  avg(power(probability - CASE WHEN rain_next_1h_actual THEN 1.0 ELSE 0.0 END, 2.0))
    FILTER (WHERE rain_next_1h_actual IS NOT NULL) AS brier_score,
  avg(
    CASE
      WHEN rain_next_1h_actual IS NULL THEN NULL
      WHEN predict_rain = rain_next_1h_actual THEN 1.0
      ELSE 0.0
    END
  ) AS decision_accuracy
FROM predictions_rain_1h
GROUP BY 1,2,3,4;


CREATE OR REPLACE VIEW rain_pipeline_health AS
WITH sites AS (
  SELECT DISTINCT site FROM observations_ws90
  UNION
  SELECT DISTINCT site FROM observations_baro
  UNION
  SELECT DISTINCT site FROM forecast_openmeteo_hourly
  UNION
  SELECT DISTINCT site FROM predictions_rain_1h
)
SELECT
  s.site,
  (SELECT max(ts) FROM observations_ws90 w WHERE w.site = s.site) AS ws90_latest_ts,
  (SELECT max(ts) FROM observations_baro b WHERE b.site = s.site) AS baro_latest_ts,
  (SELECT max(ts) FROM forecast_openmeteo_hourly f WHERE f.site = s.site) AS forecast_latest_ts,
  (SELECT max(generated_at) FROM predictions_rain_1h p WHERE p.site = s.site) AS prediction_generated_latest_ts,
  (SELECT max(evaluated_at) FROM predictions_rain_1h p WHERE p.site = s.site) AS prediction_evaluated_latest_ts
FROM sites s;