feat: implement flood calibration and prediction tasks

- Add flood calibration task to re-learn link models from historical river levels.
- Introduce flood prediction task to evaluate calibrated links and send alerts based on upstream readings.
- Create utility functions for flood prediction, including model calibration and exceedance event detection.
- Implement river station utilities for parsing sensor data and calculating distances.
- Add tests for flood calibration, prediction, river levels, and utility functions to ensure functionality and reliability.

Author: nicomontanari

.env.example

@@ -9,3 +9,5 @@ DATABASE_PORT=
 DATABASE_USERNAME=
 DATABASE_PASSWORD=
 ALERT_ZONE=D1
+# Hysteresis deadband in metres around river thresholds (anti-flapping). Default 0.05.
+RIVER_THRESHOLD_MARGIN=0.05

README.md

@@ -23,9 +23,14 @@ The system interfaces with various official and unofficial sources to provide a
 
 4. **River levels (Allerta Meteo hydrometric stations)**
    - **Feature:** Monitoring of water levels at configured hydrometric stations.
-   - **Details:** Periodically polls the Allerta Meteo time-series endpoint for each registered station and compares the latest reading against operator-defined thresholds (`soglia1`, `soglia2`, `soglia3`). Station metadata and thresholds live in the `rivers` table; each check is appended to `river_levels`.
+   - **Details:** Periodically polls the Allerta Meteo time-series endpoint for each registered station and compares the latest reading against operator-defined thresholds (`soglia1`, `soglia2`, `soglia3`). Station metadata and thresholds live in the `rivers` table; each new reading is appended to `river_levels` (de-duplicated by measurement time). A hysteresis deadband (`RIVER_THRESHOLD_MARGIN`, default 0.05 m) avoids alert flapping when a level hovers around a threshold.
    - **Notification:** Sends a Telegram message only on crossing events — when a reading rises above or falls below a threshold relative to the previous check.
 
+5. **Flood prediction (empirical precursor model)**
+   - **Feature:** Warns that a downstream point of interest is likely to exceed a threshold, with an estimated lead time, based on what upstream gauges are doing now.
+   - **Details:** For each configured `river_link` (an upstream gauge → a downstream point), a daily calibration mines the accumulated `river_levels` history for past downstream exceedances and learns, from those events only, the upstream `precursor_level` and the typical `lead_time`. A link stays inactive until it has enough historical events. Online, when the upstream gauge reaches the learned level while rising, a single prediction is emitted and recorded in `link_predictions` for later scoring. No AI — event detection plus robust statistics.
+   - **Bootstrap:** Since the live API only retains ~2.4 days, historical data is backfilled from the free ARPAE-SIMC open archive (see below). Without a backfill the model simply stays dormant until enough live events accrue.
+
 ## Scheduled Tasks (Crons)
 
 The application relies on scheduled tasks (crons) to automate the weather monitoring flow:
@@ -38,6 +43,10 @@ The application relies on scheduled tasks (crons) to automate the weather monito
   - Verifies and sends the Estofex map for the following day, following the same conditional logic based on ongoing alerts.
 - **River Levels Check**
   - Every 5 minutes, for each row in the `rivers` table, fetches the latest hydrometric reading and appends it to `river_levels`. Sends a Telegram message only when the reading crosses one of the configured thresholds since the previous check.
+- **Flood Prediction Check**
+  - Every 5 minutes, evaluates each calibrated `river_link`: if the upstream gauge has reached its learned precursor level while rising, emits a single de-duplicated flood-arrival prediction.
+- **Flood Calibration**
+  - Daily, re-learns every link's model (lead time + precursor level) from the accumulated `river_levels` history. Also runnable on demand after a backfill.
 
 ## Database bootstrap
 
@@ -55,9 +64,29 @@ Stations are managed via HTTP (port 3000):
 - `POST /rivers` — create; body `{ station_id, river_name, station_name, soglia1?, soglia2?, soglia3? }`
 - `PATCH /rivers/:id` — update any of `river_name`, `station_name`, `soglia1`, `soglia2`, `soglia3`
 - `DELETE /rivers/:id` — delete (cascades to `river_levels`)
