Async Batch Processing for Multi-Currency Shoots: Deterministic FX Rate Pinning and Dual-Currency Ledgers

When a production unit crosses three international jurisdictions in a single week, cost tracking stops being a spreadsheet exercise and becomes a distributed-systems problem. Line producers and production accountants are no longer just reconciling petty cash; they are absorbing asynchronous streams of vendor invoices, union payroll manifests, and location settlements that arrive in EUR, GBP, CAD, and USD at the same time. The specific edge case this page solves is narrow but unforgiving: how do you attach the correct foreign-exchange rate to each of those transactions inside a concurrent batch, so that the base-currency figure a completion-bond lender reads back six months later reproduces exactly, to the cent, no matter when the report is regenerated? Getting this wrong does not throw an exception — it silently drifts, and the drift is the first thing a guarantor’s auditor finds.

Prerequisites and Context

This page extends the concurrency model established in the parent guide, Async Batch Processing, and assumes you already have a semaphore-bounded worker pool fetching sources without blocking the event loop. Here we bolt currency normalization onto that pool as a per-record step that runs before any guild-rate math. It targets Python 3.11+ for standard-library zoneinfo, and leans on the same deliberate dependency set the rest of Cost Ingestion & Data Parsing Workflows uses: decimal for money, hashlib for the audit fingerprint, zoneinfo for timezone-aware timestamps, and Pydantic v2 (model_validate, field_validator) for the boundary schema. Rate application is treated as a pure function, so the FX layer needs no network access at batch time at all — only a pre-populated cache of daily snapshots.

The reference rate must come from an authoritative daily source. This implementation keys off the Federal Reserve H.10 foreign exchange rates release, which publishes one reference rate per currency per business day; the IRS publishes only yearly average rates, which cannot support transaction-level pinning and must never be used for a cost ledger. Whichever source you standardize on, the contract clause that matters is the bond agreement’s reconciliation covenant: every dollar reported must be traceable to the exact rate and rate-date that produced it. The FX values feeding this step come off the same transport described in CSV & API Sync Pipelines, and the schema-rejection semantics used when a rate is missing are the ones defined in Schema Validation & Error Handling.

Pinning a Deterministic Daily FX Rate

The primary failure mode when engineers first attempt this is not network latency — it is event-loop starvation caused by a synchronous FX lookup embedded directly inside the batch loop. If every transaction triggers a live HTTP call to a rate provider, a chunk of ten thousand line items becomes ten thousand blocking round-trips, and worse, the rate returned depends on when the batch happened to run rather than the transaction’s own date. The correct pattern inverts that entirely: load the day’s rates once into a date-keyed cache, then convert each record with a pure function that never touches the network.

Read the implementation as four responsibilities held strictly apart. DailyFXSnapshot is the immutable, frozen record of one reference rate, validated so a non-positive rate can never enter the cache. LedgerLine is the dual-currency output row — it preserves the original foreign amount and the derived base amount, plus the rate, source, and rate-date that link them. convert_to_base is the pure function: identical inputs on an identical date always yield an identical Decimal. pin_and_post ties them together, refusing to guess when a rate is absent.

from __future__ import annotations

import hashlib
import json
from datetime import date, datetime
from decimal import Decimal, ROUND_HALF_UP
from zoneinfo import ZoneInfo

from pydantic import BaseModel, ConfigDict, field_validator

# Each unit reconciles against its own operating day — an IANA zone, never a bare UTC offset.
UNIT_TZ = ZoneInfo("Europe/London")
BASE_CURRENCY = "USD"


class DailyFXSnapshot(BaseModel):
    """An immutable, date-keyed reference rate cached once at ingestion time."""

    model_config = ConfigDict(frozen=True)

    rate_date: date
    quote_currency: str            # e.g. "EUR"
    units_per_base: Decimal        # units of quote currency per 1 USD
    source: str                    # authoritative reference, e.g. "FRB_H10"

    @field_validator("units_per_base")
    @classmethod
    def rate_must_be_positive(cls, v: Decimal) -> Decimal:
        if v <= 0:
            raise ValueError("FX rate must be a positive Decimal")
        return v


