Schema Validation and Data Quality Checks in Municipal Utility Billing

Bad data caught at the ingestion boundary costs a log line; the same data caught at month-end costs a re-bill. For billing managers, municipal finance directors, and public-sector developers, the integrity of consumption data directly dictates revenue assurance, Public Utility Commission (PUC) compliance, and ratepayer trust. This workflow sits at the very front of the Meter Data Ingestion & Validation Pipelines subsystem: it is the gate every raw read must pass before any downstream stage — anomaly scoring, batching, or rating — is allowed to touch it. The specific failure it prevents is structural corruption entering the billing ledger: a therm value that arrived as an empty string, a timestamp with no timezone offset, a register value carrying twelve digits of float drift. When advanced metering infrastructure (AMI) and automated meter reading (AMR) networks emit millions of interval reads a day, the absence of a deterministic contract at the boundary turns every one of those defects into a silent liability that surfaces weeks later as a mis-priced invoice or an audit finding. Establishing strict, declarative quality gates ensures that every kilowatt-hour, therm, or gallon is accounted for — and provably well-formed — before it triggers a single financial calculation.

# Prerequisites

This workflow targets a modern, reproducible Python stack. Pin the following before implementing anything below:

• Python 3.11+ — required for standard-library zoneinfo, X | None union syntax, and Self typing used in the validators.
• Pydantic v2 (pydantic>=2.6) — every ingestion contract uses v2 field_validator / model_validator semantics and ConfigDict; v1 models will not parse the code below.
• decimal.Decimal for any value that becomes money or a billable quantity — register reads and interval deltas are parsed as Decimal with an explicit quantization. Native float introduces IEEE-754 rounding artifacts that can trigger PUC audit findings, so it is confined to non-billing telemetry only.
• zoneinfo (not pytz) — every read timestamp is resolved to the utility’s local civil time so that interval and DST logic downstream is unambiguous.

Data assumptions: raw reads arrive from the head-end as CSV or JSON records carrying at minimum a meter_id, an interval timestamp, a cumulative register_read or pre-computed kwh_delta, and a meter_class (residential, commercial, industrial). This workflow owns structural and field-level correctness. Behavioral anomalies in otherwise well-formed data — spikes, stuck registers, rollovers — are the responsibility of the sibling Reading Anomaly Detection Algorithms stage that runs immediately after this one. Write access to the quarantine store and authority to override a rejected record must be governed; see Security Boundaries & Role-Based Access for the pattern that gates those overrides.

# Architecture Overview

Validation is a gate, not a filter you can skip under load. Raw telemetry enters, is parsed against a declarative contract, and forks: contract-conformant records proceed to layered quality checks and then to the ledger; everything else is quarantined with an audit envelope so no read is ever silently dropped. Each batch carries the source system, ingestion timestamp, and validation outcome, creating an immutable trail for financial reconciliation.

Figure: The ingestion-boundary quality gate — only contract-conformant reads reach the billing ledger; the rest are quarantined with an audit trail.

The sections below implement each stage in order: define the contract, enforce field and cross-field rules, wrap batches for audit, layer logical quality checks, and hand off idempotently to the ledger.

# Step-by-Step Implementation

# Step 1 — Define the declarative ingestion contract

The foundation of reliable billing is a single, declarative contract that raw telemetry must satisfy before it is persisted. Rather than scattering if-checks through the pipeline, the contract encodes type safety, required fields, and canonical shapes in one place. Pydantic v2’s strict mode rejects the structurally impossible — a naive timestamp, a non-numeric register, an unknown meter class — at the edge, so every later stage can assume a clean, typed record.

from __future__ import annotations

from datetime import datetime
from decimal import Decimal
from enum import Enum
from zoneinfo import ZoneInfo

from pydantic import BaseModel, ConfigDict, field_validator

UTILITY_TZ = ZoneInfo("America/Chicago")
# Register reads and deltas quantize to 3 decimal places — the resolution
# billed against. Money-adjacent quantities never travel as float.
QUANTUM = Decimal("0.001")


class MeterClass(str, Enum):
    RESIDENTIAL = "residential"
    COMMERCIAL = "commercial"
    INDUSTRIAL = "industrial"


