from __future__ import annotations import calendar import hashlib import json from datetime import date, timedelta from decimal import Decimal, InvalidOperation from sqlite3 import Connection from typing import Any from fastapi import HTTPException from jarvis_finance.audit.log import record_audit_event from jarvis_finance.services.budget_common import new_id, now, row_to_dict from jarvis_finance.services.budget_recurring import calculate_annual_and_reserve from jarvis_finance.services.household_financials import get_household_financial_summary, list_household_financial_effects from jarvis_finance.services.household_review_corrections import _card_match from jarvis_finance.services.prior_year_actuals import get_prior_year_actuals, validate_comparison_year MONTHS = [f"{idx:02d}" for idx in range(1, 13)] BUDGET_CADENCES = {"monthly", "annual", "one_time", "seasonal", "fixed", "variable"} def _fmt(value: Decimal) -> str: return format(value.quantize(Decimal("0.01")), "f") def _decimal_text(value: Any, *, required: bool = False, allow_zero: bool = True) -> str | None: if value in (None, ""): if required: raise HTTPException(status_code=422, detail="decimal amount required") return None try: amount = Decimal(str(value).strip()) except (InvalidOperation, ValueError): raise HTTPException(status_code=422, detail="invalid decimal amount") from None if amount < 0: raise HTTPException(status_code=422, detail="negative budget values are not allowed") if amount == 0 and not allow_zero: raise HTTPException(status_code=422, detail="amount must be greater than zero") return _fmt(amount) def _expense_category(conn: Connection, category_id: str) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_categories WHERE category_id=? AND is_active=1", (category_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="active category not found") if row["category_type"] != "expense": raise HTTPException(status_code=422, detail="budget planning values require an active expense category") return row_to_dict(row) def _derive_budget(payload: dict[str, Any]) -> tuple[str | None, str | None]: monthly = _decimal_text(payload.get("monthly_budget_chf")) annual = _decimal_text(payload.get("annual_budget_chf")) if monthly and not annual: annual = _fmt(Decimal(monthly) * Decimal("12")) if annual and not monthly: monthly = _fmt(Decimal(annual) / Decimal("12")) return monthly, annual def _normalise_month_map(value: Any) -> dict[str, str]: if not value: return {} if not isinstance(value, dict): raise HTTPException(status_code=422, detail="previous_year_months must be a map MM -> decimal text") out: dict[str, str] = {} for month, amount in value.items(): key = str(month).zfill(2) if key not in MONTHS: raise HTTPException(status_code=422, detail="month must be 01..12") text = _decimal_text(amount) if text is not None: out[key] = text return out def _derive_previous_year(payload: dict[str, Any]) -> tuple[dict[str, str], str, list[str]]: warnings: list[str] = [] months = _normalise_month_map(payload.get("previous_year_months")) total_text = _decimal_text(payload.get("previous_year_total_chf")) distribute = bool(payload.get("distribute_previous_year_total")) if months: total = sum((Decimal(v) for v in months.values()), Decimal("0")) if total_text and Decimal(total_text) != total: warnings.append("Jahreswert 2025 weicht von Summe der Monatswerte ab; Monatswerte sind führend.") return months, _fmt(total), warnings if total_text: total = Decimal(total_text) if distribute: avg = _fmt(total / Decimal("12")) months = {m: avg for m in MONTHS} remainder = total - sum((Decimal(v) for v in months.values()), Decimal("0")) if remainder: months["12"] = _fmt(Decimal(months["12"]) + remainder) return months, _fmt(total), warnings return {}, _fmt(total), warnings return {}, "0.00", warnings FIELD_LABELS = { "monthly_budget_chf": "Budget Monat", "annual_budget_chf": "Budget Jahr", "budget_cadence": "Periodizität", "previous_year_total_chf": "Vorjahreswert Jahr", "previous_year_months": "Vorjahr Monatswerte", "notes": "Notiz", "category_id": "Kategorie-Mapping", } def _current_planning_snapshot(conn: Connection, category_id: str, year: str = "2026") -> dict[str, Any]: plan = _plan_amounts(conn, year).get(category_id, {}) budget_month = Decimal(str(plan.get("budget_month", "0"))) if plan else Decimal("0") budget_year = Decimal(str(plan.get("budget_year", "0"))) if plan else Decimal("0") previous = _previous_year_months(conn, str(int(year) - 1)).get(category_id, {}) if "__year__" in previous: prev_months = {} prev_total = previous["__year__"] else: prev_months = {m: _fmt(previous.get(m, Decimal("0"))) for m in MONTHS if previous.get(m, Decimal("0")) > 0} prev_total = sum((previous.get(m, Decimal("0")) for m in MONTHS), Decimal("0")) notes_row = conn.execute("SELECT notes FROM budget_plan_items WHERE category_id=? AND substr(plan_month,1,4)=? AND is_active=1 ORDER BY created_at LIMIT 1", (category_id, year)).fetchone() return { "monthly_budget_chf": _fmt(budget_month) if budget_month else None, "annual_budget_chf": _fmt(budget_year) if budget_year else None, "budget_cadence": str(plan.get("cadence") or "monthly") if plan else None, "previous_year_total_chf": _fmt(prev_total), "previous_year_months": prev_months, "notes": notes_row["notes"] if notes_row else None, } def _changes(old: dict[str, Any], new: dict[str, Any]) -> list[dict[str, str]]: out: list[dict[str, str]] = [] for field, label in FIELD_LABELS.items(): ov = old.get(field) nv = new.get(field) if ov != nv: out.append({ "field": field, "field_label": label, "old_value": json.dumps(ov, sort_keys=True) if isinstance(ov, dict) else ("" if ov is None else str(ov)), "new_value": json.dumps(nv, sort_keys=True) if isinstance(nv, dict) else ("" if nv is None else str(nv)), }) return out def _plausibility_checks(*, budget_year: Decimal, previous_total: Decimal, actual_ytd: Decimal, forecast: Decimal, has_budget: bool, has_previous: bool) -> list[dict[str, str]]: checks: list[dict[str, str]] = [] if not has_budget and actual_ytd > 0: checks.append({"severity": "warning", "code": "actuals_without_budget", "message": "Kategorie hat Ist-Ausgaben, aber kein Budget."}) if has_budget and actual_ytd == 0: checks.append({"severity": "info", "code": "budget_without_actuals", "message": "Kategorie hat Budget, aber noch keine Ist-Daten."}) if not has_previous: checks.append({"severity": "info", "code": "missing_previous_year", "message": "Kategorie hat keine Vorjahreswerte."}) if has_budget and has_previous and previous_total > 0: ratio = budget_year / previous_total if ratio < Decimal("0.80"): checks.append({"severity": "info", "code": "budget_lower_than_previous_year", "message": "Budget Jahr 2026 liegt mehr als 20% unter Vorjahr 2025."}) elif ratio > Decimal("1.20"): checks.append({"severity": "info", "code": "budget_higher_than_previous_year", "message": "Budget Jahr 2026 liegt mehr als 20% über Vorjahr 2025."}) if has_budget and actual_ytd > budget_year: checks.append({"severity": "warning", "code": "actual_over_budget", "message": "Ist 2026 liegt bereits über Budget."}) if has_budget and budget_year > 0 and forecast / budget_year > Decimal("1.15"): checks.append({"severity": "critical", "code": "forecast_critical", "message": "Forecast liegt kritisch über Budget."}) return checks def preview_category_budget_values(conn: Connection, category_id: str, payload: dict[str, Any]) -> dict[str, Any]: cat = _expense_category(conn, category_id) year = str(payload.get("year") or "2026") if year != "2026": raise HTTPException(status_code=422, detail="budget planning v1 writes only active year 2026") cadence = str(payload.get("budget_cadence") or payload.get("cadence") or "monthly") if cadence not in BUDGET_CADENCES: raise HTTPException(status_code=422, detail="invalid budget cadence") monthly, annual = _derive_budget(payload) previous_months, previous_total, warnings = _derive_previous_year(payload) previous_avg = Decimal(previous_total) / Decimal("12") if Decimal(previous_total) else Decimal("0") normalised = { "year": year, "category_id": category_id, "monthly_budget_chf": monthly, "annual_budget_chf": annual, "budget_cadence": cadence, "previous_year": "2025", "previous_year_months": previous_months, "previous_year_total_chf": previous_total, "source": payload.get("source") or "manual", "notes": payload.get("notes") or None, } preview_id = new_id("preview") record_audit_event(conn, source="vue_dashboard", action="budget_planning_preview", entity_type="budget_category", entity_id=category_id, new_values={"preview_id": preview_id, "category": cat["name"], "year": year}, created_by="user") conn.commit() return { "preview_id": preview_id, "summary": f"Budgetplanung {cat['name']} {year}", "warnings": warnings, "payload": normalised, "review": { "category": cat["name"], "budget_month_chf": monthly or "0.00", "budget_year_chf": annual or "0.00", "budget_cadence": cadence, "previous_year_total_chf": previous_total, "previous_year_monthly_average_chf": _fmt(previous_avg), "previous_year_months": {m: previous_months.get(m, "0.00") for m in MONTHS}, }, "requires_explicit_confirm": True, } def _upsert_plan(conn: Connection, category_id: str, payload: dict[str, Any]) -> str | None: monthly = payload.get("monthly_budget_chf") annual = payload.get("annual_budget_chf") if not monthly and not annual: return None row = conn.execute("SELECT plan_item_id FROM budget_plan_items WHERE category_id=? AND substr(plan_month,1,4)=? AND is_active=1 ORDER BY created_at LIMIT 1", (category_id, payload["year"])).fetchone() ts = now() cadence = "annual" if payload["budget_cadence"] == "annual" else "monthly" if row: plan_id = str(row["plan_item_id"]) conn.execute("UPDATE budget_plan_items SET monthly_amount_chf=?, annual_amount_chf=?, cadence=?, source_type='manual', notes=COALESCE(?, notes), updated_at=? WHERE plan_item_id=?", (monthly, annual, cadence, payload.get("notes"), ts, plan_id)) else: plan_id = new_id("bplan") cat_name = conn.execute("SELECT name FROM budget_categories WHERE category_id=?", (category_id,)).fetchone()["name"] conn.execute(""" INSERT INTO budget_plan_items(plan_item_id, plan_month, category_id, name, monthly_amount_chf, annual_amount_chf, cadence, is_fixed_cost, source_type, notes, sort_order, is_active, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, 0, 'manual', ?, 999, 1, ?, ?) """, (plan_id, f"{payload['year']}-01", category_id, f"Budget {cat_name}", monthly, annual, cadence, payload.get("notes"), ts, ts)) return plan_id def _replace_baselines(conn: Connection, category_id: str, payload: dict[str, Any]) -> None: ts = now() conn.execute("DELETE FROM budget_category_baselines WHERE category_id=? AND year=? AND baseline_type IN ('actual_previous_year','planned_budget')", (category_id, payload["previous_year"])) months = payload.get("previous_year_months") or {} for month, amount in months.items(): conn.execute(""" INSERT INTO budget_category_baselines(baseline_id, category_id, year, month, amount_text, currency, baseline_type, source, notes, created_at, updated_at) VALUES (?, ?, ?, ?, ?, 'CHF', 'actual_previous_year', ?, ?, ?, ?) """, (new_id("bbase"), category_id, payload["previous_year"], month, _fmt(Decimal(str(amount))), payload.get("source") or "manual", payload.get("notes"), ts, ts)) if not months and Decimal(str(payload.get("previous_year_total_chf") or "0")) > 0: conn.execute(""" INSERT INTO budget_category_baselines(baseline_id, category_id, year, month, amount_text, currency, baseline_type, source, notes, created_at, updated_at) VALUES (?, ?, ?, NULL, ?, 'CHF', 'actual_previous_year', ?, ?, ?, ?) """, (new_id("bbase"), category_id, payload["previous_year"], payload["previous_year_total_chf"], payload.get("source") or "manual", payload.get("notes"), ts, ts)) def confirm_category_budget_values(conn: Connection, category_id: str, payload: dict[str, Any]) -> dict[str, Any]: old_snapshot = _current_planning_snapshot(conn, category_id, str(payload.get("year") or "2026")) preview = preview_category_budget_values(conn, category_id, payload) normalised = preview["payload"] plan_id = _upsert_plan(conn, category_id, normalised) _replace_baselines(conn, category_id, normalised) new_snapshot = _current_planning_snapshot(conn, category_id, normalised["year"]) audit_id = record_audit_event( conn, source="vue_dashboard", action="budget_planning_confirmed", entity_type="budget_category", entity_id=category_id, old_values=old_snapshot, new_values={**new_snapshot, "plan_id": plan_id, "year": normalised["year"], "previous_year": normalised["previous_year"], "changes": _changes(old_snapshot, new_snapshot)}, user_text_note=normalised.get("notes"), created_by="user", ) conn.commit() return {"status": "confirmed", "entity_id": category_id, "plan_item_id": plan_id, "audit_id": audit_id, "message": "Budgetplanung gespeichert"} def preview_budget_planning_batch(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: changes = payload.get("changes") or [] if not isinstance(changes, list) or not changes: raise HTTPException(status_code=422, detail="changes required") previews = [] normalised = [] for change in changes: if not isinstance(change, dict) or not change.get("category_id"): raise HTTPException(status_code=422, detail="category_id required for every change") preview = preview_category_budget_values(conn, str(change["category_id"]), change) previews.append({"category_id": str(change["category_id"]), "summary": preview["summary"], "warnings": preview["warnings"], "review": preview["review"]}) normalised.append(preview["payload"]) preview_id = new_id("preview") return {"preview_id": preview_id, "summary": f"{len(normalised)} Budgetänderungen prüfen", "count": len(normalised), "changes": previews, "warnings": [w for p in previews for w in p["warnings"]], "payload": {"changes": normalised}, "requires_explicit_confirm": True} def confirm_budget_planning_batch(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: preview = preview_budget_planning_batch(conn, payload) audit_ids: list[str] = [] for change in preview["payload"]["changes"]: result = confirm_category_budget_values(conn, str(change["category_id"]), change) audit_ids.append(result["audit_id"]) batch_audit_id = record_audit_event(conn, source="vue_dashboard", action="budget_planning_batch_confirmed", entity_type="budget_planning_batch", entity_id=preview["preview_id"], new_values={"confirmed_count": len(audit_ids), "category_audit_ids": audit_ids}, created_by="user") conn.commit() return {"status": "confirmed", "entity_id": preview["preview_id"], "audit_id": batch_audit_id, "confirmed_count": len(audit_ids), "message": "Budgetänderungen gespeichert"} def _load_audit_json(value: Any) -> dict[str, Any]: try: return json.loads(value or "{}") except Exception: return {} def get_budget_planning_audit_timeline(conn: Connection, category_id: str, *, limit: int = 25) -> dict[str, Any]: cat = _expense_category(conn, category_id) rows = conn.execute( """ SELECT * FROM audit_log WHERE entity_type='budget_category' AND entity_id=? AND action IN ('budget_planning_confirmed','budget_planning_preview') ORDER BY timestamp DESC LIMIT ? """, (category_id, limit), ).fetchall() entries = [] for row in rows: old = _load_audit_json(row["old_values_json"]) new = _load_audit_json(row["new_values_json"]) changes = new.get("changes") if isinstance(new.get("changes"), list) else _changes(old, new) entries.append({ "timestamp": row["timestamp"], "action": row["action"], "action_label": "Budgetplanung gespeichert" if row["action"] == "budget_planning_confirmed" else "Preview erstellt", "source": row["source"], "note": row["user_text_note"] or new.get("notes") or "", "created_by": row["created_by"] or "system", "changes": changes, "technical": {"audit_id": row["audit_id"]}, }) return {"category_id": category_id, "category": cat["name"], "entries": entries} def preview_budget_planning_excel_template(conn: Connection, payload: dict[str, Any] | None = None) -> dict[str, Any]: """Read-only profile contract for the historical XLS; never mutates planning data.""" body = payload or {} year = str(body.get("year") or "2026") expenses = Decimal(str(body.get("annual_expenses_chf") or "114569.39")) income = Decimal(str(body.get("annual_income_chf") or "148775.40")) surplus = income - expenses monthly_surplus = surplus / Decimal("12") mortgage_payment = Decimal(str(body.get("mortgage_payment_chf") or "1298.75")) mortgage_annual, mortgage_reserve = calculate_annual_and_reserve(mortgage_payment, "quarterly") positions = body.get("positions") if isinstance(body.get("positions"), list) else [ { "name": "Hypothek", "position_type": "fixed_cost", "payment_amount_chf": _fmt(mortgage_payment), "cadence": "quarterly", "due_months": [3, 6, 9, 12], "annual_amount_chf": _fmt(mortgage_annual), "monthly_reserve_chf": _fmt(mortgage_reserve), "calculation_basis": "XLS-Jahreswert CHF 5’195.00 / vier Zahlungen", "certainty": "variable", "conflict": "Die XLS-Spalte bezeichnet CHF 1’298.75 irreführend als monatlich.", } ] controls = { "annual_expenses_chf": _fmt(expenses), "annual_income_chf": _fmt(income), "annual_surplus_chf": _fmt(surplus), "monthly_surplus_chf": _fmt(monthly_surplus), "mortgage_payment_chf": _fmt(mortgage_payment), "mortgage_annual_chf": _fmt(mortgage_annual), "mortgage_monthly_reserve_chf": _fmt(mortgage_reserve), "source_monthly_rest_chf": "1984.67", } return { "status": "profiled", "title": "Aus bisheriger XLS übernehmen", "summary": "Die XLS ist eine read-only Eingabequelle. Rhythmen, Konflikte, Jahreswerte und Monatsrückstellungen werden vor einer späteren Übernahme geprüft.", "year": year, "source_label": "budget_tool_2026.xls", "source_profile": {"sheet": "2026", "budget_position_count": 28, "formula_truth": False, "read_only": True}, "positions": positions, "controls": controls, "conflicts": [ {"code": "quarterly_as_monthly", "message": "Hypothek CHF 1’298.75 ist vierteljährlich; korrekte Monatsrückstellung CHF 432.92."}, {"code": "monthly_rest_wrong", "message": "CHF 1’984.67 Monatsrest ist falsch; Jahresüberschuss / 12 ergibt CHF 2’850.50."}, ], "productive_mutation": False, "requires_explicit_confirm": True, "confirm_available": False, "warnings": ["Sprint 17D führt keinen produktiven XLS-Confirm aus."], } def _plan_amounts(conn: Connection, year: str) -> dict[str, dict[str, Any]]: latest = conn.execute("SELECT snapshot_json FROM budget_plan_versions WHERE plan_year=? ORDER BY version_number DESC LIMIT 1", (year,)).fetchone() out: dict[str, dict[str, Any]] = {} if latest: snapshot = json.loads(str(latest["snapshot_json"])) for position in snapshot.get("positions") or []: if position.get("position_type") == "income" or not position.get("category_id"): continue cid = str(position["category_id"]) month = Decimal(str(position.get("monthly_reserve_chf") or "0")) annual = Decimal(str(position.get("annual_amount_chf") or "0")) item = out.setdefault(cid, {"budget_month": Decimal("0"), "budget_year": Decimal("0"), "cadence": "versioned", "positions": []}) item["budget_month"] += month item["budget_year"] += annual item["positions"].append(position) return out rows = conn.execute(""" SELECT p.*, c.name AS category_name FROM budget_plan_items p JOIN budget_categories c ON c.category_id=p.category_id WHERE p.is_active=1 AND c.is_active=1 AND c.category_type='expense' AND substr(p.plan_month,1,4)=? """, (year,)).fetchall() for row in rows: cid = str(row["category_id"]) item = out.setdefault(cid, {"budget_month": Decimal("0"), "budget_year": Decimal("0"), "cadence": str(row["cadence"] or "monthly"), "positions": []}) month = Decimal(str(row["monthly_amount_chf"] or "0")) if row["monthly_amount_chf"] else (Decimal(str(row["annual_amount_chf"] or "0")) / Decimal("12") if row["annual_amount_chf"] else Decimal("0")) year_amount = Decimal(str(row["annual_amount_chf"] or "0")) if row["annual_amount_chf"] else month * Decimal("12") item["budget_month"] += month item["budget_year"] += year_amount item["cadence"] = str(row["cadence"] or item["cadence"]) item["positions"].append({"payment_amount_chf": str(row["payment_amount_text"] or month), "monthly_reserve_chf": _fmt(month), "annual_amount_chf": _fmt(year_amount), "cadence": str(row["planning_cadence"] or row["cadence"] or "monthly"), "due_months": json.loads(row["due_months_json"] or "[]")}) return out def _remaining_position_amount(position: dict[str, Any], elapsed_month: int) -> Decimal: annual = Decimal(str(position.get("annual_amount_chf") or "0")) payment = Decimal(str(position.get("payment_amount_chf") or position.get("payment_amount_text") or "0")) cadence = str(position.get("cadence") or "monthly") due_months = position.get("due_months") or [] if due_months: return payment * Decimal(sum(1 for month in due_months if int(month) > elapsed_month)) if cadence == "monthly": return payment * Decimal(12 - elapsed_month) return annual * Decimal(12 - elapsed_month) / Decimal("12") def _remaining_plan_amount(plan: dict[str, Any], elapsed_month: int) -> Decimal: details = list(plan.get("positions") or []) if not details: return Decimal(str(plan.get("budget_month") or "0")) * Decimal(12 - elapsed_month) return sum((_remaining_position_amount(position, elapsed_month) for position in details), Decimal("0")) def _actual_months( conn: Connection, year: str, effect_rows: list[dict] | None = None ) -> dict[str, dict[str, Decimal]]: out: dict[str, dict[str, Decimal]] = {} rows = effect_rows if effect_rows is not None else list_household_financial_effects( conn, date_from=f"{year}-01-01", date_to=f"{year}-12-31" ) for row in rows: category_id = row["effective_category_id"] if not category_id: continue category = out.setdefault(str(category_id), {}) month = str(row["transaction_date"])[5:7] category[month] = category.get(month, Decimal("0")) + Decimal( str(row["expense_effect"] or "0") ) return out def _previous_year_months(conn: Connection, year: str = "2025") -> dict[str, dict[str, Decimal]]: rows = conn.execute("SELECT category_id, month, amount_text FROM budget_category_baselines WHERE year=? AND baseline_type='actual_previous_year'", (year,)).fetchall() out: dict[str, dict[str, Decimal]] = {} for row in rows: cid = str(row["category_id"]) if row["month"]: out.setdefault(cid, {})[str(row["month"]).zfill(2)] = Decimal(str(row["amount_text"] or "0")) else: out.setdefault(cid, {})["__year__"] = Decimal(str(row["amount_text"] or "0")) return out def _status(has_budget: bool, budget_year: Decimal, forecast: Decimal, actual_ytd: Decimal) -> str: if not has_budget: return "Budget fehlt" if actual_ytd > 0 else "kein Budget" if actual_ytd == 0: return "Noch keine Daten" if budget_year <= 0: return "Budget fehlt" ratio = forecast / budget_year if ratio > Decimal("1.15"): return "Kritisch" if ratio > Decimal("1"): return "Rot" if ratio >= Decimal("0.90"): return "Gelb" return "Grün" def _position_amount_for_month(position: dict[str, Any], month: int) -> Decimal: payment = Decimal(str(position.get("payment_amount_chf") or position.get("payment_amount_text") or "0")) due_months = [int(value) for value in position.get("due_months") or []] cadence = str(position.get("cadence") or "monthly") if due_months: return payment if month in due_months else Decimal("0") if cadence == "monthly": return payment if cadence in {"one_time", "yearly", "quarterly", "semiannual"}: return Decimal("0") return Decimal(str(position.get("annual_amount_chf") or "0")) / Decimal("12") def _comparison_evaluation( *, reliable: bool, forecast: Decimal, comparison: Decimal | None, comparison_year: str, ) -> str: if comparison is None: return f"Vergleichswert für {comparison_year} noch nicht erfasst" if not reliable: return "Prognose noch nicht verlässlich" deviation = forecast - comparison if comparison > 0: percent = deviation / comparison * Decimal("100") if abs(percent) < Decimal("5"): return "Im ähnlichen Bereich" if deviation > 0: return f"Voraussichtlich {abs(percent).quantize(Decimal('1'))} % mehr als {comparison_year}" rounded = abs(deviation).quantize(Decimal("0.01")) amount = ( f"{int(rounded):,}".replace(",", "'") if rounded == rounded.to_integral_value() else f"{rounded:,.2f}".replace(",", "'") ) return f"Rund CHF {amount} unter {comparison_year}" if deviation == 0: return "Im ähnlichen Bereich" return f"Vergleichsjahr {comparison_year} bestätigt mit CHF 0.00; Prozent nicht berechenbar" def _covered_closed_months( conn: Connection, year: str, current_month_number: int, effect_rows: list[dict[str, Any]], ) -> list[str]: closed = set(MONTHS[: max(0, current_month_number - 1)]) months_with_confirmed_data = { str(row["transaction_date"])[5:7] for row in effect_rows if str(row.get("transaction_date") or "").startswith(f"{year}-") } import_intervals: list[tuple[date, date]] = [] for row in conn.execute( """SELECT file_period_start, file_period_end FROM budget_import_sessions WHERE status NOT IN ('dry_run','failed') AND COALESCE(error_count,0)=0 AND file_period_start IS NOT NULL AND file_period_end IS NOT NULL ORDER BY file_period_start, file_period_end""" ).fetchall(): try: start = date.fromisoformat(str(row["file_period_start"])[:10]) end = date.fromisoformat(str(row["file_period_end"])[:10]) except ValueError: continue if end < start: continue if import_intervals and start <= import_intervals[-1][1] + timedelta(days=1): import_intervals[-1] = ( import_intervals[-1][0], max(import_intervals[-1][1], end), ) else: import_intervals.append((start, end)) import_covered = { month for month in closed if any( start <= date(int(year), int(month), 1) and end >= date( int(year), int(month), calendar.monthrange(int(year), int(month))[1], ) for start, end in import_intervals ) } open_rows = conn.execute( """SELECT DISTINCT substr(transaction_date,6,2) AS month FROM budget_transaction_candidates WHERE substr(COALESCE(transaction_date,''),1,4)=? AND status IN ('pending','needs_review','edited','auto_categorized','transfer_candidate')""", (year,), ).fetchall() incomplete = {str(row["month"]) for row in open_rows if row["month"]} eligible = closed & months_with_confirmed_data & import_covered - incomplete contiguous: list[str] = [] for month_number in range(current_month_number - 1, 0, -1): month = f"{month_number:02d}" if month not in eligible: break contiguous.append(month) return list(reversed(contiguous)) def _remaining_month_parts(data_as_of: str) -> Decimal: parsed = date.fromisoformat(data_as_of) days = calendar.monthrange(parsed.year, parsed.month)[1] current_fraction = Decimal(days - parsed.day) / Decimal(days) return Decimal(12 - parsed.month) + current_fraction def get_budget_planning_matrix( conn: Connection, *, year: str = "2026", current_month: str | None = None, comparison_year: str = "2025", _effect_rows: list[dict] | None = None, ) -> dict[str, Any]: comparison_year = validate_comparison_year(comparison_year) latest_year_row = conn.execute( """SELECT MAX(transaction_date) AS d FROM budget_transactions WHERE status='confirmed' AND substr(transaction_date,1,4)=?""", (year,), ).fetchone() latest_year_date = str(latest_year_row["d"] or "") selected_month = current_month or ( latest_year_date[:7] if latest_year_date else f"{year}-01" ) elapsed = max(1, min(12, int(selected_month[5:7]))) month_last_day = calendar.monthrange(int(year), elapsed)[1] cutoff = f"{year}-{elapsed:02d}-{month_last_day:02d}" max_date_row = conn.execute( """SELECT MAX(transaction_date) AS d FROM budget_transactions WHERE status='confirmed' AND substr(transaction_date,1,4)=? AND transaction_date<=?""", (year, cutoff), ).fetchone() max_date = str(max_date_row["d"] or "") data_as_of = max_date if max_date else f"{selected_month}-01" remaining_parts = _remaining_month_parts(data_as_of).quantize(Decimal("0.01")) cats = conn.execute( """SELECT category_id, name, sort_order FROM budget_categories WHERE is_active=1 AND category_type='expense' ORDER BY sort_order, name""" ).fetchall() effect_rows = _effect_rows if _effect_rows is not None else list_household_financial_effects( conn, date_from=f"{year}-01-01", date_to=f"{year}-12-31" ) actuals = _actual_months(conn, year, effect_rows) plans = _plan_amounts(conn, year) legacy_previous = _previous_year_months(conn, comparison_year) recurring_by_category: dict[str, dict[str, Decimal]] = {} for recurring_row in conn.execute( """SELECT category_id, expected_amount_text, frequency FROM budget_recurring_payments WHERE status='active'""" ).fetchall(): amount = Decimal(str(recurring_row["expected_amount_text"] or "0")) frequency = str(recurring_row["frequency"] or "") factors = { "monthly": (Decimal("1"), Decimal("12")), "quarterly": (Decimal("0.3333333333"), Decimal("4")), "yearly": (Decimal("0.0833333333"), Decimal("1")), "weekly": (Decimal("4.3333333333"), Decimal("52")), } month_factor, year_factor = factors.get( frequency, (Decimal("0"), Decimal("0")) ) bucket = recurring_by_category.setdefault( str(recurring_row["category_id"]), {"month": Decimal("0"), "year": Decimal("0")}, ) bucket["month"] += amount * month_factor bucket["year"] += amount * year_factor covered_months = _covered_closed_months(conn, year, elapsed, effect_rows) comparison = get_prior_year_actuals(conn, year=comparison_year) comparison_by_category = { str(row["category_id"]): row["annual_actual_chf"] for row in comparison["rows"] } quality = get_household_financial_summary( conn, date_from=f"{year}-01-01", date_to=data_as_of ) quality_warnings = list(quality.get("warnings") or []) sufficient = len(covered_months) >= 3 and not quality_warnings rows: list[dict[str, Any]] = [] for cat in cats: cid = str(cat["category_id"]) category_months = {m: actuals.get(cid, {}).get(m, Decimal("0")) for m in MONTHS} actual_ytd = sum((category_months[m] for m in MONTHS[:elapsed]), Decimal("0")) completed_actual = sum((category_months[m] for m in covered_months), Decimal("0")) average = ( (completed_actual / Decimal(len(covered_months))).quantize(Decimal("0.01")) if covered_months else Decimal("0") ) forecast = actual_ytd + average * remaining_parts forecast_display = _fmt(forecast) if sufficient else None comparison_text = comparison_by_category.get(cid) comparison_amount = Decimal(comparison_text) if comparison_text is not None else None plan = plans.get(cid, {}) budget_month = Decimal(str(plan.get("budget_month") or "0")) budget_year = Decimal(str(plan.get("budget_year") or "0")) has_budget = cid in plans and (budget_month > 0 or budget_year > 0) previous_months = legacy_previous.get(cid, {}) previous_covered = {m: v for m, v in previous_months.items() if m in MONTHS} previous_ytd = sum( (previous_covered.get(m, Decimal("0")) for m in MONTHS[:elapsed]), Decimal("0"), ) known_recurring = recurring_by_category.get( cid, {"month": Decimal("0"), "year": Decimal("0")} ) deviation = forecast - comparison_amount if sufficient and comparison_amount is not None else None deviation_percent = ( deviation / comparison_amount * Decimal("100") if deviation is not None and comparison_amount is not None and comparison_amount > 0 else None ) current_actual = category_months[f"{elapsed:02d}"] current_remaining_fraction = remaining_parts - Decimal(12 - elapsed) current_forecast = current_actual + average * current_remaining_fraction transaction_count = sum( 1 for row in effect_rows if str(row.get("effective_category_id") or "") == cid and Decimal(str(row.get("expense_effect") or "0")) != 0 and str(row.get("transaction_date") or "") <= data_as_of ) evaluation = _comparison_evaluation( reliable=sufficient, forecast=forecast, comparison=comparison_amount, comparison_year=comparison_year, ) legacy_status = _status(has_budget, budget_year, forecast, actual_ytd) status_colors = { "Kritisch": "red", "Rot": "red", "Gelb": "yellow", "Grün": "green", "Noch keine Daten": "gray", "Budget fehlt": "gray", "kein Budget": "gray", } plausibility_checks = _plausibility_checks( budget_year=budget_year, previous_total=comparison_amount or Decimal("0"), actual_ytd=actual_ytd, forecast=forecast, has_budget=has_budget, has_previous=comparison_amount is not None, ) rows.append( { "category_id": cid, "category": str(cat["name"]), "category_type": "expense", "actual_current_month_chf": _fmt(current_actual), "actual_months_2026": {m: _fmt(v) for m, v in category_months.items()}, "actual_year_to_date_chf": _fmt(actual_ytd), "actual_avg_month_chf": _fmt(average) if covered_months else None, "forecast_year_chf": _fmt(forecast), "forecast_display_chf": forecast_display, "forecast_reliable": sufficient, "forecast_is_estimate": sufficient, "forecast_basis": ( f"Ist bis {data_as_of} plus Durchschnitt aus {len(covered_months)} vollständig abgedeckten Monaten × {_fmt(remaining_parts)} verbleibende Monatsanteile" if sufficient else "Noch nicht verlässlich: mindestens drei vollständig abgedeckte abgeschlossene Monate erforderlich" ), "current_month_forecast_chf": _fmt(current_forecast) if sufficient else None, "comparison_year": comparison_year, "comparison_year_actual_chf": comparison_text, "comparison_monthly_average_chf": ( _fmt(comparison_amount / Decimal("12")) if comparison_amount is not None else None ), "deviation_chf": _fmt(deviation) if deviation is not None else None, "deviation_percent": _fmt(deviation_percent) if deviation_percent is not None else None, "evaluation": evaluation, "data_as_of": data_as_of, "covered_complete_months": covered_months, "covered_complete_month_count": len(covered_months), "confirmed_transaction_count": transaction_count, "used_monthly_average_chf": _fmt(average) if sufficient else None, "remaining_month_parts": _fmt(remaining_parts), "forecast_remaining_chf": _fmt(average * remaining_parts) if sufficient else None, "excluded_transfers": True, "excluded_credit_card_settlements": True, "previous_year_total_chf": comparison_text, "previous_year_monthly_average_chf": ( _fmt(comparison_amount / Decimal("12")) if comparison_amount is not None else None ), "has_previous_year": comparison_amount is not None, "previous_year_months": {m: _fmt(v) for m, v in previous_covered.items()}, "previous_year_coverage_months": len(previous_covered), "previous_year_coverage_sufficient": len(previous_covered) >= 3, "budget_month_chf": _fmt(budget_month) if has_budget else None, "budget_year_chf": _fmt(budget_year) if has_budget else None, "budget_cadence": str(plan.get("cadence")) if has_budget else None, "has_budget": has_budget, "linear_forecast_chf": _fmt(forecast), "budget_adjusted_forecast_chf": _fmt( forecast if sufficient else max(forecast, budget_month) ), "actual_vs_budget_chf": _fmt(actual_ytd - budget_year) if has_budget else None, "forecast_vs_budget_chf": _fmt(forecast - budget_year) if has_budget else None, "forecast_vs_previous_year_chf": _fmt(deviation) if deviation is not None else None, "current_month_vs_budget_chf": _fmt(current_actual - budget_month) if has_budget else None, "current_month_vs_previous_year_chf": ( _fmt(current_actual - previous_covered[f"{elapsed:02d}"]) if f"{elapsed:02d}" in previous_covered else None ), "ytd_vs_previous_year_chf": ( _fmt(actual_ytd - previous_ytd) if previous_covered else None ), "status": legacy_status, "status_color": status_colors.get(legacy_status, "gray"), "known_recurring_month_chf": _fmt(known_recurring["month"]), "known_recurring_year_chf": _fmt(known_recurring["year"]), "fixed_cost_quote_percent": ( _fmt(known_recurring["year"] / forecast * Decimal("100")) if forecast > 0 else "0.00" ), "variable_actual_year_to_date_chf": _fmt( max(Decimal("0"), actual_ytd - known_recurring["year"] * Decimal(elapsed) / Decimal("12")) ), "data_explorer_url": ( f"/planning/budget/analysis/data-explorer?year={year}&type=expense" f"&category_id={cid}&comparison_year={comparison_year}" ), "effective_expenses_url": f"/planning/budget/expenses/actual?year={year}&category_id={cid}", "monthly_detail_url": f"/planning/budget/status?category_id={cid}&month={selected_month}", "monthly_chart": [ { "month": m, "budget_chf": _fmt(budget_month) if has_budget else None, "actual_2025_chf": _fmt(previous_covered.get(m, Decimal("0"))), "actual_2026_chf": _fmt(category_months[m]), } for m in MONTHS ], "plausibility_checks": plausibility_checks, } ) rows.sort( key=lambda row: ( row["deviation_chf"] is None, -Decimal(str(row["deviation_chf"] or "0")), -Decimal(str(row["actual_year_to_date_chf"])), row["category"], ) ) comparison_total_text = comparison["totals"]["annual_actual_chf"] comparison_total = Decimal(comparison_total_text) if comparison_total_text is not None else None active_category_ids = {str(row["category_id"]) for row in cats} ytd_effect_rows = [ row for row in effect_rows if str(row.get("transaction_date") or "") <= cutoff ] actual_total = sum( (Decimal(str(row.get("expense_effect") or "0")) for row in ytd_effect_rows), Decimal("0"), ) unrepresented_rows = [ row for row in ytd_effect_rows if Decimal(str(row.get("expense_effect") or "0")) != 0 and str(row.get("effective_category_id") or "") not in active_category_ids ] unrepresented_actual = sum( (Decimal(str(row.get("expense_effect") or "0")) for row in unrepresented_rows), Decimal("0"), ) matrix_quality_warnings = list(quality_warnings) if unrepresented_rows: matrix_quality_warnings.append( { "code": "expenses_outside_active_categories", "message": "Bestätigte Ausgaben sind keiner aktiven kanonischen Kategorie zugeordnet.", "amount_chf": _fmt(unrepresented_actual), "transaction_count": len(unrepresented_rows), } ) reliable_rows = [row for row in rows if row["forecast_display_chf"] is not None] forecast_total = ( sum((Decimal(str(row["forecast_display_chf"])) for row in reliable_rows), Decimal("0")) if reliable_rows and not unrepresented_rows else None ) totals_comparable = ( forecast_total is not None and comparison_total is not None and comparison["totals"]["missing_category_count"] == 0 and len(reliable_rows) == len(rows) ) total_deviation = ( forecast_total - comparison_total if totals_comparable and forecast_total is not None and comparison_total is not None else None ) total_percent = ( total_deviation / comparison_total * Decimal("100") if total_deviation is not None and comparison_total and comparison_total > 0 else None ) return { "purpose": "budget_planning_forecast_v1", "year": year, "comparison_year": comparison_year, "previous_year": comparison_year, "current_month": selected_month, "data_as_of": data_as_of, "forecast_method": "Jahresprognose = Ist bis Datenstand + Durchschnitt vollständig abgedeckter Monate × verbleibende Monatsanteile", "rows": rows, "totals": { "comparison_year_actual_chf": comparison_total_text, "comparison_monthly_average_chf": comparison["totals"]["monthly_average_chf"], "actual_year_to_date_chf": _fmt(actual_total), "unrepresented_actual_chf": _fmt(unrepresented_actual), "unrepresented_transaction_count": len(unrepresented_rows), "forecast_year_chf": _fmt(forecast_total) if forecast_total is not None else None, "deviation_chf": _fmt(total_deviation) if total_deviation is not None else None, "deviation_percent": _fmt(total_percent) if total_percent is not None else None, "missing_comparison_category_count": comparison["totals"]["missing_category_count"], }, "covered_complete_months": covered_months, "quality_warnings": matrix_quality_warnings, "categories_without_budget": [row["category_id"] for row in rows if not row["has_budget"]], "categories_without_previous_year": [ row["category_id"] for row in rows if row["comparison_year_actual_chf"] is None ], "categories_with_actuals_without_budget": [ row["category_id"] for row in rows if not row["has_budget"] and Decimal(row["actual_year_to_date_chf"]) > 0 ], "categories_with_budget_without_actuals": [ row["category_id"] for row in rows if row["has_budget"] and Decimal(row["actual_year_to_date_chf"]) == 0 ], "categories_with_previous_year_without_budget": [ row["category_id"] for row in rows if not row["has_budget"] and row["comparison_year_actual_chf"] is not None ], "forecast_method_help": { "primary": "Nur bestätigte kanonische Ausgaben; keine Plan-, Abo- oder Recurring-Zuschläge", "current_month": "Der aktuelle Monat zählt im Ist; nur sein verbleibender Anteil wird prognostiziert", }, } def get_category_forecast_explanation( conn: Connection, *, category_id: str, year: str = "2026", comparison_year: str = "2025", current_month: str | None = None, ) -> dict[str, Any]: matrix = get_budget_planning_matrix( conn, year=year, comparison_year=comparison_year, current_month=current_month, ) row = next((item for item in matrix["rows"] if item["category_id"] == category_id), None) if row is None: raise HTTPException(status_code=404, detail="active expense category not found") excluded_transfer_count = int( conn.execute( """SELECT COUNT(*) FROM budget_transactions WHERE status='confirmed' AND substr(transaction_date,1,4)=? AND transaction_type='transfer'""", (year,), ).fetchone()[0] or 0 ) excluded_card_settlement_count = int( conn.execute( """SELECT COUNT(*) FROM budget_transaction_candidates WHERE substr(COALESCE(transaction_date,''),1,4)=? AND classification IN ('credit_card_payment','credit_card_payment_counterpost')""", (year,), ).fetchone()[0] or 0 ) return { "title": "So entsteht die Prognose", "category_id": category_id, "category": row["category"], "year": year, "data_as_of": row["data_as_of"], "covered_complete_months": row["covered_complete_months"], "covered_complete_month_count": row["covered_complete_month_count"], "confirmed_transaction_count": row["confirmed_transaction_count"], "actual_year_to_date_chf": row["actual_year_to_date_chf"], "used_monthly_average_chf": row["used_monthly_average_chf"], "remaining_month_parts": row["remaining_month_parts"], "forecast_remaining_chf": row["forecast_remaining_chf"], "forecast_year_chf": row["forecast_display_chf"], "forecast_reliable": row["forecast_reliable"], "comparison_year": row["comparison_year"], "comparison_year_actual_chf": row["comparison_year_actual_chf"], "deviation_chf": row["deviation_chf"], "deviation_percent": row["deviation_percent"], "excluded_transfer_count": excluded_transfer_count, "excluded_credit_card_settlement_count": excluded_card_settlement_count, "exclusions": [ "Interne Transfers zählen nicht als Ausgabe.", "Kreditkarten-Ausgleiche zählen nicht zusätzlich zu den Käufen.", "Abo- und Recurring-Kandidaten werden nicht zugeschlagen.", ], "calculation": row["forecast_basis"], } ANNUAL_POSITION_TYPES = {"income", "fixed_cost", "variable_expense", "one_time_seasonal", "savings_investment"} ANNUAL_CADENCES = {"weekly", "monthly", "quarterly", "semiannual", "yearly", "one_time", "irregular", "planned", "undetermined"} def _stable_fingerprint(value: Any) -> str: return hashlib.sha256(json.dumps(value, sort_keys=True, default=str, separators=(",", ":")).encode()).hexdigest() def _annual_source_version(conn: Connection, year: str) -> str: values: dict[str, Any] = {"year": year} for table, date_column in (("budget_plan_items", "updated_at"), ("budget_recurring_payments", "updated_at"), ("budget_transactions", "updated_at")): row = conn.execute(f"SELECT COUNT(*) AS n, MAX({date_column}) AS changed FROM {table}").fetchone() values[table] = [int(row["n"] or 0), row["changed"]] latest = conn.execute("SELECT MAX(version_number) AS n FROM budget_plan_versions WHERE plan_year=?", (year,)).fetchone() values["latest_version"] = int(latest["n"] or 0) if latest else 0 return _stable_fingerprint(values) def _normalise_annual_position(conn: Connection, raw: dict[str, Any]) -> dict[str, Any]: name = str(raw.get("name") or "").strip() if not name: raise HTTPException(status_code=422, detail="plan position name required") position_type = str(raw.get("position_type") or raw.get("item_type") or "variable_expense") if position_type not in ANNUAL_POSITION_TYPES: raise HTTPException(status_code=422, detail="invalid annual plan position type") cadence = str(raw.get("cadence") or raw.get("planning_cadence") or "monthly") if cadence == "annual": cadence = "yearly" if cadence not in ANNUAL_CADENCES: raise HTTPException(status_code=422, detail="invalid annual plan cadence") payment = Decimal(_decimal_text(raw.get("payment_amount_chf") or raw.get("payment_amount_text") or raw.get("monthly_amount_chf") or raw.get("annual_amount_chf"), required=True, allow_zero=False) or "0") annual, reserve = calculate_annual_and_reserve(payment, cadence) supplied_annual = _decimal_text(raw.get("annual_amount_chf") or raw.get("annual_amount_text")) if cadence in {"irregular", "planned", "undetermined"}: if supplied_annual is None: annual = reserve = Decimal("0") else: annual = Decimal(supplied_annual) reserve = annual / Decimal("12") due_months = raw.get("due_months") or [] if not isinstance(due_months, list) or any(not isinstance(month, int) or month not in range(1, 13) for month in due_months): raise HTTPException(status_code=422, detail="due_months must contain month numbers 1..12") category_id = str(raw.get("category_id") or "").strip() or None category = None if category_id: category = conn.execute("SELECT name, category_type FROM budget_categories WHERE category_id=? AND is_active=1", (category_id,)).fetchone() if not category: raise HTTPException(status_code=422, detail="annual plan category not found") expected_category_type = "income" if position_type == "income" else "expense" if str(category["category_type"]) != expected_category_type: raise HTTPException(status_code=422, detail="annual plan category type does not match position type") certainty = str(raw.get("certainty") or "safe") if certainty not in {"safe", "variable"}: raise HTTPException(status_code=422, detail="certainty must be safe or variable") return { "source_plan_item_id": raw.get("source_plan_item_id") or raw.get("plan_item_id"), "position_type": position_type, "name": name, "category_id": category_id, "category": str(category["name"]) if category else str(raw.get("category") or "Ohne Kategorie"), "payment_amount_chf": _fmt(payment), "cadence": cadence, "due_months": sorted(set(due_months)), "annual_amount_chf": _fmt(annual), "monthly_reserve_chf": _fmt(reserve), "calculation_basis": str(raw.get("calculation_basis") or "Manuell bestätigte Planung"), "certainty": certainty, "manual_override": bool(raw.get("manual_override", False)), } def _existing_annual_positions(conn: Connection, year: str) -> list[dict[str, Any]]: latest = conn.execute("SELECT snapshot_json FROM budget_plan_versions WHERE plan_year=? ORDER BY version_number DESC LIMIT 1", (year,)).fetchone() if latest: snapshot = json.loads(str(latest["snapshot_json"])) return list(snapshot.get("positions") or []) rows = conn.execute( """SELECT p.*, c.name AS category_name, c.category_type FROM budget_plan_items p JOIN budget_categories c ON c.category_id=p.category_id WHERE p.is_active=1 AND c.is_active=1 AND substr(p.plan_month,1,4)=? ORDER BY p.sort_order, p.name""", (year,), ).fetchall() positions: list[dict[str, Any]] = [] for row in rows: item = dict(row) cadence = str(item.get("planning_cadence") or item.get("cadence") or "monthly") if cadence == "annual": cadence = "yearly" elif cadence == "fixed": cadence = "monthly" elif cadence in {"variable", "seasonal"}: cadence = "planned" annual = Decimal(str(item.get("annual_amount_chf") or "0")) if cadence == "quarterly" and annual: payment = annual / Decimal("4") elif cadence == "semiannual" and annual: payment = annual / Decimal("2") elif cadence in {"yearly", "one_time"} and annual: payment = annual else: payment = Decimal(str(item.get("payment_amount_text") or item.get("monthly_amount_chf") or item.get("annual_amount_chf") or "0")) if payment <= 0 and annual <= 0: continue if payment <= 0: payment = annual position_type = str(item.get("item_type") or ("income" if item["category_type"] == "income" else "fixed_cost" if item.get("is_fixed_cost") else "variable_expense")) positions.append(_normalise_annual_position(conn, { "plan_item_id": item["plan_item_id"], "name": item["name"], "position_type": position_type, "category_id": item["category_id"], "payment_amount_chf": _fmt(payment), "cadence": cadence, "due_months": json.loads(item.get("due_months_json") or "[]"), "annual_amount_chf": item.get("annual_amount_chf"), "calculation_basis": item.get("calculation_basis") or "Bestehender bestätigter Budgetplan", "certainty": item.get("certainty") or "safe", "manual_override": bool(item.get("manual_override")), })) return positions def _annual_totals(positions: list[dict[str, Any]]) -> dict[str, str]: income = sum((Decimal(item["annual_amount_chf"]) for item in positions if item["position_type"] == "income"), Decimal("0")) expense = sum((Decimal(item["annual_amount_chf"]) for item in positions if item["position_type"] != "income"), Decimal("0")) return {"planned_income_chf": _fmt(income), "planned_expense_chf": _fmt(expense), "planned_surplus_chf": _fmt(income - expense)} def _annual_coverage_gaps(conn: Connection) -> list[dict[str, Any]]: gaps: list[dict[str, Any]] = [] duplicate_groups = conn.execute( """SELECT transaction_date, lower(trim(COALESCE(payee,description,''))) AS merchant, printf('%.2f', abs(CAST(COALESCE(amount_chf,amount_original,'0') AS REAL))) AS amount, COUNT(*) AS row_count FROM budget_transactions WHERE status='confirmed' AND transaction_type IN ('expense','fee') AND ((transaction_date='2026-05-12' AND lower(COALESCE(payee,description,'')) LIKE '%apple%' AND abs(CAST(COALESCE(amount_chf,amount_original,'0') AS REAL))=20.0) OR (transaction_date='2026-05-09' AND lower(COALESCE(payee,description,'')) LIKE '%netflix%' AND abs(CAST(COALESCE(amount_chf,amount_original,'0') AS REAL))=22.9)) GROUP BY transaction_date, merchant, amount HAVING COUNT(*)>1""" ).fetchall() for group in duplicate_groups: merchant = str(group["merchant"]) if "apple" in merchant or "netflix" in merchant: count = int(group["row_count"]) amount = Decimal(str(group["amount"])) gaps.append({"code": "unresolved_confirmed_duplicates", "label": "Apple" if "apple" in merchant else "Netflix", "observed_expense_chf": _fmt(amount * count), "expected_expense_chf": _fmt(amount), "duplicate_count": count - 1, "forecast_treatment": "Nicht als präzise Forecastbasis verwendet; Nutzer-Confirm ausstehend."}) card = conn.execute( """SELECT * FROM budget_transaction_candidates WHERE status IN ('pending','needs_review','edited') AND classification='credit_card_payment' AND lower(COALESCE(merchant,description,'')) LIKE '%viseca%' ORDER BY transaction_date DESC LIMIT 1""" ).fetchone() if card: try: match = _card_match(conn, card) except HTTPException: match = None bank_payment = abs(Decimal(str(card["amount_original"] or "0"))) mapped_purchases = Decimal(str(match["mapped_purchase_sum"])) if match else None gaps.append({ "code": "unresolved_card_settlement", "label": "VISECA", "bank_payment_chf": _fmt(bank_payment), "mapped_purchases_chf": _fmt(mapped_purchases) if mapped_purchases is not None else None, "remaining_difference_chf": _fmt(mapped_purchases - bank_payment) if mapped_purchases is not None else None, "financial_effect_chf": "0.00", "forecast_treatment": "Neutraler Ausgleich; Restdifferenz verhindert präzise Coverage.", }) return gaps def preview_annual_budget_plan(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: year = str(payload.get("year") or "2026") raw_positions = payload.get("positions") if not isinstance(raw_positions, list): raise HTTPException(status_code=422, detail="positions must be a list") positions = [_normalise_annual_position(conn, item) for item in raw_positions] source_version = _annual_source_version(conn, year) totals = _annual_totals(positions) binding = {"year": year, "positions": positions, "source_data_version": source_version} fingerprint = _stable_fingerprint(binding) return {"preview_id": new_id("preview"), "preview_fingerprint": fingerprint, "source_data_version": source_version, "summary": f"Jahresbudget {year}: {len(positions)} Positionen", "payload": binding | {"preview_fingerprint": fingerprint}, "review": totals, "warnings": ["Confirm erzeugt eine neue unveränderliche Budgetversion."], "requires_explicit_confirm": True} def confirm_annual_budget_plan(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: year = str(payload.get("year") or "2026") supplied_fingerprint = str(payload.get("preview_fingerprint") or "") existing = conn.execute("SELECT version_id, audit_id, version_number FROM budget_plan_versions WHERE plan_year=? AND preview_fingerprint=?", (year, supplied_fingerprint)).fetchone() if existing: return {"status": "confirmed", "entity_id": str(existing["version_id"]), "audit_id": str(existing["audit_id"]), "version_number": int(existing["version_number"]), "message": "Budgetversion bereits vorhanden"} preview = preview_annual_budget_plan(conn, {"year": year, "positions": payload.get("positions")}) if payload.get("source_data_version") != preview["source_data_version"] or supplied_fingerprint != preview["preview_fingerprint"]: raise HTTPException(status_code=409, detail="annual budget preview changed; preview again") row = conn.execute("SELECT COALESCE(MAX(version_number),0)+1 AS n FROM budget_plan_versions WHERE plan_year=?", (year,)).fetchone() version_number = int(row["n"]) version_id = new_id("bversion") snapshot = {"year": year, "version_number": version_number, "positions": preview["payload"]["positions"], "totals": preview["review"]} audit_id = record_audit_event(conn, source="vue_dashboard", action="annual_budget_version_confirmed", entity_type="budget_plan_version", entity_id=version_id, new_values={"year": year, "version_number": version_number, "preview_fingerprint": preview["preview_fingerprint"], "source_data_version": preview["source_data_version"], "totals": preview["review"]}, created_by="user") created = now() conn.execute("""INSERT INTO budget_plan_versions(version_id,plan_year,version_number,source_type,preview_fingerprint,source_data_version,summary_json,snapshot_json,audit_id,created_by,created_at) VALUES (?,?,?,?,?,?,?,?,?,'user',?)""", (version_id, year, version_number, "annual_budget_assistant", preview["preview_fingerprint"], preview["source_data_version"], json.dumps(preview["review"], sort_keys=True), json.dumps(snapshot, sort_keys=True), audit_id, created)) for item in preview["payload"]["positions"]: conn.execute("""INSERT INTO budget_plan_version_items(version_item_id,version_id,source_plan_item_id,position_type,name,category_id,payment_amount_text,cadence,due_months_json,annual_amount_text,monthly_reserve_text,calculation_basis,certainty,manual_override,created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (new_id("bvitem"), version_id, item.get("source_plan_item_id"), item["position_type"], item["name"], item.get("category_id"), item["payment_amount_chf"], item["cadence"], json.dumps(item["due_months"]), item["annual_amount_chf"], item["monthly_reserve_chf"], item["calculation_basis"], item["certainty"], 1 if item["manual_override"] else 0, created)) conn.commit() return {"status": "confirmed", "entity_id": version_id, "audit_id": audit_id, "version_number": version_number, "message": "Unveränderliche Budgetversion gespeichert"} def get_annual_budget_assistant( conn: Connection, *, year: str = "2026", current_month: str | None = None, comparison_year: str = "2025", ) -> dict[str, Any]: positions = _existing_annual_positions(conn, year) totals = _annual_totals(positions) max_date = conn.execute( "SELECT MAX(transaction_date) AS d FROM budget_transactions WHERE status='confirmed' AND substr(transaction_date,1,4)=?", (year,), ).fetchone() data_as_of = str(max_date["d"] or f"{year}-01-01") as_of = str(current_month or data_as_of[:7]) elapsed = max(1, min(12, int(as_of[5:7]))) actual = get_household_financial_summary(conn, date_from=f"{year}-01-01", date_to=f"{year}-12-31") month_actual = get_household_financial_summary(conn, period=as_of) effect_rows = list_household_financial_effects( conn, date_from=f"{year}-01-01", date_to=f"{year}-12-31" ) matrix = get_budget_planning_matrix( conn, year=year, current_month=as_of, comparison_year=comparison_year, _effect_rows=effect_rows, ) matrix_forecast_expense = sum( ( Decimal(str(row["forecast_display_chf"])) if row["forecast_display_chf"] is not None else Decimal(str(row["actual_year_to_date_chf"])) for row in matrix["rows"] ), Decimal("0"), ) uncategorized_confirmed = sum( ( Decimal(str(row.get("expense_effect") or "0")) for row in effect_rows if not row.get("effective_category_id") ), Decimal("0"), ) categorized_confirmed = sum( ( Decimal(str(row.get("expense_effect") or "0")) for row in effect_rows if row.get("effective_category_id") ), Decimal("0"), ) planned_expense = Decimal(totals["planned_expense_chf"]) planned_income = Decimal(totals["planned_income_chf"]) actual_income = Decimal(actual["income_chf"]) actual_expense = Decimal(actual["expense_chf"]) remaining_income = sum((_remaining_position_amount(position, elapsed) for position in positions if position["position_type"] == "income"), Decimal("0")) planned_month_income = sum((_position_amount_for_month(position, elapsed) for position in positions if position["position_type"] == "income"), Decimal("0")) actual_month_income = Decimal(month_actual["income_chf"]) forecast_income = ( actual_income - actual_month_income + max(actual_month_income, planned_month_income) + remaining_income if positions else ((actual_income / Decimal(elapsed) * Decimal("12")) if actual_income and elapsed >= 3 else planned_income) ) # Expense forecasting is strictly based on confirmed canonical effects. Existing # plans, recurring contracts and candidates remain available for diagnostics but # never add an amount to the normal annual forecast. uncategorized_confirmed = max(Decimal("0"), uncategorized_confirmed) uncategorized_gaps = ( [ { "code": "uncategorized_confirmed_expenses", "message": "Bestätigte Ausgaben sind noch keiner kanonischen Kategorie zugeordnet.", "amount_chf": _fmt(uncategorized_confirmed), } ] if uncategorized_confirmed > 0 else [] ) confirmed_expense_effect = categorized_confirmed + uncategorized_confirmed category_coverage_percent = ( min( Decimal("100"), categorized_confirmed / confirmed_expense_effect * Decimal("100"), ).quantize(Decimal("0.01")) if confirmed_expense_effect > 0 else Decimal("100") ) forecast_expense = matrix_forecast_expense forecast_surplus = forecast_income - forecast_expense orientation_deviation = forecast_expense - planned_expense recurring_hints: list[dict[str, Any]] = [] quality_warnings = list(actual.get("warnings") or []) unreliable_material = [ row for row in matrix["rows"] if not row["forecast_reliable"] and actual_expense > 0 and Decimal(row["actual_year_to_date_chf"]) / actual_expense >= Decimal("0.10") ] forecast_reliable = ( not quality_warnings and not unreliable_material and uncategorized_confirmed == 0 and any( Decimal(str(row["actual_year_to_date_chf"])) > 0 for row in matrix["rows"] ) and all( row["forecast_reliable"] for row in matrix["rows"] if Decimal(str(row["actual_year_to_date_chf"])) > 0 ) and bool(actual_income or actual_expense) ) estimate = forecast_reliable free_plannable = forecast_surplus if forecast_reliable else None optimization_hints: list[dict[str, Any]] = [] month_expense_forecast = sum( ( Decimal(str(row["current_month_forecast_chf"])) if row["current_month_forecast_chf"] is not None else Decimal(str(row["actual_current_month_chf"])) for row in matrix["rows"] ), Decimal("0"), ) month_income_forecast = max(Decimal(month_actual["income_chf"]), planned_month_income) month_rows = sorted( matrix["rows"], key=lambda row: Decimal(str(row["actual_current_month_chf"])), reverse=True, )[:3] latest = conn.execute( "SELECT version_id, version_number, created_at FROM budget_plan_versions WHERE plan_year=? ORDER BY version_number DESC LIMIT 1", (year,), ).fetchone() return { "purpose": "annual_budget_assistant_v1", "year": year, "comparison_year": comparison_year, "as_of": as_of, "data_as_of": data_as_of, "calculation_basis": "Ist bis Datenstand plus erwartete Werte für den verbleibenden Zeitraum", "plan": {**totals, "positions": positions, "version": dict(latest) if latest else None}, "actual": {"income_chf": actual["income_chf"], "expense_chf": actual["expense_chf"], "surplus_chf": actual["net_chf"], "semantics_version": actual["semantics_version"]}, "forecast": { "income_chf": _fmt(forecast_income) if forecast_reliable else None, "expense_chf": _fmt(forecast_expense) if forecast_reliable else None, "surplus_chf": _fmt(forecast_surplus) if forecast_reliable else None, "plan_deviation_chf": _fmt(orientation_deviation), "orientation_deviation_chf": _fmt(orientation_deviation), "free_plannable_chf": _fmt(free_plannable) if free_plannable is not None else None, "free_after_special_expenses_chf": _fmt(free_plannable) if free_plannable is not None else None, "free_investable_chf": _fmt(free_plannable) if free_plannable is not None else None, "reliable": forecast_reliable, "is_estimate": estimate, "qualifier": "ca." if estimate else "", }, "month": { "month": as_of, "actual_income_chf": month_actual["income_chf"], "actual_expense_chf": month_actual["expense_chf"], "actual_result_chf": month_actual["net_chf"], "forecast_income_chf": _fmt(month_income_forecast), "forecast_expense_chf": _fmt(month_expense_forecast), "forecast_result_chf": _fmt(month_income_forecast - month_expense_forecast), "previous_year_expense_chf": None, "top_deviations": [{ "category_id": row["category_id"], "category": row["category"], "forecast_chf": row["current_month_forecast_chf"], "previous_year_chf": None, } for row in month_rows], }, "categories": matrix["rows"], "optimization_hints": optimization_hints, "data_quality": { "status": "reliable" if forecast_reliable else "not_reliable", "label": "Schätzung" if forecast_reliable and estimate else ("Verlässlich berechenbar" if forecast_reliable else "Noch nicht verlässlich berechenbar"), "data_as_of": data_as_of, "material_gaps": quality_warnings + uncategorized_gaps + [{"message": row["evaluation"], "category": row["category"]} for row in unreliable_material], "data_hints": recurring_hints, }, "coverage": { "status": "current" if forecast_reliable else "partial", "quality_label": "Schätzung" if estimate else ("teilweise" if not forecast_reliable else "aktuell"), "category_coverage_percent": _fmt(category_coverage_percent), "gaps": recurring_hints + uncategorized_gaps, "financial_quality_warnings": quality_warnings, "free_investable_available": forecast_reliable, }, "sections": {}, "planning_steps": [], "summary_kpis": [ {"key": "actual_expense", "label": "Ist laufendes Jahr bisher", "value_chf": actual["expense_chf"], "kind": "expense"}, {"key": "forecast_expense", "label": "Voraussichtliche Jahresausgaben", "value_chf": _fmt(forecast_expense) if forecast_reliable else None, "kind": "expense"}, {"key": "comparison_actual", "label": f"Jahres-Ist {comparison_year}", "value_chf": matrix["totals"]["comparison_year_actual_chf"], "kind": "neutral"}, ], }