class LedgerLine(BaseModel):
    """Dual-currency ledger row: original amount retained, base amount derived and pinned."""

    transaction_id: str
    txn_date: date
    original_currency: str
    original_amount: Decimal
    base_currency: str
    base_amount: Decimal
    fx_rate: Decimal
    fx_source: str
    fx_rate_date: date
    payload_sha256: str
    posted_at: datetime


def _fingerprint(payload: dict) -> str:
    # Canonical, sorted serialization yields a deterministic, auditable hash.
    canonical = json.dumps(payload, sort_keys=True, default=str, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def convert_to_base(amount: Decimal, snapshot: DailyFXSnapshot) -> Decimal:
    """Pure function: same amount + same snapshot always returns the same base figure."""
    # units_per_base is quote-per-USD, so a quote-currency amount / rate yields USD.
    converted = amount / snapshot.units_per_base
    return converted.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)


def pin_and_post(
    txn: dict,
    fx_cache: dict[tuple[date, str], DailyFXSnapshot],
) -> LedgerLine:
    txn_date = date.fromisoformat(txn["date"])
    currency = txn["currency"]
    # Decimal(str(...)) — construct from the string form so no float ever touches money.
    amount = Decimal(str(txn["amount"]))

    if currency == BASE_CURRENCY:
        rate, source, rate_date = Decimal("1"), "IDENTITY", txn_date
        base_amount = amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
    else:
        # Pin by the TRANSACTION'S date — never the batch run date, never a live lookup.
        snapshot = fx_cache.get((txn_date, currency))
        if snapshot is None:
            raise LookupError(f"No cached FX snapshot for {currency} on {txn_date}")
        rate, source, rate_date = snapshot.units_per_base, snapshot.source, snapshot.rate_date
        base_amount = convert_to_base(amount, snapshot)

    return LedgerLine(
        transaction_id=txn["transaction_id"],
        txn_date=txn_date,
        original_currency=currency,
        original_amount=amount,
        base_currency=BASE_CURRENCY,
        base_amount=base_amount,
        fx_rate=rate,
        fx_source=source,
        fx_rate_date=rate_date,
        payload_sha256=_fingerprint(txn),
        posted_at=datetime.now(UNIT_TZ),
    )

Because convert_to_base is pure and the snapshot is frozen, re-running the same batch a year later against the same cache produces byte-identical base_amount values — the property a lender review depends on. Hardcoding a fallback rate or letting a live API answer during batch processing reintroduces exactly the non-determinism this design exists to remove.

The flow below shows how each transaction is keyed by its date to a cached daily FX snapshot, converted to the base currency, and written to a dual-currency ledger that retains the original amount.

Pinning a cached daily FX rate per transaction, then writing a dual-currency ledger row A foreign-currency transaction carrying an EUR, GBP or CAD amount and its own transaction date is keyed by that transaction date and checked against the day-keyed FX snapshot cache. On a cache hit the exact daily reference rate — with its source and rate-date — is pinned, then converted to the base currency by a pure Decimal function using ROUND_HALF_UP, and written to an append-only dual-currency ledger that retains both the original and the base amount. On a cache miss the record is routed to a dead-letter reconciliation queue instead; no live rate lookup is ever performed at batch time. hit · valid miss Foreign-currency transaction EUR · GBP · CAD amount + txn_date Key by transaction date Daily FX snapshot cached for this date? Pin exact daily reference rate rate · source · rate_date Convert to base currency pure function · ROUND_HALF_UP Dual-currency ledger original + base amount · append-only Dead-letter queue reconciliation · no live lookups

Audit Trail Requirements

A pinned rate is only defensible if the ledger row carries its own proof. Every LedgerLine must persist, at minimum: the original_currency and original_amount exactly as received; the derived base_amount and the fx_rate, fx_source, and fx_rate_date that produced it; the payload_sha256 fingerprint of the raw transaction; and a timezone-aware posted_at. The fingerprint is the linchpin of the reconciliation story — it lets a guarantor’s auditor tie any base-currency figure back to the exact vendor payload and the exact rate snapshot, without manual reconstruction, and it makes re-ingestion idempotent because a corrected payload that reproduces the same canonical bytes reproduces the same hash.

