147 lines
4.8 KiB
Python
147 lines
4.8 KiB
Python
"""
|
|
Budget Service — calculates spending vs budget limits per category per month.
|
|
"""
|
|
|
|
from decimal import Decimal
|
|
from sqlalchemy import func
|
|
from app.extensions import db
|
|
from app.models.budget import Budget
|
|
from app.models.transaction import Transaction
|
|
from app.models.category import Category
|
|
import calendar
|
|
from datetime import date
|
|
|
|
|
|
def get_month_spending(category_id, month_str):
|
|
"""
|
|
Return total spending for a category in a given month.
|
|
month_str: 'YYYY-MM'
|
|
"""
|
|
year, month = map(int, month_str.split('-'))
|
|
last_day = calendar.monthrange(year, month)[1]
|
|
date_from = date(year, month, 1)
|
|
date_to = date(year, month, last_day)
|
|
|
|
result = db.session.query(
|
|
func.coalesce(func.sum(Transaction.amount), 0)
|
|
).filter(
|
|
Transaction.category_id == category_id,
|
|
Transaction.transaction_type == 'expense',
|
|
Transaction.date >= date_from,
|
|
Transaction.date <= date_to,
|
|
).scalar()
|
|
|
|
return float(result)
|
|
|
|
|
|
def get_budget_summary(month_str):
|
|
"""
|
|
Return list of dicts with budget vs actual for each budgeted category.
|
|
Also includes unbudgeted categories that have spending.
|
|
"""
|
|
year, month = map(int, month_str.split('-'))
|
|
last_day = calendar.monthrange(year, month)[1]
|
|
date_from = date(year, month, 1)
|
|
date_to = date(year, month, last_day)
|
|
|
|
# All budgets for this month
|
|
budgets = Budget.query.filter_by(month=month_str).all()
|
|
budgeted_cat_ids = {b.category_id for b in budgets}
|
|
|
|
# Spending per category this month
|
|
spending_rows = db.session.query(
|
|
Transaction.category_id,
|
|
func.sum(Transaction.amount).label('total')
|
|
).filter(
|
|
Transaction.transaction_type == 'expense',
|
|
Transaction.date >= date_from,
|
|
Transaction.date <= date_to,
|
|
Transaction.category_id != None,
|
|
).group_by(Transaction.category_id).all()
|
|
|
|
spending_map = {row.category_id: float(row.total) for row in spending_rows}
|
|
|
|
summary = []
|
|
|
|
# Budgeted categories
|
|
for b in budgets:
|
|
spent = spending_map.get(b.category_id, 0.0)
|
|
limit = float(b.limit_amount) + float(b.rollover_amount or 0)
|
|
remaining = limit - spent
|
|
pct = min(round((spent / limit * 100), 1), 100) if limit > 0 else 0
|
|
summary.append({
|
|
'budget': b,
|
|
'category': b.category,
|
|
'spent': spent,
|
|
'limit': limit,
|
|
'remaining': remaining,
|
|
'pct': pct,
|
|
'is_over': spent > limit,
|
|
'has_budget': True,
|
|
})
|
|
|
|
# Unbudgeted categories with spending
|
|
for cat_id, spent in spending_map.items():
|
|
if cat_id not in budgeted_cat_ids:
|
|
cat = db.session.get(Category, cat_id)
|
|
if cat:
|
|
summary.append({
|
|
'budget': None,
|
|
'category': cat,
|
|
'spent': spent,
|
|
'limit': None,
|
|
'remaining': None,
|
|
'pct': None,
|
|
'is_over': False,
|
|
'has_budget': False,
|
|
})
|
|
|
|
# Sort: budgeted first (by % used desc), then unbudgeted
|
|
summary.sort(key=lambda x: (not x['has_budget'], -(x['pct'] or 0)))
|
|
return summary
|
|
|
|
|
|
def get_total_budget(month_str):
|
|
"""Total budgeted amount for a month."""
|
|
budgets = Budget.query.filter_by(month=month_str).all()
|
|
return sum(float(b.limit_amount) + float(b.rollover_amount or 0) for b in budgets)
|
|
|
|
|
|
def get_total_spent(month_str):
|
|
"""Total expense spending for a month (all categories)."""
|
|
year, month = map(int, month_str.split('-'))
|
|
last_day = calendar.monthrange(year, month)[1]
|
|
result = db.session.query(
|
|
func.coalesce(func.sum(Transaction.amount), 0)
|
|
).filter(
|
|
Transaction.transaction_type == 'expense',
|
|
Transaction.date >= date(year, month, 1),
|
|
Transaction.date <= date(year, month, last_day),
|
|
).scalar()
|
|
return float(result)
|
|
|
|
|
|
def apply_rollovers(from_month, to_month):
|
|
"""
|
|
Copy budgets from one month to next, applying rollover amounts.
|
|
Call on 1st of each month.
|
|
"""
|
|
from_budgets = Budget.query.filter_by(month=from_month).all()
|
|
for fb in from_budgets:
|
|
existing = Budget.query.filter_by(
|
|
month=to_month, category_id=fb.category_id
|
|
).first()
|
|
if not existing:
|
|
spent = get_month_spending(fb.category_id, from_month)
|
|
limit = float(fb.limit_amount)
|
|
rollover = max(limit - spent, 0) if fb.rollover_enabled else 0
|
|
new_budget = Budget(
|
|
category_id=fb.category_id,
|
|
month=to_month,
|
|
limit_amount=fb.limit_amount,
|
|
rollover_enabled=fb.rollover_enabled,
|
|
rollover_amount=rollover,
|
|
)
|
|
db.session.add(new_budget)
|
|
db.session.commit()
|