# ============================================================================= # Projet : AIR — Accounting Intermediate Representation # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : bank_formats.py # Description : Bank statement parsers — camt.053, MT940, CSV -> BankTransaction. # ============================================================================= """Bank statement parsers (specs: docs/research/bank-statement-formats.md, verified 2026-08-05 — ISO 20022 camt.053 is the strategic standard, MT940 is deprecated by SWIFT but still the most deployed corporate format; AIR parses both into one model). Field semantics implemented from the research: - camt.053: Ntry/Amt carries NO sign and a Ccy attribute; direction comes exclusively from CdtDbtInd (CRDT = money in, DBIT = money out); date from BookgDt/Dt (ValDt/Dt fallback); reference from NtryDtls/TxDtls/Refs/ EndToEndId, else NtryRef; description from AddtlNtryInf, else RmtInf/Ustrd. - MT940 :61: line — value date YYMMDD, optional entry date MMDD, D/C mark (D, C, RD = reversal of debit -> money in, RC = reversal of credit -> money out), optional funds code, UNSIGNED amount with COMMA decimal separator, 1!a3!c transaction type, customer reference (16x, often NONREF), optional //bank reference. Currency comes from :60F:/:62F:, never from :61:. Integrity: opening balance +/- sum of lines must equal closing balance. Amounts are Decimal from string, never floats. """ from __future__ import annotations import csv import re import xml.etree.ElementTree as ET from datetime import date, datetime from decimal import Decimal, InvalidOperation from pathlib import Path from kernel.reconcile import BankTransaction class BankFormatError(ValueError): pass # --- camt.053 (ISO 20022 BankToCustomerStatement) -------------------------------- def _local(tag: str) -> str: """Strip the XML namespace: '{urn:...}Ntry' -> 'Ntry'.""" return tag.rsplit("}", 1)[-1] def _find(element: ET.Element, *path: str) -> ET.Element | None: """Namespace-agnostic descent (camt versions differ only in namespace).""" current: ET.Element | None = element for name in path: if current is None: return None current = next((c for c in current if _local(c.tag) == name), None) return current def _text(element: ET.Element | None) -> str: return (element.text or "").strip() if element is not None else "" def parse_camt053(path: Path) -> list[BankTransaction]: root = ET.parse(path).getroot() statement = _find(root, "BkToCstmrStmt", "Stmt") if statement is None: raise BankFormatError(f"{path}: no BkToCstmrStmt/Stmt — not a camt.053 file") transactions: list[BankTransaction] = [] for entry in statement: if _local(entry.tag) != "Ntry": continue amt = _find(entry, "Amt") if amt is None: raise BankFormatError(f"{path}: Ntry without Amt") amount = Decimal(_text(amt)) # unsigned by spec currency = amt.get("Ccy", "") direction = _text(_find(entry, "CdtDbtInd")) if direction == "DBIT": amount = -amount elif direction != "CRDT": raise BankFormatError(f"{path}: CdtDbtInd must be CRDT|DBIT, got {direction!r}") when = _text(_find(entry, "BookgDt", "Dt")) or _text(_find(entry, "ValDt", "Dt")) if not when: raise BankFormatError(f"{path}: Ntry without BookgDt/ValDt date") reference = ( _text(_find(entry, "NtryDtls", "TxDtls", "Refs", "EndToEndId")) or _text(_find(entry, "NtryRef")) ) description = ( _text(_find(entry, "AddtlNtryInf")) or _text(_find(entry, "NtryDtls", "TxDtls", "RmtInf", "Ustrd")) ) transactions.append(BankTransaction( date=date.fromisoformat(when[:10]), amount=amount, currency=currency, description=description, reference=reference, )) return transactions # --- MT940 (SWIFT customer statement) ----------------------------------------------- # :61: value-date(6) [entry-date(4)] D/C-mark [funds-code] amount type(4) refs _LINE_61 = re.compile( r"^:61:(?P\d{6})(?P\d{4})?(?PRC|RD|C|D)" r"(?P[A-Z])?(?P\d{1,15},\d*)" r"(?P[A-Z][A-Z0-9]{3})(?P[^/\n]{0,16})(?://(?P.{0,16}))?" ) _BALANCE = re.compile( r"^:6[02][FM]:(?P[CD])(?P\d{6})(?P[A-Z]{3})(?P\d{1,15},\d*)" ) def _yy_to_date(yymmdd: str) -> date: year = int(yymmdd[:2]) year += 1900 if year >= 70 else 2000 # SWIFT 2-digit pivot return date(year, int(yymmdd[2:4]), int(yymmdd[4:6])) def _comma_decimal(text: str) -> Decimal: try: return Decimal(text.replace(",", ".")) except InvalidOperation as exc: raise BankFormatError(f"bad MT940 amount {text!r}") from exc def parse_mt940(path: Path) -> list[BankTransaction]: lines = path.read_text(encoding="utf-8").splitlines() currency = "" opening: Decimal | None = None closing: Decimal | None = None transactions: list[BankTransaction] = [] i = 0 while i < len(lines): line = lines[i].strip() if line.startswith((":60F:", ":60M:", ":62F:", ":62M:")): match = _BALANCE.match(line) if not match: raise BankFormatError(f"{path}: malformed balance line {line!r}") currency = match["ccy"] balance = _comma_decimal(match["amount"]) if match["mark"] == "D": balance = -balance if line.startswith(":60"): opening = balance else: closing = balance elif line.startswith(":61:"): match = _LINE_61.match(line) if not match: raise BankFormatError(f"{path}: malformed :61: line {line!r}") amount = _comma_decimal(match["amount"]) # D = money out; C = money in; RD reverses a debit (in); RC (out) if match["mark"] in ("D", "RC"): amount = -amount reference = match["custref"].strip() if reference.upper() == "NONREF": reference = match["bankref"] or "" description = "" if i + 1 < len(lines) and lines[i + 1].startswith(":86:"): description = lines[i + 1][4:].strip() i += 1 transactions.append(BankTransaction( date=_yy_to_date(match["valdate"]), amount=amount, currency=currency, description=description, reference=reference, )) i += 1 # research §2.2: :60F: +/- sum(:61:) must equal :62F: if opening is not None and closing is not None: total = opening + sum(t.amount for t in transactions) if total != closing: raise BankFormatError( f"{path}: statement does not balance — opening {opening} + " f"movements = {total}, but closing balance is {closing}" ) if not currency: raise BankFormatError(f"{path}: no :60F:/:62F: balance line — currency unknown") return transactions # --- generic CSV ---------------------------------------------------------------------- def parse_bank_csv(path: Path) -> list[BankTransaction]: """Columns: date, amount (signed, money in positive), currency, description, reference.""" transactions: list[BankTransaction] = [] with path.open(encoding="utf-8") as f: for row in csv.DictReader(f): try: transactions.append(BankTransaction( date=datetime.strptime(row["date"].strip(), "%Y-%m-%d").date(), amount=Decimal(row["amount"].strip()), currency=row["currency"].strip(), description=(row.get("description") or "").strip(), reference=(row.get("reference") or "").strip(), )) except (KeyError, ValueError, InvalidOperation) as exc: raise BankFormatError(f"{path}: bad CSV row {row!r}: {exc}") from exc return transactions # --- dispatcher -------------------------------------------------------------------------- def parse_statement(path: Path) -> list[BankTransaction]: """Detect the format by suffix, then by content sniffing.""" suffix = path.suffix.lower() if suffix == ".xml": return parse_camt053(path) if suffix in (".mt940", ".sta", ".940"): return parse_mt940(path) if suffix == ".csv": return parse_bank_csv(path) head = path.read_text(encoding="utf-8", errors="replace")[:2000] if "