Mapping ISO 20022 to Internal GL Formats

This page solves one narrow but high-stakes scenario: taking a single deeply nested ISO 20022 message — a camt.053 account statement, a camt.054 debit/credit notification, or a pacs.008 credit transfer — and turning each booked entry into a deterministic, double-entry posting in a flat internal general ledger (GL). It sits inside the Multi-Currency Ledger Mapping stage of the pipeline: the transaction has already crossed the ingestion boundary, but its hierarchical XML still has to be flattened, routed to a GL account, FX-converted, and posted without introducing reconciliation drift, rounding artifacts, or a fractured audit trail. Everything below assumes the same non-negotiables as the rest of the Core Architecture & Bank Feed Ingestion layer — determinism, idempotency, and tamper-evident lineage.

ISO 20022 message to internal GL posting — deterministic mapping pipeline One Ntry block from a camt or pacs message flows left to right through seven stages. Stage 1 validates the raw XML against its XSD; an invalid payload is diverted to the dead-letter queue. Stage 2 flattens the nested entry to one canonical Decimal-typed dict. Stage 3 routes the proprietary BkTxCd code to a GL account and resolves CRDT or DBIT into a posting side. Stage 4 converts the amount to the functional currency using a content-addressed FX snapshot. Stage 5 is an idempotency gate keyed on the idempotency_key; a duplicate retransmission is diverted to the dead-letter queue rather than re-posted. Stage 6 atomically writes the balanced double entry, and stage 7 appends an immutable audit record carrying the raw payload hash, applied rate, and rule version. The XSD gate and the idempotency gate both feed one shared dead-letter queue for schema, field, route, rate, and duplicate failures. 1 2 3 4 5 6 7 XSD Pre-flight validate raw XML camt / pacs Ntry Flatten Entry one canonical dict Decimal · Ccy attr GL Routing side + account BkTxCd → acct FX Convert to functional ccy Decimal · snapshot Idempotency first-seen gate claim idem_key Double Entry atomic · balanced debit = credit Audit Record write-once hash · rate · ver Dead-Letter Queue schema · field · route · rate · duplicate schema invalid duplicate

Prerequisites

Confirm every upstream dependency below is satisfied before this mapping stage runs. A transaction that fails any item must be quarantined, never coerced.

Throughout, a structured audit logger is assumed in scope. Every step emits at least trace_id, source_hash, and match_decision so the decision can be reconstructed years later.

Step-by-step implementation

Step 1 — XSD pre-flight validation

Validate every payload against the official schema before a single field is read. An entry that does not validate must go to the dead-letter queue (DLQ) — the same exception path used across Exception Routing & Human-in-the-Loop Workflows — rather than risk a partial map.

python
from lxml import etree
import structlog

audit = structlog.get_logger("iso20022.gl_map")

def validate_camt(raw_payload: bytes, xsd_path: str, *, trace_id: str, source_hash: str) -> etree._Element:
    """Validate raw ISO 20022 XML against its XSD; raise on the first defect."""
    schema = etree.XMLSchema(etree.parse(xsd_path))
    parser = etree.XMLParser(resolve_entities=False, no_network=True)  # XXE-safe
    root = etree.fromstring(raw_payload, parser=parser)
    if not schema.validate(root):
        audit.error("xsd_invalid", trace_id=trace_id, source_hash=source_hash,
                    match_decision="DLQ_SCHEMA_INVALID",
                    detail=str(schema.error_log.last_error))
        raise ValueError("CAMT_SCHEMA_INVALID")
    audit.info("xsd_valid", trace_id=trace_id, source_hash=source_hash,
               match_decision="VALIDATED")
    return root

Step 2 — Flatten the Ntry block to a canonical record

ISO 20022 nests optional repeating groups (Ntry, NtryDtls, TxDtls) that must collapse to exactly one canonical record per ledger line. Read monetary precision immediately with Decimal. Note the trap on the currency code: findtext(".//ns:Amt/@Ccy", ...) does not work because findtext() resolves element text, not attributes — you must call .find(...) then .get("Ccy") on the returned element.

python
from decimal import Decimal, ROUND_HALF_EVEN
import hashlib
from typing import Dict

NS = {"ns": "urn:iso:std:iso:20022:tech:xsd:camt.053.001.08"}

