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 : optimize.py6# Description : Optimization passes — duplicate detection, netting, fusion (Phase 6).7# =============================================================================8"""Optimization passes (the LLVM optimization-pass analogue).910Opt-in (compile with optimize=True); the default pipeline is untouched.11All three preserve the double-entry invariant, which the pass manager12re-verifies after each of them, and full provenance: every optimized line's13provenance node points back to the lines it absorbed.1415- DuplicateDetectionPass — events with identical economic content but16 different ids get a warning (AIR-W800); a human decides, nothing is dropped.17- NettingPass — a Refund referencing a Sale compiled in the same batch is18 netted against it: one net entry (or none, when fully offset) replaces both.19- FusionPass — N same-day payments with the same posting shape become one20 batch entry (50 identical payouts -> 1 entry, amounts summed per line).21"""22from __future__ import annotations2324import hashlib25import json26from collections import defaultdict27from decimal import Decimal2829from aic.diagnostics import Diagnostic, Severity30from aic.passes.base import Pass31from aic.unit import CompilationUnit32from core.events import EconomicEvent, EventType33from core.journal import JournalEntry, JournalLine, Side34from core.money import Money353637def _event_fingerprint(event: EconomicEvent) -> str:38 """Economic-content hash: identical business facts, ignoring id and meta."""39 payload = json.loads(event.model_dump_json(exclude_none=True))40 payload.pop("id", None)41 payload.pop("meta", None)42 return hashlib.sha256(43 json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()444546class DuplicateDetectionPass(Pass):47 name = "duplicate-detection"4849 def run(self, unit: CompilationUnit) -> list[Diagnostic]:50 diags: list[Diagnostic] = []51 by_content: dict[str, list[str]] = defaultdict(list)52 for event in unit.document.events:53 by_content[_event_fingerprint(event)].append(event.id)54 for ids in by_content.values():55 if len(ids) > 1:56 diags.append(Diagnostic(57 code="AIR-W800", severity=Severity.WARNING,58 message=(59 f"{len(ids)} events carry identical economic content: "60 + ", ".join(ids)61 ),62 location=f"events {ids[0]} .. {ids[-1]}",63 suggestion="if these are duplicates, remove all but one and "64 "recompile; if legitimate repeats, add a "65 "distinguishing description",66 origin_pass=self.name,67 ))68 return diags697071class NettingPass(Pass):72 """Net Refund entries against the Sale they reverse (same compilation)."""7374 name = "netting"7576 def run(self, unit: CompilationUnit) -> list[Diagnostic]:77 diags: list[Diagnostic] = []78 by_event = {e.source_event_id: e for e in unit.entries}7980 for event in unit.document.events:81 if event.type is not EventType.REFUND or not event.related_event:82 continue83 refund_entry = by_event.get(event.id)84 sale_entry = by_event.get(event.related_event)85 if refund_entry is None or sale_entry is None:86 continue8788 # per-(account, sale side): refund lines mirror sale lines, so a89 # refund debit offsets the sale's credit on the same account90 sale_amts: dict[tuple[str, Side], Money] = {91 (l.account.code, l.side): l.amount for l in sale_entry.lines}92 refund_amts: dict[tuple[str, Side], Money] = {}93 refund_prov: dict[tuple[str, Side], str | None] = {}94 for l in refund_entry.lines:95 mirrored = Side.CREDIT if l.side is Side.DEBIT else Side.DEBIT96 refund_amts[(l.account.code, mirrored)] = l.amount97 refund_prov[(l.account.code, mirrored)] = l.provenance_id98 if set(refund_amts) - set(sale_amts):99 continue # shapes differ; not safely nettable100 if any((sale_amts[k] - refund_amts[k]).is_negative()101 for k in refund_amts):102 continue # refund exceeds sale; leave both entries alone103104 net_lines: list[JournalLine] = []105 for sale_line in sale_entry.lines:106 key = (sale_line.account.code, sale_line.side)107 refunded = refund_amts.get(key)108 net = (sale_line.amount - refunded) if refunded else sale_line.amount109 if net.is_zero():110 continue111 inputs = tuple(p for p in (sale_line.provenance_id,112 refund_prov.get(key)) if p)113 node = unit.provenance.define(114 kind="netting",115 operation=f"net:{event.related_event}-{event.id}",116 amount=net,117 inputs=inputs,118 )119 net_lines.append(JournalLine(120 account=sale_line.account, side=sale_line.side, amount=net,121 memo=f"net of {sale_entry.id} minus {refund_entry.id}",122 provenance_id=node.id,123 ))124125 unit.entries.remove(sale_entry)126 unit.entries.remove(refund_entry)127 if net_lines:128 unit.entries.append(JournalEntry(129 id=f"je_net_{event.related_event}",130 date=refund_entry.date,131 description=(f"Netting: {sale_entry.description} "132 f"minus {refund_entry.description}"),133 lines=tuple(net_lines),134 source_event_id=f"net({event.related_event},{event.id})",135 policy_set=sale_entry.policy_set,136 policy_version=sale_entry.policy_version,137 ))138 detail = "netted into one entry"139 else:140 detail = "fully offset: no entry posted"141 diags.append(Diagnostic(142 code="AIR-N801", severity=Severity.NOTE,143 message=(f"refund '{event.id}' netted against sale "144 f"'{event.related_event}' ({detail})"),145 location=f"events {event.related_event}, {event.id}",146 origin_pass=self.name,147 ))148 return diags149150151class FusionPass(Pass):152 """Fuse same-day payments with identical posting shape into one batch."""153154 name = "fusion"155 FUSABLE = (EventType.PAYMENT_RECEIVED, EventType.PAYMENT_SENT)156157 def run(self, unit: CompilationUnit) -> list[Diagnostic]:158 diags: list[Diagnostic] = []159 events_by_id = {e.id: e for e in unit.document.events}160161 groups: dict[tuple, list[JournalEntry]] = defaultdict(list)162 for entry in unit.entries:163 event = events_by_id.get(entry.source_event_id)164 if event is None or event.type not in self.FUSABLE:165 continue166 shape = tuple(sorted(167 (l.account.code, l.side.value, l.amount.currency)168 for l in entry.lines))169 groups[(event.type, entry.date, shape)].append(entry)170171 for (etype, date, shape), entries in groups.items():172 if len(entries) < 2:173 continue174 summed: dict[tuple[str, str, str], Decimal] = defaultdict(Decimal)175 accounts = {}176 inputs: dict[tuple[str, str, str], list[str]] = defaultdict(list)177 for entry in entries:178 for line in entry.lines:179 key = (line.account.code, line.side.value,180 line.amount.currency)181 summed[key] += line.amount.amount182 accounts[line.account.code] = line.account183 if line.provenance_id:184 inputs[key].append(line.provenance_id)185186 batch_hash = hashlib.sha256(187 ",".join(sorted(e.id for e in entries)).encode("utf-8")188 ).hexdigest()[:8]189 lines = []190 for (code, side, ccy), amount in sorted(summed.items()):191 money = Money(amount, ccy)192 node = unit.provenance.define(193 kind="fusion",194 operation=f"fuse:{len(entries)}x{etype.value}",195 amount=money,196 inputs=tuple(inputs[(code, side, ccy)]),197 )198 lines.append(JournalLine(199 account=accounts[code], side=Side(side), amount=money,200 memo=f"batch of {len(entries)} {etype.value} entries",201 provenance_id=node.id,202 ))203 for entry in entries:204 unit.entries.remove(entry)205 unit.entries.append(JournalEntry(206 id=f"je_batch_{batch_hash}",207 date=date,208 description=f"Batch: {len(entries)} x {etype.value}",209 lines=tuple(lines),210 source_event_id=(211 "batch(" + ",".join(e.source_event_id for e in entries) + ")"),212 policy_set=entries[0].policy_set,213 policy_version=entries[0].policy_version,214 ))215 diags.append(Diagnostic(216 code="AIR-N800", severity=Severity.NOTE,217 message=(f"fused {len(entries)} {etype.value} entries dated "218 f"{date} into one batch entry (je_batch_{batch_hash})"),219 location=f"entries {', '.join(e.id for e in entries)}",220 origin_pass=self.name,221 ))222 return diags223