# ============================================================================= # Projet : AIR — Accounting Intermediate Representation # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : posting.py # Description : Posting pass — lowers economic events into balanced journal entries. # ============================================================================= """Posting pass: the lowering stage (AIR -> journal entries). Maps each economic event to a balanced double-entry journal entry using the account roles defined in the ALSL policy set. Every line carries a provenance node so any posted figure traces back to its origin. Posting rules (v0.1): - Sale DR cash/AR (total) CR revenue, CR tax payables - Refund DR revenue, DR taxes CR cash/AR (mirror of Sale) - Purchase DR expense/asset (+non-recoverable tax), DR recoverable taxes CR cash/AP - PaymentReceived DR cash (settled) CR AR (booked) [± realized FX] - PaymentSent DR AP (booked) CR cash (settled) [± realized FX] - OwnerContribution DR cash CR owner capital - LoanReceived DR cash CR loan payable """ from __future__ import annotations from aic.diagnostics import Diagnostic, Severity from aic.passes.base import Pass from aic.unit import CompilationUnit, EventState from core.events import EconomicEvent, EventType from core.journal import Account, JournalEntry, JournalLine, Side from core.money import Money class PostingPass(Pass): name = "posting" def run(self, unit: CompilationUnit) -> list[Diagnostic]: diags: list[Diagnostic] = [] for event in unit.document.events: state = unit.event_state(event.id) if state.post_subtotal is None: continue # a previous pass already rejected this event try: lines = self._lower(unit, event, state, diags) except KeyError as exc: diags.append(Diagnostic( code="AIR-E600", severity=Severity.ERROR, message=str(exc.args[0]), location=f"event {event.id}", suggestion="add the missing account role to the policy set's " "accounts section", origin_pass=self.name, )) continue if not lines: continue unit.entries.append(JournalEntry( id=f"je_{event.id}", date=event.date, description=event.description or event.type.value, lines=tuple(lines), source_event_id=event.id, policy_set=unit.policies.name, policy_version=unit.policies.version, reverses=( f"je_{event.related_event}" if event.type is EventType.REFUND and event.related_event else None ), )) return diags # -- helpers --------------------------------------------------------------- def _line( self, unit: CompilationUnit, account: Account, side: Side, amount: Money, memo: str, parent_node: str | None, ) -> JournalLine: node = unit.provenance.define( kind="journal_line", operation=f"post:{side.value}:{account.code}", amount=amount, inputs=(parent_node,) if parent_node else (), ) return JournalLine( account=account, side=side, amount=amount, memo=memo, provenance_id=node.id, ) def _settlement_account(self, unit: CompilationUnit, event: EconomicEvent, counterparty_role: str) -> Account: immediate = bool(event.payment and event.payment.immediate) return unit.policies.account("cash" if immediate else counterparty_role) def _check_gross(self, event: EconomicEvent, computed_total: Money, diags: list[Diagnostic], pass_name: str) -> None: """Warn when the document's stated gross disagrees with the computed total.""" if event.payment is None or event.payment.gross is None: return gross = event.payment.gross.to_money() if gross.currency != computed_total.currency: return # gross stated in transaction currency, total in functional if gross.amount != computed_total.amount: diags.append(Diagnostic( code="AIR-W600", severity=Severity.WARNING, message=( f"stated gross {gross} differs from computed total " f"{computed_total}; the computed total is authoritative" ), location=f"event {event.id}, field payment.gross", suggestion="check the source document; a small delta usually means " "the issuer rounded differently", origin_pass=pass_name, )) # -- lowering per event type ------------------------------------------------- def _lower( self, unit: CompilationUnit, event: EconomicEvent, state: EventState, diags: list[Diagnostic], ) -> list[JournalLine]: pol = unit.policies assert state.post_subtotal is not None subtotal = state.post_subtotal sub_node = state.post_subtotal_node taxes = state.post_taxes total = subtotal for t in taxes: total = total + t.amount if event.type is EventType.SALE: self._check_gross(event, total, diags, self.name) counter = self._settlement_account(unit, event, "accounts_receivable") lines = [self._line(unit, counter, Side.DEBIT, total, "sale: consideration incl. taxes", sub_node)] lines.append(self._line(unit, pol.account("revenue"), Side.CREDIT, subtotal, "sale: revenue", sub_node)) for t in taxes: lines.append(self._line( unit, pol.account(t.payable_role), Side.CREDIT, t.amount, f"{t.code} collected ({t.policy})", t.node_id)) return lines if event.type is EventType.REFUND: counter = self._settlement_account(unit, event, "accounts_receivable") lines = [self._line(unit, pol.account("revenue"), Side.DEBIT, subtotal, "refund: revenue reversal", sub_node)] for t in taxes: lines.append(self._line( unit, pol.account(t.payable_role), Side.DEBIT, t.amount, f"{t.code} refunded ({t.policy})", t.node_id)) lines.append(self._line(unit, counter, Side.CREDIT, total, "refund: consideration incl. taxes", sub_node)) return lines if event.type is EventType.PURCHASE: debit_role = state.classification_role or "expense_default" debit_base = subtotal recoverable = [t for t in taxes if t.recoverable] for t in taxes: if not t.recoverable: debit_base = debit_base + t.amount # non-recoverable tax is a cost memo = ( f"purchase: capitalized ({state.classification_policy})" if state.classify_as == "asset" else "purchase: cost" ) lines = [self._line(unit, pol.account(debit_role), Side.DEBIT, debit_base, memo, sub_node)] for t in recoverable: lines.append(self._line( unit, pol.account(t.receivable_role), Side.DEBIT, t.amount, f"{t.code} recoverable ({t.policy})", t.node_id)) counter = self._settlement_account(unit, event, "accounts_payable") lines.append(self._line(unit, counter, Side.CREDIT, total, "purchase: consideration incl. taxes", sub_node)) self._check_gross(event, total, diags, self.name) return lines if event.type is EventType.PAYMENT_RECEIVED: settled = subtotal booked = state.booked_amount or settled lines = [self._line(unit, pol.account("cash"), Side.DEBIT, settled, "payment received", sub_node)] lines.append(self._line( unit, pol.account("accounts_receivable"), Side.CREDIT, booked, f"settles {event.related_event or 'receivable'}", sub_node)) diff = state.fx_diff if diff is not None and not diff.is_zero(): if diff.is_negative(): lines.append(self._line(unit, pol.account("fx_loss"), Side.DEBIT, -diff, "realized FX loss", state.fx_diff_node)) else: lines.append(self._line(unit, pol.account("fx_gain"), Side.CREDIT, diff, "realized FX gain", state.fx_diff_node)) return lines if event.type is EventType.PAYMENT_SENT: settled = subtotal booked = state.booked_amount or settled lines = [self._line( unit, pol.account("accounts_payable"), Side.DEBIT, booked, f"settles {event.related_event or 'payable'}", sub_node)] diff = state.fx_diff if diff is not None and not diff.is_zero(): if diff.is_negative(): # paying fewer functional units than booked -> gain lines.append(self._line(unit, pol.account("fx_gain"), Side.CREDIT, -diff, "realized FX gain", state.fx_diff_node)) else: lines.append(self._line(unit, pol.account("fx_loss"), Side.DEBIT, diff, "realized FX loss", state.fx_diff_node)) lines.append(self._line(unit, pol.account("cash"), Side.CREDIT, settled, "payment sent", sub_node)) return lines if event.type is EventType.OWNER_CONTRIBUTION: return [ self._line(unit, pol.account("cash"), Side.DEBIT, subtotal, "owner contribution", sub_node), self._line(unit, pol.account("owner_capital"), Side.CREDIT, subtotal, "owner contribution", sub_node), ] if event.type is EventType.LOAN_RECEIVED: return [ self._line(unit, pol.account("cash"), Side.DEBIT, subtotal, "loan proceeds", sub_node), self._line(unit, pol.account("loan_payable"), Side.CREDIT, subtotal, "loan principal", sub_node), ] diags.append(Diagnostic( code="AIR-E601", severity=Severity.ERROR, message=f"no posting rule for event type '{event.type.value}'", location=f"event {event.id}", suggestion="extend the posting pass or use a supported event type", origin_pass=self.name, )) return []