from __future__ import annotations from decimal import Decimal from sqlite3 import Connection from jarvis_finance.services.budget_common import row_to_dict from jarvis_finance.services.analytics_envelope import analytics_envelope from jarvis_finance.services.budget_overview import _fmt, get_budget_status_by_category from jarvis_finance.services.budget_transactions import list_budget_transactions EXPENSE_TYPES = ("expense", "fee") INCOME_TYPES = ("income", "refund") FLOW_TYPES = EXPENSE_TYPES + INCOME_TYPES def _month_rows(conn: Connection, month: str): return conn.execute( f""" SELECT t.*, c.name AS category_name, c.color AS category_color, c.icon AS category_icon, COALESCE(t.payee, t.description, t.source_type, 'Quelle') AS merchant_display_name FROM budget_transactions t LEFT JOIN budget_categories c ON c.category_id=t.category_id WHERE t.status='confirmed' AND substr(t.transaction_date,1,7)=? AND t.transaction_type IN ({','.join(['?'] * len(FLOW_TYPES))}) ORDER BY t.transaction_date DESC, t.created_at DESC """, (month, *FLOW_TYPES), ).fetchall() def _year_rows(conn: Connection, year: str): return conn.execute( f""" SELECT t.*, c.name AS category_name, c.category_type, c.color AS category_color, c.icon AS category_icon, COALESCE(t.payee, t.description, t.source_type, 'Quelle') AS merchant_display_name FROM budget_transactions t LEFT JOIN budget_categories c ON c.category_id=t.category_id WHERE t.status='confirmed' AND substr(t.transaction_date,1,4)=? AND t.transaction_type IN ({','.join(['?'] * len(FLOW_TYPES))}) ORDER BY t.transaction_date """, (year, *FLOW_TYPES), ).fetchall() def _amount(row) -> Decimal: return Decimal(str(row["amount_chf"] or row["amount_original"] or "0")) def _review_candidate_count(conn: Connection, year: str) -> int: return int(conn.execute("SELECT COUNT(*) AS c FROM budget_transaction_candidates WHERE substr(transaction_date,1,4)=? AND status IN ('pending','needs_review','auto_categorized')", (year,)).fetchone()["c"] or 0) def get_budget_dashboard_cockpit(conn: Connection, *, month: str = "2026-05") -> dict: year = month[:4] rows = _month_rows(conn, month) income = Decimal("0") expense = Decimal("0") category_totals: dict[str, dict] = {} for row in rows: amount = _amount(row) if row["transaction_type"] in INCOME_TYPES: income += amount elif row["transaction_type"] in EXPENSE_TYPES: expense += amount cid = str(row["category_id"] or "uncategorized") item = category_totals.setdefault(cid, {"category_id": cid, "category": row["category_name"] or "Review nötig", "amount": Decimal("0"), "transaction_count": 0, "color": row["category_color"], "icon": row["category_icon"]}) item["amount"] += amount item["transaction_count"] += 1 status_rows = get_budget_status_by_category(conn, year=year) planned_expense_month = sum((Decimal(str(r.get("budget_month") or "0")) for r in status_rows if r.get("category_type") == "expense"), Decimal("0")) budget_consumption = Decimal("0") if planned_expense_month == 0 else (expense / planned_expense_month * Decimal("100")) savings_rate = Decimal("0") if income == 0 else ((income - expense) / income * Decimal("100")) total_expense = expense or Decimal("1") top_categories = [] for item in sorted(category_totals.values(), key=lambda x: x["amount"], reverse=True)[:5]: top_categories.append({ "category_id": item["category_id"], "category": item["category"], "amount_chf": _fmt(item["amount"]), "share_percent": _fmt(item["amount"] / total_expense * Decimal("100")), "transaction_count": item["transaction_count"], "color": item["color"], "icon": item["icon"], }) critical = [] for r in status_rows: if r.get("category_type") != "expense": continue month_over = Decimal(str(r.get("actual_current_month") or "0")) > Decimal(str(r.get("budget_month") or "0")) > 0 if r.get("status") in {"Rot", "Kritisch", "ohne Budget"} or month_over: critical.append(r) critical = critical[:6] categories_without_budget = sum(1 for r in status_rows if r.get("category_type") == "expense" and not r.get("has_budget") and Decimal(str(r.get("actual_year_to_date") or "0")) > 0) from jarvis_finance.services.budget_recurring import get_recurring_dashboard recurring = get_recurring_dashboard(conn, today=f"{month}-28") recurring_kpis = recurring.get("kpis", {}) recent_expenses = [] for row in rows: if row["transaction_type"] in EXPENSE_TYPES: d = row_to_dict(row) d["merchant_display_name"] = row["merchant_display_name"] recent_expenses.append(d) return { "purpose": "monthly_cockpit", "month": month, "year": year, "chart_library": "native-svg", "kpis": { "income_current_month_chf": _fmt(income), "expense_current_month_chf": _fmt(expense), "net_cashflow_current_month_chf": _fmt(income - expense), "budget_consumption_percent": _fmt(budget_consumption), "savings_rate_percent": _fmt(savings_rate), "open_review_transactions": _review_candidate_count(conn, year), "fixed_costs_month_chf": recurring_kpis.get("monthly_fixed_costs_chf", "0.00"), "fixed_cost_quote_percent": recurring_kpis.get("fixed_cost_quote_percent", "0.00"), "open_fixed_cost_candidate_count": recurring_kpis.get("open_candidate_count", 0), "missing_fixed_cost_payment_count": recurring_kpis.get("missing_payment_count", 0), }, "monthly_cashflow": {"income_chf": _fmt(income), "expense_chf": _fmt(expense), "net_cashflow_chf": _fmt(income - expense)}, "top_expense_categories": top_categories, "critical_categories": critical, "recent_confirmed_expenses": recent_expenses[:8], "open_todos": {"review_candidates": _review_candidate_count(conn, year), "categories_without_budget": categories_without_budget, "unconfirmed_transactions": _review_candidate_count(conn, year)}, } def get_budget_vs_actual_analysis(conn: Connection, *, year: str = "2026") -> dict: status_rows = get_budget_status_by_category(conn, year=year) rows = _year_rows(conn, year) monthly = {f"{year}-{idx:02d}": {"income": Decimal("0"), "expense": Decimal("0")} for idx in range(1, 13)} category_expense: dict[str, dict] = {} for row in rows: amount = _amount(row) month = str(row["transaction_date"])[:7] if row["transaction_type"] in INCOME_TYPES: monthly[month]["income"] += amount elif row["transaction_type"] in EXPENSE_TYPES: monthly[month]["expense"] += amount cid = str(row["category_id"] or "uncategorized") item = category_expense.setdefault(cid, {"category_id": cid, "category": row["category_name"] or "Review nötig", "amount": Decimal("0"), "color": row["category_color"]}) item["amount"] += amount cashflow = [] trend = [] cum_income = cum_expense = Decimal("0") for month, values in sorted(monthly.items()): income = values["income"]; expense = values["expense"]; net = income - expense cum_income += income; cum_expense += expense cashflow.append({"month": month, "income_chf": _fmt(income), "expense_chf": _fmt(expense), "net_cashflow_chf": _fmt(net)}) trend.append({"month": month, "cumulative_income_chf": _fmt(cum_income), "cumulative_expense_chf": _fmt(cum_expense), "cumulative_net_chf": _fmt(cum_income - cum_expense)}) budget_vs_actual = [r | {"purpose": "plan_vs_reality_analysis"} for r in status_rows] expense_rows = [r for r in budget_vs_actual if r.get("category_type") == "expense"] top_overruns = [] top_under = [] without_budget = [] without_previous_year = [] for r in expense_rows: year_dev = Decimal(str(r.get("deviation_year") or "0")) month_dev = Decimal(str(r.get("actual_current_month") or "0")) - Decimal(str(r.get("budget_month") or "0")) over = max(year_dev, month_dev) if not r.get("has_previous_year"): without_previous_year.append(r | {"amount_chf": r.get("actual_2025_year") or "0", "deviation_scope": "missing_previous_year"}) if not r.get("has_budget") and Decimal(str(r.get("actual_year_to_date") or "0")) > 0: without_budget.append(r | {"amount_chf": r.get("actual_year_to_date") or "0", "deviation_scope": "missing_budget"}) elif over > 0: top_overruns.append(r | {"amount_chf": _fmt(over), "deviation_scope": "month" if month_dev >= year_dev else "year"}) elif year_dev < 0: top_under.append(r | {"amount_chf": _fmt(abs(year_dev)), "deviation_scope": "year"}) top_overruns = sorted(top_overruns, key=lambda r: Decimal(str(r["amount_chf"])), reverse=True)[:5] top_under = sorted(top_under, key=lambda r: Decimal(str(r["amount_chf"])), reverse=True)[:5] total_expense = sum((item["amount"] for item in category_expense.values()), Decimal("0")) or Decimal("1") comparison_bars = [ { "category_id": r.get("category_id"), "category": r.get("category"), "budget_2026_chf": r.get("budget_year") or "0.00", "actual_ytd_2026_chf": r.get("actual_year_to_date") or "0.00", "forecast_2026_chf": r.get("forecast_year") or "0.00", "actual_2025_chf": r.get("actual_2025_year") or "0.00", "status": r.get("status"), } for r in expense_rows ] donut = [{"category_id": item["category_id"], "category": item["category"], "amount_chf": _fmt(item["amount"]), "share_percent": _fmt(item["amount"] / total_expense * Decimal("100")), "color": item.get("color")} for item in sorted(category_expense.values(), key=lambda x: x["amount"], reverse=True)] legacy = { "purpose": "plan_vs_reality_analysis", "year": year, "chart_library": "native-svg", "empty_state": not rows, "budget_vs_actual_by_category": budget_vs_actual, "comparison_bars": comparison_bars, "top_overruns": top_overruns, "top_under_budget": top_under, "categories_without_budget": without_budget, "categories_without_previous_year": without_previous_year, "cashflow_by_month": cashflow, "category_donut": donut, "trend_by_month": trend, } return analytics_envelope(legacy, view="budget_vs_actual", filters={"year": year}, totals={"category_count": len(budget_vs_actual), "empty_state": not rows}, series=cashflow + trend, rows=budget_vs_actual)