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 : ledger.py6# Description : Native ledger — append-only, hash-chained event store with balance projections.7# =============================================================================8"""AIR's own ledger: the standalone mode.910AIR does not require any third-party system. Compiled journals can be posted11to this native ledger: an append-only, hash-chained JSONL store (each record12carries the SHA-256 of the previous record over canonical JSON — tamper13evidence, per docs/research/ledger-engines.md). Balances, trial balance and14financial statements are projections computed from the store.1516Corrections never mutate: reversals append contra entries.17"""18from __future__ import annotations1920import hashlib21import json22from collections import defaultdict23from dataclasses import dataclass, field24from decimal import Decimal25from pathlib import Path2627from core.journal import (28 Account,29 AccountType,30 CompiledJournal,31 JournalEntry,32 JournalLine,33 Side,34)35from core.money import Money3637GENESIS_HASH = "0" * 64383940def _canonical(record: dict[str, object]) -> str:41 return json.dumps(record, sort_keys=True, separators=(",", ":"), default=str)424344def _entry_record(entry: JournalEntry, idempotency_key: str) -> dict[str, object]:45 return {46 "id": entry.id,47 "date": entry.date.isoformat(),48 "description": entry.description,49 "source_event_id": entry.source_event_id,50 "policy_set": entry.policy_set,51 "policy_version": entry.policy_version,52 "reverses": entry.reverses,53 "idempotency_key": idempotency_key,54 "lines": [55 {56 "account_code": line.account.code,57 "account_name": line.account.name,58 "account_type": line.account.type.value,59 "side": line.side.value,60 "amount": str(line.amount.amount),61 "currency": line.amount.currency,62 "memo": line.memo,63 "provenance_id": line.provenance_id,64 }65 for line in entry.lines66 ],67 }686970@dataclass71class Ledger:72 """Append-only ledger with optional JSONL persistence."""7374 path: Path | None = None75 _entries: list[JournalEntry] = field(default_factory=list)76 _records: list[dict[str, object]] = field(default_factory=list)77 _keys: set[str] = field(default_factory=set)7879 def __post_init__(self) -> None:80 if self.path is not None and self.path.exists():81 for raw in self.path.read_text(encoding="utf-8").splitlines():82 if raw.strip():83 self._load_record(json.loads(raw))8485 # -- append ------------------------------------------------------------------86 def post_journal(self, journal: CompiledJournal, idempotency_key: str) -> int:87 """Append all entries of a compiled journal. Idempotent per key.8889 Returns the number of entries actually appended (0 on a replay).90 """91 if idempotency_key in self._keys:92 return 093 appended = 094 for entry in journal.entries:95 self._append(entry, idempotency_key)96 appended += 197 self._keys.add(idempotency_key)98 return appended99100 def reverse_entry(self, entry_id: str, idempotency_key: str) -> JournalEntry:101 """Append a contra entry that reverses a posted entry. Never deletes."""102 original = next((e for e in self._entries if e.id == entry_id), None)103 if original is None:104 raise KeyError(f"ledger has no entry '{entry_id}'")105 contra = JournalEntry(106 id=f"rev_{original.id}",107 date=original.date,108 description=f"REVERSAL: {original.description}",109 lines=tuple(110 JournalLine(111 account=line.account,112 side=Side.CREDIT if line.side is Side.DEBIT else Side.DEBIT,113 amount=line.amount,114 memo=f"reversal of {original.id}: {line.memo}",115 provenance_id=line.provenance_id,116 )117 for line in original.lines118 ),119 source_event_id=original.source_event_id,120 policy_set=original.policy_set,121 policy_version=original.policy_version,122 reverses=original.id,123 )124 self._append(contra, idempotency_key)125 return contra126127 def _append(self, entry: JournalEntry, idempotency_key: str) -> None:128 record = _entry_record(entry, idempotency_key)129 prev = self._records[-1]["hash"] if self._records else GENESIS_HASH130 record["prev_hash"] = prev131 record["hash"] = hashlib.sha256(132 (str(prev) + _canonical({k: v for k, v in record.items() if k != "hash"}))133 .encode("utf-8")134 ).hexdigest()135 self._records.append(record)136 self._entries.append(entry)137 if self.path is not None:138 self.path.parent.mkdir(parents=True, exist_ok=True)139 with self.path.open("a", encoding="utf-8") as f:140 f.write(_canonical(record) + "\n")141142 def _load_record(self, record: dict[str, object]) -> None:143 lines = tuple(144 JournalLine(145 account=Account(146 code=str(l["account_code"]),147 name=str(l["account_name"]),148 type=AccountType(str(l["account_type"])),149 ),150 side=Side(str(l["side"])),151 amount=Money(Decimal(str(l["amount"])), str(l["currency"])),152 memo=str(l.get("memo", "")),153 provenance_id=(str(l["provenance_id"]) if l.get("provenance_id") else None),154 )155 for l in record["lines"] # type: ignore[union-attr]156 )157 from datetime import date as _date158 entry = JournalEntry(159 id=str(record["id"]),160 date=_date.fromisoformat(str(record["date"])),161 description=str(record["description"]),162 lines=lines,163 source_event_id=str(record["source_event_id"]),164 policy_set=str(record.get("policy_set", "")),165 policy_version=str(record.get("policy_version", "")),166 reverses=(str(record["reverses"]) if record.get("reverses") else None),167 )168 self._records.append(record)169 self._entries.append(entry)170 self._keys.add(str(record.get("idempotency_key", "")))171172 # -- integrity & projections ---------------------------------------------------173 def verify_chain(self) -> bool:174 prev = GENESIS_HASH175 for record in self._records:176 expected = hashlib.sha256(177 (prev + _canonical({k: v for k, v in record.items() if k != "hash"}))178 .encode("utf-8")179 ).hexdigest()180 if record.get("hash") != expected or record.get("prev_hash") != prev:181 return False182 prev = str(record["hash"])183 return True184185 @property186 def entries(self) -> list[JournalEntry]:187 return list(self._entries)188189 def accounts(self) -> dict[str, Account]:190 out: dict[str, Account] = {}191 for entry in self._entries:192 for line in entry.lines:193 out[line.account.code] = line.account194 return dict(sorted(out.items()))195196 def balances(self) -> dict[str, dict[str, Decimal]]:197 """Normal-side balance per account code per currency."""198 out: dict[str, dict[str, Decimal]] = defaultdict(lambda: defaultdict(Decimal))199 for entry in self._entries:200 for line in entry.lines:201 sign = 1 if line.side is line.account.type.normal_side else -1202 out[line.account.code][line.amount.currency] += sign * line.amount.amount203 return {code: dict(per) for code, per in sorted(out.items())}204