Validating CSV Meter Exports with Pydantic Models

Vendor CSV exports promise clean columns and deliver edge cases. Automated metering infrastructure (AMI) and automated meter reading (AMR) vendors routinely ship files that deviate from their own published specification — a truncated decimal, a thousands separator, a naive timestamp, a null spelled N/A. A single malformed cell that slips past ingestion cascades into wrong tiered charges, revenue leakage, and audit findings, so the fix belongs at the boundary: reject or coerce every row before it reaches the ledger. This page sits under Schema Validation & Data Quality Checks, part of the broader Meter Data Ingestion & Validation Pipelines subsystem, and shows how a single Pydantic v2 model turns an untrusted CSV row into a typed, decimal-safe, deduplicated record — or diverts it to a dead-letter queue with a precise reason.

# Minimal Prerequisites

• Python 3.11+ — for zoneinfo in the standard library and X | None union syntax.
• Pydantic v2 (pydantic>=2.6) — the model below uses v2 field_validator / model_validator semantics; v1 will not parse it.
• decimal.Decimal for all usage math — never float. IEEE-754 rounding drift compounds across thousands of interval reads and corrupts tiered-rate boundaries.
• zoneinfo (not pytz) — the utility’s civil time zone resolves which billing day a late-night read lands on.

Data assumptions: rows arrive from a csv.DictReader as dict[str, str], one dict per line, with a header the vendor may reorder or extend between drops. Every value is a string until this model coerces it. Nothing downstream should ever see a raw CSV field.

# Annotated Implementation

The whole contract lives in one model. Field constraints reject out-of-range values declaratively; mode="before" validators sanitize vendor formatting quirks before type coercion; a model_validator enforces cross-field billing logic that no single field can express on its own.

from __future__ import annotations

from datetime import datetime
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from enum import Enum
from zoneinfo import ZoneInfo
import hashlib

from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

UTILITY_TZ = ZoneInfo("America/Chicago")  # the utility's civil time zone
TWO_PLACES = Decimal("0.01")
NULL_TOKENS = {"", "null", "na", "n/a", "none", "-"}


class MeterType(str, Enum):
    RESIDENTIAL = "RES"
    COMMERCIAL = "COM"
    INDUSTRIAL = "IND"


class MeterRead(BaseModel):
    # Reject unexpected columns instead of silently ignoring a vendor schema drift.
    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")

    meter_id: str = Field(pattern=r"^MTR-\d{8}$")           # municipal asset identifier
    read_ts_utc: datetime                                   # normalized to UTC on the way in
    consumption_kwh: Decimal = Field(ge=0, le=Decimal("99999.99"))
    demand_kw: Decimal | None = Field(default=None, ge=0, le=Decimal("5000.00"))
    meter_type: MeterType
    status_flag: int = Field(ge=0, le=4)                    # 0 normal .. 4 tamper/error
    source_system: str = "AMI_EXPORT_V2"

    @field_validator("read_ts_utc", mode="before")
    @classmethod
    def normalize_timestamp(cls, v: object) -> object:
        # Vendors mix ISO 8601, US locale, and naive local time. Coerce all to UTC.
        if not isinstance(v, str):
            return v
        raw = v.strip()
        for fmt in ("%Y-%m-%dT%H:%M:%S%z", "%m/%d/%Y %H:%M:%S", "%Y-%m-%d %H:%M:%S"):
            try:
                dt = datetime.strptime(raw, fmt)
            except ValueError:
                continue
            # A naive timestamp is the vendor's LOCAL time, not UTC — attach civil tz first.
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=UTILITY_TZ)
            return dt.astimezone(ZoneInfo("UTC"))
        raise ValueError(f"unrecognized timestamp format: {raw!r}")

    @field_validator("consumption_kwh", "demand_kw", mode="before")
    @classmethod
    def sanitize_decimal(cls, v: object) -> object:
        # Strip thousands separators and normalize the many spellings of "no value".
        if not isinstance(v, str):
            return v
        cleaned = v.replace(",", "").strip()
        if cleaned.lower() in NULL_TOKENS:
            return None
        try:
            # Quantize once, here, so the ledger never inherits float drift.
            return Decimal(cleaned).quantize(TWO_PLACES, rounding=ROUND_HALF_UP)
        except InvalidOperation:
            raise ValueError(f"invalid decimal for meter value: {v!r}")

    @model_validator(mode="after")
    def check_billing_invariants(self) -> "MeterRead":
        # A residential meter reporting demand with zero consumption is physically incoherent.
        if (
            self.meter_type is MeterType.RESIDENTIAL
            and self.demand_kw
            and self.consumption_kwh == 0
        ):
            raise ValueError("residential demand without consumption")
        return self

    @property
    def read_hash(self) -> str:
        # Deterministic idempotency key: re-syncing the same read never double-bills.
        key = f"{self.meter_id}|{self.read_ts_utc.isoformat()}|{self.consumption_kwh}"
        return hashlib.sha256(key.encode()).hexdigest()


