86 lines
2.5 KiB
Python
86 lines
2.5 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
|
|
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."""
|
|
# Income credited to this account
|
|
income = db.session.query(
|
|
func.coalesce(func.sum(Transaction.amount), 0)
|
|
).filter(
|
|
Transaction.account_id == account_id,
|
|
Transaction.transaction_type == 'income'
|
|
).scalar()
|
|
|
|
# Expenses debited from this account
|
|
expense = db.session.query(
|
|
func.coalesce(func.sum(Transaction.amount), 0)
|
|
).filter(
|
|
Transaction.account_id == account_id,
|
|
Transaction.transaction_type == 'expense'
|
|
).scalar()
|
|
|
|
# Transfers out (this account is source)
|
|
transfer_out = db.session.query(
|
|
func.coalesce(func.sum(Transaction.amount), 0)
|
|
).filter(
|
|
Transaction.account_id == account_id,
|
|
Transaction.transaction_type == 'transfer'
|
|
).scalar()
|
|
|
|
# Transfers in (this account is destination)
|
|
transfer_in = db.session.query(
|
|
func.coalesce(func.sum(Transaction.amount), 0)
|
|
).filter(
|
|
Transaction.to_account_id == account_id,
|
|
Transaction.transaction_type == 'transfer'
|
|
).scalar()
|
|
|
|
balance = Decimal(str(income)) - Decimal(str(expense)) \
|
|
- Decimal(str(transfer_out)) + Decimal(str(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))
|