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.
96 lines
3.1 KiB
96 lines
3.1 KiB
"""Read models and shared calculations for reconciliation checkpoints.""" |
|
|
|
from dataclasses import dataclass |
|
from datetime import date, datetime |
|
from decimal import Decimal |
|
from uuid import UUID |
|
|
|
from django.db.models import DecimalField, F, Sum, Value, Window |
|
from django.db.models.functions import Coalesce, RowNumber |
|
|
|
from finance.models import ReconciliationCheckpoint, Transaction |
|
|
|
|
|
_MONEY_FIELD = DecimalField(max_digits=24, decimal_places=2) |
|
_ZERO = Decimal("0.00") |
|
|
|
|
|
@dataclass(frozen=True) |
|
class CheckpointStatus: |
|
checkpoint_id: UUID |
|
account_id: UUID |
|
account_name: str |
|
currency: str |
|
as_of_date: date |
|
created_at: datetime |
|
institution_balance: Decimal |
|
snapshot_ledger_balance: Decimal |
|
snapshot_difference: Decimal |
|
live_ledger_balance: Decimal |
|
live_difference: Decimal |
|
state: str |
|
|
|
|
|
def ledger_balance_through(account_id: UUID, as_of_date: date) -> Decimal: |
|
from finance.services.base import money |
|
|
|
total = Transaction.objects.filter( |
|
account_id=account_id, |
|
posting_date__lte=as_of_date, |
|
).aggregate( |
|
total=Coalesce( |
|
Sum("amount"), |
|
Value(_ZERO), |
|
output_field=_MONEY_FIELD, |
|
) |
|
)["total"] |
|
return money(total, allow_zero=True) |
|
|
|
|
|
def checkpoint_status(checkpoint: ReconciliationCheckpoint) -> CheckpointStatus: |
|
account = checkpoint.account |
|
live_ledger_balance = ledger_balance_through(checkpoint.account_id, checkpoint.as_of_date) |
|
live_difference = checkpoint.institution_balance - live_ledger_balance |
|
if live_difference == _ZERO: |
|
state = "reconciled" |
|
elif checkpoint.difference_at_creation == _ZERO: |
|
state = "changed_since_reconciliation" |
|
else: |
|
state = "difference" |
|
return CheckpointStatus( |
|
checkpoint_id=checkpoint.pk, |
|
account_id=account.pk, |
|
account_name=account.name, |
|
currency=account.currency, |
|
as_of_date=checkpoint.as_of_date, |
|
created_at=checkpoint.created_at, |
|
institution_balance=checkpoint.institution_balance, |
|
snapshot_ledger_balance=checkpoint.ledger_balance_at_creation, |
|
snapshot_difference=checkpoint.difference_at_creation, |
|
live_ledger_balance=live_ledger_balance, |
|
live_difference=live_difference, |
|
state=state, |
|
) |
|
|
|
|
|
def checkpoint_history(account_id: UUID) -> list[CheckpointStatus]: |
|
checkpoints = ReconciliationCheckpoint.objects.filter( |
|
account_id=account_id, |
|
).select_related("account").order_by("-as_of_date", "-created_at", "-id") |
|
return [checkpoint_status(checkpoint) for checkpoint in checkpoints] |
|
|
|
|
|
def latest_checkpoint_statuses(currency: str) -> list[CheckpointStatus]: |
|
checkpoints = ReconciliationCheckpoint.objects.filter( |
|
account__archived=False, |
|
account__currency=currency, |
|
).annotate( |
|
account_rank=Window( |
|
expression=RowNumber(), |
|
partition_by=[F("account_id")], |
|
order_by=["-as_of_date", "-created_at", "-id"], |
|
) |
|
).filter(account_rank=1).select_related("account").order_by( |
|
"-as_of_date", "-created_at", "-id" |
|
) |
|
return [checkpoint_status(checkpoint) for checkpoint in checkpoints]
|
|
|