gizmoduck/finance/reports.py

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
    group_name: str | None
    category_name: str
    is_group_total: bool
    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", "category__group__name")
        .annotate(
            expense=_sum(-F("amount"), Q(kind=Transaction.Kind.EXPENSE)),
            refund=_sum("amount", Q(kind=Transaction.Kind.REFUND)),
        )
    )
    leaves = tuple(
        SpendingCategory(
            category_id=row["category_id"],
            group_name=row["category__group__name"],
            category_name=row["category__name"] or "Uncategorized",
            is_group_total=False,
            expense=row["expense"],
            refund=row["refund"],
            net_spending=row["expense"] - row["refund"],
        )
        for row in grouped
    )
    grouped_rows = []
    for group_name in sorted({row.group_name for row in leaves if row.group_name}):
        children = [row for row in leaves if row.group_name == group_name]
        grouped_rows.append(SpendingCategory(None, group_name, group_name, True,
            sum((row.expense for row in children), _ZERO), sum((row.refund for row in children), _ZERO),
            sum((row.net_spending for row in children), _ZERO)))
        grouped_rows.extend(sorted(children, key=lambda row: (-row.net_spending, row.category_name)))
    grouped_rows.extend(sorted((row for row in leaves if not row.group_name), key=lambda row: (-row.net_spending, row.category_name)))
    return tuple(grouped_rows)


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"
    elif month.year == date.max.year and month.month == 12:
        errors["month"] = "invalid_month"
    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