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.
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.
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.
Related Guides
- Async Batch Processing — the parent guide’s semaphore-bounded worker pool and dead-letter routing that this FX step plugs into.
- CSV & API Sync Pipelines — the deterministic transport that delivers both the cost batches and the daily rate feeds this page caches.
- Schema Validation & Error Handling — the Pydantic boundary contracts and quarantine semantics reused for missing-rate rejection.
- Compliance Fallback Chains — the controlled path a transaction takes when no cached snapshot exists for its date and currency.
- Automating Daily Cost Report Ingestion with Python — a sibling walkthrough of chunked, audit-first ingestion with the same Decimal and quarantine discipline.
Up one level: Async Batch Processing, part of Cost Ingestion & Data Parsing Workflows.