dodil/erp-brick-gl

DODIL ERP brick: general ledger

src/brick_gl/infrastructure/ai_steps.py

"""The AI steps (module.yaml ai_steps): each is a SUGGESTION a person accepts or dismisses, never an action.

  ai_check_journal               findings on a proposed manual journal: the same lines and amount as an earlier one, an account
                                 pair never paired before, an amount out of pattern. Each has its evidence.
  explain_reconciliation_break   a likely cause for a break between a control account and its open-item owner.

Both have a fixed BASELINE (this module: deterministic, from the books, no model) and an optional model port. With no provider
wired the baseline answers; a model can only ADD findings. Tests never call a model. The mode is a setting
(ai.<step>.mode: off | suggest) the owner turns on; every finding is stored in gl_ai_finding with its inputs digest."""
import hashlib
import json
from datetime import date, timedelta
from decimal import Decimal
from statistics import median
from typing import Optional

STEPS = {
    "ai_check_journal": {"benefit": "a duplicate, an odd account pair or an amount out of pattern is caught before the journal posts",
                         "safe_for_auto": False, "fallback": "the approver reviews without findings"},
    "explain_reconciliation_break": {"benefit": "a break comes with a likely cause", "safe_for_auto": False,
                                     "fallback": "the bookkeeper works the break by hand"},
}


class Provider:
    """Port for a model: returns extra findings [{finding, evidence}] or None. No provider = the baseline only."""

    def check_journal(self, journal: dict, history: list) -> Optional[list]:
        return None

    def explain_break(self, brk: dict, context: dict) -> Optional[dict]:
        return None


provider: Provider = Provider()


def digest(*parts) -> str:
    return hashlib.sha256(json.dumps(parts, default=str, sort_keys=True).encode()).hexdigest()


def _sig(lines: list) -> tuple:
    return tuple(sorted((int(ln["account_id"]), "d" if (ln["debit"] or 0) > 0 else "c", str(ln["debit"] if (ln["debit"] or 0) > 0 else ln["credit"]))
                        for ln in lines))


def baseline_check(journal: dict, lines: list, history: list, today: date) -> list:
    """Findings on a proposed journal against the last 12 months of manual journals: [{finding, evidence}]."""
    out = []
    cutoff = today - timedelta(days=366)
    past = [h for h in history if h["journal"]["journal_date"] >= cutoff and h["journal"]["journal_id"] != journal["journal_id"]]
    mine = _sig(lines)
    for h in past:
        if _sig(h["lines"]) == mine:
            out.append({"finding": "duplicate", "evidence": f"same lines and amount as {h['number']} on {h['journal']['journal_date']}"})
            break
    pairs_seen = {(a, b) for h in past for a in {ln["account_id"] for ln in h["lines"] if (ln["debit"] or 0) > 0}
                  for b in {ln["account_id"] for ln in h["lines"] if (ln["credit"] or 0) > 0}}
    if len(past) >= 5:
        for d in {ln["account_id"] for ln in lines if (ln["debit"] or 0) > 0}:
            for c in {ln["account_id"] for ln in lines if (ln["credit"] or 0) > 0}:
                if (d, c) not in pairs_seen:
                    out.append({"finding": "odd_pair", "evidence": f"accounts {d} and {c} were never paired in the {len(past)} manual journals of the last 12 months"})
    totals = [h["journal"]["functional_total"] for h in past if h["journal"]["functional_total"]]
    if len(totals) >= 3:
        usual = Decimal(median(totals))
        if usual > 0 and journal["functional_total"] >= usual * 10:
            out.append({"finding": "amount_out_of_pattern", "evidence": f"{journal['functional_total']} is at least ten times the usual {usual:.2f} "
                                                                           f"(median of {len(totals)} manual journals)"})
    extra = provider.check_journal(journal, past)
    return out + (extra or [])


def baseline_explain(brk: dict, context: dict) -> dict:
    kind = brk.get("kind", "")
    docs = context.get("journals", [])
    if kind == "unattributed":
        cause, evidence = ("a receipt was posted without the invoice it pays (no against_document), so the ledger cannot attribute it",
                           f"{len(docs)} journal(s) on {brk.get('document')}: allocate the receipt to its invoice in payments")
    elif kind == "settled_not_posted":
        cause, evidence = ("the open-item owner shows less outstanding than the ledger: a receipt, credit note or write-off settled there has "
                           "no journal here (or was posted to another document)", f"ledger {brk.get('ours')} vs owner {brk.get('theirs')}")
    elif kind.startswith("posted_not_in_"):
        cause, evidence = ("the document is in the ledger but the owner does not know it: it may have been posted twice, voided there, or "
                           "carries another id", f"{len(docs)} journal(s) on {brk.get('document')}")
    elif kind == "open_not_posted":
        cause, evidence = ("the owner has an open document the ledger never received: its posting was refused (a closed period, an unmapped "
                           "purpose) and is waiting to be retried", f"owner outstanding {brk.get('theirs')}")
    else:
        cause, evidence = ("the amounts differ: a part payment, a rounding difference or a credit note in the wrong period",
                           f"ledger {brk.get('ours')} vs owner {brk.get('theirs')}")
    extra = provider.explain_break(brk, context)
    return extra or {"cause": cause, "evidence": evidence}