dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/application/forecast.py
"""Cash position and forecast (PAY-29), DERIVED on read - a snapshot is kept only when someone saves one. Per currency, never summed across
currencies:
today's position each open account's latest statement closing balance, plus the unreconciled entries (live receipts on that account
that no statement line has yet claimed and that were received after that statement);
per week expected RECEIPTS - open invoices by due date shifted by the customer's average days late over the last 12 months
(never earlier than today), direct-debit collections on their collection dates; expected PAYMENTS - purchasing's open
bills by due date, tax's VAT and CIS liabilities; and the running balance.
Every figure lists the documents it is made of. A neighbour that cannot answer leaves its side out and says so (never a silent zero).
Not in this release: payment requests (PAY-36) and purchasing's approved payment runs (no route to read them yet)."""
import json
import uuid
from collections import defaultdict
from datetime import date, timedelta
from decimal import Decimal
from typing import Annotated, Optional
from pydantic import Field
from brick_payments.application import use_cases as uc
from brick_payments.application.inputs import Ref
from brick_payments.domain import settlement as stl
from brick_payments.domain.money import ZERO
from brick_payments.infrastructure import seams
PAY = uc.PAY
Weeks = Annotated[int, Field(ge=1, le=52)]
FORECAST = "pay:forecast:read"
HMRC_CURRENCY = "GBP"
def _avg_days_late(docs: list, by: dict, today: date) -> dict:
"""Per customer: the whole days late the invoices fully paid in the last 12 months were paid (never negative); no history = 0."""
late: dict = defaultdict(list)
since = today - timedelta(days=365)
for d in docs:
rows = by.get(d.invoice_id, [])
if not rows or d.due_date is None or stl.outstanding(d.total, rows) != ZERO:
continue
paid_on = max(r.allocated_on for r in rows if r.amount > ZERO)
if paid_on >= since:
late[d.bp_id].append(max(0, (paid_on - d.due_date).days))
return {bp: round(sum(v) / len(v)) for bp, v in late.items()}
def build(repo, caller: dict, weeks: int) -> dict:
today = uc._today(repo)
horizon = today + timedelta(days=7 * weeks)
pos: dict = defaultdict(lambda: {"total": ZERO, "items": []})
flows: dict = defaultdict(lambda: defaultdict(lambda: {"receipts": [], "payments": []})) # currency -> week -> {receipts, payments}
notes, beyond = [], defaultdict(int)
def add(currency: str, side: str, on: date, amount: Decimal, **item):
on = max(on, today)
if on >= horizon:
beyond[currency] += 1
return
flows[currency][(on - today).days // 7][side].append({"expected_on": on.isoformat(), "amount": f"{amount:.2f}", **item})
for b in repo.config.banks():
if b.get("closed_from") and b["closed_from"] <= today:
continue
sts = [s for s in repo.bank.statements(b["bank_account_id"]) if s["closing_balance"] is not None]
last = sts[-1] if sts else None
base = Decimal(str(last["closing_balance"])) if last else ZERO
pend = repo.deposits.live_receipts_of_account(b["bank_account_id"], last["to_date"] if last and last["to_date"] else None)
extra = sum((Decimal(str(r["amount"])) for r in pend), ZERO)
pos[b["currency"]]["total"] += base + extra
pos[b["currency"]]["items"].append({"kind": "account", "ref": b["bank_account_id"], "closing_balance": f"{base:.2f}",
"statement": last["statement_id"] if last else None, "unreconciled_entries": f"{extra:.2f}",
"entries": [r["payment_id"] for r in pend], "amount": f"{base + extra:.2f}"})
try: # receivables: open invoices
docs = seams.invoices(caller=caller)
by = uc._group(repo.allocations.for_invoices([d.invoice_id for d in docs]))
late = _avg_days_late(docs, by, today)
flying = repo.dd.invoices_in_flight()
for d in docs:
out = stl.outstanding(d.total, by.get(d.invoice_id, []))
if out > ZERO and d.invoice_id not in flying and d.due_date is not None:
add(d.currency, "receipts", d.due_date + timedelta(days=late.get(d.bp_id, 0)), out, kind="invoice", ref=d.invoice_id,
bp_id=d.bp_id, due_date=d.due_date.isoformat(), days_late_habit=late.get(d.bp_id, 0))
except seams.Unavailable as down:
notes.append({"side": "invoices", "reason": down.reason})
for run in repo.dd.runs(): # direct debits on their collection dates
if run["lifecycle"] in ("draft", "approved", "submitted") and run["collection_date"] >= today:
planned = [c for c in repo.dd.collections(run["run_id"]) if c["status"] == "planned"]
if planned:
add(run["currency"], "receipts", run["collection_date"], sum((Decimal(str(c["amount"])) for c in planned), ZERO),
kind="direct_debit_run", ref=run["run_id"], status=run["lifecycle"], collections=len(planned))
try: # payables: purchasing's open bills
for i in seams.open_bills(today, uc._who(caller)):
due = date.fromisoformat(str(i["due_date"])[:10])
add(str(i["currency"]).upper(), "payments", due, Decimal(str(i["outstanding"])), kind="bill", ref=i["bill_id"], bill_no=i.get("bill_no"),
bp_id=i.get("bp_id"), due_date=due.isoformat())
except seams.Unavailable as down:
notes.append({"side": "bills", "reason": down.reason})
except (KeyError, ValueError, TypeError):
notes.append({"side": "bills", "reason": "purchasing's open bills could not be read"})
try: # tax: VAT and CIS
data = seams.vat_liabilities(uc._who(caller))
for x in data.get("liabilities", []):
add(HMRC_CURRENCY, "payments", date.fromisoformat(str(x["due_date"])[:10]), Decimal(str(x["outstanding_amount"])),
kind="vat_liability", ref=f"{x['period_key']}:{x['charge_type']}", due_date=str(x["due_date"])[:10])
for x in data.get("cis_due", []):
add(HMRC_CURRENCY, "payments", date.fromisoformat(str(x["due_date"])[:10]), Decimal(str(x["amount"])), kind="cis_due",
ref=str(x["tax_month"]), due_date=str(x["due_date"])[:10])
except seams.Unavailable as down:
notes.append({"side": "tax", "reason": down.reason})
except (KeyError, ValueError, TypeError):
notes.append({"side": "tax", "reason": "tax's liabilities could not be read"})
out = []
for cur in sorted(set(pos) | set(flows)):
running = pos[cur]["total"] if cur in pos else ZERO
rows = []
for w in range(weeks):
f = flows[cur][w] if cur in flows and w in flows[cur] else {"receipts": [], "payments": []}
rin = sum((Decimal(i["amount"]) for i in f["receipts"]), ZERO)
pay = sum((Decimal(i["amount"]) for i in f["payments"]), ZERO)
running += rin - pay
rows.append({"week": w + 1, "from": (today + timedelta(days=7 * w)).isoformat(), "to": (today + timedelta(days=7 * w + 6)).isoformat(),
"expected_receipts": {"total": f"{rin:.2f}", "items": f["receipts"]}, "expected_payments": {"total": f"{pay:.2f}", "items": f["payments"]},
"net": f"{rin - pay:.2f}", "running_balance": f"{running:.2f}"})
out.append({"currency": cur, "position": {"total": f"{pos[cur]['total']:.2f}" if cur in pos else "0.00", "items": pos[cur]["items"] if cur in pos else []},
"weeks": rows, "beyond_horizon": beyond.get(cur, 0)})
return {"as_of": today.isoformat(), "weeks": weeks, "currencies": out, "unavailable": notes,
"not_included": ["payment requests (PAY-36)", "purchasing's approved payment runs"]}
@PAY.use_case(kind="query", route="GET /cash-forecast", permission=FORECAST)
def cash_forecast(caller: dict, weeks: Optional[Weeks] = None) -> dict:
"""Cash position: per currency, today's position and, week by week for forecast_weeks, expected receipts and payments with the running
balance; every figure names the documents it is made of. Never summed across currencies. Derived on read."""
with uc.repository() as repo, uc._decide():
n = weeks or repo.settings.current().forecast_weeks
return build(repo, caller, n)
@PAY.use_case(kind="command", route="POST /cash-forecast/snapshots", permission=FORECAST)
def save_cash_forecast(caller: dict, weeks: Optional[Weeks] = None) -> dict:
"""Keep the forecast as it reads now (a snapshot), to compare with what happened. Nothing else is stored."""
with uc.repository() as repo, uc._decide():
n = weeks or repo.settings.current().forecast_weeks
body = build(repo, caller, n)
today, who = uc._today(repo), uc._who(caller)
sid = f"forecast:{today.isoformat()}:{uuid.uuid4().hex[:8]}"
repo.accounts.add_snapshot({"snapshot_id": sid, "saved_on": today, "saved_by": who, "body": json.dumps(body)}, uc._now())
return {"snapshot_id": sid, "saved_on": today.isoformat(), "saved_by": who, "weeks": n}
@PAY.use_case(kind="query", route="GET /cash-forecast/snapshots", permission=FORECAST)
def list_cash_forecasts(caller: dict) -> dict:
"""The forecasts someone saved, newest first."""
with uc.repository() as repo:
return {"snapshots": [{"snapshot_id": s["snapshot_id"], "saved_on": s["saved_on"].isoformat(), "saved_by": s["saved_by"]} for s in repo.accounts.snapshots()]}
@PAY.use_case(kind="query", route="GET /cash-forecast/snapshots/{snapshot_id}", permission=FORECAST)
def get_cash_forecast(caller: dict, snapshot_id: Ref) -> dict:
"""One saved forecast as it read when it was saved."""
from erp_platform.bricks import Refused
with uc.repository() as repo:
s = next((x for x in repo.accounts.snapshots() if x["snapshot_id"] == snapshot_id), None)
if s is None:
raise Refused(404, f"no saved forecast {snapshot_id}")
return {"snapshot_id": s["snapshot_id"], "saved_on": s["saved_on"].isoformat(), "saved_by": s["saved_by"], "forecast": json.loads(s["body"])}