dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/application/fx.py
"""The realised FX difference on an allocation (PAY-33, PAY-42 v3).
A foreign-currency receipt is booked at the ledger's rate on the day it arrived (kept on the receipt); the invoice it settles was
booked at its own FROZEN rate (invoicing's totals route). When the two differ, settling leaves a functional-currency residue on the
receivable: (receipt rate - invoice rate) x the allocated amount. It is posted when the allocation is written, in the SAME
transaction, as a journal keyed alloc:<payment_id>:<invoice_id>:<seq> (event fx): receivable against fx_gain when positive, fx_loss
when negative. The undo posts the opposite (event fx_reversal, keyed on the reversing row) from the amount KEPT in pay_fx_posting -
no rate is read again. A same-rate allocation, or one with no rate on either side, posts nothing. The ledger refusing the journal
refuses the allocation (nothing is allocated); no ledger installed is a reply, not an error.
"""
from decimal import ROUND_HALF_UP, Decimal
from typing import Optional
from brick_payments.domain.errors import RuleBroken
from brick_payments.domain.money import ZERO
from brick_payments.infrastructure import seams
CENT = Decimal("0.01")
def difference(receipt_rate: Optional[Decimal], invoice_rate: Optional[Decimal], amount: Decimal) -> Decimal:
"""(receipt rate - invoice rate) x amount, to the cent; zero when either rate is unknown or they are the same."""
if receipt_rate is None or invoice_rate is None or receipt_rate == invoice_rate:
return ZERO
return ((receipt_rate - invoice_rate) * amount).quantize(CENT, rounding=ROUND_HALF_UP)
def _functional(company: str, actor: str) -> Optional[str]:
"""The company's functional currency from the ledger; "" when no ledger is installed; None when it is but cannot say."""
got = seams.company(company, actor=actor) if company else None
if got == {}:
return ""
return str(got.get("functional_currency") or "") or None if got else None
def _post(repo, r, row, *, event: str, direction: str, amount: Decimal, rates: tuple, actor: str, at) -> dict:
"""One journal for one allocation row. Receivable Dr + fx_gain Cr (a gain), fx_loss Dr + receivable Cr (a loss); a reversal swaps."""
bank = repo.config.bank(r.bank_account_id) if r.bank_account_id else None
company = (bank or {}).get("company_code") or ""
functional = _functional(company, actor)
if functional == "":
return {"posted": False, "reason": "no ledger is installed, so the FX difference was not posted"}
if functional is None:
raise RuleBroken(f"The ledger could not give company {company or '(none)'}'s functional currency, so the FX difference cannot be posted. "
"Nothing was allocated; give the bank account a company and try again")
doc = f"alloc:{r.payment_id}:{row.invoice_id}:{row.seq}"
g = f"{abs(amount):.2f}"
receivable = {**seams.receivable_line(r.bp_id), "against_document": row.invoice_id}
pl = {"purpose": "fx_gain" if direction == "gain" else "fx_loss"}
# the original: a gain is Dr receivable / Cr fx_gain, a loss Dr fx_loss / Cr receivable; the reversal is the mirror image
dr, cr = (receivable, pl) if direction == "gain" else (pl, receivable)
if event == "fx_reversal":
dr, cr = cr, dr
lines = [{**dr, "debit": g}, {**cr, "credit": g}]
posted = seams.post_journal(doc, event, row.allocated_on, functional, f"FX difference {r.payment_id} / {row.invoice_id}", lines, actor, company)
if not posted.get("posted"):
if "no ledger" in str(posted.get("reason", "")):
return posted
raise RuleBroken(f"The ledger did not take the FX difference ({posted.get('reason', 'no reason given')}). Nothing was allocated; try again")
repo.fx.add({"payment_id": r.payment_id, "invoice_id": row.invoice_id, "seq": row.seq, "event": event, "direction": direction,
"source_document": doc, "journal_id": posted["journal_id"], "amount": abs(amount), "currency": functional,
"receipt_rate": rates[0], "invoice_rate": rates[1], "posted_on": row.allocated_on}, at)
return {"posted": True, "journal_id": posted["journal_id"], "event": event, "direction": direction, "amount": g, "currency": functional}
def post_allocations(repo, r, rows: list, docs: dict, actor: str, at) -> list:
"""The FX journals for freshly written receipt allocation rows (positive rows) and for their reversals (negative rows)."""
out = []
for row in rows:
if row.source_kind != "receipt":
continue
if row.amount > ZERO:
ir = docs[row.invoice_id].fx_rate if row.invoice_id in docs else None
if ir is None and r.fx_rate is not None and row.invoice_id in docs:
raise RuleBroken(f"Invoice {row.invoice_id} ({docs[row.invoice_id].currency}) came from invoicing without its frozen fx_rate, "
"so the realised FX difference cannot be computed (it would silently be 0). Nothing was allocated; "
"fix the invoice's rate in invoicing and try again")
diff = difference(r.fx_rate, ir, row.amount)
if diff != ZERO:
out.append({"invoice_id": row.invoice_id, "seq": row.seq,
**_post(repo, r, row, event="fx", direction="gain" if diff > ZERO else "loss", amount=diff, rates=(r.fx_rate, ir), actor=actor, at=at)})
elif row.reverses_seq is not None:
was = repo.fx.get(r.payment_id, row.invoice_id, row.reverses_seq)
if was is not None and was["event"] == "fx":
out.append({"invoice_id": row.invoice_id, "seq": row.seq,
**_post(repo, r, row, event="fx_reversal", direction=was["direction"], amount=Decimal(str(was["amount"])),
rates=(Decimal(str(was["receipt_rate"])), Decimal(str(was["invoice_rate"]))), actor=actor, at=at)})
return out