# ============================================================================= # Projet : AIR — Accounting Intermediate Representation # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : optimize.py # Description : Optimization passes — duplicate detection, netting, fusion (Phase 6). # ============================================================================= """Optimization passes (the LLVM optimization-pass analogue). Opt-in (compile with optimize=True); the default pipeline is untouched. All three preserve the double-entry invariant, which the pass manager re-verifies after each of them, and full provenance: every optimized line's provenance node points back to the lines it absorbed. - DuplicateDetectionPass — events with identical economic content but different ids get a warning (AIR-W800); a human decides, nothing is dropped. - NettingPass — a Refund referencing a Sale compiled in the same batch is netted against it: one net entry (or none, when fully offset) replaces both. - FusionPass — N same-day payments with the same posting shape become one batch entry (50 identical payouts -> 1 entry, amounts summed per line). """ from __future__ import annotations import hashlib import json from collections import defaultdict from decimal import Decimal from aic.diagnostics import Diagnostic, Severity from aic.passes.base import Pass from aic.unit import CompilationUnit from core.events import EconomicEvent, EventType from core.journal import JournalEntry, JournalLine, Side from core.money import Money def _event_fingerprint(event: EconomicEvent) -> str: """Economic-content hash: identical business facts, ignoring id and meta.""" payload = json.loads(event.model_dump_json(exclude_none=True)) payload.pop("id", None) payload.pop("meta", None) return hashlib.sha256( json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest() class DuplicateDetectionPass(Pass): name = "duplicate-detection" def run(self, unit: CompilationUnit) -> list[Diagnostic]: diags: list[Diagnostic] = [] by_content: dict[str, list[str]] = defaultdict(list) for event in unit.document.events: by_content[_event_fingerprint(event)].append(event.id) for ids in by_content.values(): if len(ids) > 1: diags.append(Diagnostic( code="AIR-W800", severity=Severity.WARNING, message=( f"{len(ids)} events carry identical economic content: " + ", ".join(ids) ), location=f"events {ids[0]} .. {ids[-1]}", suggestion="if these are duplicates, remove all but one and " "recompile; if legitimate repeats, add a " "distinguishing description", origin_pass=self.name, )) return diags class NettingPass(Pass): """Net Refund entries against the Sale they reverse (same compilation).""" name = "netting" def run(self, unit: CompilationUnit) -> list[Diagnostic]: diags: list[Diagnostic] = [] by_event = {e.source_event_id: e for e in unit.entries} for event in unit.document.events: if event.type is not EventType.REFUND or not event.related_event: continue refund_entry = by_event.get(event.id) sale_entry = by_event.get(event.related_event) if refund_entry is None or sale_entry is None: continue # per-(account, sale side): refund lines mirror sale lines, so a # refund debit offsets the sale's credit on the same account sale_amts: dict[tuple[str, Side], Money] = { (l.account.code, l.side): l.amount for l in sale_entry.lines} refund_amts: dict[tuple[str, Side], Money] = {} refund_prov: dict[tuple[str, Side], str | None] = {} for l in refund_entry.lines: mirrored = Side.CREDIT if l.side is Side.DEBIT else Side.DEBIT refund_amts[(l.account.code, mirrored)] = l.amount refund_prov[(l.account.code, mirrored)] = l.provenance_id if set(refund_amts) - set(sale_amts): continue # shapes differ; not safely nettable if any((sale_amts[k] - refund_amts[k]).is_negative() for k in refund_amts): continue # refund exceeds sale; leave both entries alone net_lines: list[JournalLine] = [] for sale_line in sale_entry.lines: key = (sale_line.account.code, sale_line.side) refunded = refund_amts.get(key) net = (sale_line.amount - refunded) if refunded else sale_line.amount if net.is_zero(): continue inputs = tuple(p for p in (sale_line.provenance_id, refund_prov.get(key)) if p) node = unit.provenance.define( kind="netting", operation=f"net:{event.related_event}-{event.id}", amount=net, inputs=inputs, ) net_lines.append(JournalLine( account=sale_line.account, side=sale_line.side, amount=net, memo=f"net of {sale_entry.id} minus {refund_entry.id}", provenance_id=node.id, )) unit.entries.remove(sale_entry) unit.entries.remove(refund_entry) if net_lines: unit.entries.append(JournalEntry( id=f"je_net_{event.related_event}", date=refund_entry.date, description=(f"Netting: {sale_entry.description} " f"minus {refund_entry.description}"), lines=tuple(net_lines), source_event_id=f"net({event.related_event},{event.id})", policy_set=sale_entry.policy_set, policy_version=sale_entry.policy_version, )) detail = "netted into one entry" else: detail = "fully offset: no entry posted" diags.append(Diagnostic( code="AIR-N801", severity=Severity.NOTE, message=(f"refund '{event.id}' netted against sale " f"'{event.related_event}' ({detail})"), location=f"events {event.related_event}, {event.id}", origin_pass=self.name, )) return diags class FusionPass(Pass): """Fuse same-day payments with identical posting shape into one batch.""" name = "fusion" FUSABLE = (EventType.PAYMENT_RECEIVED, EventType.PAYMENT_SENT) def run(self, unit: CompilationUnit) -> list[Diagnostic]: diags: list[Diagnostic] = [] events_by_id = {e.id: e for e in unit.document.events} groups: dict[tuple, list[JournalEntry]] = defaultdict(list) for entry in unit.entries: event = events_by_id.get(entry.source_event_id) if event is None or event.type not in self.FUSABLE: continue shape = tuple(sorted( (l.account.code, l.side.value, l.amount.currency) for l in entry.lines)) groups[(event.type, entry.date, shape)].append(entry) for (etype, date, shape), entries in groups.items(): if len(entries) < 2: continue summed: dict[tuple[str, str, str], Decimal] = defaultdict(Decimal) accounts = {} inputs: dict[tuple[str, str, str], list[str]] = defaultdict(list) for entry in entries: for line in entry.lines: key = (line.account.code, line.side.value, line.amount.currency) summed[key] += line.amount.amount accounts[line.account.code] = line.account if line.provenance_id: inputs[key].append(line.provenance_id) batch_hash = hashlib.sha256( ",".join(sorted(e.id for e in entries)).encode("utf-8") ).hexdigest()[:8] lines = [] for (code, side, ccy), amount in sorted(summed.items()): money = Money(amount, ccy) node = unit.provenance.define( kind="fusion", operation=f"fuse:{len(entries)}x{etype.value}", amount=money, inputs=tuple(inputs[(code, side, ccy)]), ) lines.append(JournalLine( account=accounts[code], side=Side(side), amount=money, memo=f"batch of {len(entries)} {etype.value} entries", provenance_id=node.id, )) for entry in entries: unit.entries.remove(entry) unit.entries.append(JournalEntry( id=f"je_batch_{batch_hash}", date=date, description=f"Batch: {len(entries)} x {etype.value}", lines=tuple(lines), source_event_id=( "batch(" + ",".join(e.source_event_id for e in entries) + ")"), policy_set=entries[0].policy_set, policy_version=entries[0].policy_version, )) diags.append(Diagnostic( code="AIR-N800", severity=Severity.NOTE, message=(f"fused {len(entries)} {etype.value} entries dated " f"{date} into one batch entry (je_batch_{batch_hash})"), location=f"entries {', '.join(e.id for e in entries)}", origin_pass=self.name, )) return diags