Real-Time vs Batch Ingestion for Automated Financial Reconciliation

Choosing between streaming and batch ingestion is the first irreversible decision in a reconciliation pipeline, and it is invoked before any matching logic runs. The mode you select governs how quickly a transaction reaches the ledger, how out-of-order delivery is tolerated, and how much determinism survives into the audit trail. This stage sits immediately downstream of the bank connection layer described in Core Architecture & Bank Feed Ingestion and immediately upstream of the matching cascade in Transaction Matching Algorithms & Logic: every payload that crosses it must already carry a deterministic source_hash and idempotency_key, because both modes converge on the same canonical schema and the same double-posting guarantees. Get the mode selection wrong and you either pay unbounded latency on settlement-critical flows or import the full complexity of distributed ordering onto records that never needed it.

The engineering question is therefore not “streaming or batch” as a religion, but a deterministic routing decision per source, per account, and per reconciliation SLA. High-velocity card authorizations want a streaming path with watermark windowing; end-of-day settlement files want a batch path with snapshot delta comparison. This page defines the mechanism for both, gives a production-grade router with audit logging, and enumerates the failure codes each path emits.

Prerequisites: Pipeline State Before Mode Selection

Mode selection is only safe once the ingestion boundary has already established a normalised contract. Before the router executes, the following upstream state must hold:

• A canonical, typed payload. Raw OFX, MT940, ISO 20022 camt.053, and CSV variants must already be decoded into the canonical transaction model. The deterministic tokenisation that produces these fields is owned by OFX & MT940 Parser Design; the router never parses, it only routes typed records.
• Monetary values as Decimal. No float may reach this stage. FX-bearing records additionally require the conversion-matrix lineage from Multi-Currency Ledger Mapping so that streamed and batched records reconcile against the same signed spot rate.
• A live, rotated credential. Both paths authenticate continuously against the institution; the token lifecycle and circuit-breaker behaviour are defined in Secure API Token Management, and streaming consumers must additionally respect the pull cadence documented in Best Practices for Handling Bank API Rate Limits.
• A deterministic idempotency key per record. Computed once at the boundary from tx_id, normalised amount, currency, and value timestamp, this key is what makes a retried stream event and a re-read batch row collapse to a single ledger posting.

If any of these are missing, the router must reject the payload rather than guess — a partially-normalised record routed into a streaming partition corrupts ordering for every record behind it.

Mechanism Deep-Dive: Streaming Windows vs Batch Boundaries

Streaming ingestion

Real-time ingestion is an event-driven topology. Transactions are consumed immediately from a distributed log (Kafka, Kinesis) as they arrive from webhook endpoints or polled APIs, then admitted to the matching engine the moment ordering can be proven. The hard problem is not throughput; it is that financial events arrive out of order across network paths while reconciliation demands a strict per-account sequence.

The mechanism that resolves this is watermark windowing. Each partition tracks a monotonically advancing watermark W — the timestamp below which no further events are expected. An event with value time t is admissible when t >= W - allowed_lateness; anything older is diverted, never silently dropped. Within a partition, a sequence validator enforces that each record’s sequence strictly exceeds the last committed sequence for that tx_id lineage, which is what prevents cross-account interleaving. Partitions must be keyed on the bank account identifier (or settlement currency) so that ordering is preserved within each ledger context; key on anything coarser and two accounts share a watermark and corrupt each other’s sequence.

Complexity is O(1) amortised admission per event, but the latent cost is state: every partition holds a sequence cache and a deferred-resolution queue for partial matches awaiting their counterpart. Streaming matching is therefore inherently probabilistic — a debit may stream in seconds before its settlement leg, so the engine admits the partial and reconciles asynchronously, exactly the deferred-resolution behaviour that feeds Multi-Step Reconciliation Chains.

Batch ingestion

Batch ingestion aggregates transactions over a fixed window (hourly, daily, or aligned to the settlement cycle) and processes them as an immutable file against partitioned workers. The window boundary is the source of its determinism: the set of records is closed before processing begins, so the matcher can run a deterministic multi-pass scan and a snapshot delta against the prior window. This is O(n log n) over the window for the sort-merge join, but every pass is reproducible — re-running the same window yields byte-identical results, which is precisely what an auditor wants to replay.

