dodil/erp-brick-payments

DODIL ERP brick: payments

src/brick_payments/domain/reconcile.py

"""Statement-line reconciliation, pure: the rule matcher (PAY-23) and the history matcher behind the AI suggestion (PAY-22).

A rule only ever names an ACCOUNT and a TAX CODE; it never allocates to a customer's invoice (receipts go through allocation).
The suggestion is the baseline step - the payee text and the lines the business posted before, no model call - so it is useful
with nothing wired and always explainable. It only SUGGESTS; a person decides.
"""
import re
from dataclasses import dataclass
from decimal import Decimal
from typing import Optional

MODEL = "history-payee-v1"


def _has(haystack: str, needle: str) -> bool:
    return not needle or needle.strip().lower() in (haystack or "").lower()


def rule_matches(rule: dict, line: dict) -> bool:
    """direction, text_contains, counterparty_contains, amount band and the bank account (or any) - ALL that are set must hold."""
    amount = abs(Decimal(str(line["amount"])))
    if rule["bank_account_id"] and rule["bank_account_id"] != line["bank_account_id"]:
        return False
    if rule["direction"] != ("credit" if Decimal(str(line["amount"])) > 0 else "debit"):
        return False
    if not (_has(line.get("text", ""), rule.get("text_contains", "")) and _has(line.get("counterparty_name", ""), rule.get("counterparty_contains", ""))):
        return False
    if rule.get("amount_min") is not None and amount < Decimal(str(rule["amount_min"])):
        return False
    if rule.get("amount_max") is not None and amount > Decimal(str(rule["amount_max"])):
        return False
    return True


def matching_rules(rules: list, line: dict) -> list:
    return [r for r in rules if rule_matches(r, line)]


def _payee(line: dict) -> str:
    base = (line.get("counterparty_name") or line.get("text") or "").lower()
    return " ".join(re.sub(r"[^a-z ]", " ", base).split()[:2])


@dataclass(frozen=True)
class Suggestion:
    account_id: str
    tax_code: str
    cost_centre: str
    receipt_needed: bool
    evidence: dict
    confidence: Decimal
    model: str = MODEL


def suggest(line: dict, history: list) -> Optional[Suggestion]:
    """The account and tax code of the same payee's past postings (most common first); None when we have never paid them."""
    payee = _payee(line)
    if not payee:
        return None
    past = [h for h in history if _payee(h) == payee and h.get("account_id")]
    if not past:
        return None
    tally: dict = {}
    for h in past:
        tally.setdefault((h["account_id"], h["tax_code"] or "", h.get("cost_centre") or ""), []).append(h["line_id"])
    (account, tax, cc), ids = max(tally.items(), key=lambda kv: (len(kv[1]), kv[0]))
    share = Decimal(len(ids)) / Decimal(len(past))
    confidence = (share * Decimal(min(len(past), 5)) / Decimal(5)).quantize(Decimal("0.01"))
    return Suggestion(account, tax, cc, tax not in ("", "E", "Z", "O", "N"),
                      {"payee": payee, "same_payee_postings": len(past), "posted_to_this_account": len(ids), "examples": sorted(ids)[:3]},
                      confidence)


# ── a statement balances and follows the one before it (PAY-34) ──────────────────────────────────────────

def signed_total(lines) -> Decimal:
    """The statement's lines as the bank sent them: credits plus, debits minus."""
    return sum((ln.amount if ln.credit else -ln.amount for ln in lines), Decimal("0.00"))


def balance_gap(opening: Optional[Decimal], closing: Optional[Decimal], total: Decimal) -> Optional[Decimal]:
    """closing - (opening + lines): None when the figures are not both given, zero when the statement adds up."""
    if opening is None or closing is None:
        return None
    return closing - (opening + total)


def continuity_gap(previous_closing: Optional[Decimal], opening: Optional[Decimal]) -> Optional[Decimal]:
    """opening - the previous statement's closing: None when either is not known, zero when the statements follow."""
    if previous_closing is None or opening is None:
        return None
    return opening - previous_closing