Two storage rules are non-negotiable. First, the ledger is append-only: an FX correction is a new compensating line, never an in-place edit of a posted row, so the historical figure a report was built on remains recoverable. Second, the audit record is written to write-once storage before the ledger transaction commits, so a crash mid-batch leaves a replayable record of intent rather than a silent gap. When a transaction references a date and currency with no cached snapshot, the LookupError above is caught by the orchestrator and the record is routed to the dead-letter queue with its payload, its fingerprint, and a machine-readable reason code — never dropped, never resolved with a guessed rate. That controlled fallback path is the same one specified in Compliance Fallback Chains, applied here to a missing rate table rather than a missing guild scale.

Gotchas and Production Edge Cases

Rate-date versus post-date. The single most common bug is keying the lookup off the batch’s run date or the record’s posted_at instead of the transaction’s own txn_date. An invoice for a location day two weeks ago must be converted at that day’s rate. Pin to the economic date of the cost, not the date your pipeline happened to process it.

Daylight-saving and multi-location units. txn_date is a calendar date, but posted_at is a timezone-aware datetime anchored to an IANA zone through zoneinfo. A bare UTC offset silently drifts across a daylight-saving boundary, so a London unit and a Vancouver unit shooting the same week must each stamp records in their own production zone; the ledger normalizes on read, never by mutating the ingested value. When two units disagree about which calendar day a late-night settlement belongs to, the transaction date on the vendor document — not the wall clock at ingestion — is authoritative.

Two units in different IANA zones pin the same FX snapshot by transaction date, not post time Two production units shoot the same wrap week: one stamps records in Europe/London, the other in America/Vancouver, eight hours behind. For each of three consecutive dates the London unit posts a record at a late-night local wall-clock time (23:50, 22:15, 23:40) while the Vancouver unit posts the same-day record at 15:50, 14:15 and 15:40 local. Although the posted_at timestamps differ by zone and sit near the day boundary, each record's txn_date pins it to a single day-keyed FX snapshot in the shared cache — 0.9188, 0.9205 and 0.9213 EUR per USD. The rate is chosen by the economic transaction date on the vendor document, never by the local post time, so both units converge on the identical base-currency figure. Europe/London unit posted_at zone FX snapshot cache keyed by txn_date America/Vancouver unit posted_at zone · UTC−8h txn_date txn_date 23:50 London · 06-10 2026-06-10 0.9188 EUR per USD 15:50 Vancouver · 06-10 22:15 London · 06-11 2026-06-11 0.9205 EUR per USD 14:15 Vancouver · 06-11 23:40 London · 06-12 2026-06-12 0.9213 EUR per USD 15:40 Vancouver · 06-12 Same vendor txn_date pins the same snapshot in both zones — posted_at records the wall clock, but never moves the rate.

Weekend and holiday gaps. Reference sources publish only on business days, so a Saturday transaction has no same-day H.10 rate. Decide the roll rule once — typically the most recent prior published rate — encode it when you build the cache, and record which rule fired in fx_source, so the roll is auditable rather than an implicit convenience.

Float contamination at the boundary. Constructing Decimal from a float (Decimal(0.1)) imports the float’s binary error; always go through the string form, Decimal(str(...)), or better, read the raw payload field as a string and let Pydantic coerce it. A single fractional cent, compounded across tens of thousands of daily transactions, becomes exactly the variance a guarantor will ask you to explain.

Idempotency on replay. Because the fingerprint is deterministic and the ledger is append-only, re-ingesting a batch after a partial failure is safe only if the consumer deduplicates on (transaction_id, payload_sha256). Without that guard, a retried chunk double-posts. This is the same quarantine-and-replay discipline that Schema Validation & Error Handling enforces at the point of entry, extended to the currency layer.

Once every foreign transaction carries a pinned rate and a dual-currency ledger row, the downstream guild-rate math — overtime multipliers, meal penalties, and the fringe percentages computed in Pension & Health Fund Calculations — runs against a base-currency figure that is stable and reproducible, which is the whole point of pushing FX normalization to the front of the batch.

Up one level: Async Batch Processing, part of Cost Ingestion & Data Parsing Workflows.