Codex
2 months ago
5 changed files with 535 additions and 0 deletions
@ -0,0 +1,24 @@
|
||||
from django.db import migrations |
||||
|
||||
|
||||
class Migration(migrations.Migration): |
||||
dependencies = [ |
||||
("finance", "0001_initial"), |
||||
] |
||||
|
||||
operations = [ |
||||
migrations.RunSQL( |
||||
sql=( |
||||
"CREATE INDEX finance_tx_posting_date_idx " |
||||
"ON finance_transaction (posting_date)" |
||||
), |
||||
reverse_sql="DROP INDEX finance_tx_posting_date_idx", |
||||
), |
||||
migrations.RunSQL( |
||||
sql=( |
||||
"CREATE INDEX finance_tx_account_posting_idx " |
||||
"ON finance_transaction (account_id, posting_date)" |
||||
), |
||||
reverse_sql="DROP INDEX finance_tx_account_posting_idx", |
||||
), |
||||
] |
||||
@ -0,0 +1,124 @@
|
||||
from dataclasses import dataclass |
||||
from datetime import date |
||||
import re |
||||
from typing import Mapping |
||||
from uuid import UUID |
||||
|
||||
from django.db.models import QuerySet |
||||
|
||||
from finance.models import Transaction |
||||
|
||||
|
||||
_CURRENCY_RE = re.compile(r"[A-Z]{3}", re.ASCII) |
||||
|
||||
|
||||
class QueryInputError(ValueError): |
||||
def __init__(self, errors: dict[str, str]): |
||||
super().__init__("Invalid transaction filters") |
||||
self.errors = errors |
||||
|
||||
|
||||
@dataclass(frozen=True) |
||||
class TransactionFilters: |
||||
date_from: date | None = None |
||||
date_to: date | None = None |
||||
account_id: UUID | None = None |
||||
kind: str | None = None |
||||
payee_id: UUID | None = None |
||||
category_id: UUID | None = None |
||||
currency: str | None = None |
||||
uncategorized: bool | None = None |
||||
|
||||
@classmethod |
||||
def from_mapping(cls, mapping: Mapping[str, object]) -> "TransactionFilters": |
||||
errors: dict[str, str] = {} |
||||
date_from = _parse_date(mapping.get("date_from"), "date_from", errors) |
||||
date_to = _parse_date(mapping.get("date_to"), "date_to", errors) |
||||
account_id = _parse_uuid(mapping.get("account"), "account", errors) |
||||
payee_id = _parse_uuid(mapping.get("payee"), "payee", errors) |
||||
category_id = _parse_uuid(mapping.get("category"), "category", errors) |
||||
|
||||
kind = _optional_string(mapping.get("kind")) |
||||
if kind is not None and kind not in Transaction.Kind.values: |
||||
errors["kind"] = "invalid_choice" |
||||
|
||||
currency = _optional_string(mapping.get("currency")) |
||||
if currency is not None and _CURRENCY_RE.fullmatch(currency) is None: |
||||
errors["currency"] = "invalid_currency" |
||||
|
||||
uncategorized_value = _optional_string(mapping.get("uncategorized")) |
||||
uncategorized = None |
||||
if uncategorized_value == "yes": |
||||
uncategorized = True |
||||
elif uncategorized_value == "no": |
||||
uncategorized = False |
||||
elif uncategorized_value is not None: |
||||
errors["uncategorized"] = "invalid_choice" |
||||
|
||||
if date_from is not None and date_to is not None and date_to < date_from: |
||||
errors["date_to"] = "before_date_from" |
||||
|
||||
if errors: |
||||
raise QueryInputError(errors) |
||||
return cls( |
||||
date_from=date_from, |
||||
date_to=date_to, |
||||
account_id=account_id, |
||||
kind=kind, |
||||
payee_id=payee_id, |
||||
category_id=category_id, |
||||
currency=currency, |
||||
uncategorized=uncategorized, |
||||
) |
||||
|
||||
|
||||
def filtered_transactions(filters: TransactionFilters) -> QuerySet[Transaction]: |
||||
rows = Transaction.objects.select_related( |
||||
"account", "payee", "category", "statement", "refund_of", "installment_plan" |
||||
) |
||||
if filters.date_from is not None: |
||||
rows = rows.filter(posting_date__gte=filters.date_from) |
||||
if filters.date_to is not None: |
||||
rows = rows.filter(posting_date__lte=filters.date_to) |
||||
if filters.account_id is not None: |
||||
rows = rows.filter(account_id=filters.account_id) |
||||
if filters.kind is not None: |
||||
rows = rows.filter(kind=filters.kind) |
||||
if filters.payee_id is not None: |
||||
rows = rows.filter(payee_id=filters.payee_id) |
||||
if filters.category_id is not None: |
||||
rows = rows.filter(category_id=filters.category_id) |
||||
if filters.currency is not None: |
||||
rows = rows.filter(account__currency=filters.currency) |
||||
if filters.uncategorized is not None: |
||||
rows = rows.filter(category__isnull=filters.uncategorized) |
||||
return rows.order_by("-posting_date", "-id") |
||||
|
||||
|
||||
def _optional_string(value: object) -> str | None: |
||||
if value is None: |
||||
return None |
||||
text = str(value).strip() |
||||
return text or None |
||||
|
||||
|
||||
def _parse_date(value: object, field: str, errors: dict[str, str]) -> date | None: |
||||
text = _optional_string(value) |
||||
if text is None: |
||||
return None |
||||
try: |
||||
return date.fromisoformat(text) |
||||
except ValueError: |
||||
errors[field] = "invalid_date" |
||||
return None |
||||
|
||||
|
||||
def _parse_uuid(value: object, field: str, errors: dict[str, str]) -> UUID | None: |
||||
text = _optional_string(value) |
||||
if text is None: |
||||
return None |
||||
try: |
||||
return UUID(text) |
||||
except (ValueError, AttributeError): |
||||
errors[field] = "invalid_uuid" |
||||
return None |
||||
@ -0,0 +1,159 @@
|
||||
from dataclasses import dataclass |
||||
from datetime import date |
||||
from decimal import Decimal |
||||
import re |
||||
from uuid import UUID |
||||
|
||||
from django.db.models import DecimalField, F, Q, Sum, Value |
||||
from django.db.models.functions import Coalesce |
||||
|
||||
from finance.models import Account, Transaction |
||||
|
||||
|
||||
_CURRENCY_RE = re.compile(r"[A-Z]{3}", re.ASCII) |
||||
_MONEY_FIELD = DecimalField(max_digits=24, decimal_places=2) |
||||
_ZERO = Decimal("0.00") |
||||
|
||||
|
||||
class ReportInputError(ValueError): |
||||
def __init__(self, errors: dict[str, str]): |
||||
super().__init__("Invalid report inputs") |
||||
self.errors = errors |
||||
|
||||
|
||||
@dataclass(frozen=True) |
||||
class MonthlySummary: |
||||
month: date |
||||
currency: str |
||||
inflow: Decimal |
||||
expense: Decimal |
||||
refund: Decimal |
||||
net_spending: Decimal |
||||
net_activity: Decimal |
||||
|
||||
|
||||
@dataclass(frozen=True) |
||||
class SpendingCategory: |
||||
category_id: UUID | None |
||||
category_name: str |
||||
expense: Decimal |
||||
refund: Decimal |
||||
net_spending: Decimal |
||||
|
||||
|
||||
@dataclass(frozen=True) |
||||
class AccountBalance: |
||||
account_id: UUID |
||||
account_name: str |
||||
balance: Decimal |
||||
|
||||
|
||||
def available_currencies() -> tuple[str, ...]: |
||||
return tuple( |
||||
Account.objects.order_by("currency") |
||||
.values_list("currency", flat=True) |
||||
.distinct() |
||||
) |
||||
|
||||
|
||||
def monthly_summary(month: date, currency: str) -> MonthlySummary: |
||||
month, currency, month_after = _validated_period(month, currency) |
||||
totals = _monthly_rows(month, month_after, currency).aggregate( |
||||
inflow=_sum("amount", Q(kind=Transaction.Kind.INCOME)), |
||||
expense=_sum(-F("amount"), Q(kind=Transaction.Kind.EXPENSE)), |
||||
refund=_sum("amount", Q(kind=Transaction.Kind.REFUND)), |
||||
) |
||||
inflow = totals["inflow"] |
||||
expense = totals["expense"] |
||||
refund = totals["refund"] |
||||
net_spending = expense - refund |
||||
return MonthlySummary( |
||||
month=month, |
||||
currency=currency, |
||||
inflow=inflow, |
||||
expense=expense, |
||||
refund=refund, |
||||
net_spending=net_spending, |
||||
net_activity=inflow - net_spending, |
||||
) |
||||
|
||||
|
||||
def spending_by_category(month: date, currency: str) -> tuple[SpendingCategory, ...]: |
||||
month, currency, month_after = _validated_period(month, currency) |
||||
grouped = ( |
||||
_monthly_rows(month, month_after, currency) |
||||
.filter(kind__in=[Transaction.Kind.EXPENSE, Transaction.Kind.REFUND]) |
||||
.values("category_id", "category__name") |
||||
.annotate( |
||||
expense=_sum(-F("amount"), Q(kind=Transaction.Kind.EXPENSE)), |
||||
refund=_sum("amount", Q(kind=Transaction.Kind.REFUND)), |
||||
) |
||||
) |
||||
rows = tuple( |
||||
SpendingCategory( |
||||
category_id=row["category_id"], |
||||
category_name=row["category__name"] or "Uncategorized", |
||||
expense=row["expense"], |
||||
refund=row["refund"], |
||||
net_spending=row["expense"] - row["refund"], |
||||
) |
||||
for row in grouped |
||||
) |
||||
return tuple(sorted(rows, key=lambda row: (-row.net_spending, row.category_name))) |
||||
|
||||
|
||||
def account_balances(month: date, currency: str) -> tuple[AccountBalance, ...]: |
||||
month, currency, month_after = _validated_period(month, currency) |
||||
del month |
||||
rows = ( |
||||
Account.objects.filter(currency=currency) |
||||
.annotate( |
||||
report_balance=Coalesce( |
||||
Sum("transactions__amount", filter=Q(transactions__posting_date__lt=month_after)), |
||||
Value(_ZERO), |
||||
output_field=_MONEY_FIELD, |
||||
) |
||||
) |
||||
.order_by("name", "id") |
||||
) |
||||
return tuple( |
||||
AccountBalance( |
||||
account_id=row.pk, |
||||
account_name=row.name, |
||||
balance=row.report_balance, |
||||
) |
||||
for row in rows |
||||
) |
||||
|
||||
|
||||
def _monthly_rows(month: date, month_after: date, currency: str): |
||||
return Transaction.objects.filter( |
||||
posting_date__gte=month, |
||||
posting_date__lt=month_after, |
||||
account__currency=currency, |
||||
) |
||||
|
||||
|
||||
def _sum(expression, condition: Q): |
||||
return Coalesce( |
||||
Sum(expression, filter=condition), |
||||
Value(_ZERO), |
||||
output_field=_MONEY_FIELD, |
||||
) |
||||
|
||||
|
||||
def _validated_period(month: date, currency: str) -> tuple[date, str, date]: |
||||
errors: dict[str, str] = {} |
||||
if not isinstance(month, date): |
||||
errors["month"] = "invalid_date" |
||||
elif month.day != 1: |
||||
errors["month"] = "must_be_first_day" |
||||
if not isinstance(currency, str) or _CURRENCY_RE.fullmatch(currency) is None: |
||||
errors["currency"] = "invalid_currency" |
||||
if errors: |
||||
raise ReportInputError(errors) |
||||
if month.month == 12: |
||||
month_after = date(month.year + 1, 1, 1) |
||||
else: |
||||
month_after = date(month.year, month.month + 1, 1) |
||||
return month, currency, month_after |
||||
@ -0,0 +1,107 @@
|
||||
from datetime import date |
||||
|
||||
import pytest |
||||
|
||||
from finance.models import Account, Category, Payee, Transaction |
||||
from finance.queries import ( |
||||
QueryInputError, |
||||
TransactionFilters, |
||||
filtered_transactions, |
||||
) |
||||
|
||||
|
||||
pytestmark = pytest.mark.django_db |
||||
|
||||
|
||||
def make_account(name="Synthetic Bank", currency="BRL"): |
||||
return Account.objects.create(name=name, kind=Account.Kind.BANK, currency=currency) |
||||
|
||||
|
||||
def make_transaction(account, *, posting_date, amount="-10.00", **values): |
||||
return Transaction.objects.create( |
||||
account=account, |
||||
amount=amount, |
||||
posting_date=posting_date, |
||||
kind=values.pop("kind", Transaction.Kind.EXPENSE), |
||||
**values, |
||||
) |
||||
|
||||
|
||||
def test_filtered_transactions_combines_filters_and_orders_stably(): |
||||
account = make_account() |
||||
category = Category.objects.create(name="Synthetic Food", kind=Category.Kind.EXPENSE) |
||||
older = make_transaction(account, posting_date=date(2026, 1, 5)) |
||||
newer = make_transaction(account, posting_date=date(2026, 1, 20)) |
||||
make_transaction(account, posting_date=date(2026, 1, 25), category=category) |
||||
make_transaction(account, posting_date=date(2026, 2, 1)) |
||||
make_transaction(make_account("Synthetic USD", "USD"), posting_date=date(2026, 1, 22)) |
||||
|
||||
filters = TransactionFilters.from_mapping({ |
||||
"date_from": "2026-01-01", |
||||
"date_to": "2026-01-31", |
||||
"currency": "BRL", |
||||
"uncategorized": "yes", |
||||
}) |
||||
|
||||
rows = list(filtered_transactions(filters)) |
||||
|
||||
assert [row.pk for row in rows] == [newer.pk, older.pk] |
||||
|
||||
|
||||
def test_filtered_transactions_supports_reference_kind_and_categorized_filters(): |
||||
account = make_account() |
||||
other_account = make_account("Synthetic Other") |
||||
category = Category.objects.create(name="Synthetic Food", kind=Category.Kind.EXPENSE) |
||||
payee = Payee.objects.create(name="Synthetic Merchant") |
||||
expected = make_transaction( |
||||
account, |
||||
posting_date=date(2026, 1, 10), |
||||
category=category, |
||||
payee=payee, |
||||
) |
||||
make_transaction(account, posting_date=date(2026, 1, 11), category=category) |
||||
make_transaction(other_account, posting_date=date(2026, 1, 12), category=category, payee=payee) |
||||
|
||||
filters = TransactionFilters.from_mapping({ |
||||
"account": str(account.pk), |
||||
"category": str(category.pk), |
||||
"payee": str(payee.pk), |
||||
"kind": "EXPENSE", |
||||
"uncategorized": "no", |
||||
}) |
||||
|
||||
assert list(filtered_transactions(filters)) == [expected] |
||||
|
||||
|
||||
@pytest.mark.parametrize( |
||||
("mapping", "errors"), |
||||
[ |
||||
({"date_from": "31/01/2026"}, {"date_from": "invalid_date"}), |
||||
({"date_to": "not-a-date"}, {"date_to": "invalid_date"}), |
||||
({"account": "bad"}, {"account": "invalid_uuid"}), |
||||
({"category": "bad"}, {"category": "invalid_uuid"}), |
||||
({"payee": "bad"}, {"payee": "invalid_uuid"}), |
||||
({"kind": "UNKNOWN"}, {"kind": "invalid_choice"}), |
||||
({"currency": "brl"}, {"currency": "invalid_currency"}), |
||||
({"uncategorized": "maybe"}, {"uncategorized": "invalid_choice"}), |
||||
], |
||||
) |
||||
def test_invalid_filter_is_explicit(mapping, errors): |
||||
with pytest.raises(QueryInputError) as caught: |
||||
TransactionFilters.from_mapping(mapping) |
||||
|
||||
assert caught.value.errors == errors |
||||
|
||||
|
||||
def test_filter_errors_are_reported_together_and_date_order_is_checked(): |
||||
with pytest.raises(QueryInputError) as caught: |
||||
TransactionFilters.from_mapping({ |
||||
"date_from": "2026-02-01", |
||||
"date_to": "2026-01-01", |
||||
"account": "bad", |
||||
}) |
||||
|
||||
assert caught.value.errors == { |
||||
"account": "invalid_uuid", |
||||
"date_to": "before_date_from", |
||||
} |
||||
@ -0,0 +1,121 @@
|
||||
from datetime import date |
||||
from decimal import Decimal |
||||
|
||||
import pytest |
||||
|
||||
from finance.models import Account, Category, Transaction |
||||
from finance.reports import ( |
||||
ReportInputError, |
||||
account_balances, |
||||
available_currencies, |
||||
monthly_summary, |
||||
spending_by_category, |
||||
) |
||||
|
||||
|
||||
pytestmark = pytest.mark.django_db |
||||
|
||||
|
||||
def make_account(name="Synthetic Bank", *, kind=Account.Kind.BANK, currency="BRL"): |
||||
return Account.objects.create(name=name, kind=kind, currency=currency) |
||||
|
||||
|
||||
def post(account, amount, kind, posting_date, *, category=None): |
||||
return Transaction.objects.create( |
||||
account=account, |
||||
amount=Decimal(amount), |
||||
kind=kind, |
||||
posting_date=posting_date, |
||||
category=category, |
||||
) |
||||
|
||||
|
||||
def test_monthly_summary_counts_card_purchase_once_and_refund_in_posting_month(): |
||||
bank = make_account() |
||||
card = make_account("Synthetic Card", kind=Account.Kind.CREDIT_CARD) |
||||
post(bank, "500.00", Transaction.Kind.OPENING, date(2025, 12, 31)) |
||||
post(bank, "1000.00", Transaction.Kind.INCOME, date(2026, 1, 2)) |
||||
post(card, "-300.00", Transaction.Kind.EXPENSE, date(2026, 1, 3)) |
||||
post(bank, "-300.00", Transaction.Kind.TRANSFER, date(2026, 1, 20)) |
||||
post(card, "300.00", Transaction.Kind.TRANSFER, date(2026, 1, 20)) |
||||
post(card, "50.00", Transaction.Kind.REFUND, date(2026, 1, 25)) |
||||
post(card, "20.00", Transaction.Kind.REFUND, date(2026, 2, 1)) |
||||
|
||||
result = monthly_summary(date(2026, 1, 1), "BRL") |
||||
|
||||
assert result.inflow == Decimal("1000.00") |
||||
assert result.expense == Decimal("300.00") |
||||
assert result.refund == Decimal("50.00") |
||||
assert result.net_spending == Decimal("250.00") |
||||
assert result.net_activity == Decimal("750.00") |
||||
|
||||
|
||||
def test_spending_has_uncategorized_bucket_and_never_combines_currency(): |
||||
bank = make_account() |
||||
usd = make_account("Synthetic USD", currency="USD") |
||||
food = Category.objects.create(name="Synthetic Food", kind=Category.Kind.EXPENSE) |
||||
post(bank, "-100.00", Transaction.Kind.EXPENSE, date(2026, 1, 2), category=food) |
||||
post(bank, "20.00", Transaction.Kind.REFUND, date(2026, 1, 3), category=food) |
||||
post(bank, "-20.00", Transaction.Kind.EXPENSE, date(2026, 1, 4)) |
||||
post(usd, "-999.00", Transaction.Kind.EXPENSE, date(2026, 1, 5), category=food) |
||||
|
||||
rows = spending_by_category(date(2026, 1, 1), "BRL") |
||||
|
||||
assert [(row.category_name, row.net_spending) for row in rows] == [ |
||||
("Synthetic Food", Decimal("80.00")), |
||||
("Uncategorized", Decimal("20.00")), |
||||
] |
||||
assert rows[0].expense == Decimal("100.00") |
||||
assert rows[0].refund == Decimal("20.00") |
||||
|
||||
|
||||
def test_empty_month_returns_explicit_zero_totals_and_no_spending_rows(): |
||||
result = monthly_summary(date(2026, 1, 1), "BRL") |
||||
|
||||
assert result.inflow == Decimal("0.00") |
||||
assert result.expense == Decimal("0.00") |
||||
assert result.refund == Decimal("0.00") |
||||
assert result.net_spending == Decimal("0.00") |
||||
assert result.net_activity == Decimal("0.00") |
||||
assert spending_by_category(date(2026, 1, 1), "BRL") == () |
||||
|
||||
|
||||
def test_account_balances_include_openings_and_transfers_through_month_end(): |
||||
beta = make_account("Synthetic Beta") |
||||
alpha = make_account("Synthetic Alpha") |
||||
usd = make_account("Synthetic USD", currency="USD") |
||||
post(alpha, "100.00", Transaction.Kind.OPENING, date(2025, 12, 1)) |
||||
post(alpha, "-20.00", Transaction.Kind.TRANSFER, date(2026, 1, 20)) |
||||
post(beta, "20.00", Transaction.Kind.TRANSFER, date(2026, 1, 20)) |
||||
post(beta, "-5.00", Transaction.Kind.EXPENSE, date(2026, 2, 1)) |
||||
post(usd, "1000.00", Transaction.Kind.OPENING, date(2025, 12, 1)) |
||||
|
||||
rows = account_balances(date(2026, 1, 1), "BRL") |
||||
|
||||
assert [(row.account_name, row.balance) for row in rows] == [ |
||||
("Synthetic Alpha", Decimal("80.00")), |
||||
("Synthetic Beta", Decimal("20.00")), |
||||
] |
||||
|
||||
|
||||
def test_available_currencies_are_distinct_and_sorted(): |
||||
make_account("Synthetic BRL", currency="BRL") |
||||
make_account("Synthetic USD", currency="USD") |
||||
make_account("Synthetic BRL 2", currency="BRL") |
||||
|
||||
assert available_currencies() == ("BRL", "USD") |
||||
|
||||
|
||||
@pytest.mark.parametrize( |
||||
("month", "currency", "errors"), |
||||
[ |
||||
(date(2026, 1, 2), "BRL", {"month": "must_be_first_day"}), |
||||
(date(2026, 1, 1), "brl", {"currency": "invalid_currency"}), |
||||
("2026-01", "BRL", {"month": "invalid_date"}), |
||||
], |
||||
) |
||||
def test_report_inputs_are_validated(month, currency, errors): |
||||
with pytest.raises(ReportInputError) as caught: |
||||
monthly_summary(month, currency) |
||||
|
||||
assert caught.value.errors == errors |
||||
Loading…
Reference in new issue