feat: enhance flood calibration and prediction with new river readings and tests

Author: nicomontanari

src/models/river-level.ts

@@ -49,6 +49,21 @@ export const getRiverLevelsSince = async (riverId: number, since: string): Promi
     return database.query<RiverLevel>(query, [riverId, since])
 }
 
+export interface RiverReading {
+    value: number
+    measured_at: string
+}
+
+/**
+ * Lightweight reading fetch for calibration: only the two columns the model needs, ordered in time.
+ * Keeps memory low on small hosts when scanning years of history (vs SELECT * full rows).
+ */
+export const getRiverReadingsSince = async (riverId: number, since: string): Promise<RiverReading[]> => {
+    const query = `SELECT value, measured_at FROM ${tableName} WHERE river_id = $1 AND measured_at >= $2 ORDER BY measured_at ASC`
+
+    return database.query<RiverReading>(query, [riverId, since])
+}
+
 export type BackfillRiverLevel = Omit<CreatableRiverLevel, never>
 
 /**

src/tasks/flood-calibration.ts

@@ -1,12 +1,15 @@
 import { getRiverById } from '../models/river'
-import { getRiverLevelsSince } from '../models/river-level'
+import { getRiverReadingsSince } from '../models/river-level'
 import { getRiverLinks, updateRiverLinkModel } from '../models/river-link'
 import { calibrateLink, isLinkActive, toReadings } from '../utilities/flood-prediction'
 import logger from '../logger'
 
 const log = logger.child({ task: 'flood-calibration' })
 
-const EPOCH_START = '1970-01-01T00:00:00.000Z'
+// Bound how far back calibration reads, so memory stays modest on small hosts and the model
+// reflects recent channel behaviour (~6 years still spans many flood seasons).
+const CALIBRATION_LOOKBACK_DAYS = 6 * 365
+const DAY_MS = 24 * 60 * 60 * 1000
 
 export interface FloodCalibrationSummary {
     calibrated: number
@@ -31,6 +34,7 @@ const thresholdFor = (
 export const runFloodCalibration = async (): Promise<FloodCalibrationSummary> => {
     const links = await getRiverLinks()
     const summary: FloodCalibrationSummary = { calibrated: 0, active: 0, skipped: 0 }
+    const since = new Date(Date.now() - CALIBRATION_LOOKBACK_DAYS * DAY_MS).toISOString()
 
     for (const link of links) {
         try {
@@ -51,8 +55,8 @@ export const runFloodCalibration = async (): Promise<FloodCalibrationSummary> =>
             }
 
             const [downstreamRows, upstreamRows] = await Promise.all([
-                getRiverLevelsSince(link.downstream_river_id, EPOCH_START),
-                getRiverLevelsSince(link.upstream_river_id, EPOCH_START),
+                getRiverReadingsSince(link.downstream_river_id, since),
+                getRiverReadingsSince(link.upstream_river_id, since),
             ])
 
             const model = calibrateLink(toReadings(upstreamRows), toReadings(downstreamRows), threshold)

src/utilities/flood-prediction.ts

@@ -47,13 +47,17 @@ export interface FloodModelOptions {
     minLeadMinutes: number
     minSamples: number
     precursorPercentile: number
+    minSeparationMinutes: number
 }
 
 export const DEFAULT_FLOOD_OPTIONS: FloodModelOptions = {
     maxLookbackMinutes: 48 * 60,
     minLeadMinutes: 15,
     minSamples: 3,
     precursorPercentile: 0.25,
+    // Treat brief dips back under the threshold as part of the same flood, so one flood is one
+    // event (not many micro-events when the level hugs the threshold).
+    minSeparationMinutes: 6 * 60,
 }
 
 const MINUTE_MS = 60_000
@@ -83,22 +87,37 @@ export const percentile = (values: number[], p: number): number => {
  * Splits a level series into maximal runs strictly above `threshold`. Each run yields one event
  * with its onset (first sample above), peak time and peak value.
  */