def parse_rows(rows: list[dict[str, str]]) -> tuple[list[MeterRead], list[dict]]:
    """Validate a CSV batch. Good rows proceed; bad rows go to a dead-letter list."""
    valid: list[MeterRead] = []
    dead_letter: list[dict] = []
    for line_no, row in enumerate(rows, start=2):  # line 1 is the header
        try:
            valid.append(MeterRead(**row))
        except Exception as exc:  # pydantic.ValidationError carries per-field detail
            dead_letter.append({"line": line_no, "raw": row, "error": str(exc)})
    return valid, dead_letter

Three decisions carry the billing weight. extra="forbid" turns a silent vendor schema drift (an added or renamed column) into a loud rejection instead of a dropped field. Quantizing inside sanitize_decimal guarantees the ledger only ever sees two-decimal Decimal values, closing the door on the fractional-cent variance that reconciliation cannot later resolve. And attaching UTILITY_TZ to naive timestamps before converting to UTC prevents a whole-file offset error — a naive 23:30 read is local civil time, and treating it as UTC would misfile it against the wrong billing day for the same tiered rate calculation it feeds. The read_hash property is the idempotency contract that lets AMI/AMR feed synchronization re-deliver a batch without inflating consumption totals.

# Edge Cases and Billing Gotchas

Vendor data breaks in predictable, billing-specific ways. These are the ones that most often turn a “clean” import into a wrong bill.

• Thousands separators and vendor null tokens. A commercial read of 12,480.5 and a missing demand of N/A both arrive as strings. sanitize_decimal strips the comma, maps the null token to None, and quantizes — so 12,480.5 becomes Decimal("12480.50") rather than raising or truncating to 12.
• Naive vs. offset timestamps in one file. A single export can mix 2026-06-30T23:30:00-05:00 with a naive 06/30/2026 23:30:00. The offset-aware value converts directly; the naive one is stamped with the utility’s civil zone first. Both land on the correct UTC instant, so no read is misattributed across a day boundary.
• DST fall-back ambiguity. During the autumn repeated hour, a naive 01:30 is genuinely ambiguous in local time. zoneinfo resolves it deterministically, but reads inside that window should be cross-checked by reading anomaly detection algorithms for duplicate intervals before they price out.
• Negative or rollover registers. A meter that wraps past its maximum can emit a negative delta. The ge=0 bound rejects it at the field level so it is diverted to the dead-letter queue rather than laundered into a plausible bill — a genuinely absent read is instead handled by fallback routing for missing rate data, never invented here.
• Silent column drift. A vendor that inserts a new firmware_rev column will pass a positional parser and shift every downstream field by one. extra="forbid" catches this on the first row instead of after a full cycle posts against misaligned data.

# Verification Snippet

Validate the model against a fixture of known-bad rows and assert both that good rows survive and that each bad row lands in the dead-letter queue for the right reason.

def test_parse_rows_separates_valid_and_bad() -> None:
    rows = [
        # valid: comma separator + naive local timestamp
        {"meter_id": "MTR-00010001", "read_ts_utc": "06/30/2026 23:30:00",
         "consumption_kwh": "12,480.5", "demand_kw": "N/A",
         "meter_type": "COM", "status_flag": "0"},
        # bad: negative register (rollover corruption)
        {"meter_id": "MTR-00010002", "read_ts_utc": "2026-06-30T23:30:00-05:00",
         "consumption_kwh": "-4.0", "demand_kw": "1.0",
         "meter_type": "RES", "status_flag": "0"},
    ]
    valid, dead_letter = parse_rows(rows)

    assert len(valid) == 1
    assert valid[0].consumption_kwh == Decimal("12480.50")  # comma stripped, quantized
    assert valid[0].demand_kw is None                       # "N/A" -> None
    assert valid[0].read_ts_utc.tzinfo is not None          # timezone-aware, in UTC

    assert len(dead_letter) == 1
    assert dead_letter[0]["line"] == 3                      # header is line 1
    assert "consumption_kwh" in dead_letter[0]["error"]     # bound rejected the negative

    # Idempotency: the same read hashes identically across re-syncs.
    assert valid[0].read_hash == MeterRead(**rows[0]).read_hash

Running this against a fixture drawn from real historical exports — not synthetic data — is what surfaces vendor-specific quirks before a batch reaches production. Rows that repeatedly dead-letter for the same field point to a spec change worth escalating to the vendor.

# FAQ

# Related Topics

• Schema Validation & Data Quality Checks — the parent guide covering validation strategy across the ingestion tier.
• Meter Data Ingestion & Validation Pipelines — the end-to-end subsystem this validation boundary belongs to.
• Async Batch Processing for High-Volume Reads — scaling this per-row validation across millions of daily intervals.
• Error Handling & Retry Workflows — dead-letter routing, backoff, and circuit breakers for malformed feed drops.
• Reading Anomaly Detection Algorithms — statistical checks that run on the clean, bounded records this model emits.

Up: Schema Validation & Data Quality Checks · Meter Data Ingestion & Validation Pipelines