spb/air Public MIT
AIR — The Language of Accounting.
Python 100%
1# =============================================================================2# Projet : AIR — Accounting Intermediate Representation3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : posting.py6# Description : Posting pass — lowers economic events into balanced journal entries.7# =============================================================================8"""Posting pass: the lowering stage (AIR -> journal entries).910Maps each economic event to a balanced double-entry journal entry using the11account roles defined in the ALSL policy set. Every line carries a provenance12node so any posted figure traces back to its origin.1314Posting rules (v0.1):15- Sale DR cash/AR (total) CR revenue, CR tax payables16- Refund DR revenue, DR taxes CR cash/AR (mirror of Sale)17- Purchase DR expense/asset (+non-recoverable tax), DR recoverable taxes18 CR cash/AP19- PaymentReceived DR cash (settled) CR AR (booked) [± realized FX]20- PaymentSent DR AP (booked) CR cash (settled) [± realized FX]21- OwnerContribution DR cash CR owner capital22- LoanReceived DR cash CR loan payable23"""24from __future__ import annotations2526from aic.diagnostics import Diagnostic, Severity27from aic.passes.base import Pass28from aic.unit import CompilationUnit, EventState29from core.events import EconomicEvent, EventType30from core.journal import Account, JournalEntry, JournalLine, Side31from core.money import Money323334class PostingPass(Pass):35 name = "posting"3637 def run(self, unit: CompilationUnit) -> list[Diagnostic]:38 diags: list[Diagnostic] = []39 for event in unit.document.events:40 state = unit.event_state(event.id)41 if state.post_subtotal is None:42 continue # a previous pass already rejected this event43 try:44 lines = self._lower(unit, event, state, diags)45 except KeyError as exc:46 diags.append(Diagnostic(47 code="AIR-E600", severity=Severity.ERROR,48 message=str(exc.args[0]),49 location=f"event {event.id}",50 suggestion="add the missing account role to the policy set's "51 "accounts section",52 origin_pass=self.name,53 ))54 continue55 if not lines:56 continue57 unit.entries.append(JournalEntry(58 id=f"je_{event.id}",59 date=event.date,60 description=event.description or event.type.value,61 lines=tuple(lines),62 source_event_id=event.id,63 policy_set=unit.policies.name,64 policy_version=unit.policies.version,65 reverses=(66 f"je_{event.related_event}"67 if event.type is EventType.REFUND and event.related_event68 else None69 ),70 ))71 return diags7273 # -- helpers ---------------------------------------------------------------74 def _line(75 self, unit: CompilationUnit, account: Account, side: Side,76 amount: Money, memo: str, parent_node: str | None,77 ) -> JournalLine:78 node = unit.provenance.define(79 kind="journal_line",80 operation=f"post:{side.value}:{account.code}",81 amount=amount,82 inputs=(parent_node,) if parent_node else (),83 )84 return JournalLine(85 account=account, side=side, amount=amount,86 memo=memo, provenance_id=node.id,87 )8889 def _settlement_account(self, unit: CompilationUnit, event: EconomicEvent,90 counterparty_role: str) -> Account:91 immediate = bool(event.payment and event.payment.immediate)92 return unit.policies.account("cash" if immediate else counterparty_role)9394 def _check_gross(self, event: EconomicEvent, computed_total: Money,95 diags: list[Diagnostic], pass_name: str) -> None:96 """Warn when the document's stated gross disagrees with the computed total."""97 if event.payment is None or event.payment.gross is None:98 return99 gross = event.payment.gross.to_money()100 if gross.currency != computed_total.currency:101 return # gross stated in transaction currency, total in functional102 if gross.amount != computed_total.amount:103 diags.append(Diagnostic(104 code="AIR-W600", severity=Severity.WARNING,105 message=(106 f"stated gross {gross} differs from computed total "107 f"{computed_total}; the computed total is authoritative"108 ),109 location=f"event {event.id}, field payment.gross",110 suggestion="check the source document; a small delta usually means "111 "the issuer rounded differently",112 origin_pass=pass_name,113 ))114115 # -- lowering per event type -------------------------------------------------116 def _lower(117 self, unit: CompilationUnit, event: EconomicEvent,118 state: EventState, diags: list[Diagnostic],119 ) -> list[JournalLine]:120 pol = unit.policies121 assert state.post_subtotal is not None122 subtotal = state.post_subtotal123 sub_node = state.post_subtotal_node124 taxes = state.post_taxes125 total = subtotal126 for t in taxes:127 total = total + t.amount128129 if event.type is EventType.SALE:130 self._check_gross(event, total, diags, self.name)131 counter = self._settlement_account(unit, event, "accounts_receivable")132 lines = [self._line(unit, counter, Side.DEBIT, total,133 "sale: consideration incl. taxes", sub_node)]134 lines.append(self._line(unit, pol.account("revenue"), Side.CREDIT,135 subtotal, "sale: revenue", sub_node))136 for t in taxes:137 lines.append(self._line(138 unit, pol.account(t.payable_role), Side.CREDIT, t.amount,139 f"{t.code} collected ({t.policy})", t.node_id))140 return lines141142 if event.type is EventType.REFUND:143 counter = self._settlement_account(unit, event, "accounts_receivable")144 lines = [self._line(unit, pol.account("revenue"), Side.DEBIT,145 subtotal, "refund: revenue reversal", sub_node)]146 for t in taxes:147 lines.append(self._line(148 unit, pol.account(t.payable_role), Side.DEBIT, t.amount,149 f"{t.code} refunded ({t.policy})", t.node_id))150 lines.append(self._line(unit, counter, Side.CREDIT, total,151 "refund: consideration incl. taxes", sub_node))152 return lines153154 if event.type is EventType.PURCHASE:155 debit_role = state.classification_role or "expense_default"156 debit_base = subtotal157 recoverable = [t for t in taxes if t.recoverable]158 for t in taxes:159 if not t.recoverable:160 debit_base = debit_base + t.amount # non-recoverable tax is a cost161 memo = (162 f"purchase: capitalized ({state.classification_policy})"163 if state.classify_as == "asset" else "purchase: cost"164 )165 lines = [self._line(unit, pol.account(debit_role), Side.DEBIT,166 debit_base, memo, sub_node)]167 for t in recoverable:168 lines.append(self._line(169 unit, pol.account(t.receivable_role), Side.DEBIT, t.amount,170 f"{t.code} recoverable ({t.policy})", t.node_id))171 counter = self._settlement_account(unit, event, "accounts_payable")172 lines.append(self._line(unit, counter, Side.CREDIT, total,173 "purchase: consideration incl. taxes", sub_node))174 self._check_gross(event, total, diags, self.name)175 return lines176177 if event.type is EventType.PAYMENT_RECEIVED:178 settled = subtotal179 booked = state.booked_amount or settled180 lines = [self._line(unit, pol.account("cash"), Side.DEBIT, settled,181 "payment received", sub_node)]182 lines.append(self._line(183 unit, pol.account("accounts_receivable"), Side.CREDIT, booked,184 f"settles {event.related_event or 'receivable'}", sub_node))185 diff = state.fx_diff186 if diff is not None and not diff.is_zero():187 if diff.is_negative():188 lines.append(self._line(unit, pol.account("fx_loss"),189 Side.DEBIT, -diff,190 "realized FX loss", state.fx_diff_node))191 else:192 lines.append(self._line(unit, pol.account("fx_gain"),193 Side.CREDIT, diff,194 "realized FX gain", state.fx_diff_node))195 return lines196197 if event.type is EventType.PAYMENT_SENT:198 settled = subtotal199 booked = state.booked_amount or settled200 lines = [self._line(201 unit, pol.account("accounts_payable"), Side.DEBIT, booked,202 f"settles {event.related_event or 'payable'}", sub_node)]203 diff = state.fx_diff204 if diff is not None and not diff.is_zero():205 if diff.is_negative():206 # paying fewer functional units than booked -> gain207 lines.append(self._line(unit, pol.account("fx_gain"),208 Side.CREDIT, -diff,209 "realized FX gain", state.fx_diff_node))210 else:211 lines.append(self._line(unit, pol.account("fx_loss"),212 Side.DEBIT, diff,213 "realized FX loss", state.fx_diff_node))214 lines.append(self._line(unit, pol.account("cash"), Side.CREDIT, settled,215 "payment sent", sub_node))216 return lines217218 if event.type is EventType.OWNER_CONTRIBUTION:219 return [220 self._line(unit, pol.account("cash"), Side.DEBIT, subtotal,221 "owner contribution", sub_node),222 self._line(unit, pol.account("owner_capital"), Side.CREDIT, subtotal,223 "owner contribution", sub_node),224 ]225226 if event.type is EventType.LOAN_RECEIVED:227 return [228 self._line(unit, pol.account("cash"), Side.DEBIT, subtotal,229 "loan proceeds", sub_node),230 self._line(unit, pol.account("loan_payable"), Side.CREDIT, subtotal,231 "loan principal", sub_node),232 ]233234 diags.append(Diagnostic(235 code="AIR-E601", severity=Severity.ERROR,236 message=f"no posting rule for event type '{event.type.value}'",237 location=f"event {event.id}",238 suggestion="extend the posting pass or use a supported event type",239 origin_pass=self.name,240 ))241 return []242