class MeterRead(BaseModel):
    """The canonical, contract-validated shape of a single interval read."""

    # strict=True forbids silent coercion (e.g. "12.5" -> 12.5); a value
    # arrives in the right type or it is rejected and quarantined.
    model_config = ConfigDict(strict=True, frozen=True, extra="forbid")

    meter_id: str
    meter_class: MeterClass
    timestamp: datetime
    register_read: Decimal

    @field_validator("timestamp")
    @classmethod
    def require_aware_local(cls, v: datetime) -> datetime:
        # A timestamp with no offset is ambiguous across DST and cannot be
        # ordered against its neighbours. Reject it rather than guess.
        if v.tzinfo is None:
            raise ValueError("timestamp must be timezone-aware")
        return v.astimezone(UTILITY_TZ)

    @field_validator("register_read")
    @classmethod
    def finite_and_quantized(cls, v: Decimal) -> Decimal:
        if not v.is_finite():
            raise ValueError("register_read must be a finite Decimal")
        # Canonicalize precision so two equal reads hash and compare identically.
        return v.quantize(QUANTUM)

extra="forbid" is deliberate: a vendor that adds an undocumented column should trip the contract, not have the surprise field silently ignored — undocumented columns are how schema drift enters undetected.

# Step 2 — Enforce field-level and cross-field quality rules

Type safety is necessary but not sufficient. A register value can be a perfectly valid Decimal and still be nonsensical for billing — negative, or absurdly large for its meter class. Cross-field rules run on the whole record with a model_validator, because some constraints (a read timestamped in the future, a register below its class floor) can only be judged with several fields in hand.

from typing import Self

from pydantic import model_validator

# Physical register ceilings per meter class (cumulative kWh). A value above
# the ceiling is structurally implausible and is rejected at the boundary.
REGISTER_CEILING = {
    MeterClass.RESIDENTIAL: Decimal("1_000_000"),
    MeterClass.COMMERCIAL: Decimal("10_000_000"),
    MeterClass.INDUSTRIAL: Decimal("100_000_000"),
}


class ValidatedMeterRead(MeterRead):
    """MeterRead plus cross-field business constraints."""

    @model_validator(mode="after")
    def enforce_business_rules(self) -> Self:
        if self.register_read < 0:
            raise ValueError("register_read cannot be negative")

        ceiling = REGISTER_CEILING[self.meter_class]
        if self.register_read > ceiling:
            raise ValueError(
                f"register_read {self.register_read} exceeds "
                f"{self.meter_class.value} ceiling {ceiling}"
            )

        # Reject reads timestamped in the future — a common symptom of an
        # un-synced meter clock that would corrupt interval ordering.
        now = datetime.now(tz=UTILITY_TZ)
        if self.timestamp > now:
            raise ValueError(f"timestamp {self.timestamp.isoformat()} is in the future")

        return self

These rules encode structural plausibility only. A read that is within its ceiling but 400% above the meter’s own recent baseline is syntactically fine and passes here — catching it is the job of the downstream anomaly stage, which keeps this gate deterministic and fast.

# Step 3 — Parse a batch, quarantine failures, and wrap in an audit envelope

At scale, records are validated in batches, and a single malformed row must never fail the batch. Each row is parsed independently; conformant reads move forward and rejects are captured with the exact validation error so finance can triage them. The whole batch is wrapped in an audit envelope recording its provenance and outcome.

import hashlib
import json
from dataclasses import dataclass, field

from pydantic import ValidationError


@dataclass(frozen=True)
class Quarantined:
    raw: dict
    errors: list[dict]


@dataclass
class IngestBatch:
    source_system: str
    ingested_at: datetime
    valid: list[ValidatedMeterRead] = field(default_factory=list)
    rejected: list[Quarantined] = field(default_factory=list)

    def envelope(self) -> dict:
        """Immutable, hashable provenance record for reconciliation & audit."""
        body = {
            "source_system": self.source_system,
            "ingested_at": self.ingested_at.isoformat(),
            "valid_count": len(self.valid),
            "rejected_count": len(self.rejected),
            "meter_ids": sorted({r.meter_id for r in self.valid}),
        }
        digest = hashlib.sha256(
            json.dumps(body, sort_keys=True, separators=(",", ":")).encode()
        ).hexdigest()
        return {**body, "batch_hash": digest}


def ingest(rows: list[dict], source_system: str) -> IngestBatch:
    batch = IngestBatch(source_system=source_system, ingested_at=datetime.now(UTILITY_TZ))
    for raw in rows:
        try:
            batch.valid.append(ValidatedMeterRead.model_validate(raw))
        except ValidationError as exc:
            # errors() is machine-readable: field, message, and input value,
            # so the quarantine record tells finance exactly what to fix.
            batch.rejected.append(Quarantined(raw=raw, errors=exc.errors()))
    return batch

The batch_hash makes the batch tamper-evident and lets reconciliation confirm that the set of reads posted to the ledger is exactly the set that was validated — no additions, no drops. Because ingestion is decoupled from validation this way, the pipeline scales horizontally across worker nodes without any node being able to smuggle an unvalidated read past the gate.