def flatten_camt_entry(entry: etree._Element, *, trace_id: str, source_hash: str) -> Dict:
    """Deterministically flatten one Ntry block to a canonical GL dict."""
    amt = entry.find(".//ns:Amt", namespaces=NS)
    if amt is None or amt.text is None:
        audit.error("missing_amount", trace_id=trace_id, source_hash=source_hash,
                    match_decision="DLQ_MISSING_FIELD")
        raise ValueError("MISSING_AMOUNT")

    record = {
        "amount": Decimal(amt.text).quantize(Decimal("0.01"), rounding=ROUND_HALF_EVEN),
        "currency": amt.get("Ccy"),                                  # attribute, not child
        "direction": entry.findtext(".//ns:CdtDbtInd", namespaces=NS),  # CRDT / DBIT
        "value_date": entry.findtext(".//ns:ValDt/ns:Dt", namespaces=NS),
        "booking_date": entry.findtext(".//ns:BookgDt/ns:Dt", namespaces=NS),
        "reference": entry.findtext(".//ns:AcctSvcrRef", namespaces=NS),
        "bk_tx_cd": entry.findtext(".//ns:BkTxCd/ns:Prtry/ns:Cd", namespaces=NS),
        "raw_xml_hash": hashlib.sha256(etree.tostring(entry)).hexdigest(),
    }
    audit.info("entry_flattened", trace_id=trace_id, source_hash=source_hash,
               match_decision="FLATTENED", reference=record["reference"])
    return record

Default fallbacks must never be applied to monetary or date fields; a missing critical field triggers the DLQ with an explicit code, not a silent zero.

Step 3 — Resolve direction and route to a GL account

GL routing is declarative, version-controlled, and strictly typed so a rule change is reviewable and reproducible. Validate the rulebook with pydantic, then match the proprietary bank-transaction code (BkTxCd) to a GL account and resolve CRDT/DBIT into a posting side.

python
import re
from pydantic import BaseModel, Field

class RouteRule(BaseModel):
    pattern: str
    gl_account: str = Field(..., pattern=r"^\d{4}-\d{3}$")
    description: str

CRDR = {"CRDT": "CREDIT", "DBIT": "DEBIT"}  # ISO 20022 uses 4-letter codes, not C/D

def route_to_gl(record: Dict, rules: list[RouteRule], *,
                trace_id: str, source_hash: str) -> Dict:
    side = CRDR.get(record["direction"])
    if side is None:
        audit.error("bad_cdtdbt", trace_id=trace_id, source_hash=source_hash,
                    match_decision="DLQ_DIRECTION_UNKNOWN", value=record["direction"])
        raise ValueError("DIRECTION_UNKNOWN")
    for rule in rules:
        if record["bk_tx_cd"] and re.match(rule.pattern, record["bk_tx_cd"]):
            audit.info("gl_routed", trace_id=trace_id, source_hash=source_hash,
                       match_decision="ROUTED", gl_account=rule.gl_account, side=side)
            return {**record, "gl_account": rule.gl_account, "side": side}
    audit.warning("no_route", trace_id=trace_id, source_hash=source_hash,
                  match_decision="DLQ_NO_ROUTE", bk_tx_cd=record["bk_tx_cd"])
    raise ValueError("GL_ROUTE_MISSING")

The rulebook itself stays in version-controlled YAML, for example a ^PMNT|SEPA pattern mapping to 1010-000 (customer payments) and ^FEES|CHRG mapping to 6020-000 (bank fees).

Step 4 — Apply FX conversion with Decimal

When the entry currency differs from the entity’s functional currency, convert against the snapshot rate for the value_date before posting, so the GL never holds a mixed-currency imbalance. Realized and unrealized FX differences route to dedicated P&L accounts rather than contaminating operational balances.

python
def convert_to_functional(record: Dict, snapshot: Dict[str, Decimal], base_ccy: str, *,
                          trace_id: str, source_hash: str) -> Dict:
    if record["currency"] == base_ccy:
        return {**record, "functional_amount": record["amount"], "applied_rate": Decimal("1")}
    pair = f"{record['currency']}/{base_ccy}"
    rate = snapshot.get(pair)
    if rate is None:
        audit.error("rate_missing", trace_id=trace_id, source_hash=source_hash,
                    match_decision="DLQ_RATE_MISSING", pair=pair)
        raise ValueError("RATE_MISSING")
    converted = (record["amount"] * rate).quantize(Decimal("0.01"), rounding=ROUND_HALF_EVEN)
    audit.info("fx_applied", trace_id=trace_id, source_hash=source_hash,
               match_decision="CONVERTED", pair=pair, applied_rate=str(rate))
    return {**record, "functional_amount": converted, "applied_rate": rate}

Step 5 — Enforce idempotency before posting

Idempotency is non-negotiable: a bank retransmission must never double-post. Check a distributed cache or a unique DB constraint keyed on the idempotency_key before writing the journal entry. This is the same guarantee leaned on by downstream exact-match and hash comparison during matching.

python
import redis