The financial-domain caveat is window-edge transactions. A payment posted at 23:59:58 and its fee at 00:00:03 straddle the boundary and will appear in different windows; the batch reconciler must carry an overlap region (a small look-back into the previous window’s unmatched tail) or it will manufacture a phantom exception every night. This is the batch-side analogue of streaming’s allowed_lateness.

Production-Grade Python Implementation

The router below admits a canonical payload, decides its mode, enforces idempotency and per-partition ordering for the streaming path, and emits a structured audit record on every decision. It uses pydantic for the typed contract, Decimal for money, and a structlog-style logger that emits trace_id, source_hash, and match_decision for every branch so that no record changes state without an audit line.

from __future__ import annotations

import hashlib
import logging
from decimal import Decimal, getcontext
from enum import Enum
from typing import Protocol

from pydantic import BaseModel, ConfigDict, Field

getcontext().prec = 28
audit_log = logging.getLogger("reconciliation.ingestion")


class IngestionMode(str, Enum):
    STREAM = "stream"
    BATCH = "batch"


class RouteDecision(str, Enum):
    ADMITTED = "ADMITTED"
    DEFERRED = "DEFERRED"
    REJECTED = "REJECTED"
    DUPLICATE = "DUPLICATE"


class TransactionPayload(BaseModel):
    model_config = ConfigDict(frozen=True)

    tx_id: str
    account_id: str
    amount: Decimal
    currency: str = Field(min_length=3, max_length=3)
    value_ts: int            # epoch millis, normalised to UTC upstream
    sequence: int
    source_hash: str
    realtime_capable: bool


class StateStore(Protocol):
    def last_sequence(self, account_id: str) -> int: ...
    def watermark(self, account_id: str) -> int: ...
    def seen(self, idem_key: str) -> bool: ...
    def commit(self, payload: "TransactionPayload", idem_key: str) -> None: ...


class DeferredQueue(Protocol):
    def push(self, payload: "TransactionPayload", reason: str) -> None: ...


class IngestionRouter:
    """Deterministic stream/batch router with idempotent admission."""

    def __init__(
        self,
        store: StateStore,
        dlq: DeferredQueue,
        *,
        allowed_lateness_ms: int = 5_000,
        realtime_amount_ceiling: Decimal = Decimal("250000.00"),
    ) -> None:
        self._store = store
        self._dlq = dlq
        self._allowed_lateness_ms = allowed_lateness_ms
        self._ceiling = realtime_amount_ceiling

    @staticmethod
    def _idempotency_key(p: TransactionPayload) -> str:
        canonical = f"{p.tx_id}:{p.amount.normalize()}:{p.currency}:{p.value_ts}"
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    def select_mode(self, p: TransactionPayload) -> IngestionMode:
        # High-value or non-realtime sources settle through the deterministic
        # batch path; only low-risk, streaming-capable sources go real-time.
        if not p.realtime_capable or abs(p.amount) > self._ceiling:
            return IngestionMode.BATCH
        return IngestionMode.STREAM

    def route(self, p: TransactionPayload, trace_id: str) -> RouteDecision:
        idem_key = self._idempotency_key(p)
        mode = self.select_mode(p)
        decision = self._evaluate(p, idem_key, mode)

        audit_log.info(
            "ingestion.route",
            extra={
                "trace_id": trace_id,
                "source_hash": p.source_hash,
                "idempotency_key": idem_key,
                "account_id": p.account_id,
                "mode": mode.value,
                "match_decision": decision.value,
            },
        )
        return decision

    def _evaluate(
        self, p: TransactionPayload, idem_key: str, mode: IngestionMode
    ) -> RouteDecision:
        if self._store.seen(idem_key):
            return RouteDecision.DUPLICATE

        if mode is IngestionMode.STREAM:
            wm = self._store.watermark(p.account_id)
            if p.value_ts < wm - self._allowed_lateness_ms:
                self._dlq.push(p, reason="LATE_ARRIVAL_REJECTED")
                return RouteDecision.REJECTED

            last_seq = self._store.last_sequence(p.account_id)
            if p.sequence <= last_seq:
                self._dlq.push(p, reason="OUT_OF_ORDER_SEQUENCE")
                return RouteDecision.REJECTED

        self._store.commit(p, idem_key)
        return RouteDecision.ADMITTED

Every return path is reachable by an auditor: a DUPLICATE proves the idempotency guard fired, a REJECTED carries its diversion reason into the deferred queue, and an ADMITTED record is now owned by the matcher. The same idem_key derivation is reused verbatim by the batch path, which is what lets a streamed event and a re-read settlement row collapse to one posting.