-export const detectExceedanceEvents = (series: Reading[], threshold: number): ExceedanceEvent[] => {
+export const detectExceedanceEvents = (
+    series: Reading[],
+    threshold: number,
+    minSeparationMinutes = 0
+): ExceedanceEvent[] => {
     const sorted = sortByTime(series)
+    const separationMs = minSeparationMinutes * MINUTE_MS
     const events: ExceedanceEvent[] = []
     let current: ExceedanceEvent | null = null
+    let belowSince: number | null = null
 
     for (const point of sorted) {
         if (point.value > threshold) {
+            belowSince = null
             if (!current) {
                 current = { onsetAt: point.measuredAt, peakAt: point.measuredAt, peakValue: point.value }
             } else if (point.value > current.peakValue) {
                 current.peakValue = point.value
                 current.peakAt = point.measuredAt
             }
         } else if (current) {
-            events.push(current)
-            current = null
+            // Only close the event once the level has stayed below the threshold for at least the
+            // separation window; shorter dips are absorbed into the same flood.
+            if (belowSince === null) {
+                belowSince = point.measuredAt
+            }
+            if (point.measuredAt - belowSince >= separationMs) {
+                events.push(current)
+                current = null
+                belowSince = null
+            }
         }
     }
 
@@ -147,7 +166,7 @@ export const calibrateLink = (
     threshold: number,
     options: FloodModelOptions = DEFAULT_FLOOD_OPTIONS
 ): LinkModel => {
-    const events = detectExceedanceEvents(downstream, threshold)
+    const events = detectExceedanceEvents(downstream, threshold, options.minSeparationMinutes)
     const observations = events
         .map((event) => findPrecursor(upstream, event, options))
         .filter((obs): obs is PrecursorObservation => obs !== null)

src/utilities/telegram.ts

@@ -103,7 +103,8 @@ export const sendFloodPredictionMessage = async (
     const textMessage = `🌊⚠️ Possibile piena in arrivo
 Monte: ${upstream.river_name} — ${upstream.station_name}: ${payload.upstreamValue} m
 Storicamente questo livello a monte ha preceduto il superamento della soglia ${payload.targetThreshold} a ${downstream.river_name} — ${downstream.station_name}.
-Arrivo stimato: tra ~${formatLeadTime(payload.leadTimeMinutes)} (≈ ${eta})`
+Arrivo stimato: tra ~${formatLeadTime(payload.leadTimeMinutes)} (≈ ${eta})
+ℹ️ Stima statistica, non una previsione ufficiale: verificare sempre le fonti ufficiali (allertameteo.regione.emilia-romagna.it, Protezione Civile).`
 
     await sendTelegramMessage(config.chat_id, textMessage)
 }

tests/tasks/flood-calibration.ts

@@ -0,0 +1,145 @@
+import { expect } from 'chai'
+import { afterEach, beforeEach, describe, it } from 'mocha'
+import sinon, { SinonStub } from 'sinon'
+import * as riverModel from '../../src/models/river'
+import * as riverLevelModel from '../../src/models/river-level'
+import * as riverLinkModel from '../../src/models/river-link'
+import { River } from '../../src/models/river'
+import { RiverLink } from '../../src/models/river-link'
+import { runFloodCalibration } from '../../src/tasks/flood-calibration'
+
+const T0 = new Date('2026-01-01T00:00:00.000Z').getTime()
+const MIN = 60_000
+const DAY = 24 * 60 * MIN
+
+const iso = (ms: number) => new Date(ms).toISOString()
+
+const river = (id: number, soglia1: number | null): River => ({
+    id,
+    station_id: String(id),
+    river_name: 'Idice',
+    station_name: `st-${id}`,
+    soglia1,
+    soglia2: null,
+    soglia3: null,
+    created_on: iso(T0),
+    updated_on: iso(T0),
+})
+
+const link = (overrides: Partial<RiverLink> = {}): RiverLink => ({
+    id: 1,
+    upstream_river_id: 10,
+    downstream_river_id: 20,
+    target_threshold: 1,
+    lead_time_minutes: null,
+    precursor_level: null,
+    sample_size: 0,
+    model_json: null,
+    last_calibrated_on: null,
+    created_on: iso(T0),
+    updated_on: iso(T0),
+    ...overrides,
+})
+
+// Three downstream exceedances (>10) 3 days apart, each preceded by an upstream peak.
+const buildPair = (dayOffset: number, leadMin: number, upstreamPeak: number) => {
+    const onset = T0 + dayOffset * DAY
+    return {
+        downstream: [
+            { value: 9, measured_at: iso(onset - 30 * MIN) },
+            { value: 11, measured_at: iso(onset) },
+            { value: 12, measured_at: iso(onset + 30 * MIN) },
+            { value: 8, measured_at: iso(onset + 60 * MIN) },
+        ],
+        upstream: [
+            { value: upstreamPeak - 2, measured_at: iso(onset - (leadMin + 60) * MIN) },
+            { value: upstreamPeak, measured_at: iso(onset - leadMin * MIN) },
+            { value: upstreamPeak - 1, measured_at: iso(onset - (leadMin - 30) * MIN) },
+        ],
+    }
+}
+
+const e = [buildPair(2, 120, 5), buildPair(5, 150, 6), buildPair(8, 135, 5.5)]
+const downstreamRows = e.flatMap((p) => p.downstream)
+const upstreamRows = e.flatMap((p) => p.upstream)
+
+describe('tests/tasks/flood-calibration', () => {
+    let getLinks: SinonStub
+    let getRiver: SinonStub
+    let getLevels: SinonStub
+    let update: SinonStub
+
+    beforeEach(() => {
+        getLinks = sinon.stub(riverLinkModel, 'getRiverLinks')
+        getRiver = sinon.stub(riverModel, 'getRiverById')
+        getLevels = sinon.stub(riverLevelModel, 'getRiverReadingsSince')
+        update = sinon.stub(riverLinkModel, 'updateRiverLinkModel').resolves({} as never)
+    })
+
+    afterEach(() => sinon.restore())
+
+    it('learns and persists an active model from historical events', async () => {
+        getLinks.resolves([link()])
+        getRiver.withArgs(20).resolves(river(20, 10))
+        getRiver.withArgs(10).resolves(river(10, 5))
+        getLevels.withArgs(20).resolves(downstreamRows)
+        getLevels.withArgs(10).resolves(upstreamRows)
+
+        const summary = await runFloodCalibration()
+
+        expect(update.calledOnce).to.equal(true)
+        const patch = update.firstCall.args[1]
+        expect(patch.sample_size).to.equal(3)
+        expect(patch.lead_time_minutes).to.equal(135)
+        expect(patch.precursor_level).to.be.closeTo(5.25, 1e-9)
+        expect(summary).to.deep.equal({ calibrated: 1, active: 1, skipped: 0 })
+    })
+
+    it('skips a link whose downstream soglia is not set', async () => {
+        getLinks.resolves([link()])
+        getRiver.withArgs(20).resolves(river(20, null))
+        getRiver.withArgs(10).resolves(river(10, 5))
+
+        const summary = await runFloodCalibration()
+
+        expect(update.called).to.equal(false)
+        expect(summary).to.deep.equal({ calibrated: 0, active: 0, skipped: 1 })
+    })
+
+    it('skips a link with a missing river', async () => {
+        getLinks.resolves([link()])
+        getRiver.withArgs(20).resolves(undefined)
+        getRiver.withArgs(10).resolves(river(10, 5))
+
+        const summary = await runFloodCalibration()
+
+        expect(summary.skipped).to.equal(1)
+        expect(update.called).to.equal(false)
+    })
+
+    it('isolates a per-link failure and continues', async () => {
+        getLinks.resolves([link({ id: 1 }), link({ id: 2 })])
+        getRiver.withArgs(20).resolves(river(20, 10))
+        getRiver.withArgs(10).resolves(river(10, 5))
+        getLevels.withArgs(20).rejects(new Error('db down'))
+
+        const summary = await runFloodCalibration()
+
+        expect(summary.skipped).to.equal(2)
+        expect(summary.calibrated).to.equal(0)
+    })
+
+    it('persists an inactive model when there are too few events', async () => {
+        getLinks.resolves([link()])
+        getRiver.withArgs(20).resolves(river(20, 10))
+        getRiver.withArgs(10).resolves(river(10, 5))
+        getLevels.withArgs(20).resolves(e[0].downstream) // single event only
+        getLevels.withArgs(10).resolves(e[0].upstream)
+
+        const summary = await runFloodCalibration()
+
+        const patch = update.firstCall.args[1]
+        expect(patch.sample_size).to.equal(1)
+        expect(summary).to.deep.equal({ calibrated: 1, active: 0, skipped: 0 })
+    })
+})