Data Pipeline

The engine requires six distinct data streams feeding a time-series store. Each stream serves a different role in the calibration and product pipeline.

Stream Architecture

Six Data Streams

Stream 1 — NWP Ensemble Ingest

Implemented endpoints:

Implementation — src/data_ingestion/open_meteo.py:

Unit conversions applied on ingestion:

Stream 2 — Observed Ground Truth (IEM + ERA5)

IEM ASOS Daily API — src/data_ingestion/iem_observed.py:

ERA5 Reanalysis — src/data_ingestion/open_meteo_reanalysis.py:

ERA5 is a global atmospheric reanalysis that assimilates millions of observations. Its irradiance values are independent of GFS/ECMWF forecasts, making it a valid ground truth source for backtest evaluation.

Stream 3 — Satellite Ground Truth (SatSure Sparta)

Satellite data provides independent ground truth that NWP models don't see. It feeds two functions: improving bias correction (what actually happened vs what was forecast) and providing verification data for warranty settlement.

Stream 4 — Market Price Feed (ForecastEx)

ForecastEx prices represent market consensus probabilities. Comparing the engine's calibrated probability against market price reveals the edge — and provides an independent validation of calibration quality.

Stream 5 — Settlement / Verification

This is the ground truth stream for ForecastEx settlement. After each observation period closes, verified outcomes are compared against the engine's predictions to calculate Brier scores and update the reliability diagram.

Stream 6 — Engine Output

Daily Automation Pipeline

scripts/daily_pipeline.sh — triggered via launchd, 8 steps:

Step 1: Score Yesterday's Predictions

Step 2: Generate Tomorrow's Temperature Signals

Step 3: D+0 METAR Blending

Steps 4–6: Non-Temperature Variable Signals

run_variable_signals.py is the generic signal generator for non-temperature variables. It:

1. Loads the variable via get_variable(name) from the registry
2. Fetches ensemble forecasts via var.ingest_ensemble()
3. Applies bias correction via var.bias_correct_ensemble()
4. Computes probabilities via ensemble member counting
5. Optionally generates risk scores via RiskScorer
6. Saves signals to data/signals/{variable}/signals_YYYY-MM-DD.csv

Step 7: Divergence Monitor

Output actions: HOLD (normal), FLAG (significant shift >= threshold), REVERSE (direction flipped YES↔NO).

Step 8: Dashboard Export

Export outputs (written to dashboard/public/data/):

Midday Automation Pipeline

scripts/midday_pipeline.sh — triggered via LaunchAgent at noon local time:

Generic Backtest Pipeline

scripts/run_phase_generic.py — unified Phase 0/1 orchestrator for all variables:

python scripts/run_phase_generic.py --variable rainfall --phase 0+1
python scripts/run_phase_generic.py --variable wind_speed --phase 0+1
python scripts/run_phase_generic.py --variable irradiance --phase 0+1
python scripts/run_phase_generic.py --variable temperature_high --phase 0+1

Phase 0 (Baseline):

1. var.ingest_historical() — download historical forecasts from Open-Meteo
2. var.observe() — download observations from IEM or ERA5
3. Merge on [station, date]
4. Compute MAE, bias, RMSE per city and overall
5. Save paired data to data/backtest/{variable}/forecast_vs_observed.csv

Phase 1 (Calibration):

1. Split train/test using TRAIN_END_DATE / TEST_START_DATE
2. Train bias parameters (additive or multiplicative based on var.config.bias_method)
3. Generate backtest probabilities via generate_backtest_probabilities_generic()
4. Train isotonic calibration on raw probabilities
5. Compute Brier scores (raw and calibrated)
6. Save artifacts: bias_params.json, calibration_model.pkl, phase1_summary.json

Storage: TimescaleDB

The engine uses TimescaleDB — a PostgreSQL extension with native time-series support.

Why TimescaleDB:

• Standard SQL (no proprietary query language)
• Automatic time-partitioning (hypertables)
• Efficient compression for historical data
• Native continuous aggregates for roll-up calculations
• PostgreSQL ecosystem (PostGIS for spatial queries, standard tooling)

Schema design stores data in time-series hypertables per stream type, with location and variable as indexed dimensions. Each prediction record includes full provenance metadata for audit trail compliance.

Data Directories

Deployment

Initial deployment runs on a single VPS with Docker containers:

• Ingestion service container
• TimescaleDB container
• Engine pipeline container (scheduled via launchd/cron)
• FastAPI backend container (Risk API + probability endpoints)
• Nginx reverse proxy

This is sufficient for initial scale — NWP data volume is modest (megabytes per ingest cycle, not gigabytes). Kubernetes migration is deferred until multi-project scale justifies the operational complexity.