Configuration Rules and Threshold Calibration

These parameters are the dials that move a source between latency and determinism. Start at the recommended values and tune against observed lateness and exception rates, not in the abstract.

Multi-Dimensional Validation

Mode selection is necessary but never sufficient — an admitted record still has to satisfy the matching constraints downstream, and the two paths must converge on identical answers. A streamed record and its batched counterpart are reconciled against the same composite of checks: deterministic identity via Exact Match & Hash Comparison, temporal and value tolerance via Date-Window & Amount Tolerance Rules, and descriptor similarity when references are noisy. The router’s job is to guarantee that, whichever path a record took, it arrives at this composite with the same source_hash, the same signed Decimal amount, and the same UTC value_ts. If streaming applied allowed_lateness while batch applied batch_overlap_ms, the two windows must be calibrated so a record on the boundary lands in exactly one — divergent tolerance bands between the paths are the most common cause of a “matches in batch, exception in stream” discrepancy.

Async and High-Throughput Execution

Streaming partitions are consumed concurrently, one bounded worker per partition, so that a slow account cannot stall the others. Backpressure is explicit: an asyncio.Queue with maxsize caps in-flight records, and await queue.put(...) blocks the consumer when the matcher falls behind rather than letting memory grow unbounded. Batch chunks fan out the same way over an object-storage prefix, but with a fixed worker pool sized to the window rather than to live traffic.

import asyncio
import logging
from typing import AsyncIterator

audit_log = logging.getLogger("reconciliation.ingestion")


async def consume_partition(
    account_id: str,
    events: AsyncIterator[TransactionPayload],
    router: IngestionRouter,
    sink: asyncio.Queue[TransactionPayload],
    trace_id: str,
) -> None:
    """One worker per partition preserves per-account ordering under load."""
    admitted = 0
    async for payload in events:
        decision = router.route(payload, trace_id=trace_id)
        if decision is RouteDecision.ADMITTED:
            await sink.put(payload)        # backpressure: blocks when matcher lags
            admitted += 1

    audit_log.info(
        "ingestion.partition.drained",
        extra={
            "trace_id": trace_id,
            "source_hash": account_id,        # partition lineage key
            "account_id": account_id,
            "admitted": admitted,
            "match_decision": "PARTITION_DRAINED",
        },
    )


async def run_consumers(
    partitions: dict[str, AsyncIterator[TransactionPayload]],
    router: IngestionRouter,
    sink: asyncio.Queue[TransactionPayload],
    trace_id: str,
) -> None:
    async with asyncio.TaskGroup() as tg:
        for account_id, events in partitions.items():
            tg.create_task(
                consume_partition(account_id, events, router, sink, trace_id)
            )

The non-negotiable rule is one worker per ordering key. Fan out two coroutines over the same account_id and you reintroduce exactly the interleaving the sequence validator exists to prevent. Throughput scales by adding partitions, never by parallelising within one.

Failure Modes Specific to This Stage

Compliance and Audit Trail Requirements

Both paths must satisfy SOC 2 Type II, GAAP/IFRS recognition, and data-residency mandates, and they do so by emitting the same evidence regardless of mode. Every routing decision — admit, defer, reject, duplicate — appends an immutable, append-only event carrying trace_id, source_hash, idempotency_key, the selected mode, and the match_decision, cryptographically chained to its predecessor. Deterministic reprocessing is the load-bearing guarantee: batch snapshots and streaming watermarks together let an auditor reconstruct the exact ledger state at any point in time and replay it to byte-identical output. Ingestion workers, the matcher, and the ledger committer run under distinct IAM roles so that no single principal can both admit and post a record, and every transaction carries a provenance_chain naming its parser version, the ingestion mode, and the normalisation rules applied. Records that fall out of either path land in the human review workflow described under Exception Routing & Human-in-the-Loop Workflows, where the same audit line follows the item through resolution. For FX-bearing records, the signed spot rate and its versioned matrix must appear in the audit emission so that a streamed conversion and a batched conversion are provably identical; the precision rules behind that are the Decimal contexts documented in Python Decimal Contexts.

Related

• OFX & MT940 Parser Design
• Multi-Currency Ledger Mapping
• Secure API Token Management
• Exact Match & Hash Comparison
• Multi-Step Reconciliation Chains

Part of Core Architecture & Bank Feed Ingestion.