AMI Batch vs Streaming Ingestion Tradeoffs

Every municipal utility that moves off manual reads eventually faces the same architectural fork: should the billing system pull AMI meter data on a schedule, or should the head-end push each read as an event the moment it lands? Vendors sell streaming as the modern default, and for a prepay or outage-detection product it is. But most municipal billing runs monthly, and a monthly cycle rewards the operational simplicity of a scheduled batch far more than it rewards sub-second latency. Choosing the wrong topology means either paying for always-on streaming infrastructure you read from twelve times a year, or bolting real-time guarantees onto a batch that was never designed for them. This page sits under AMI/AMR feed synchronization, part of the broader Meter Data Ingestion & Validation Pipelines subsystem, and gives you a side-by-side decision framework, a skeleton for each style, and a recommendation you can defend in a budget meeting.

# Minimal Prerequisites

• Python 3.11+ — for zoneinfo in the standard library, X | None unions, and asyncio.TaskGroup in the streaming skeleton.
• A head-end contract you can read — either a pollable export endpoint (batch) or a durable topic/queue subscription with an offset or acknowledgement model (streaming). You need to know which one the vendor actually offers before you design.
• decimal.Decimal for every metered quantity — never float, in either topology. The ingestion style changes when a read arrives, never how consumption is represented.
• zoneinfo (not pytz) — batch windows and streaming watermarks are both defined in the utility’s civil time, so the window that closes at midnight is a wall-clock boundary, not a UTC one.

Data assumptions: each read carries a stable meter_id, a timezone-aware timestamp, a Decimal register value, and a monotonically increasing per-meter sequence number the head-end assigns. That sequence number is what makes ordering and deduplication tractable in both designs.

# Annotated Implementation

The comparison below is the whole decision in one table. Read it top to bottom, then look at the two skeletons — they exist to show how differently the same read is handled, not to be copied wholesale.

The batch skeleton is a pull: it names a closed window, asks the head-end for everything in it, sorts deterministically, and hands a bounded list to validation.

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta
from decimal import Decimal
from zoneinfo import ZoneInfo

UTILITY_TZ = ZoneInfo("America/Chicago")


@dataclass(frozen=True)
class Read:
    meter_id: str
    seq: int                 # head-end per-meter sequence number
    read_ts: datetime        # timezone-aware
    register: Decimal        # cumulative register value, never float


def batch_window(day: datetime) -> tuple[datetime, datetime]:
    # Window boundaries are civil-time midnights, not UTC midnights.
    start = day.astimezone(UTILITY_TZ).replace(hour=0, minute=0, second=0, microsecond=0)
    return start, start + timedelta(days=1)


def pull_batch(head_end, day: datetime) -> list[Read]:
    """Scheduled pull: fetch a closed window, then order deterministically."""
    start, end = batch_window(day)
    raw = head_end.export(start=start, end=end)          # one bounded request
    reads = [Read(**r) for r in raw]
    # A complete window means we can sort the entire set before anything downstream runs.
    reads.sort(key=lambda r: (r.meter_id, r.seq))
    return reads

The streaming skeleton is a push: an async consumer receives events as they arrive, gates each behind per-meter ordering, and acknowledges only after the read is durably handed off.

from __future__ import annotations

import asyncio
from decimal import Decimal


async def consume_stream(subscription, handoff) -> None:
    """Push consumer: process each event on arrival, preserving per-meter order."""
    last_seq: dict[str, int] = {}          # highest seq handed off per meter
    async with asyncio.TaskGroup() as tg:
        async for event in subscription:   # durable topic; ack after handoff
            read = Read(**event.payload)
            prev = last_seq.get(read.meter_id, 0)
            if read.seq <= prev:
                await event.ack()          # duplicate or late — already processed
                continue
            if read.seq != prev + 1:
                # Gap: an earlier read has not arrived. Park, don't process out of order.
                await subscription.park(read.meter_id, event)
                continue
            last_seq[read.meter_id] = read.seq
            tg.create_task(_handoff_then_ack(handoff, read, event))


async def _handoff_then_ack(handoff, read: Read, event) -> None:
    await handoff.write(read)              # idempotent, keyed on (meter_id, seq)
    await event.ack()                      # ack only after durable write

The two skeletons differ in exactly one way that matters: the batch job owns a complete set before it does anything, so ordering and deduplication are a sort and a group-by. The streaming consumer never sees the whole set, so it has to carry per-meter state (last_seq) and make a decision — process, drop, or park — on every single event with no guarantee about what comes next. That state is the source of most streaming operational cost. Both paths converge on the same downstream: the validated, ordered reads feed the same schema validation and data quality checks regardless of how they arrived.

