diff --git a/app/routes/dashboard.py b/app/routes/dashboard.py index 4f9ae12..e51d647 100644 --- a/app/routes/dashboard.py +++ b/app/routes/dashboard.py @@ -253,6 +253,18 @@ def reconcile_api(): }) +@dashboard_bp.route('/api/health-score') +@login_required +def health_score_api(): + from app.services.health_score_service import compute_health_score + try: + data = compute_health_score() + except Exception as e: + log.warning('[dashboard] health_score_api failed: %s', e) + return jsonify({'error': str(e)}), 500 + return jsonify(data) + + @dashboard_bp.route('/api/anomalies') @login_required def anomalies_api(): diff --git a/app/routes/settings.py b/app/routes/settings.py index 4b46167..de58080 100644 --- a/app/routes/settings.py +++ b/app/routes/settings.py @@ -1,7 +1,7 @@ import os import uuid from flask import (Blueprint, render_template, redirect, url_for, flash, - request, current_app, send_from_directory) + request, current_app, send_from_directory, jsonify) from flask_login import login_required, current_user from app.utils.audit import audit from flask_wtf import FlaskForm @@ -208,6 +208,21 @@ def recurring(): return render_template('settings/recurring.html', rules=rules, upcoming=upcoming) +@settings_bp.route('/recurring/projection') +@login_required +def recurring_projection(): + days = request.args.get('days', 30, type=int) + if days not in (30, 60, 90): + days = 30 + from app.services.recurring_service import projected_cash_flow + data = projected_cash_flow(days) + # Serialize events (date objects → string) + data['events'] = [ + {**ev, 'date': ev['date'].strftime('%b %d')} for ev in data['events'] + ] + return jsonify(data) + + @settings_bp.route('/recurring/new', methods=['GET', 'POST']) @login_required def recurring_new(): diff --git a/app/routes/transactions.py b/app/routes/transactions.py index 0fb66bd..4ea4a36 100644 --- a/app/routes/transactions.py +++ b/app/routes/transactions.py @@ -443,12 +443,21 @@ def ocr_receipt_file(): from app.models.category import Category from flask import current_app + from app.models.receipt import Receipt + data = request.get_json() if not data or not data.get('filename'): return jsonify({'error': 'No filename provided'}), 400 # Security: only allow basenames, no path traversal filename = os.path.basename(data['filename']) + + # Ownership check: filename must exist in receipts table (single-user, but + # prevents OCR extraction from arbitrary files on disk via a crafted request) + receipt_record = Receipt.query.filter_by(filename=filename).first() + if not receipt_record: + return jsonify({'error': 'Receipt not found'}), 404 + upload_dir = current_app.config.get('UPLOAD_FOLDER', '/home/pfm/app/uploads') file_path = os.path.join(upload_dir, filename) diff --git a/app/services/health_score_service.py b/app/services/health_score_service.py new file mode 100644 index 0000000..b147541 --- /dev/null +++ b/app/services/health_score_service.py @@ -0,0 +1,245 @@ +""" +Financial Health Score — synthesises savings rate, budget adherence, goal +progress, and emergency fund coverage into a single 0–100 score. + +Each component is worth 25 points. Returns the total score plus a breakdown +so the UI can show per-component detail and suggestions. +""" + +from datetime import date +from dateutil.relativedelta import relativedelta +from sqlalchemy import func +from app.extensions import db +from app.models.transaction import Transaction +from app.models.goal import Goal + + +# ── Component scorers ──────────────────────────────────────────────────────── + +def _score_savings(max_pts=25): + """ + Avg savings rate over the last 3 full months. + ≥ 20% → full marks; 10–20% → 18; 1–10% → 10; ≤ 0% → 0. + """ + today = date.today() + month_start = today.replace(day=1) + + # Collect last 3 complete months + incomes, expenses = [], [] + for i in range(1, 4): + mo_end = (month_start - relativedelta(days=1)) + mo_start = mo_end.replace(day=1) + month_start = mo_start + + inc = float(db.session.query(func.coalesce(func.sum(Transaction.amount), 0)) + .filter(Transaction.transaction_type == 'income', + Transaction.date >= mo_start, + Transaction.date <= mo_end).scalar()) + exp = float(db.session.query(func.coalesce(func.sum(Transaction.amount), 0)) + .filter(Transaction.transaction_type == 'expense', + Transaction.date >= mo_start, + Transaction.date <= mo_end).scalar()) + incomes.append(inc) + expenses.append(exp) + + total_inc = sum(incomes) + total_exp = sum(expenses) + rate = ((total_inc - total_exp) / total_inc * 100) if total_inc > 0 else 0 + + if rate >= 20: + pts = max_pts + elif rate >= 10: + pts = round(max_pts * 0.72) # 18/25 + elif rate > 0: + pts = round(max_pts * 0.40) # 10/25 + else: + pts = 0 + + return { + 'points': pts, + 'max': max_pts, + 'value': round(rate, 1), + 'label': f'{rate:+.1f}% savings rate (3-mo avg)', + 'tip': None if rate >= 20 else ( + 'Aim for 20%+ savings rate.' if rate < 10 else + 'Good start — push toward 20%.'), + } + + +def _score_budgets(max_pts=25): + """ + What fraction of budgeted categories are currently under their limit? + All under → full marks; scales linearly. + """ + from app.services.budget_service import get_budget_summary + month_str = date.today().strftime('%Y-%m') + summary = [s for s in get_budget_summary(month_str) if s['has_budget']] + + if not summary: + return { + 'points': max_pts, # no budgets set → not penalised + 'max': max_pts, + 'value': None, + 'label': 'No budgets set', + 'tip': 'Set monthly budgets to track spending limits.', + } + + under = sum(1 for s in summary if not s['is_over']) + ratio = under / len(summary) + pts = round(ratio * max_pts) + + over_cats = [s['category'].name for s in summary if s['is_over']] + tip = None + if over_cats: + tip = f'Over budget: {", ".join(over_cats[:3])}{"…" if len(over_cats) > 3 else ""}.' + + return { + 'points': pts, + 'max': max_pts, + 'value': round(ratio * 100, 1), + 'label': f'{under}/{len(summary)} categories under budget', + 'tip': tip, + } + + +def _score_goals(max_pts=25): + """ + Average completion % across active (non-completed) goals. + 100% avg → full marks; scales linearly. + """ + goals = Goal.query.filter_by(is_completed=False).all() + if not goals: + completed = Goal.query.filter_by(is_completed=True).count() + return { + 'points': max_pts if completed else round(max_pts * 0.5), + 'max': max_pts, + 'value': 100.0 if completed else 0.0, + 'label': 'All goals completed!' if completed else 'No savings goals set', + 'tip': None if completed else 'Create a savings goal to track progress.', + } + + pcts = [] + for g in goals: + target = float(g.target_amount) + if target > 0: + pcts.append(min(float(g.current_amount) / target * 100, 100)) + + avg = (sum(pcts) / len(pcts)) if pcts else 0 + pts = round(avg / 100 * max_pts) + + behind = [g.name for g, p in zip(goals, pcts) if p < 25] + tip = None + if behind: + tip = f'Behind on: {", ".join(behind[:2])}{"…" if len(behind) > 2 else ""}.' + + return { + 'points': pts, + 'max': max_pts, + 'value': round(avg, 1), + 'label': f'{round(avg, 0):.0f}% avg goal progress ({len(goals)} active)', + 'tip': tip, + } + + +def _score_emergency_fund(max_pts=25): + """ + Liquid assets vs 3-month expense target. + ≥ 3 months → full marks; scales linearly up to 3 months. + """ + from app.services.goal_service import get_emergency_fund_status + ef = get_emergency_fund_status() + + liquid = ef['liquid_assets'] + target3 = ef['target_3mo'] + covered = ef['months_covered'] + pct3 = ef['pct_3mo'] # 0–100, capped + + pts = round(pct3 / 100 * max_pts) + + if covered >= 3: + tip = None + elif covered >= 1: + short = target3 - liquid + tip = f'Build to 3-month emergency fund (need ${short:,.0f} more).' + else: + tip = 'Start an emergency fund — aim for 1 month of expenses first.' + + return { + 'points': pts, + 'max': max_pts, + 'value': round(covered, 1), + 'label': f'{covered:.1f} months emergency fund', + 'tip': tip, + } + + +# ── Public API ─────────────────────────────────────────────────────────────── + +def compute_health_score(): + """ + Compute the overall financial health score. + + Returns: + { + score: int 0–100 + grade: str 'A' | 'B' | 'C' | 'D' | 'F' + color: str CSS color + components: list of component dicts + } + """ + components = { + 'savings': _score_savings(), + 'budgets': _score_budgets(), + 'goals': _score_goals(), + 'emergency_fund': _score_emergency_fund(), + } + + score = sum(c['points'] for c in components.values()) + + if score >= 85: + grade, color = 'A', '#10b981' + elif score >= 70: + grade, color = 'B', '#3b82f6' + elif score >= 55: + grade, color = 'C', '#f59e0b' + elif score >= 40: + grade, color = 'D', '#f97316' + else: + grade, color = 'F', '#ef4444' + + # Add display names for the template + labels = { + 'savings': 'Savings Rate', + 'budgets': 'Budget Adherence', + 'goals': 'Goal Progress', + 'emergency_fund': 'Emergency Fund', + } + icons = { + 'savings': 'bi-piggy-bank', + 'budgets': 'bi-pie-chart', + 'goals': 'bi-bullseye', + 'emergency_fund': 'bi-shield-check', + } + + component_list = [ + { + 'key': key, + 'name': labels[key], + 'icon': icons[key], + 'points': c['points'], + 'max': c['max'], + 'value': c['value'], + 'label': c['label'], + 'tip': c['tip'], + 'pct': round(c['points'] / c['max'] * 100), + } + for key, c in components.items() + ] + + return { + 'score': score, + 'grade': grade, + 'color': color, + 'components': component_list, + 'tips': [c['tip'] for c in component_list if c['tip']], + } diff --git a/app/services/recurring_service.py b/app/services/recurring_service.py index d94a213..cf1bc88 100644 --- a/app/services/recurring_service.py +++ b/app/services/recurring_service.py @@ -94,6 +94,104 @@ def process_due_rules(dry_run=False): return created +def projected_cash_flow(days=90): + """ + Build a projected cash flow from all active recurring rules over the next + N days. Returns weekly-bucketed chart data plus a flat event list. + + Returns dict: + labels — list of 'Mon DD' strings (week-start dates) + income — list of floats (income per week bucket) + expense — list of floats (expense per week bucket) + balance — list of floats (running balance at end of each bucket) + events — list of {date, description, amount, type, rule_id} + starting_balance — float + ending_balance — float + total_income — float + total_expense — float + net — float + """ + from app.models.account import Account + from sqlalchemy import func + from app.extensions import db + + today = date.today() + cutoff = today + timedelta(days=days) + + # Starting balance = sum of all active account balances + starting_balance = float( + db.session.query(func.coalesce(func.sum(Account.balance), 0)) + .filter(Account.is_active == True) + .scalar() + ) + + # Enumerate all occurrences of active rules within the window + rules = RecurringRule.query.filter_by(is_active=True).all() + events = [] + for rule in rules: + run_date = rule.next_run or rule.start_date + # Advance to window start if rule fires before today + while run_date < today: + run_date = next_occurrence(run_date, rule.frequency) + while run_date <= cutoff: + if rule.end_date and run_date > rule.end_date: + break + events.append({ + 'date': run_date, + 'description': rule.description, + 'amount': float(rule.amount), + 'type': rule.transaction_type, + 'rule_id': rule.id, + }) + run_date = next_occurrence(run_date, rule.frequency) + + events.sort(key=lambda e: e['date']) + + # Build weekly buckets: each bucket starts on Monday + # Find the Monday on or before today + week_start = today - timedelta(days=today.weekday()) + buckets = [] + ws = week_start + while ws <= cutoff: + buckets.append(ws) + ws += timedelta(weeks=1) + + bucket_income = [0.0] * len(buckets) + bucket_expense = [0.0] * len(buckets) + + for ev in events: + # Find which bucket this event falls in + idx = (ev['date'] - week_start).days // 7 + if 0 <= idx < len(buckets): + if ev['type'] == 'income': + bucket_income[idx] += ev['amount'] + else: + bucket_expense[idx] += ev['amount'] + + # Running balance + running = starting_balance + bucket_balance = [] + for inc, exp in zip(bucket_income, bucket_expense): + running += inc - exp + bucket_balance.append(round(running, 2)) + + total_income = sum(bucket_income) + total_expense = sum(bucket_expense) + + return { + 'labels': [b.strftime('%b %d') for b in buckets], + 'income': [round(v, 2) for v in bucket_income], + 'expense': [round(v, 2) for v in bucket_expense], + 'balance': bucket_balance, + 'events': events, + 'starting_balance': round(starting_balance, 2), + 'ending_balance': round(bucket_balance[-1], 2) if bucket_balance else round(starting_balance, 2), + 'total_income': round(total_income, 2), + 'total_expense': round(total_expense, 2), + 'net': round(total_income - total_expense, 2), + } + + def get_upcoming(days=30): """Return list of upcoming recurring transactions in the next N days.""" today = date.today() diff --git a/app/templates/bank_import/index.html b/app/templates/bank_import/index.html index 93168a5..32a997e 100644 --- a/app/templates/bank_import/index.html +++ b/app/templates/bank_import/index.html @@ -197,6 +197,17 @@ + {# Import progress bar (shown only during chunked import) #} +