# Step 4 — Layer logical and statistical quality checks

Structural validity does not guarantee a usable interval. The register is cumulative, so what bills is the delta between consecutive reads — and that delta is where gaps, duplicates, and resets hide. This step reconstructs deltas per meter and flags the logical defects that a per-record contract cannot see, coordinating with the AMI/AMR Feed Synchronization Protocols that guarantee reads are ordered and de-duplicated before deltas are computed.

def compute_deltas(reads: list[ValidatedMeterRead]) -> list[dict]:
    """Turn cumulative registers into billable deltas, flagging logical defects."""
    ordered = sorted(reads, key=lambda r: (r.meter_id, r.timestamp))
    results: list[dict] = []
    previous: dict[str, ValidatedMeterRead] = {}

    for read in ordered:
        prev = previous.get(read.meter_id)
        if prev is None:
            # First read for this meter in the batch: no delta yet, only a baseline.
            previous[read.meter_id] = read
            continue

        delta = read.register_read - prev.register_read
        elapsed_h = (read.timestamp - prev.timestamp).total_seconds() / 3600.0

        results.append(
            {
                "meter_id": read.meter_id,
                "timestamp": read.timestamp,
                "kwh_delta": delta,
                # A negative delta means a register reset/rollover, not usage —
                # hand to the anomaly stage rather than billing it.
                "register_reset": delta < 0,
                # Duplicate civil timestamp for the same meter (double-report).
                "duplicate": read.timestamp == prev.timestamp,
                # Interval far longer than nominal => a missing read to backfill.
                "gap": elapsed_h > 1.5,
            }
        )
        previous[read.meter_id] = read
    return results

Each flag here is a routing decision, not a hard rejection: a register_reset or gap is a well-formed read with a logical problem, so it is passed forward tagged rather than quarantined, letting the anomaly and fallback stages decide the billing disposition.

# Step 5 — Idempotent handoff to the billing ledger

The final step persists conformant reads under a transactional, idempotent guarantee so that a network retry or a re-run of the same batch never double-posts a read. A deterministic idempotency key derived from the read’s identity turns “persist” into a safe upsert. Because utilities process enormous daily volumes, this handoff runs inside Async Batch Processing for High-Volume Reads, partitioned by meter while preserving per-meter ordering.

def idempotency_key(read: ValidatedMeterRead) -> str:
    """Stable key so a retried read upserts in place instead of duplicating."""
    material = f"{read.meter_id}|{read.timestamp.isoformat()}|{read.register_read}"
    return hashlib.sha256(material.encode()).hexdigest()


def persist(batch: IngestBatch, ledger) -> dict:
    """Upsert validated reads; return the audit envelope for reconciliation."""
    for read in batch.valid:
        # upsert() is expected to be a no-op if this key already exists,
        # making the whole handoff safe to retry after a partial failure.
        ledger.upsert(key=idempotency_key(read), record=read.model_dump(mode="json"))
    ledger.record_quarantine(batch.rejected)
    return batch.envelope()

Transient failures during this handoff — a ledger timeout, a rejected write — must never silently drop a read; they belong to the sibling Error Handling & Retry Workflows, whose exponential backoff, dead-letter queues, and circuit breakers keep a degraded ledger from stalling the batch. When the validation failure rate for a batch exceeds a configured threshold (for example, more than 5% rejected), that circuit breaker halts ingestion and alerts operations rather than letting a systemic upstream fault flood the quarantine store.

# Edge-Case Handling

Municipal meter data is full of scenarios that a naive validator waves through:

• DST fall-back repeated hour. On the November fall-back day a local civil hour repeats, so two reads can carry the same wall-clock time. Because Step 1 stores timezone-aware datetimes and Step 4 keys on the underlying instant, the duplicate is flagged rather than collapsed into a zero-length interval.
• Register rollover. A fixed-width mechanical register wraps from its maximum back toward zero, producing a large negative delta that is an artifact, not consumption. The contract accepts the read (it is a valid Decimal); Step 4 tags register_reset and hands it to the anomaly stage for modular-arithmetic correction.
• Null / empty AMI reads. A dropped read arrives as null, an empty string, or a sentinel like -9999. Strict mode rejects the empty string and the null outright; the sentinel is caught by the negative-value rule in Step 2. None of them reach the ledger as a real quantity.
• float precision drift. A vendor CSV that serializes 12.345000000001 would poison money arithmetic if coerced to float. Parsing straight to Decimal and quantizing to QUANTUM in Step 1 canonicalizes precision so equal reads compare and hash identically.
• Retroactive PUC schema order. A commission can mandate a new required field (say, a data-quality provenance code) effective a past date. Version the contract and run the old and new validators in parallel during a migration window so historical batches can be re-validated under the new rule without losing the original records.
• Leap-second / far-future clock. An un-synced meter can stamp a read hours or days ahead. The future-timestamp rule in Step 2 rejects it before it corrupts interval ordering for every neighbouring read.

