83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
"""
|
|
Account Service — balance calculated from transactions.
|
|
Balance = sum of income - sum of expenses for an account,
|
|
plus any incoming transfers minus outgoing transfers.
|
|
"""
|
|
|
|
from decimal import Decimal
|
|
from sqlalchemy import func, case, or_, and_
|
|
from app.extensions import db
|
|
from app.models.account import Account
|
|
from app.models.transaction import Transaction
|
|
|
|
|
|
def calc_balance(account_id):
|
|
"""
|
|
Recalculate and persist the balance for a given account.
|
|
Uses a single aggregation query with CASE expressions instead of 4 queries.
|
|
"""
|
|
row = db.session.query(
|
|
func.coalesce(func.sum(case(
|
|
(and_(Transaction.account_id == account_id,
|
|
Transaction.transaction_type == 'income'), Transaction.amount),
|
|
else_=0
|
|
)), 0),
|
|
func.coalesce(func.sum(case(
|
|
(and_(Transaction.account_id == account_id,
|
|
Transaction.transaction_type == 'expense'), Transaction.amount),
|
|
else_=0
|
|
)), 0),
|
|
func.coalesce(func.sum(case(
|
|
(and_(Transaction.account_id == account_id,
|
|
Transaction.transaction_type == 'transfer'), Transaction.amount),
|
|
else_=0
|
|
)), 0),
|
|
func.coalesce(func.sum(case(
|
|
(and_(Transaction.to_account_id == account_id,
|
|
Transaction.transaction_type == 'transfer'), Transaction.amount),
|
|
else_=0
|
|
)), 0),
|
|
).filter(
|
|
or_(Transaction.account_id == account_id,
|
|
Transaction.to_account_id == account_id)
|
|
).one()
|
|
|
|
income, expense, transfer_out, transfer_in = (Decimal(str(v)) for v in row)
|
|
balance = income - expense - transfer_out + transfer_in
|
|
|
|
account = db.session.get(Account, account_id)
|
|
if account:
|
|
account.balance = balance
|
|
db.session.commit()
|
|
return balance
|
|
|
|
|
|
def recalc_all():
|
|
"""Recalculate balances for all accounts."""
|
|
for account in Account.query.filter_by(is_active=True).all():
|
|
calc_balance(account.id)
|
|
|
|
|
|
def get_total_assets():
|
|
"""Sum of all positive-balance accounts (non-credit)."""
|
|
result = db.session.query(
|
|
func.coalesce(func.sum(Account.balance), 0)
|
|
).filter(
|
|
Account.is_active == True,
|
|
Account.account_type != 'credit_card',
|
|
Account.balance > 0
|
|
).scalar()
|
|
return float(result)
|
|
|
|
|
|
def get_total_liabilities():
|
|
"""Sum of credit card balances (negative = owed)."""
|
|
result = db.session.query(
|
|
func.coalesce(func.sum(Account.balance), 0)
|
|
).filter(
|
|
Account.is_active == True,
|
|
Account.account_type == 'credit_card',
|
|
Account.balance < 0
|
|
).scalar()
|
|
return abs(float(result))
|