spb/air Public MIT

1# =============================================================================2# Projet : AIR — Accounting Intermediate Representation3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : reconcile.py6# Description : Bank reconciliation — match statement lines against ledger cash movements.7# =============================================================================8"""Bank reconciliation (Phase 6).910A bank statement (camt.053, MT940, or CSV — parsers in kernel/bank_formats.py)11is reduced to a list of BankTransaction records, then matched against the12ledger's cash-account movements:1314- signed amounts: positive = money into our account (bank credit = our debit);15- a match is exact on (signed amount, currency) within a configurable16  date-tolerance window (default 3 days);17- greedy one-to-one matching, earliest ledger candidate first;18- everything unmatched — on either side — is reported, never dropped.1920Amounts are Decimal end to end; floats are rejected at the parser boundary.21"""22from __future__ import annotations2324from dataclasses import dataclass, field25from datetime import date26from decimal import Decimal2728from core.journal import Side29from kernel.ledger import Ledger303132@dataclass(frozen=True, slots=True)33class BankTransaction:34    """One statement line, format-agnostic. Positive amount = money in."""3536    date: date37    amount: Decimal38    currency: str39    description: str = ""40    reference: str = ""414243@dataclass(frozen=True, slots=True)44class LedgerMovement:45    """One cash-account line in the books. Positive amount = money in (debit)."""4647    date: date48    amount: Decimal49    currency: str50    entry_id: str51    memo: str = ""525354@dataclass55class ReconciliationResult:56    matched: list[tuple[BankTransaction, LedgerMovement]] = field(default_factory=list)57    unmatched_bank: list[BankTransaction] = field(default_factory=list)58    unmatched_ledger: list[LedgerMovement] = field(default_factory=list)5960    @property61    def is_clean(self) -> bool:62        return not self.unmatched_bank and not self.unmatched_ledger6364    def summary(self) -> dict[str, int]:65        return {66            "matched": len(self.matched),67            "unmatched_bank": len(self.unmatched_bank),68            "unmatched_ledger": len(self.unmatched_ledger),69        }7071    def render(self) -> str:72        lines = ["Bank Reconciliation", "-" * 72]73        for txn, movement in self.matched:74            lines.append(75                f"MATCH  {txn.date}  {txn.amount:>12} {txn.currency}  "76                f"bank:{txn.reference or txn.description[:24]:<24} "77                f"ledger:{movement.entry_id}"78            )79        for txn in self.unmatched_bank:80            lines.append(81                f"BANK?  {txn.date}  {txn.amount:>12} {txn.currency}  "82                f"{txn.description[:40]}  <- in the bank, not in the books"83            )84        for movement in self.unmatched_ledger:85            lines.append(86                f"BOOK?  {movement.date}  {movement.amount:>12} "87                f"{movement.currency}  {movement.entry_id}  "88                f"<- in the books, not at the bank"89            )90        lines.append("-" * 72)91        s = self.summary()92        lines.append(93            f"{s['matched']} matched, {s['unmatched_bank']} unexplained bank, "94            f"{s['unmatched_ledger']} outstanding ledger — "95            + ("CLEAN" if self.is_clean else "DIFFERENCES FOUND")96        )97        return "\n".join(lines)9899100def cash_movements(ledger: Ledger, cash_account: str = "1000") -> list[LedgerMovement]:101    """Every line touching the cash account, debit-positive (money in)."""102    movements: list[LedgerMovement] = []103    for entry in ledger.entries:104        for line in entry.lines:105            if line.account.code != cash_account:106                continue107            signed = (line.amount.amount if line.side is Side.DEBIT108                      else -line.amount.amount)109            movements.append(LedgerMovement(110                date=entry.date, amount=signed,111                currency=line.amount.currency,112                entry_id=entry.id, memo=line.memo,113            ))114    return movements115116117def reconcile(118    transactions: list[BankTransaction],119    movements: list[LedgerMovement],120    tolerance_days: int = 3,121) -> ReconciliationResult:122    """Greedy one-to-one matching on (signed amount, currency) within the123    date window; among candidates, the closest date wins."""124    result = ReconciliationResult()125    remaining = list(movements)126    for txn in sorted(transactions, key=lambda t: t.date):127        candidates = [128            m for m in remaining129            if m.amount == txn.amount and m.currency == txn.currency130            and abs((m.date - txn.date).days) <= tolerance_days131        ]132        if candidates:133            best = min(candidates, key=lambda m: abs((m.date - txn.date).days))134            remaining.remove(best)135            result.matched.append((txn, best))136        else:137            result.unmatched_bank.append(txn)138    result.unmatched_ledger = remaining139    return result140