Add real TOU day/night split groundwork ahead of the IoT pulse reader

Prepares for a future electric pulse-reading device without needing
one yet: a new meter_window_consumption table stores real per-window
(Day/Night) daily kWh computed from hourly device telemetry before it
gets purged, and computeCost() now prefers real data over the tariff's
fallback_split_pct for whatever it covers, filling any gap (or the
whole thing, today) with the existing % split. Zero behaviour change
until a meter has mqtt_topic set and a TOU tariff assigned — the new
table stays empty and every calculation degrades to exactly the
current fallback-only path.

scheduler.js's computeDailyWindowSplit() buckets each hour's telemetry
delta into whichever tariff_rate_windows row its time-of-day falls
into (previously-unused start_time/end_time/days_of_week columns),
runs once daily before the hourly-aggregate purge.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

backend/src/db.js

@@ -149,6 +149,21 @@ export async function initDb() {
       UNIQUE (meter_id, bucket_start)
     );
     CREATE INDEX IF NOT EXISTS telemetry_hourly_meter_time_idx ON meter_telemetry_hourly (meter_id, bucket_start DESC);
+
+    -- real per-window (e.g. Day/Night) daily consumption for TOU-tariffed
+    -- device-fed meters, computed once from hourly telemetry by scheduler.js's
+    -- computeDailyWindowSplit() before that telemetry gets purged — this is
+    -- what survives long-term once a meter has real device data, versus the
+    -- tariff's fallback_split_pct guess that's all we have without one.
+    CREATE TABLE IF NOT EXISTS meter_window_consumption (
+      id               SERIAL PRIMARY KEY,
+      meter_id         INT NOT NULL REFERENCES meters(id) ON DELETE CASCADE,
+      consumption_date DATE NOT NULL,
+      window_label     TEXT NOT NULL,
+      consumption      NUMERIC(14,3) NOT NULL,
+      UNIQUE (meter_id, consumption_date, window_label)
+    );
+    CREATE INDEX IF NOT EXISTS meter_window_consumption_meter_date_idx ON meter_window_consumption (meter_id, consumption_date);
   `)
 
   await seedDefaults()

backend/src/lib/cost-calc.js

@@ -73,7 +73,7 @@ async function getMeterGasConfig(meterId) {
 // Converts a raw reading-delta to the category's billed unit — a no-op for
 // every category except gas meters with a volume unit configured, where the
 // raw m3/ft3 delta becomes kWh so it lines up with tariff rates (p/kWh).
-async function convertMeterConsumption(meterId, rawConsumption) {
+export async function convertMeterConsumption(meterId, rawConsumption) {
   const gasConfig = await getMeterGasConfig(meterId)
   if (gasConfig?.category_key === 'gas' && gasConfig.gas_volume_unit) {
     return gasVolumeToKwh(rawConsumption, gasConfig.gas_volume_unit, gasConfig.gas_calorific_value)
@@ -219,11 +219,13 @@ export async function getRateWindows(tariffId) {
   return rows
 }
 
-// Core layered calc: usage (split by fallback % across TOU windows, or single
+// Core layered calc: usage (real per-window consumption from device telemetry
-// rate) + standing charge + CCL (electric/gas only, skipped if exempt) + VAT.
+// where available, falling back to a fixed % split across TOU windows for
-// Mirrors the usage + fixed costs + levy -> total layering already used by
+// whatever it doesn't cover, or a single rate outside TOU) + standing charge +
-// the Directors report's dry/wet forecast.
+// CCL (electric/gas only, skipped if exempt) + VAT. Mirrors the usage + fixed
-export function computeCost({ tariff, windows, consumption, daysInPeriod }) {
+// costs + levy -> total layering already used by the Directors report's
+// dry/wet forecast.
+export function computeCost({ tariff, windows, consumption, daysInPeriod, realWindowSplit }) {
   if (!tariff || consumption == null) {
     return {
       usage_cost_pence: 0, standing_cost_pence: 0, ccl_cost_pence: 0, rab_levy_cost_pence: 0,
@@ -234,15 +236,48 @@ export function computeCost({ tariff, windows, consumption, daysInPeriod }) {
   let usageCostPence = 0
   let split = null
 
-  if (tariff.is_time_of_use && windows.length > 1 && tariff.fallback_split_pct && Object.keys(tariff.fallback_split_pct).length) {
+  const hasFallbackSplit = !!(tariff.fallback_split_pct && Object.keys(tariff.fallback_split_pct).length)
+  const hasRealSplit = !!(realWindowSplit && Object.keys(realWindowSplit.byWindow || {}).length)
+
+  if (tariff.is_time_of_use && windows.length > 1 && (hasRealSplit || hasFallbackSplit)) {
     split = {}
-    for (const [key, pct] of Object.entries(tariff.fallback_split_pct)) {
+    let coveredConsumption = 0
-      const win = windows.find(w => w.label.toLowerCase() === key.toLowerCase())
+
-      if (!win) continue
+    if (hasRealSplit) {
-      const share = consumption * (Number(pct) / 100)
+      for (const w of windows) {
-      const cost = share * Number(win.unit_rate_pence_per_unit)
+        const real = realWindowSplit.byWindow[w.label]
-      split[win.label] = { share, rate: Number(win.unit_rate_pence_per_unit), cost_pence: cost }
+        if (!real) continue
-      usageCostPence += cost
+        const cost = real * Number(w.unit_rate_pence_per_unit)
+        split[w.label] = { share: real, rate: Number(w.unit_rate_pence_per_unit), cost_pence: cost }
+        usageCostPence += cost
+        coveredConsumption += real
+      }
+    }
+
+    // Whatever real device data doesn't cover — no device yet, or a gap in
+    // telemetry — is split by the tariff's assumed fallback %, same as before
+    // real data existed at all.
+    const uncovered = consumption - coveredConsumption
+    if (uncovered > 0 && hasFallbackSplit) {
+      for (const [key, pct] of Object.entries(tariff.fallback_split_pct)) {
+        const win = windows.find(w => w.label.toLowerCase() === key.toLowerCase())
+        if (!win) continue
+        const share = uncovered * (Number(pct) / 100)
+        const cost = share * Number(win.unit_rate_pence_per_unit)
+        const existing = split[win.label]
+        split[win.label] = {
+          share: (existing?.share || 0) + share,
+          rate: Number(win.unit_rate_pence_per_unit),
+          cost_pence: (existing?.cost_pence || 0) + cost,
+        }
+        usageCostPence += cost
+      }
+    } else if (uncovered > 0) {
+      // Real data exists but doesn't cover the whole period and there's no
+      // fallback % configured — price the gap at a simple blended average
+      // rather than dropping it from the bill.
+      const avgRate = windows.reduce((s, w) => s + Number(w.unit_rate_pence_per_unit), 0) / windows.length
+      usageCostPence += uncovered * avgRate
     }
   } else {
     const rate = windows.length ? Number(windows[0].unit_rate_pence_per_unit) : 0
@@ -286,6 +321,25 @@ const ZERO_TOTALS = {
   metering_cost_pence: 0, other_charges_cost_pence: 0, subtotal_pence: 0, vat_pence: 0, total_pence: 0,
 }
 
+// Real per-window (e.g. Day/Night) consumption over [rangeStart, rangeEnd],
+// summed from scheduler.js's daily device-telemetry breakdown — null if the
+// meter has none for this range at all (no device installed yet, or a
+// non-TOU tariff), which computeCost() treats as "fall back to the tariff's
+// assumed % split" entirely, same as before this table existed.
+export async function getRealWindowSplit(meterId, rangeStart, rangeEnd) {
+  const { rows } = await pool.query(
+    `SELECT window_label, SUM(consumption) AS total
+     FROM meter_window_consumption
+     WHERE meter_id = $1 AND consumption_date >= $2 AND consumption_date <= $3
+     GROUP BY window_label`,
+    [meterId, rangeStart, rangeEnd]
+  )
+  if (!rows.length) return null
+  const byWindow = {}
+  for (const r of rows) byWindow[r.window_label] = Number(r.total)
+  return { byWindow }
+}
+
 // Cost a single amount of consumption over [rangeStart, rangeEnd], splitting it
 // pro-rata (by day count) across every tariff segment active in that range. Used
 // both for a closed historical period and for a projected future range — in both
@@ -299,7 +353,8 @@ async function splitCostAcrossSegments(meterId, rangeStart, rangeEnd, consumptio
   if (!segments.length) {
     const tariff = await getTariffForMeter(meterId, asOfDate || rangeEnd)
     const windows = tariff ? await getRateWindows(tariff.id) : []
-    const cost = computeCost({ tariff, windows, consumption, daysInPeriod: rangeDays })
+    const realWindowSplit = await getRealWindowSplit(meterId, rangeStart, rangeEnd)
+    const cost = computeCost({ tariff, windows, consumption, daysInPeriod: rangeDays, realWindowSplit })
     return { tariff, windows, segments: [], rate_changed_mid_period: false, ...cost }
   }
 
@@ -311,7 +366,8 @@ async function splitCostAcrossSegments(meterId, rangeStart, rangeEnd, consumptio
     const segDays = daysInclusive(seg.seg_start, seg.seg_end)
     const segConsumption = dailyAvg == null ? null : dailyAvg * segDays
     const windows = await getRateWindows(seg.id)
-    const cost = computeCost({ tariff: seg, windows, consumption: segConsumption, daysInPeriod: segDays })
+    const realWindowSplit = await getRealWindowSplit(meterId, seg.seg_start, seg.seg_end)
+    const cost = computeCost({ tariff: seg, windows, consumption: segConsumption, daysInPeriod: segDays, realWindowSplit })
     for (const k of Object.keys(agg)) agg[k] += cost[k]
     segmentResults.push({
       tariff_id: seg.id, tariff_name: seg.name,

backend/src/lib/scheduler.js

@@ -3,6 +3,7 @@
 // stack has no cron library anywhere, "run once daily" is always a 1-minute
 // wall-clock poll guarded by a last-run-date check.
 import { pool, getConfig } from '../db.js'
+import { getTariffForMeter, getRateWindows, convertMeterConsumption } from './cost-calc.js'
 
 const HOURLY_MS = 60 * 60 * 1000
 const DAILY_CHECK_MS = 60 * 1000
@@ -66,6 +67,84 @@ async function insertDailyDeviceReadings() {
   }
 }
 
+function toISODate(d) {
+  return new Date(d).toISOString().slice(0, 10)
+}
+
+// Does rate window `w` cover this hour-of-day (0-23) / day-of-week (0=Sunday,
+// matching JS getUTCDay())? Handles windows that wrap past midnight (e.g. a
+// Night window from 23:00 to 07:00). Hour-of-day is read off bucket_start in
+// UTC, same as the rest of this file's daily-key logic — a deliberate
+// simplification that can be an hour off right around a BST/GMT change,
+// acceptable given day/night windows are normally hours wide either side.
+function windowCoversHour(w, hour, dow) {
+  if (w.days_of_week && w.days_of_week.length && !w.days_of_week.includes(dow)) return false
+  if (!w.start_time || !w.end_time) return true
+  const start = Number(String(w.start_time).slice(0, 2))
+  const end = Number(String(w.end_time).slice(0, 2))
+  if (start === end) return true
+  if (start < end) return hour >= start && hour < end
+  return hour >= start || hour < end
+}
+
+// Real per-window (Day/Night/...) consumption for TOU-tariffed device-fed
+// meters, computed from yesterday's hourly telemetry before
+// purgeOldHourlyAggregates() removes it — this is what lets cost-calc.js use
+// a real split instead of the tariff's fallback_split_pct guess once a pulse
+// reader is actually installed. A no-op today for every meter (no mqtt_topic
+// set yet), and harmless once one exists on a non-TOU meter (skipped).
+async function computeDailyWindowSplit() {
+  const yesterday = new Date()
+  yesterday.setDate(yesterday.getDate() - 1)
+  const dateStr = yesterday.toISOString().slice(0, 10)
+
+  const { rows: meters } = await pool.query(
+    `SELECT id FROM meters WHERE mqtt_topic IS NOT NULL AND active = TRUE`
+  )
+
+  for (const m of meters) {
+    const tariff = await getTariffForMeter(m.id, dateStr)
+    if (!tariff || !tariff.is_time_of_use) continue
+    const windows = await getRateWindows(tariff.id)
+    if (windows.length <= 1) continue
+
+    // Include the bucket just before midnight so the first hour of the day
+    // still gets a delta to attribute.
+    const { rows: buckets } = await pool.query(
+      `SELECT bucket_start, value FROM meter_telemetry_hourly
+       WHERE meter_id = $1
+         AND bucket_start >= ($2::date - INTERVAL '1 hour')
+         AND bucket_start < ($2::date + INTERVAL '1 day')
+       ORDER BY bucket_start ASC`,
+      [m.id, dateStr]
+    )
+    if (buckets.length < 2) continue
+
+    const perWindow = {}
+    for (let i = 1; i < buckets.length; i++) {
+      const prev = buckets[i - 1]
+      const cur = buckets[i]
+      if (toISODate(cur.bucket_start) !== dateStr) continue
+      const delta = Number(cur.value) - Number(prev.value)
+      if (delta <= 0) continue // rollover / stale sample — skip rather than record a bogus negative
+      const hour = new Date(prev.bucket_start).getUTCHours()
+      const dow = new Date(prev.bucket_start).getUTCDay()
+      const win = windows.find(w => windowCoversHour(w, hour, dow)) || windows[0]
+      const kwh = await convertMeterConsumption(m.id, delta)
+      perWindow[win.label] = (perWindow[win.label] || 0) + kwh
+    }
+
+    for (const [label, kwh] of Object.entries(perWindow)) {
+      await pool.query(
+        `INSERT INTO meter_window_consumption (meter_id, consumption_date, window_label, consumption)
+         VALUES ($1, $2, $3, $4)
+         ON CONFLICT (meter_id, consumption_date, window_label) DO UPDATE SET consumption = EXCLUDED.consumption`,
+        [m.id, dateStr, label, kwh]
+      )
+    }
+  }
+}
+
 async function purgeOldHourlyAggregates() {
   const cfg = await getConfig()
   const days = cfg.telemetry_hourly_retention_days ?? 60
@@ -104,6 +183,11 @@ export function startScheduler() {
     } catch (err) {
       console.error('[scheduler] daily reading insert failed:', err.message)
     }
+    try {
+      await computeDailyWindowSplit()
+    } catch (err) {
+      console.error('[scheduler] daily window split failed:', err.message)
+    }
     try {
       await purgeOldHourlyAggregates()
     } catch (err) {