You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
161 lines
4.7 KiB
161 lines
4.7 KiB
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" |
|
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
|
|
|