# Recommendation

For monthly municipal billing, a scheduled batch pull is the right default — and usually the only thing you need. The billing cycle is inherently periodic, so latency below “before the next cycle runs” has no business value, and the batch’s native replay makes reconciliation and PUC-mandated re-rates a re-run rather than an engineering project. You get deterministic ordering for free, one place to catch a failed window, and a bill that near-zero infrastructure between runs.

Reach for streaming only when a downstream product genuinely consumes reads in real time — prepay accounts that must disconnect on a zero balance, demand-response events, or outage detection. Even then, keep billing on the batch: run streaming for the real-time surface and let the same reads settle into the nightly batch for the authoritative ledger. That hybrid gives you real-time where it pays and keeps the audited money path simple. Do not adopt streaming for its own sake; the per-meter ordering state and always-on brokers are a permanent operational tax that a monthly cycle never earns back.

# Edge Cases and Billing Gotchas

• Late and out-of-order reads in streaming. A meter on a flaky backhaul can deliver seq=8 before seq=7. Processing them in arrival order corrupts the register delta. The consumer must park the gap (as the skeleton does) or hold a watermark, and never rate a read until its predecessor is durably in — otherwise a transient network reorder produces a permanently wrong bill.
• Batch window gaps. If a nightly job silently skips a window — the trigger misfired, the head-end was down — a full day of reads vanishes until someone notices a low bill. Track the last successfully closed window and alert on any gap, then backfill the missing range; a missing read is a job for fallback routing for missing rate data, never a silent zero.
• Replay and backfill double-counting. Both topologies re-deliver: a batch re-run for a corrected date range, a streaming offset rewound after a consumer bug. Without an idempotency key on (meter_id, seq) the handoff appends duplicates and inflates consumption. Make the ledger write idempotent so any replay is safe to run twice.
• Deduplication across the boundary. In a hybrid deployment the same physical read arrives once by stream and once by batch. Dedup on the head-end’s stable (meter_id, seq) pair rather than arrival time, so the batch settlement overwrites the streamed provisional value instead of adding to it.

# Verification Snippet

Assert the two properties that distinguish the topologies: the batch always emits a fully ordered set, and the streaming consumer refuses to process an out-of-order read until its predecessor arrives.

from decimal import Decimal
from datetime import datetime
from zoneinfo import ZoneInfo

TZ = ZoneInfo("America/Chicago")


def test_batch_is_fully_ordered() -> None:
    reads = [
        Read("MTR-2", 2, datetime(2026, 7, 1, 3, tzinfo=TZ), Decimal("120.00")),
        Read("MTR-1", 1, datetime(2026, 7, 1, 1, tzinfo=TZ), Decimal("50.00")),
        Read("MTR-1", 2, datetime(2026, 7, 1, 2, tzinfo=TZ), Decimal("55.00")),
    ]
    reads.sort(key=lambda r: (r.meter_id, r.seq))
    keys = [(r.meter_id, r.seq) for r in reads]
    assert keys == [("MTR-1", 1), ("MTR-1", 2), ("MTR-2", 2)]  # total order


def test_streaming_parks_out_of_order() -> None:
    # A read whose predecessor has not arrived must not advance last_seq.
    last_seq: dict[str, int] = {"MTR-1": 1}
    incoming_seq = 3                         # seq 2 is still missing
    prev = last_seq["MTR-1"]
    assert incoming_seq != prev + 1          # gap detected → park, do not process
    assert last_seq["MTR-1"] == 1            # state unchanged until the gap fills

The first test proves a complete window can always be totally ordered before anything downstream runs — the property that makes batch simple. The second proves the streaming consumer holds the line on ordering, the property that makes streaming correct but stateful.

# FAQ

# Related Topics

• AMI/AMR Feed Synchronization Protocols — the parent guide on exactly-once transport and idempotent handoff for both topologies.
• Meter Data Ingestion & Validation Pipelines — the end-to-end subsystem these ingestion styles feed.
• Async Batch Processing for High-Volume Reads — how to scale the batch path across 100k+ daily reads without exhausting resources.
• Schema Validation & Data Quality Checks — the shared validation gate both batch and streaming reads pass through.
• Error Handling & Retry Workflows — retries, backoff, and dead-letter routing that both topologies depend on.

Up: AMI/AMR Feed Synchronization Protocols · Meter Data Ingestion & Validation Pipelines