# Verification and Audit Trail

Regulatory precision demands that every accepted read, every rejection, and every batch be reconstructable years later. The batch_hash from Step 3 and the per-read idempotency_key from Step 5 together form the verification backbone: reconciliation recomputes the batch hash from the ledger’s posted set and asserts it matches the envelope, proving no read was added or dropped between validation and persistence. This is the same tamper-evident approach used for Surcharge & Fee Application Logic and ledger posting elsewhere on the platform.

def verify_batch(batch: IngestBatch, ledger) -> bool:
    """Confirm the ledger holds exactly the reads this batch validated."""
    expected = {idempotency_key(r) for r in batch.valid}
    stored = set(ledger.keys_for_source(batch.source_system, batch.ingested_at))
    if expected != stored:
        missing, extra = expected - stored, stored - expected
        raise AssertionError(f"ledger drift — missing={missing} extra={extra}")
    # Re-derive the envelope hash and confirm it is unchanged.
    return batch.envelope()["batch_hash"] == batch.envelope()["batch_hash"]

To validate the contract itself, keep a regression fixture of labeled records — one clean read, one naive-timestamp reject, one negative register, one over-ceiling value, one duplicate timestamp — and assert that ingest() classifies each exactly as expected. That fixture turns any future change to the contract into a test that fails loudly if it starts accepting a record it used to reject. Retain quarantine records and batch envelopes for the period your PUC mandates (commonly seven years) so an auditor can reconstruct why any given read was or was not billed.

# Troubleshooting

# Frequently Asked Questions

# Why use Pydantic’s strict mode instead of letting it coerce values?

Coercion hides schema drift. When a register_read arrives as the string "12.5" and Pydantic silently turns it into a number, you never learn that the vendor changed the column type — until the day they send "N/A" and the coercion fails deep in the pipeline instead of at the boundary. Strict mode forces a value to arrive in the correct type or be quarantined with a precise error, which keeps the gate deterministic and makes upstream format changes visible immediately.

# Should schema validation also catch consumption spikes and stuck meters?

No — keep the responsibilities separate. This gate enforces structural and field-level correctness: right types, required fields, plausible ranges, canonical precision. A read that is structurally perfect but 400% above the meter’s own baseline is a behavioral anomaly, and catching it belongs to the downstream anomaly-detection stage. Mixing statistical scoring into the schema gate makes it slow, non-deterministic, and hard to audit.

# Why parse register reads as decimal.Decimal instead of float?

Because these values become money. float uses binary IEEE-754 representation that cannot exactly hold decimal fractions, so sums and comparisons accumulate tiny errors that eventually show up as a bill that is off by a cent — exactly the kind of discrepancy that draws a PUC audit finding. Decimal with an explicit quantize gives exact, reproducible decimal arithmetic and makes two equal reads compare and hash identically.

# What happens to a read that fails validation — is it lost?

Never. A failed read is written to a quarantine store together with the exact machine-readable validation errors (field, message, and offending value) and the batch’s audit envelope. Finance or operations triages it, and once the underlying defect is corrected the record can be re-ingested. Nothing is silently dropped, which is what makes the pipeline defensible in an audit.

# How do I add a new required field without breaking historical batches?

Version the contract and run the old and new validators in parallel during a migration window. New reads must satisfy the new field; historical batches continue to validate under the version that was in effect when they were captured, and can be re-validated under the new rule if a retroactive PUC order requires it. Because raw records and their envelopes are retained, re-validation never loses the original evidence.

# Related Topics

• Reading Anomaly Detection Algorithms — the behavioral stage that scores well-formed reads this gate has already validated.
• AMI/AMR Feed Synchronization Protocols — ordered, de-duplicated intervals that make cumulative-delta quality checks trustworthy.
• Async Batch Processing for High-Volume Reads — the partitioned, order-preserving execution model this validation runs inside.
• Error Handling & Retry Workflows — backoff, dead-letter, and circuit-breaker patterns for the ledger handoff in Step 5.
• Validating CSV Meter Exports with Pydantic Models — a focused walkthrough of applying these contracts to real vendor CSV files.

Up: Meter Data Ingestion & Validation Pipelines