Storing Interval Reads in a Time-Series Schema
An interval that has been correctly modeled still has to land in a table without ever duplicating, silently overwriting, or losing a read — and a meter feed re-delivers the same interval constantly, so the write path itself must be idempotent. This page sits under Time-Series Storage & Interval Modeling, part of the broader Meter Data Ingestion & Validation Pipelines subsystem, and it narrows to one concrete artifact: a time-series table schema plus a typed upsert path keyed on (meter_id, interval_start_utc) that stores both the cumulative registers and the derived Decimal interval energy, and that turns a re-delivered read into a no-op while turning a genuine correction into a versioned supersede.
Minimal Prerequisites
- Python 3.11+ — for
zoneinfo, timezone-awaredatetime, andX | Noneunions. - A relational store with
INSERT ... ON CONFLICT DO UPDATE— SQLite (shown here), PostgreSQL, or any engine with upsert on a unique key. decimal.Decimalfor all metered values, persisted as text — neverfloat, and never a binary column that would re-introduce rounding drift.- Validated, UTC-anchored intervals — reads reaching this write path already carry timezone-aware bounds and a non-negative register delta from the modeling layer.
Data assumptions: each interval is identified uniquely by its meter and its UTC start instant; interval energy is always register_end - register_start; and the same interval will be delivered more than once, so the primary key — not application logic — is what enforces uniqueness.
Annotated Implementation
The schema stores the interval’s identity, both register endpoints, the derived energy, and a version. The PRIMARY KEY (meter_id, interval_start_utc) is the idempotency contract: a second delivery of the same read collides on the key instead of inserting a duplicate row. The upsert bumps the version only when the energy actually changes, so a re-sync is a no-op and a real re-read supersedes in place.
from __future__ import annotations
import sqlite3
from datetime import datetime, timezone
from decimal import Decimal
# --- Schema -----------------------------------------------------------------
# One row per (meter, interval). Instants are ISO-8601 UTC strings so ordering
# and equality are exact; Decimals are stored as text so no float drift enters.
DDL = """
CREATE TABLE IF NOT EXISTS interval_read (
meter_id TEXT NOT NULL,
interval_start_utc TEXT NOT NULL, -- ISO-8601, always UTC
interval_end_utc TEXT NOT NULL,
register_start TEXT NOT NULL, -- Decimal-as-text, never float
register_end TEXT NOT NULL,
interval_kwh TEXT NOT NULL, -- derived = register_end - register_start
version INTEGER NOT NULL DEFAULT 1,
quality TEXT NOT NULL DEFAULT 'ACTUAL',
ingested_utc TEXT NOT NULL,
PRIMARY KEY (meter_id, interval_start_utc) -- idempotency contract
);
"""
def upsert_interval(
conn: sqlite3.Connection,
meter_id: str,
interval_start_utc: datetime,
interval_end_utc: datetime,
register_start: Decimal,
register_end: Decimal,
) -> None:
"""Idempotent write keyed on (meter_id, interval_start_utc).
Re-delivering an unchanged read is a no-op; a genuine change supersedes the
prior value by bumping the version, keeping the row's identity stable.
"""
if interval_start_utc.tzinfo is None or interval_end_utc.tzinfo is None:
raise ValueError("interval bounds must be timezone-aware (stored as UTC)")
interval_kwh = register_end - register_start # Decimal subtraction, no float
if interval_kwh < 0:
raise ValueError("negative interval energy — resolve rollover before storing")
conn.execute(
"""
INSERT INTO interval_read (
meter_id, interval_start_utc, interval_end_utc,
register_start, register_end, interval_kwh,
version, quality, ingested_utc
)
VALUES (?, ?, ?, ?, ?, ?, 1, 'ACTUAL', ?)
ON CONFLICT (meter_id, interval_start_utc) DO UPDATE SET
interval_end_utc = excluded.interval_end_utc,
register_start = excluded.register_start,
register_end = excluded.register_end,
interval_kwh = excluded.interval_kwh,
quality = 'CORRECTED',
version = interval_read.version + 1,
ingested_utc = excluded.ingested_utc
WHERE interval_read.interval_kwh <> excluded.interval_kwh -- only on a real change
""",
(
meter_id,
interval_start_utc.astimezone(timezone.utc).isoformat(),
interval_end_utc.astimezone(timezone.utc).isoformat(),
str(register_start), str(register_end), str(interval_kwh),
datetime.now(timezone.utc).isoformat(),
),
)
conn.commit()
Two clauses do the heavy lifting. The primary key makes a duplicate delivery structurally impossible — the second write hits ON CONFLICT instead of creating a second row. The WHERE interval_read.interval_kwh <> excluded.interval_kwh guard on the update makes the conflict path conditional: an identical re-delivery matches nothing to update and leaves the version untouched, while a re-read whose energy differs bumps version and stamps CORRECTED. Storing Decimal values with str() keeps them exact across the round trip, and normalizing every instant with astimezone(timezone.utc).isoformat() guarantees the key is byte-identical no matter which offset the read arrived in — the same idempotency identity established during AMI/AMR feed synchronization.
Edge Cases
- Duplicate interval upsert. The same read arriving twice in a batch, or a nightly re-sync of an unchanged day, hits
ON CONFLICTand — because the energy matches — theWHEREguard suppresses the update entirely. No duplicate row, no version churn, no inflated totals. - Out-of-order arrival. A late interval whose start is earlier than rows already stored simply upserts on its own key; ordering is a property of
interval_start_utc, not of insertion order, so a query sorting by that column always reconstructs the correct sequence regardless of when each read landed. - DST-length interval. A nominal daily interval spanning a daylight-saving change is 23 or 25 hours wide. Because the key and bounds are UTC instants, the row stores and sorts correctly with no special case; only downstream gap counting needs
zoneinfoto know the civil day was short or long — the arithmetic is covered in computing seasonal rate boundaries across DST transitions. - Correction supersedes. A re-read that changes the register endpoint bumps
versionand flags the rowCORRECTEDin place. The row’s identity — its(meter_id, interval_start_utc)key — never changes, so anything referencing that interval keeps resolving to it while reflecting the corrected energy, the append-only discipline shared with error handling and retry workflows.
Verification Snippet
Exercise the three paths that matter: a first write, an identical re-delivery that must not change anything, and a genuine correction that must supersede — all resolving to exactly one row.
def test_upsert_is_idempotent_and_versions_corrections() -> None:
conn = sqlite3.connect(":memory:")
conn.execute(DDL)
start = datetime(2026, 7, 1, 5, 0, tzinfo=timezone.utc)
end = datetime(2026, 7, 1, 6, 0, tzinfo=timezone.utc)
# 1) first write
upsert_interval(conn, "MTR-00010001", start, end,
Decimal("1000.00"), Decimal("1012.50"))
# 2) identical re-delivery: must NOT bump the version
upsert_interval(conn, "MTR-00010001", start, end,
Decimal("1000.00"), Decimal("1012.50"))
row = conn.execute(
"SELECT version, interval_kwh FROM interval_read "
"WHERE meter_id = ? AND interval_start_utc = ?",
("MTR-00010001", start.isoformat()),
).fetchone()
assert row == (1, "12.50") # still v1, exact Decimal preserved
# 3) a re-read correcting the register endpoint: supersede as v2
upsert_interval(conn, "MTR-00010001", start, end,
Decimal("1000.00"), Decimal("1015.00"))
row = conn.execute(
"SELECT version, interval_kwh, quality FROM interval_read "
"WHERE meter_id = ? AND interval_start_utc = ?",
("MTR-00010001", start.isoformat()),
).fetchone()
assert row == (2, "15.00", "CORRECTED")
# exactly one row per (meter, interval), no matter how many deliveries
(count,) = conn.execute("SELECT COUNT(*) FROM interval_read").fetchone()
assert count == 1
The assertions encode the contract directly: the version stays at 1 across a duplicate delivery, advances to 2 on a real correction, the Decimal energy survives as an exact string, and the row count never exceeds one for a given key.
FAQ
Related Topics
- Time-Series Storage & Interval Modeling — the parent guide covering the canonical record, cumulative-vs-delta storage, and partitioning.
- AMI/AMR Feed Synchronization Protocols — where the per-read idempotency identity this upsert relies on originates.
- Schema Validation & Data Quality Checks — the boundary gate that guarantees the reads reaching this write path are typed and bounded.
- Error Handling & Retry Workflows — the retry and dead-letter patterns that make a re-delivered write safe to attempt.
Up: Time-Series Storage & Interval Modeling · Meter Data Ingestion & Validation Pipelines