spb/air Public MIT

1# =============================================================================2# Projet : AIR — Accounting Intermediate Representation3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : invariant.py6# Description : The accounting verifier — double-entry balance checked after every pass.7# =============================================================================8"""Accounting invariants (AIR's equivalent of the LLVM IR verifier).910Two levels, both hard errors when violated:11121. Entry level — every journal entry balances per currency:13   sum(debits) == sum(credits). This is double-entry itself.142. Ledger level — the accounting equation:15   Assets = Liabilities + Equity + (Revenue - Expenses)16   computed from typed account balances across all entries.1718The pass manager runs these after EVERY pass that touches journal entries;19a violation aborts compilation with a clang-style diagnostic.20"""21from __future__ import annotations2223from collections import defaultdict24from decimal import Decimal2526from aic.diagnostics import Diagnostic, Severity27from core.journal import AccountType, JournalEntry, Side282930def entry_imbalances(entry: JournalEntry) -> dict[str, Decimal]:31    """Per-currency imbalance (debits - credits). Empty dict == balanced."""32    balance: dict[str, Decimal] = defaultdict(Decimal)33    for line in entry.lines:34        balance[line.amount.currency] += line.signed()35    return {ccy: diff for ccy, diff in balance.items() if diff != 0}363738def verify_entries(entries: list[JournalEntry], pass_name: str) -> list[Diagnostic]:39    """Entry-level double-entry check. Returns error diagnostics (empty == OK)."""40    diags: list[Diagnostic] = []41    for entry in entries:42        if not entry.lines:43            diags.append(Diagnostic(44                code="AIR-E100",45                severity=Severity.ERROR,46                message=f"journal entry '{entry.id}' has no lines",47                location=f"entry {entry.id} (event {entry.source_event_id})",48                suggestion="the posting pass must emit at least two lines per entry",49                origin_pass=pass_name,50            ))51            continue52        for ccy, diff in entry_imbalances(entry).items():53            diags.append(Diagnostic(54                code="AIR-E101",55                severity=Severity.ERROR,56                message=(57                    f"journal entry '{entry.id}' is unbalanced in {ccy}: "58                    f"debits - credits = {diff}"59                ),60                location=f"entry {entry.id} (event {entry.source_event_id})",61                suggestion="every entry must satisfy sum(debits) == sum(credits) per currency",62                origin_pass=pass_name,63            ))64    return diags656667def balances_by_type(entries: list[JournalEntry]) -> dict[AccountType, dict[str, Decimal]]:68    """Normal-side balances per account type per currency.6970    Asset/expense balances are debit-positive; liability/equity/revenue71    balances are credit-positive.72    """73    out: dict[AccountType, dict[str, Decimal]] = {74        t: defaultdict(Decimal) for t in AccountType75    }76    for entry in entries:77        for line in entry.lines:78            sign = 1 if line.side is line.account.type.normal_side else -179            out[line.account.type][line.amount.currency] += sign * line.amount.amount80    return out818283def accounting_equation_residual(entries: list[JournalEntry]) -> dict[str, Decimal]:84    """Assets - (Liabilities + Equity + Revenue - Expenses), per currency.8586    Zero everywhere iff the ledger satisfies the accounting equation.87    """88    b = balances_by_type(entries)89    currencies = {ccy for per_ccy in b.values() for ccy in per_ccy}90    residual: dict[str, Decimal] = {}91    for ccy in currencies:92        assets = b[AccountType.ASSET][ccy]93        liabilities = b[AccountType.LIABILITY][ccy]94        equity = b[AccountType.EQUITY][ccy]95        revenue = b[AccountType.REVENUE][ccy]96        expenses = b[AccountType.EXPENSE][ccy]97        residual[ccy] = assets - (liabilities + equity + revenue - expenses)98    return {ccy: r for ccy, r in residual.items() if r != 0}99100101def verify_equation(entries: list[JournalEntry], pass_name: str) -> list[Diagnostic]:102    """Ledger-level accounting-equation check."""103    diags: list[Diagnostic] = []104    for ccy, res in accounting_equation_residual(entries).items():105        diags.append(Diagnostic(106            code="AIR-E102",107            severity=Severity.ERROR,108            message=(109                f"accounting equation violated in {ccy}: "110                f"Assets - (Liabilities + Equity + Revenue - Expenses) = {res}"111            ),112            location="ledger",113            suggestion="an entry posted to a mistyped account or an unbalanced entry slipped through",114            origin_pass=pass_name,115        ))116    return diags117