54 lines
2.2 KiB
Python
54 lines
2.2 KiB
Python
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.tables import Account, Transaction
|
|
|
|
|
|
def _confirmed_sum(session: Session, account_id: int,
|
|
after: date | None, upto: date | None) -> Decimal:
|
|
"""Summe bestaetigter Buchungen fuer `account_id` mit `after < booking_date
|
|
<= upto` (Raender werden nur gesetzt, wenn nicht None)."""
|
|
q = select(func.coalesce(func.sum(Transaction.amount), 0)).where(
|
|
Transaction.account_id == account_id, Transaction.status == "confirmed")
|
|
if after is not None:
|
|
q = q.where(Transaction.booking_date > after)
|
|
if upto is not None:
|
|
q = q.where(Transaction.booking_date <= upto)
|
|
return Decimal(session.execute(q).scalar_one())
|
|
|
|
|
|
def account_balance(session: Session, account: Account, at: date | None = None) -> Decimal:
|
|
"""Kontostand am Ende des Tages `at` (Default: heute).
|
|
|
|
Konto-Saldo-Anker (Ausbaustufe 3 Task 1) ersetzt die bisherige
|
|
Statement-closing-Logik ersatzlos. Konvention: der Anker gilt PER
|
|
TAGESENDE des Ankerdatums - er enthaelt bereits alle Buchungen bis
|
|
einschliesslich diesem Tag.
|
|
|
|
Mit Anker: anchor_balance + Sum(tx: anchor_date < booking_date <= at)
|
|
- Sum(tx: at < booking_date <= anchor_date)
|
|
(jeweils nur eine der beiden Summen ist nicht-leer, je nachdem ob `at`
|
|
vor oder nach dem Anker liegt; bei at == anchor_date sind beide leer.)
|
|
|
|
Ohne Anker: Sum(tx: booking_date <= at) - bisheriges Verhalten mit Basis 0.
|
|
"""
|
|
if at is None:
|
|
at = date.today()
|
|
|
|
if account.anchor_date is not None and account.anchor_balance is not None:
|
|
anchor_date = account.anchor_date
|
|
anchor_balance = Decimal(account.anchor_balance)
|
|
if at >= anchor_date:
|
|
return anchor_balance + _confirmed_sum(session, account.id, anchor_date, at)
|
|
return anchor_balance - _confirmed_sum(session, account.id, at, anchor_date)
|
|
|
|
return _confirmed_sum(session, account.id, None, at)
|
|
|
|
|
|
def total_balance(session: Session, at: date | None = None) -> Decimal:
|
|
accounts = session.execute(select(Account)).scalars().all()
|
|
return sum((account_balance(session, a, at) for a in accounts), Decimal("0"))
|