-- `POST /river-levels` — trigger an on-demand check (returns `{ checked, crossings, skipped }`)
+- `POST /river-levels` — trigger an on-demand check (returns `{ checked, crossings, skipped, unchanged }`)
+- `GET /rivers/nearest?lat=&lon=&limit=` — discovery: nearest stations to a point, with coordinates and the official soglie the Allerta sensor list reports (suggestions only)
+
+The `station_id` is the Allerta Meteo `idstazione`; threshold values are operator-defined (the `nearest` endpoint surfaces the portal's official soglie as suggestions, but they are not auto-applied).
+
+## Flood prediction (links, calibration, backfill)
+
+Both ends of a link must be registered as `rivers` so their readings accumulate in `river_levels`.
 
-The `station_id` is the Allerta Meteo `idstazione`; threshold values must be looked up manually from the [Allerta Meteo portal](https://allertameteo.regione.emilia-romagna.it/) and stored alongside the station.
+- `GET /river-links` — list links and their learned models
+- `POST /river-links` — create; body `{ upstream_river_id, downstream_river_id, target_threshold? }` (`target_threshold` 1–3, default 1)
+- `DELETE /river-links/:id` — delete a link
+- `POST /flood-calibration` — re-learn all link models from history (returns `{ calibrated, active, skipped }`)
+- `POST /flood-prediction` — evaluate all links now (returns `{ evaluated, predictions, skipped }`)
+- `POST /flood-backfill` — body `{ from: "YYYY-MM", to: "YYYY-MM" }`; backfills `river_levels` from the ARPAE-SIMC open archive (`https://dati-simc.arpae.it/opendata/osservati/meteo/storico/`) for every registered station. Returns `202` immediately and runs in the background (it streams ~20 MB/month). Run it once, then `POST /flood-calibration`.
+
+Example bootstrap for the Molinella stations (Idice S. Antonio + Reno Gandazzolo and their upstream gauges):
+
+```bash
+# register stations, link them, then:
+curl -X POST localhost:3000/flood-backfill -H 'content-type: application/json' -d '{"from":"2020-01","to":"2024-12"}'
+curl -X POST localhost:3000/flood-calibration
+```
 
 ## Configuration and Installation
 

sql/rivers.sql

@@ -18,8 +18,53 @@ CREATE TABLE IF NOT EXISTS river_levels (
     soglia1_above BOOLEAN,
     soglia2_above BOOLEAN,
     soglia3_above BOOLEAN,
-    created_on TIMESTAMPTZ NOT NULL DEFAULT now()
+    created_on TIMESTAMPTZ NOT NULL DEFAULT now(),
+    -- One stored reading per (station, measurement time): the poller runs more often than the
+    -- sensor publishes, so this de-duplicates repeated reads of the same measurement.
+    UNIQUE (river_id, measured_at)
 );
 
+-- For databases created before the UNIQUE clause above existed:
+--   ALTER TABLE river_levels ADD CONSTRAINT river_levels_river_id_measured_at_key
+--       UNIQUE (river_id, measured_at);
+
 CREATE INDEX IF NOT EXISTS river_levels_river_id_created_on_idx
     ON river_levels (river_id, created_on DESC);
+
+CREATE INDEX IF NOT EXISTS river_levels_river_id_measured_at_idx
+    ON river_levels (river_id, measured_at ASC);
+
+-- Flood prediction: an upstream gauge whose readings historically preceded a threshold
+-- exceedance at a downstream point of interest. Both ends are registered rivers so the poller
+-- collects their readings into river_levels. Learned parameters are filled by calibration.
+CREATE TABLE IF NOT EXISTS river_links (
+    id SERIAL PRIMARY KEY,
+    upstream_river_id INTEGER NOT NULL REFERENCES rivers(id) ON DELETE CASCADE,
+    downstream_river_id INTEGER NOT NULL REFERENCES rivers(id) ON DELETE CASCADE,
+    -- Which downstream soglia (1/2/3) defines the exceedance event we predict.
+    target_threshold SMALLINT NOT NULL DEFAULT 1,
+    lead_time_minutes INTEGER,
+    precursor_level NUMERIC,
+    sample_size INTEGER NOT NULL DEFAULT 0,
+    model_json JSONB,
+    last_calibrated_on TIMESTAMPTZ,
+    created_on TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_on TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE (upstream_river_id, downstream_river_id),
+    CHECK (upstream_river_id <> downstream_river_id)
+);
+
+-- One row per prediction emitted, scored later against what actually happened downstream.
+CREATE TABLE IF NOT EXISTS link_predictions (
+    id SERIAL PRIMARY KEY,
+    link_id INTEGER NOT NULL REFERENCES river_links(id) ON DELETE CASCADE,
+    predicted_at TIMESTAMPTZ NOT NULL,
+    predicted_exceedance_at TIMESTAMPTZ NOT NULL,
+    upstream_value NUMERIC NOT NULL,
+    actual_exceedance_at TIMESTAMPTZ,
+    outcome TEXT NOT NULL DEFAULT 'pending',
+    created_on TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS link_predictions_link_id_predicted_at_idx
+    ON link_predictions (link_id, predicted_at DESC);

src/config/config.ts

@@ -12,6 +12,7 @@ export interface Config {
     telegram_token: string
     chat_id: string
     alert_zone: string
+    river_threshold_margin: number
 }
 
 export const getNodeEnv = (): NodeEnv => process.env.NODE_ENV as NodeEnv

src/config/development.ts

@@ -1,4 +1,5 @@
 import { Config } from './config'
+import { getRiverThresholdMargin } from './env'
 import dotenv from 'dotenv'
 
 dotenv.config()
@@ -14,4 +15,5 @@ export const developmentConfig: Config = {
     telegram_token: process.env.TELEGRAM_2_TOKEN as string,
     chat_id: process.env.CHAT_ID as string,
     alert_zone: process.env.ALERT_ZONE || 'D1',
+    river_threshold_margin: getRiverThresholdMargin(),
 }

src/config/env.ts

@@ -0,0 +1,22 @@
+// Environment-variable helpers shared by the per-environment config objects. Kept in its own
+// module (importing nothing from ./config) so it does not create a runtime import cycle.
+
+const DEFAULT_RIVER_THRESHOLD_MARGIN = 0.05
+
+/**
+ * Hysteresis deadband (in metres) applied around river thresholds to avoid alert flapping.
+ * Falls back to a small default when RIVER_THRESHOLD_MARGIN is unset or invalid.
+ */
+export const getRiverThresholdMargin = (): number => {
+    const raw = process.env.RIVER_THRESHOLD_MARGIN
+
+    // Treat unset/blank as "use the default" — Number('') is 0, which would silently disable the
+    // deadband rather than fall back.
+    if (raw === undefined || raw.trim() === '') {
+        return DEFAULT_RIVER_THRESHOLD_MARGIN
+    }
+
+    const margin = Number(raw)
+
+    return Number.isFinite(margin) && margin >= 0 ? margin : DEFAULT_RIVER_THRESHOLD_MARGIN
+}

src/config/production.ts

@@ -1,4 +1,5 @@
 import { Config } from './config'
+import { getRiverThresholdMargin } from './env'
 
 export const productionConfig: Config = {
     database: {
@@ -11,4 +12,5 @@ export const productionConfig: Config = {
     telegram_token: process.env.TELEGRAM_TOKEN as string,
     chat_id: process.env.CHANNEL_ID as string,
     alert_zone: process.env.ALERT_ZONE || 'D1',
+    river_threshold_margin: getRiverThresholdMargin(),
 }

src/models/link-prediction.ts

@@ -0,0 +1,36 @@
+import { database } from '..'
+
+const tableName = 'link_predictions'
+
+export type PredictionOutcome = 'pending' | 'hit' | 'miss'
+
+export interface LinkPrediction {
+    id: number
+    link_id: number
+    predicted_at: string
+    predicted_exceedance_at: string
+    upstream_value: number
+    actual_exceedance_at: string | null
+    outcome: PredictionOutcome
+    created_on: string
+}
+
+export type CreatableLinkPrediction = Omit<LinkPrediction, 'id' | 'outcome' | 'actual_exceedance_at' | 'created_on'>
+
+export const createLinkPrediction = async (prediction: CreatableLinkPrediction): Promise<LinkPrediction> => {
+    return database.create<LinkPrediction>(tableName, {
+        ...prediction,
+        actual_exceedance_at: null,
+        outcome: 'pending',
+        created_on: new Date().toISOString(),
+    })
+}
+
+export const getLatestLinkPrediction = async (linkId: number): Promise<LinkPrediction | undefined> => {
+    const rows = await database.query<LinkPrediction>(
+        `SELECT * FROM ${tableName} WHERE link_id = $1 ORDER BY predicted_at DESC, id DESC LIMIT 1`,
+        [linkId]
+    )
+
+    return rows[0]
+}

src/models/river-level.ts

@@ -16,7 +16,9 @@ export interface RiverLevel {
 export type CreatableRiverLevel = Omit<RiverLevel, 'id' | 'created_on'>
 
 export const getLatestRiverLevel = async (riverId: number): Promise<RiverLevel | undefined> => {
-    const query = `SELECT * FROM ${tableName} WHERE river_id = $1 ORDER BY created_on DESC, id DESC LIMIT 1`
+    // Order by measured_at (not created_on): a historical backfill inserts old measurements with a
+    // current created_on, so created_on no longer tracks measurement recency.
+    const query = `SELECT * FROM ${tableName} WHERE river_id = $1 ORDER BY measured_at DESC, id DESC LIMIT 1`
 
     const rows = await database.query<RiverLevel>(query, [riverId])
 
@@ -26,3 +28,62 @@ export const getLatestRiverLevel = async (riverId: number): Promise<RiverLevel |
 export const createRiverLevel = async (level: CreatableRiverLevel): Promise<RiverLevel> => {
     return database.create<RiverLevel>(tableName, { ...level, created_on: new Date().toISOString() })
 }
+
+/**
+ * Inserts a reading only if no row already exists for the same (river_id, measured_at). Returns
+ * the created row, or undefined when the reading was already stored. Relies on the
+ * UNIQUE (river_id, measured_at) constraint and keeps the table from accumulating duplicate rows
+ * when the sensor has not published a newer reading between polls.
+ */
+export const createRiverLevelIfNew = async (level: CreatableRiverLevel): Promise<RiverLevel | undefined> => {
+    return database.createOrIgnore<RiverLevel>(
+        tableName,
+        { ...level, created_on: new Date().toISOString() },
+        'river_id, measured_at'
+    )
+}
+
+export const getRiverLevelsSince = async (riverId: number, since: string): Promise<RiverLevel[]> => {
+    const query = `SELECT * FROM ${tableName} WHERE river_id = $1 AND measured_at >= $2 ORDER BY measured_at ASC`
+
+    return database.query<RiverLevel>(query, [riverId, since])
+}
+
+export type BackfillRiverLevel = Omit<CreatableRiverLevel, never>
+
+/**
+ * Inserts many readings in one statement, skipping any (river_id, measured_at) already present.
+ * Returns the number of rows actually inserted. Used by the historical backfill; callers must keep
+ * each batch under the Postgres parameter limit (7 params per row).
+ */
+export const bulkInsertRiverLevels = async (rows: BackfillRiverLevel[]): Promise<number> => {
+    if (rows.length === 0) {
+        return 0
+    }
+
+    const now = new Date().toISOString()
+    const values: (string | number | boolean | null)[] = []
+    const tuples = rows.map((row, index) => {
+        const base = index * 7
+        values.push(
+            row.river_id,
+            row.value,
+            row.measured_at,
+            row.soglia1_above,
+            row.soglia2_above,
+            row.soglia3_above,
+            now
+        )
+        return `($${base + 1}, $${base + 2}, $${base + 3}, $${base + 4}, $${base + 5}, $${base + 6}, $${base + 7})`
+    })
+
+    const query = `INSERT INTO ${tableName}
+        (river_id, value, measured_at, soglia1_above, soglia2_above, soglia3_above, created_on)
+        VALUES ${tuples.join(', ')}
+        ON CONFLICT (river_id, measured_at) DO NOTHING
+        RETURNING id`
+
+    const inserted = await database.query<{ id: number }>(query, values)
+
+    return inserted.length
+}

src/models/river-link.ts

@@ -0,0 +1,76 @@
+import { database } from '..'
+
+const tableName = 'river_links'
+
+export interface RiverLink {
+    id: number
+    upstream_river_id: number
+    downstream_river_id: number
+    target_threshold: number
+    lead_time_minutes: number | null
+    precursor_level: number | null
+    sample_size: number
+    model_json: unknown | null
+    last_calibrated_on: string | null
+    created_on: string
+    updated_on: string
+}
+
+export interface CreatableRiverLink {
+    upstream_river_id: number
+    downstream_river_id: number
+    target_threshold?: number
+}
+
+export interface RiverLinkModelPatch {
+    lead_time_minutes: number | null
+    precursor_level: number | null
+    sample_size: number
+    model_json: unknown
+}
+
+export const getRiverLinks = async (): Promise<RiverLink[]> => {
+    return database.query<RiverLink>(`SELECT * FROM ${tableName} ORDER BY id ASC`)
+}
+
+export const getRiverLinkById = async (id: number): Promise<RiverLink | undefined> => {
+    const rows = await database.query<RiverLink>(`SELECT * FROM ${tableName} WHERE id = $1`, [id])
+
+    return rows[0]
+}
+
+export const createRiverLink = async (link: CreatableRiverLink): Promise<RiverLink> => {
+    const now = new Date().toISOString()
+
+    return database.create<RiverLink>(tableName, {
+        upstream_river_id: link.upstream_river_id,
+        downstream_river_id: link.downstream_river_id,
+        target_threshold: link.target_threshold ?? 1,
+        lead_time_minutes: null,
+        precursor_level: null,
+        sample_size: 0,
+        model_json: null,
+        last_calibrated_on: null,
+        created_on: now,
+        updated_on: now,
+    })
+}
+
+export const updateRiverLinkModel = async (id: number, patch: RiverLinkModelPatch): Promise<RiverLink> => {
+    return database.edit<RiverLink>(
+        tableName,
+        {
+            lead_time_minutes: patch.lead_time_minutes,
+            precursor_level: patch.precursor_level,
+            sample_size: patch.sample_size,
+            model_json: patch.model_json === undefined ? null : JSON.stringify(patch.model_json),
+            last_calibrated_on: new Date().toISOString(),
+            updated_on: new Date().toISOString(),
+        },
+        id
+    )
+}
+
+export const deleteRiverLink = async (id: number): Promise<void> => {
+    await database.delete(tableName, id)
+}