486 lines
17 KiB
Python
486 lines
17 KiB
Python
"""
|
||
Report Service — aggregates data for monthly, quarterly, yearly,
|
||
net worth history, category trends, and tax year reports.
|
||
"""
|
||
|
||
import calendar
|
||
from collections import defaultdict
|
||
from datetime import date, datetime, timedelta
|
||
from sqlalchemy import func, extract
|
||
from app.extensions import db
|
||
from app.models.transaction import Transaction
|
||
from app.models.category import Category
|
||
from app.models.account import Account
|
||
from app.models.net_worth_snapshot import NetWorthSnapshot
|
||
from app.models.investment import Investment
|
||
|
||
|
||
# ── Helpers ───────────────────────────────────────────────────────────────────
|
||
|
||
def _month_range(year, month):
|
||
last = calendar.monthrange(year, month)[1]
|
||
return date(year, month, 1), date(year, month, last)
|
||
|
||
|
||
def _quarter_range(year, quarter):
|
||
start_month = (quarter - 1) * 3 + 1
|
||
end_month = start_month + 2
|
||
_, last = calendar.monthrange(year, end_month)
|
||
return date(year, start_month, 1), date(year, end_month, last)
|
||
|
||
|
||
def _year_range(year):
|
||
return date(year, 1, 1), date(year, 12, 31)
|
||
|
||
|
||
def _totals(date_from, date_to):
|
||
"""Return (income, expense) totals for a date range."""
|
||
inc = db.session.query(
|
||
func.coalesce(func.sum(Transaction.amount), 0)
|
||
).filter(
|
||
Transaction.transaction_type == 'income',
|
||
Transaction.date >= date_from,
|
||
Transaction.date <= date_to,
|
||
).scalar()
|
||
|
||
exp = db.session.query(
|
||
func.coalesce(func.sum(Transaction.amount), 0)
|
||
).filter(
|
||
Transaction.transaction_type == 'expense',
|
||
Transaction.date >= date_from,
|
||
Transaction.date <= date_to,
|
||
).scalar()
|
||
|
||
return float(inc), float(exp)
|
||
|
||
|
||
def _category_breakdown(date_from, date_to, txn_type='expense'):
|
||
rows = db.session.query(
|
||
Category.name,
|
||
Category.color,
|
||
Category.icon,
|
||
func.sum(Transaction.amount).label('total')
|
||
).join(Transaction, Transaction.category_id == Category.id)\
|
||
.filter(
|
||
Transaction.transaction_type == txn_type,
|
||
Transaction.date >= date_from,
|
||
Transaction.date <= date_to,
|
||
).group_by(Category.id)\
|
||
.order_by(func.sum(Transaction.amount).desc())\
|
||
.all()
|
||
return [{'name': r.name, 'color': r.color, 'icon': r.icon, 'total': float(r.total)} for r in rows]
|
||
|
||
|
||
# ── Monthly report ────────────────────────────────────────────────────────────
|
||
|
||
def monthly_report(year, month):
|
||
date_from, date_to = _month_range(year, month)
|
||
income, expense = _totals(date_from, date_to)
|
||
expense_cats = _category_breakdown(date_from, date_to, 'expense')
|
||
income_cats = _category_breakdown(date_from, date_to, 'income')
|
||
|
||
transactions = Transaction.query\
|
||
.filter(
|
||
Transaction.transaction_type.in_(['income', 'expense']),
|
||
Transaction.date >= date_from,
|
||
Transaction.date <= date_to,
|
||
).order_by(Transaction.date.desc()).all()
|
||
|
||
return {
|
||
'period': f'{date_from.strftime("%B %Y")}',
|
||
'date_from': date_from,
|
||
'date_to': date_to,
|
||
'income': income,
|
||
'expense': expense,
|
||
'net': income - expense,
|
||
'savings_rate': round(((income - expense) / income * 100), 1) if income > 0 else 0,
|
||
'expense_categories': expense_cats,
|
||
'income_categories': income_cats,
|
||
'transactions': transactions,
|
||
'transaction_count': len(transactions),
|
||
}
|
||
|
||
|
||
# ── Quarterly report ──────────────────────────────────────────────────────────
|
||
|
||
def quarterly_report(year, quarter):
|
||
date_from, date_to = _quarter_range(year, quarter)
|
||
income, expense = _totals(date_from, date_to)
|
||
expense_cats = _category_breakdown(date_from, date_to, 'expense')
|
||
|
||
# Monthly breakdown within the quarter
|
||
months = []
|
||
start_month = (quarter - 1) * 3 + 1
|
||
for m in range(start_month, start_month + 3):
|
||
mf, mt = _month_range(year, m)
|
||
mi, me = _totals(mf, mt)
|
||
months.append({
|
||
'label': date(year, m, 1).strftime('%B'),
|
||
'income': mi,
|
||
'expense': me,
|
||
'net': mi - me,
|
||
})
|
||
|
||
return {
|
||
'period': f'Q{quarter} {year}',
|
||
'date_from': date_from,
|
||
'date_to': date_to,
|
||
'income': income,
|
||
'expense': expense,
|
||
'net': income - expense,
|
||
'savings_rate': round(((income - expense) / income * 100), 1) if income > 0 else 0,
|
||
'expense_categories': expense_cats,
|
||
'months': months,
|
||
}
|
||
|
||
|
||
# ── Yearly report ─────────────────────────────────────────────────────────────
|
||
|
||
def yearly_report(year):
|
||
date_from, date_to = _year_range(year)
|
||
income, expense = _totals(date_from, date_to)
|
||
expense_cats = _category_breakdown(date_from, date_to, 'expense')
|
||
income_cats = _category_breakdown(date_from, date_to, 'income')
|
||
|
||
# Monthly breakdown
|
||
months = []
|
||
for m in range(1, 13):
|
||
mf, mt = _month_range(year, m)
|
||
mi, me = _totals(mf, mt)
|
||
months.append({
|
||
'label': date(year, m, 1).strftime('%b'),
|
||
'income': mi,
|
||
'expense': me,
|
||
'net': mi - me,
|
||
})
|
||
|
||
return {
|
||
'period': str(year),
|
||
'date_from': date_from,
|
||
'date_to': date_to,
|
||
'income': income,
|
||
'expense': expense,
|
||
'net': income - expense,
|
||
'savings_rate': round(((income - expense) / income * 100), 1) if income > 0 else 0,
|
||
'avg_monthly_income': round(income / 12, 2),
|
||
'avg_monthly_expense': round(expense / 12, 2),
|
||
'expense_categories': expense_cats,
|
||
'income_categories': income_cats,
|
||
'months': months,
|
||
}
|
||
|
||
|
||
# ── Net worth history ─────────────────────────────────────────────────────────
|
||
|
||
def net_worth_history():
|
||
from dateutil.relativedelta import relativedelta
|
||
|
||
snapshots = NetWorthSnapshot.query\
|
||
.order_by(NetWorthSnapshot.snapshot_date.asc())\
|
||
.all()
|
||
|
||
result = {
|
||
'snapshots': snapshots,
|
||
'labels': [s.snapshot_date.strftime('%b %Y') for s in snapshots],
|
||
'values': [float(s.net_worth) for s in snapshots],
|
||
'assets': [float(s.total_assets) for s in snapshots],
|
||
'liabilities': [float(s.total_liabilities) for s in snapshots],
|
||
'count': len(snapshots),
|
||
'proj_labels': [],
|
||
'proj_values': [],
|
||
'projected_1yr': None,
|
||
'monthly_delta': None,
|
||
}
|
||
|
||
if len(snapshots) >= 3:
|
||
recent = snapshots[-6:] # up to last 6 data points
|
||
deltas = [
|
||
float(recent[i].net_worth) - float(recent[i - 1].net_worth)
|
||
for i in range(1, len(recent))
|
||
]
|
||
avg_delta = sum(deltas) / len(deltas)
|
||
|
||
last_nw = float(snapshots[-1].net_worth)
|
||
last_date = snapshots[-1].snapshot_date
|
||
|
||
proj_labels, proj_values = [], []
|
||
for i in range(1, 13):
|
||
proj_labels.append((last_date + relativedelta(months=i)).strftime('%b %Y'))
|
||
proj_values.append(round(last_nw + avg_delta * i, 2))
|
||
|
||
result['proj_labels'] = proj_labels
|
||
result['proj_values'] = proj_values
|
||
result['projected_1yr'] = proj_values[-1]
|
||
result['monthly_delta'] = round(avg_delta, 2)
|
||
|
||
return result
|
||
|
||
|
||
# ── Category spending trends (last 6 months) ──────────────────────────────────
|
||
|
||
def category_trends(months_back=6):
|
||
today = date.today()
|
||
result = {}
|
||
labels = []
|
||
|
||
for i in range(months_back - 1, -1, -1):
|
||
# Walk back i months
|
||
if today.month - i <= 0:
|
||
yr = today.year - 1
|
||
mo = 12 + (today.month - i)
|
||
else:
|
||
yr = today.year
|
||
mo = today.month - i
|
||
mf, mt = _month_range(yr, mo)
|
||
label = date(yr, mo, 1).strftime('%b %Y')
|
||
labels.append(label)
|
||
|
||
rows = db.session.query(
|
||
Category.name,
|
||
Category.color,
|
||
func.sum(Transaction.amount).label('total')
|
||
).join(Transaction, Transaction.category_id == Category.id)\
|
||
.filter(
|
||
Transaction.transaction_type == 'expense',
|
||
Transaction.date >= mf,
|
||
Transaction.date <= mt,
|
||
).group_by(Category.id).all()
|
||
|
||
for row in rows:
|
||
if row.name not in result:
|
||
result[row.name] = {'color': row.color, 'data': [0] * months_back}
|
||
idx = months_back - 1 - i
|
||
result[row.name]['data'][idx] = float(row.total)
|
||
|
||
# Keep top 6 categories by total
|
||
sorted_cats = sorted(result.items(), key=lambda x: sum(x[1]['data']), reverse=True)[:6]
|
||
|
||
datasets = []
|
||
for name, info in sorted_cats:
|
||
datasets.append({
|
||
'label': name,
|
||
'data': info['data'],
|
||
'borderColor': info['color'],
|
||
'backgroundColor': info['color'] + '22',
|
||
'tension': 0.3,
|
||
'fill': False,
|
||
})
|
||
|
||
return {'labels': labels, 'datasets': datasets}
|
||
|
||
|
||
# ── Tax year summary ──────────────────────────────────────────────────────────
|
||
|
||
def tax_year_summary(year):
|
||
date_from, date_to = _year_range(year)
|
||
income, expense = _totals(date_from, date_to)
|
||
|
||
income_cats = _category_breakdown(date_from, date_to, 'income')
|
||
expense_cats = _category_breakdown(date_from, date_to, 'expense')
|
||
|
||
# All income transactions for the year
|
||
income_txns = Transaction.query\
|
||
.filter(
|
||
Transaction.transaction_type == 'income',
|
||
Transaction.date >= date_from,
|
||
Transaction.date <= date_to,
|
||
).order_by(Transaction.date.asc()).all()
|
||
|
||
return {
|
||
'year': year,
|
||
'date_from': date_from,
|
||
'date_to': date_to,
|
||
'total_income': income,
|
||
'total_expense': expense,
|
||
'net': income - expense,
|
||
'income_categories': income_cats,
|
||
'expense_categories': expense_cats,
|
||
'income_transactions': income_txns,
|
||
}
|
||
|
||
|
||
# ── Month-over-month category comparison ─────────────────────────────────────
|
||
|
||
def category_mom_comparison():
|
||
"""
|
||
Returns per-category expense totals for: this month, last month, and the
|
||
3-month rolling average (last 3 complete months). Sorted by this-month
|
||
spend descending.
|
||
"""
|
||
today = date.today()
|
||
|
||
this_start = today.replace(day=1)
|
||
this_end = today
|
||
|
||
last_end = this_start - timedelta(days=1)
|
||
last_start = last_end.replace(day=1)
|
||
|
||
# Build month ranges for the 3-month rolling average (the 3 complete months
|
||
# ending with last month)
|
||
avg_ranges = []
|
||
cursor = last_start
|
||
for _ in range(3):
|
||
me = cursor - timedelta(days=1)
|
||
ms = me.replace(day=1)
|
||
avg_ranges.append((ms, me))
|
||
cursor = ms
|
||
|
||
def _totals_by_cat(start, end):
|
||
rows = db.session.query(
|
||
Category.id,
|
||
Category.name,
|
||
Category.color,
|
||
Category.icon,
|
||
func.sum(Transaction.amount).label('total'),
|
||
).join(Transaction, Transaction.category_id == Category.id)\
|
||
.filter(
|
||
Transaction.transaction_type == 'expense',
|
||
Transaction.date >= start,
|
||
Transaction.date <= end,
|
||
).group_by(Category.id).all()
|
||
return {r.id: {'name': r.name, 'color': r.color, 'icon': r.icon,
|
||
'total': float(r.total)} for r in rows}
|
||
|
||
this_data = _totals_by_cat(this_start, this_end)
|
||
last_data = _totals_by_cat(last_start, last_end)
|
||
avg_data = [_totals_by_cat(s, e) for s, e in avg_ranges]
|
||
|
||
# Collect all known category IDs + their metadata
|
||
cat_meta = {}
|
||
for src in [this_data, last_data] + avg_data:
|
||
for cid, info in src.items():
|
||
if cid not in cat_meta:
|
||
cat_meta[cid] = {k: info[k] for k in ('name', 'color', 'icon')}
|
||
|
||
rows = []
|
||
for cid, meta in cat_meta.items():
|
||
this_amt = this_data.get(cid, {}).get('total', 0.0)
|
||
last_amt = last_data.get(cid, {}).get('total', 0.0)
|
||
avg_monthly = (
|
||
sum(md.get(cid, {}).get('total', 0.0) for md in avg_data) / len(avg_data)
|
||
if avg_data else 0.0
|
||
)
|
||
change_pct = (
|
||
round((this_amt - last_amt) / last_amt * 100, 1)
|
||
if last_amt > 0 else None
|
||
)
|
||
rows.append({
|
||
**meta,
|
||
'id': cid,
|
||
'this_month': round(this_amt, 2),
|
||
'last_month': round(last_amt, 2),
|
||
'avg_3mo': round(avg_monthly, 2),
|
||
'change_pct': change_pct,
|
||
})
|
||
|
||
rows.sort(key=lambda r: r['this_month'], reverse=True)
|
||
return rows
|
||
|
||
|
||
# ── Spending anomaly detection ────────────────────────────────────────────────
|
||
|
||
def spending_anomalies(days_back=30, multiplier=2.0, min_avg=10.0, min_amount=10.0):
|
||
"""
|
||
Find transactions in the last `days_back` days whose amount is more than
|
||
`multiplier` × the category's average monthly spend over the prior 3 months.
|
||
Returns a list of dicts with transaction details + context, capped at 5.
|
||
"""
|
||
today = date.today()
|
||
window_start = today - timedelta(days=days_back)
|
||
|
||
# Baseline: the 3 complete months before today's month
|
||
base_end = today.replace(day=1) - timedelta(days=1)
|
||
base_start = (base_end.replace(day=1) - timedelta(days=60)).replace(day=1)
|
||
|
||
# Per-category, per-month totals over baseline
|
||
rows = db.session.query(
|
||
Transaction.category_id,
|
||
extract('year', Transaction.date).label('yr'),
|
||
extract('month', Transaction.date).label('mo'),
|
||
func.sum(Transaction.amount).label('total'),
|
||
).filter(
|
||
Transaction.transaction_type == 'expense',
|
||
Transaction.date >= base_start,
|
||
Transaction.date <= base_end,
|
||
Transaction.category_id.isnot(None),
|
||
).group_by(
|
||
Transaction.category_id,
|
||
extract('year', Transaction.date),
|
||
extract('month', Transaction.date),
|
||
).all()
|
||
|
||
cat_month_totals = defaultdict(list)
|
||
for r in rows:
|
||
cat_month_totals[r.category_id].append(float(r.total))
|
||
|
||
cat_avg = {
|
||
cid: sum(totals) / len(totals)
|
||
for cid, totals in cat_month_totals.items()
|
||
}
|
||
|
||
# Recent transactions in the window
|
||
recent = (
|
||
Transaction.query
|
||
.filter(
|
||
Transaction.transaction_type == 'expense',
|
||
Transaction.date >= window_start,
|
||
Transaction.date <= today,
|
||
Transaction.category_id.isnot(None),
|
||
)
|
||
.order_by(Transaction.date.desc())
|
||
.all()
|
||
)
|
||
|
||
anomalies = []
|
||
for txn in recent:
|
||
avg = cat_avg.get(txn.category_id)
|
||
amt = float(txn.amount)
|
||
if avg and avg >= min_avg and amt >= min_amount and amt > avg * multiplier:
|
||
anomalies.append({
|
||
'id': txn.id,
|
||
'date': txn.date.strftime('%b %d'),
|
||
'description': txn.description,
|
||
'amount': amt,
|
||
'category': txn.category.name if txn.category else 'Other',
|
||
'category_color': txn.category.color if txn.category else '#94a3b8',
|
||
'category_icon': txn.category.icon if txn.category else 'bi-tag',
|
||
'avg': round(avg, 2),
|
||
'multiple': round(amt / avg, 1),
|
||
})
|
||
|
||
# Sort by multiple desc (biggest outliers first), cap at 5
|
||
anomalies.sort(key=lambda a: a['multiple'], reverse=True)
|
||
return anomalies[:5]
|
||
|
||
|
||
# ── Snapshot helpers ──────────────────────────────────────────────────────────
|
||
|
||
def take_net_worth_snapshot():
|
||
"""Save today's net worth snapshot. Called by cron on 1st of month."""
|
||
today = date.today()
|
||
|
||
existing = NetWorthSnapshot.query.filter_by(snapshot_date=today).first()
|
||
if existing:
|
||
return existing
|
||
|
||
accounts = Account.query.filter_by(is_active=True).all()
|
||
total_assets = sum(float(a.balance) for a in accounts if float(a.balance) > 0 and a.account_type != 'credit_card')
|
||
total_liab = sum(abs(float(a.balance)) for a in accounts if float(a.balance) < 0)
|
||
|
||
investments = Investment.query.filter_by(is_active=True).all()
|
||
inv_value = sum(i.current_value for i in investments)
|
||
|
||
account_balances = {a.name: float(a.balance) for a in accounts}
|
||
|
||
snapshot = NetWorthSnapshot(
|
||
snapshot_date=today,
|
||
total_assets=total_assets + inv_value,
|
||
total_liabilities=total_liab,
|
||
net_worth=(total_assets + inv_value) - total_liab,
|
||
account_balances=account_balances,
|
||
investment_value=inv_value,
|
||
)
|
||
db.session.add(snapshot)
|
||
db.session.commit()
|
||
return snapshot
|