def claim_idempotency(key: str, client: redis.Redis, *,
                      trace_id: str, source_hash: str) -> bool:
    """Returns True if this is the first time we have seen `key`."""
    is_new = bool(client.set(f"idem:{key}", "1", ex=604800, nx=True))  # 7-day TTL
    audit.info("idempotency_check", trace_id=trace_id, source_hash=source_hash,
               match_decision="NEW" if is_new else "DUPLICATE", idem_key=key)
    return is_new

A duplicate returns HTTP 409 Conflict or is silently dropped with an audit line — never reprocessed.

Step 6 — Post the double entry and append the audit record

The final write is atomic: the balanced double entry and its immutable audit record commit together or not at all. Hash the normalized line and append it to a write-once table that captures the raw payload hash, the applied rate, the rule version, and the posting status.

python
def post_and_audit(record: Dict, *, idem_key: str, rule_version: str,
                   trace_id: str, source_hash: str) -> None:
    audit.info("gl_posted",
               trace_id=trace_id,
               source_hash=source_hash,
               match_decision="POSTED",
               idem_key=idem_key,
               gl_account=record["gl_account"],
               side=record["side"],
               original_amount=str(record["amount"]),
               original_currency=record["currency"],
               functional_amount=str(record["functional_amount"]),
               applied_rate=str(record["applied_rate"]),
               rule_version=rule_version,
               raw_xml_hash=record["raw_xml_hash"])

Configuration boundary table

Parameter Default Valid range Notes
rounding_mode ROUND_HALF_EVEN any decimal rounding constant Banker’s rounding; keep identical to the matching stage.
precision (minor units) 2 0–4 Per-currency: 0 for JPY, 3 for KWD/BHD.
tolerance 0.01 0.00–0.05 Two-decimal currencies; widen to ±1 minor unit for zero-decimal.
idempotency_ttl 604800 s 86400–2592000 Redis key lifetime; must exceed the bank’s retransmission window.
snapshot_max_age 86400 s 3600–172800 Reject the batch if the FX snapshot is older.
dlq_on_missing_field true true / false Must stay true for monetary and date fields.

Verification and testing

Run the mapper against a fixed sample statement and assert byte-for-byte stable output. A golden-fixture test catches rule drift before it reaches production.

python
def test_camt053_maps_to_balanced_gl():
    raw = open("fixtures/camt053_multiccy.xml", "rb").read()
    root = validate_camt(raw, "xsd/camt.053.001.08.xsd",
                         trace_id="t-1", source_hash="h-1")
    entries = root.findall(".//ns:Ntry", namespaces=NS)
    snapshot = {"EUR/USD": Decimal("1.0850")}
    rules = [RouteRule(pattern="^PMNT|SEPA", gl_account="1010-000", description="Payments")]

    posted = []
    for e in entries:
        rec = flatten_camt_entry(e, trace_id="t-1", source_hash="h-1")
        rec = route_to_gl(rec, rules, trace_id="t-1", source_hash="h-1")
        rec = convert_to_functional(rec, snapshot, "USD", trace_id="t-1", source_hash="h-1")
        posted.append(rec)

    debits = sum(r["functional_amount"] for r in posted if r["side"] == "DEBIT")
    credits = sum(r["functional_amount"] for r in posted if r["side"] == "CREDIT")
    assert debits == credits                 # double entry must balance
    assert all(r["functional_amount"].as_tuple().exponent == -2 for r in posted)

Confirm three things: the entry count equals the number of Ntry blocks, debits equal credits in the functional currency, and every amount carries exactly the configured precision. Then re-run the suite against a historical statement archive to prove a rule change has not altered any past mapping — deploy rule updates as canaries and watch reconciliation success rates on a live dashboard.

Troubleshooting

Symptom / code Root cause Fix
CAMT_SCHEMA_INVALID Payload fails XSD; truncated transfer or wrong message version. Bind the namespace and XSD to the actual camt/pacs version; re-pull from the bank; DLQ until valid.
currency is None after flatten findtext(".//ns:Amt/@Ccy") used to read the Ccy attribute. Use amt.find(...).get("Ccy"); findtext() never resolves attributes.
DIRECTION_UNKNOWN Feed uses single-letter C/D (older camt.052) instead of CRDT/DBIT. Extend the CRDR map to the feed’s actual codes; do not assume the four-letter form.
RATE_MISSING No snapshot rate for the pair on the value_date (holiday, new rail). Fall back to the nearest prior business-day rate only if policy allows; otherwise hold for manual entry with an audit note.
GL_ROUTE_MISSING A BkTxCd value not covered by any rule pattern. Add or widen the routing rule, bump rule_version, and replay the DLQ by idempotency_key.
Debits ≠ credits in fixture FX conversion applied after, not before, posting. Always convert in Step 4 before the Step 6 write so the GL never holds mixed-currency lines.

Part of Multi-Currency Ledger Mapping, within Core Architecture & Bank Feed Ingestion.