dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/application/bankrec.py
"""A statement balances, follows the one before, and is reconciled to the ledger (PAY-34); the open-banking feed (PAY-35).
* imported -> balanced: opening + lines = closing (enforced at import: the file is refused otherwise) and opening = the previous
statement's closing (gate statement_continuous; a gap is named and the statement stays imported until the missing one arrives).
* balanced -> reconciled: every line matched or posted, AND the ledger balance of the bank account's gl account at the closing date
(gl's account balance route) equals the closing balance - or the difference is shown with the items that explain it. Reconciled
locks the statement: an unmatch of one of its lines is refused, and the reconcile and the unmatch take the statement's one guard
key (pay_claim 'statement'), so they cannot both land.
* the feed (PAY-35) is the same import as a file: POST /statements/ingest turns the provider's transactions and booked balance into a
statement (format feed) keyed by the bank's transaction id; the provider holds an owner's read-only consent for at most 90 days,
shown 7 days ahead; the provider never writes pay_ tables.
"""
from datetime import timedelta
from decimal import Decimal
from types import SimpleNamespace
from typing import Annotated, Optional
from erp_platform.bricks import Refused
from pydantic import Field
from brick_payments.application import statements as st
from brick_payments.application import use_cases as uc
from brick_payments.application.inputs import Balance, Code, Day, FeedLineIn, Ref
from brick_payments.domain import bankfile, reconcile
from brick_payments.domain.errors import RuleBroken
from brick_payments.domain.money import ZERO
from brick_payments.domain.readiness import fmt
from brick_payments.infrastructure import seams
PAY = uc.PAY
CONSENT_DAYS, WARN_DAYS = 90, 7
def _statement(repo, statement_id: str) -> dict:
row = repo.bank.statement(statement_id)
if row is None:
raise Refused(404, f"no statement {statement_id}")
return row
def _continuity(repo, row: dict) -> dict:
return st.continuity(repo, row["bank_account_id"], row["statement_id"],
SimpleNamespace(from_date=row["from_date"], opening_balance=None if row["opening_balance"] is None else Decimal(str(row["opening_balance"]))))
def _balanced(repo, row: dict, claim: bool = True) -> tuple:
"""(gate, row): the balance gate, the statement moved imported -> balanced when it now holds (its claim taken in this transaction)."""
if row["opening_balance"] is None or row["closing_balance"] is None:
return ({"gate": "statement_balances", "met": False, "reason": f"statement {row['statement_id']} carries no opening and closing balance, "
"so it cannot be proven to add up", "fix": "import it from a camt.053 or MT940 file, or a feed with the booked balance"}, row)
cont = _continuity(repo, row)
if not cont["met"]:
return {"gate": "statement_continuous", "met": False, "reason": cont["reason"], "fix": cont["fix"], "gap": cont["gap"]}, row
if row["lifecycle"] == "imported":
if claim: # one take per key per transaction: it counts only COMMITTED claims
repo.claims.take("statement", row["statement_id"], uc._now(), "system", "balanced")
row = {**row, "lifecycle": "balanced"}
repo.bank.save_statement(row)
return {"gate": "statement_balances", "met": True}, row
@PAY.use_case(kind="command", route="POST /statements/{statement_id}/check-balance", permission=st.RECONCILE, idempotent=True)
def check_statement_balance(caller: dict, statement_id: Ref) -> dict:
"""Does the statement add up (opening + lines = closing) and follow the one before (its opening is that one's closing)? Moves it from
imported to balanced when it does; otherwise names the difference or the gap."""
with uc.repository() as repo, uc._decide():
row = _statement(repo, statement_id)
gate, row = _balanced(repo, row)
return {"statement_id": statement_id, "lifecycle": row["lifecycle"], "gates": [gate]}
def _reconciliation(repo, row: dict, actor: str) -> dict:
"""The statement against the ledger: open lines, the ledger's balance at the closing date vs the statement's, and what explains a gap."""
lines = repo.bank.lines(statement_id=row["statement_id"], limit=100_000)
open_lines = [ln for ln in lines if ln["lifecycle"] == "open"]
bank = repo.config.bank(row["bank_account_id"]) or {}
as_of = row["to_date"] or max((ln["booked_on"] for ln in lines), default=None)
out = {"statement_id": row["statement_id"], "lifecycle": row["lifecycle"], "bank_account_id": row["bank_account_id"],
"closing_balance": st._s(row["closing_balance"]), "as_of": as_of.isoformat() if as_of else None,
"open_lines": [{"line_id": ln["line_id"], "amount": st._s(ln["amount"]), "text": ln["text"]} for ln in open_lines]}
if row["closing_balance"] is None or as_of is None:
return {**out, "book_balance": None, "difference": None, "gate": {"gate": "book_equals_bank", "met": False,
"reason": "the statement has no closing balance or closing date to compare the ledger with", "fix": "import it with its balances and dates"}}
if not bank.get("gl_account_id"):
return {**out, "book_balance": None, "difference": None, "gate": {"gate": "book_equals_bank", "met": False,
"reason": f"bank account {row['bank_account_id']} has no ledger account", "fix": "set its gl_account_id in the settings"}}
got = seams.account_balance(bank["gl_account_id"], as_of, bank.get("company_code") or "", actor)
if got is None:
return {**out, "book_balance": None, "difference": None, "gate": {"gate": "book_equals_bank", "met": False,
"reason": f"the ledger does not know account {bank['gl_account_id']}", "fix": "check the bank account's gl_account_id"}}
book, closing = Decimal(str(got["balance"])), Decimal(str(row["closing_balance"]))
diff = book - closing
items = [{"kind": "in_transit", "line_id": ln["line_id"], "amount": st._s(ln["amount"])} for ln in lines
if ln["matched_kind"] == "transfer" and not ln["matched_ref"]]
items += [{"kind": "receipt_not_on_a_statement", "payment_id": r.payment_id, "amount": st._s(r.amount), "received_on": r.received_on.isoformat()}
for r in repo.receipts.not_on_a_statement(row["bank_account_id"], as_of)]
met = diff == ZERO
return {**out, "book_balance": st._s(book), "difference": st._s(diff), "unreconciled_items": items,
"gate": {"gate": "book_equals_bank", "met": met,
"reason": (f"the ledger shows {fmt(book)} and the bank {fmt(closing)} on {as_of}" if met else
f"the ledger shows {fmt(book)} on {as_of}, the bank {fmt(closing)}: {fmt(abs(diff))} {'more in the books' if diff > 0 else 'less in the books'}"),
"fix": "" if met else "find the items below (a receipt typed in but not on the statement, a transfer in transit, a posting dated differently)"}}
@PAY.use_case(kind="query", route="GET /statements/{statement_id}/reconciliation", permission=uc.READ)
def statement_reconciliation(caller: dict, statement_id: Ref) -> dict:
"""Where a statement stands against the ledger: its lines still open, the ledger's balance of the bank account at the closing date
against the statement's closing balance, the difference and the unreconciled items that may explain it. Changes nothing."""
with uc.repository() as repo, uc._decide():
row = _statement(repo, statement_id)
return {**_reconciliation(repo, row, uc._who(caller)), "balance_gate": {"gate": "statement_balances", "met": row["lifecycle"] != "imported"}}
@PAY.use_case(kind="command", route="POST /statements/{statement_id}/reconcile", permission=st.RECONCILE, idempotent=True)
def reconcile_statement(caller: dict, statement_id: Ref) -> dict:
"""Complete the bank reconciliation: the statement is reconciled only when it balances and follows the previous one, every line is
matched or posted, and the ledger's balance of the bank account at the closing date equals the closing balance. Otherwise nothing
changes and the difference is shown with the unreconciled items. A reconciled statement is locked (an unmatch of its lines is refused)."""
with uc.repository() as repo, uc._decide():
row = _statement(repo, statement_id)
if row["lifecycle"] == "reconciled":
return {"statement_id": statement_id, "lifecycle": "reconciled", "replay": True, "reconciled_by": row["reconciled_by"]}
who, now, today = uc._who(caller), uc._now(), uc._today(repo)
gate, row = _balanced(repo, row, claim=False)
if not gate["met"]:
raise RuleBroken(f"{gate['gate']}: {gate['reason']}. {gate['fix']}")
rec = _reconciliation(repo, row, who)
if rec["open_lines"]:
shown = ", ".join(f"{x['line_id']} ({x['amount']})" for x in rec["open_lines"][:10])
raise RuleBroken(f"lines_resolved: {len(rec['open_lines'])} line(s) are not matched or posted yet: {shown}. Match or post each one first")
if not rec["gate"]["met"]:
items = "; ".join(f"{i['kind']} {i.get('payment_id') or i.get('line_id')} {i['amount']}" for i in rec.get("unreconciled_items", [])) or "none found"
raise RuleBroken(f"book_equals_bank: {rec['gate']['reason']}. Unreconciled items: {items}")
repo.claims.take("statement", statement_id, now, who, "reconcile")
repo.bank.save_statement({**row, "lifecycle": "reconciled", "reconciled_by": who, "reconciled_at": uc._now().replace(tzinfo=None)})
return {"statement_id": statement_id, "lifecycle": "reconciled", "replay": False, "reconciled_by": who, "book_balance": rec["book_balance"],
"closing_balance": rec["closing_balance"], "as_of": rec["as_of"], "on": today.isoformat()}
# ═══ PAY-35: the open-banking feed ═══════════════════════════════════════════════════════════════════════════
def _consent_view(c: dict, today) -> dict:
left = (c["expires_on"] - today).days
return {"bank_account_id": c["bank_account_id"], "provider": c["provider"], "granted_on": c["granted_on"].isoformat(),
"expires_on": c["expires_on"].isoformat(), "days_left": left, "expired": left <= 0, "expires_soon": 0 < left <= WARN_DAYS,
**({"warning": f"the owner's consent for {c['bank_account_id']} ends in {left} day(s) ({c['expires_on']}): ask the owner to renew it"}
if 0 < left <= WARN_DAYS else {})}
def _current(repo, bank_account_id: str, today) -> Optional[dict]:
rows = repo.feeds.consents(bank_account_id)
return max(rows, key=lambda c: c["expires_on"]) if rows else None
@PAY.use_case(kind="command", route="POST /bank-accounts/{bank_account_id}/feed-consent", permission="pay:bank_account:write")
def record_feed_consent(caller: dict, bank_account_id: Code, provider: Annotated[str, Field(min_length=1, max_length=64)],
granted_on: Optional[Day] = None, days: Annotated[int, Field(ge=1, le=CONSENT_DAYS)] = CONSENT_DAYS) -> dict:
"""The owner granted an account-information provider read-only access to this account for up to 90 days (UK Open Banking). The
feed is accepted only while the consent runs; its expiry is shown 7 days ahead. A renewal is a new row, never an edit."""
with uc.repository() as repo, uc._decide():
today, now = uc._today(repo), uc._now()
if repo.config.bank(bank_account_id) is None:
raise Refused(404, f"no bank account {bank_account_id}")
start = uc._date(granted_on, "granted_on") or today
if start > today:
raise RuleBroken(f"A consent granted on {start} is in the future. It cannot be recorded before it was given")
if (today - start).days > CONSENT_DAYS:
raise RuleBroken(f"A consent granted on {start} is more than {CONSENT_DAYS} days old, so it has already run out. Ask the owner to grant it again")
repo.claims.take("feed_consent", bank_account_id, now, uc._who(caller), provider)
row = {"bank_account_id": bank_account_id, "granted_on": start, "provider": provider.strip(), "expires_on": start + timedelta(days=days),
"recorded_by": uc._who(caller)}
repo.feeds.add(row, now)
return _consent_view(row, today)
@PAY.use_case(kind="query", route="GET /feed-consents", permission=uc.READ)
def list_feed_consents(caller: dict) -> dict:
"""Every account's current feed consent with the days it has left; one ending within 7 days carries a warning for the owner."""
with uc.repository() as repo:
today = uc._today(repo)
cur = {}
for c in repo.feeds.consents():
if c["bank_account_id"] not in cur or c["expires_on"] > cur[c["bank_account_id"]]["expires_on"]:
cur[c["bank_account_id"]] = c
views = [_consent_view(c, today) for c in cur.values()]
return {"consents": views, "expiring": [v for v in views if v["expires_soon"]]}
@PAY.use_case(kind="command", route="POST /statements/ingest", permission=uc.STATEMENT_INGEST, tool=False)
def ingest_feed(caller: dict, bank_account_id: Code, as_of: Day, booked_balance: Balance, lines: Annotated[list[FeedLineIn], Field(max_length=5000)],
provider: Optional[Annotated[str, Field(min_length=1, max_length=64)]] = None) -> dict:
"""The feed provider's daily post (PAY-35): the day's transactions and the booked balance, imported exactly as a statement file is,
keyed by the bank's transaction id - a transaction delivered twice, or again in a later delivery, is one line. Accepted only while the
owner's read-only consent runs. The statement's opening is the booked balance less the delivered transactions, so a delivery that
misses a transaction shows as a gap against the previous statement."""
with uc.repository() as repo, uc._decide():
today = uc._today(repo)
bank = repo.config.bank(bank_account_id)
if bank is None:
raise Refused(422, f"there is no bank account {bank_account_id} - set it up in the settings first")
consent = _current(repo, bank_account_id, today)
if consent is None:
raise RuleBroken(f"feed_consent: no consent is recorded for {bank_account_id}. Record the owner's consent (POST /bank-accounts/{bank_account_id}/feed-consent)")
if consent["expires_on"] <= today: # 90 days from the grant: the 91st day is out
raise RuleBroken(f"feed_consent: the owner's consent for {bank_account_id} ended on {consent['expires_on']}. Ask the owner to grant it again")
if provider is not None and provider.strip().lower() != consent["provider"].strip().lower():
raise RuleBroken(f"feed_consent: the owner's consent for {bank_account_id} was given to {consent['provider']}, not {provider}. "
"Only the provider the owner named may post this account's feed")
day_ = uc._date(as_of, "as_of")
seen, parsed = set(), []
for ln in lines:
if ln.transaction_id in seen:
continue # delivered twice in one post: one line
seen.add(ln.transaction_id)
booked = uc._date(ln.booked_on, "booked_on")
parsed.append(bankfile.Line(ln.transaction_id, booked, ln.amount, (ln.currency or bank["currency"]).upper(), ln.direction == "credit",
ln.remittance or "", ln.counterparty_name or "", None, ln.counterparty_account or "", "", booked))
closing = booked_balance
# a transaction already held (an earlier delivery) is not delivered again: the opening is the booked balance less the NEW lines only,
# so a full re-delivery names no false gap against the previous statement's closing (PAY-34/PAY-35)
held = repo.bank.existing_lines([f"bank:{bank_account_id}:{p.entry_ref}" for p in parsed])
fresh = [p for p in parsed if f"bank:{bank_account_id}:{p.entry_ref}" not in held]
mine = sorted(repo.bank.statements(bank_account_id), key=lambda s_: (s_["to_date"] or day_, s_["statement_id"]))
if not fresh and mine and Decimal(str(mine[-1]["closing_balance"])) == closing:
return {"lines": len(parsed), "created": 0, "skipped_duplicate": len(parsed), "lines_created": 0, "statement": mine[-1]["statement_id"],
"statement_lifecycle": mine[-1]["lifecycle"], "reconcile": {"suggested": 0, "rule_proposals": 0, "auto_posted": 0},
"consent": _consent_view(consent, today), "replay": True}
# one statement per delivery with its own opening and closing: the first of a day is FEED-<day>, a later one FEED-<day>-<n>
base = f"FEED-{day_.isoformat()}"
used = [x for x in repo.bank.statements(bank_account_id) if x["statement_id"].split(":", 1)[-1] == base
or x["statement_id"].split(":", 1)[-1].startswith(base + "-")]
ref = base if not used else f"{base}-{len(used) + 1}"
feed = bankfile.Statement(fresh, ref, closing - reconcile.signed_total(fresh), closing,
min((p.booked_on for p in fresh), default=day_), day_)
out = st.import_parsed(repo, caller, "feed", feed, "", bank_account_id)
out["lines"] += len(held)
out["skipped_duplicate"] += len(held) # already held from an earlier delivery: one line each
out = st.after_import(out, caller)
out["consent"] = _consent_view(consent, today)
return out