from __future__ import annotations import hashlib import json import re from collections import defaultdict 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, require_decimal_text, validate_currency from jarvis_finance.services.budget_overview import _fmt CADENCES = {"weekly", "monthly", "quarterly", "semiannual", "yearly", "one_time", "irregular", "planned", "undetermined"} LEGACY_CADENCE = {"weekly": "weekly", "monthly": "monthly", "quarterly": "quarterly", "yearly": "yearly"} RECURRING_TYPES = {"fixed_cost", "subscription", "variable_recurring", "income", "one_time"} LEGACY_TYPE = {"fixed_cost": "fixed_cost", "subscription": "subscription", "variable_recurring": "variable_recurring"} STATUSES = {"candidate", "active", "ignored", "paused", "archived"} EXCLUDED_TERMS = ("migros", "migrol", "restaurant", "coop", "galaxus", "digitec", "tank", "shell", "bp ", "true wealth", "investment", "transfer", "kreditkartenausgleich", "credit card", "credit_card_payment", "investment_transfer", "covered_by") VARIABLE_TERMS = ("migros", "coop", "restaurant", "tank", "fuel", "galaxus", "digitec") FIXED_TERMS = ("insurance", "versicherung", "krankenkasse", "miete", "wohnen", "tax", "steuer", "leasing", "schule", "school", "telekom", "sunrise", "salt", "swisscom") SUBSCRIPTION_TERMS = ("netflix", "apple", "google", "disney", "amazon", "prime", "openai", "chatgpt", "dropbox", "norton", "bitdefender", "tradingview", "babbel", "spotify") SEMANTIC_YEARLY_TERMS = ("autobahnvignette", "vignette", "serafe", "jahresbeitrag", "annual") def _dec(value: object) -> Decimal: try: return Decimal(str(value if value not in (None, "") else "0")) except (InvalidOperation, ValueError): return Decimal("0") def _norm(value: str | None) -> str: return re.sub(r"\s+", " ", re.sub(r"[^a-z0-9]+", " ", str(value or "").lower())).strip() def _similar_key(value: str | None) -> str: return _norm(value).replace(" ", "")[:18] def _date(value: str) -> date: return date.fromisoformat(str(value)[:10]) def _safe_date(value: Any) -> date | None: try: if value in (None, ""): return None return _date(str(value)) except (TypeError, ValueError): return None def _add_months(d: date, months: int, day: int | None = None) -> date: month = d.month - 1 + months year = d.year + month // 12 month = month % 12 + 1 desired_day = day or d.day last = 28 while True: try: return date(year, month, min(desired_day, last)) except ValueError: last -= 1 def next_expected_date(last_seen: date, cadence: str, expected_day: int | None = None, expected_month: int | None = None) -> str: if cadence == "weekly": return (last_seen + timedelta(days=7)).isoformat() if cadence == "monthly": return _add_months(last_seen, 1, expected_day).isoformat() if cadence == "quarterly": return _add_months(last_seen, 3, expected_day).isoformat() if cadence == "semiannual": return _add_months(last_seen, 6, expected_day).isoformat() if cadence == "yearly": month = expected_month or last_seen.month return date(last_seen.year + 1, month, min(expected_day or last_seen.day, 28)).isoformat() return "" def _infer_frequency(dates: list[date]) -> str: if len(dates) < 2: return "undetermined" gaps = sorted((dates[i] - dates[i - 1]).days for i in range(1, len(dates))) avg = sum(gaps) / len(gaps) if 24 <= avg <= 38: return "monthly" if 75 <= avg <= 105: return "quarterly" if 150 <= avg <= 220: return "semiannual" if 330 <= avg <= 400: return "yearly" if 5 <= avg <= 10: return "weekly" return "irregular" def _infer_type(name: str, category: str, amount: Decimal, frequency: str) -> str: text = f"{name} {category}".lower() if any(t in text for t in VARIABLE_TERMS): return "variable_recurring" if any(t in text for t in FIXED_TERMS) or amount >= Decimal("100") or frequency in {"quarterly", "yearly"}: return "fixed_cost" if any(t in text for t in SUBSCRIPTION_TERMS) or amount < Decimal("100"): return "subscription" return "variable_recurring" def _excluded(name: str, description: str = "") -> bool: text = f"{name} {description}".lower() return any(term in text for term in EXCLUDED_TERMS) def _row_amount(row: Any) -> Decimal: return _dec(row["amount_chf"] or row["amount_original"] or "0") def calculate_annual_and_reserve(amount: Decimal, cadence: str) -> tuple[Decimal, Decimal]: """Canonical cadence calculation shared by recurring, annual plan and XLS preview.""" if cadence == "monthly": annual = amount * Decimal("12") elif cadence == "quarterly": annual = amount * Decimal("4") elif cadence == "semiannual": annual = amount * Decimal("2") elif cadence == "yearly" or cadence == "one_time": annual = amount elif cadence == "weekly": annual = amount * Decimal("52") else: return Decimal("0"), Decimal("0") return annual, annual / Decimal("12") def _expected_budget(amount: Decimal, cadence: str) -> tuple[Decimal, Decimal]: annual, reserve = calculate_annual_and_reserve(amount, cadence) return reserve, annual def _fingerprint(value: dict[str, Any]) -> str: return hashlib.sha256(json.dumps(value, sort_keys=True, default=str, separators=(",", ":")).encode()).hexdigest() def _legacy_cadence(cadence: str) -> str: return LEGACY_CADENCE.get(cadence, "irregular") def _legacy_type(recurring_type: str) -> str: return LEGACY_TYPE.get(recurring_type, "fixed_cost") def _upsert_candidate(conn: Connection, item: dict[str, Any]) -> str: existing = conn.execute( "SELECT recurring_id, data_version FROM budget_recurring_payments WHERE status='candidate' AND lower(name)=lower(?) AND category_id=?", (item["name"], item["category_id"]), ).fetchone() rid = str(existing["recurring_id"]) if existing else new_id("rec") cadence = str(item.get("planning_cadence") or item.get("frequency") or "undetermined") planning_type = str(item.get("planning_type") or item.get("recurring_type") or "one_time") values = { **item, "recurring_id": rid, "currency": item.get("currency") or "CHF", "frequency": _legacy_cadence(cadence), "recurring_type": _legacy_type(planning_type), "planning_cadence": cadence, "planning_type": planning_type, "amount_min_text": item.get("amount_min_text") or item["expected_amount_text"], "amount_max_text": item.get("amount_max_text") or item["expected_amount_text"], "due_months_json": json.dumps(item.get("due_months") or []), "periodicity_status": item.get("periodicity_status") or "unconfirmed", "amount_tolerance_pct": str(item.get("amount_tolerance_pct") or "10"), "tolerance_percent": str(item.get("tolerance_percent") or item.get("amount_tolerance_pct") or "10"), "tolerance_amount_text": item.get("tolerance_amount_text"), "date_tolerance_days": int(item.get("date_tolerance_days") or 5), "status": "candidate", "source": "detected", "confidence": str(item.get("confidence") or "0.50"), "updated_at": now(), } if existing: conn.execute( """UPDATE budget_recurring_payments SET expected_amount_text=:expected_amount_text, amount_min_text=:amount_min_text, amount_max_text=:amount_max_text, frequency=:frequency, planning_cadence=:planning_cadence, expected_day_of_month=:expected_day_of_month, expected_month=:expected_month, due_months_json=:due_months_json, recurring_type=:recurring_type, planning_type=:planning_type, periodicity_status=:periodicity_status, confidence=:confidence, last_seen_date=:last_seen_date, next_expected_date=NULL, candidate_evidence_json=:candidate_evidence_json, merchant_name=:merchant_name, tolerance_percent=:tolerance_percent, tolerance_amount_text=:tolerance_amount_text, data_version=data_version+1, updated_at=:updated_at WHERE recurring_id=:recurring_id""", values, ) else: conn.execute( """INSERT INTO budget_recurring_payments( recurring_id,name,merchant_name,merchant_id,category_id,account_id, expected_amount_text,currency,frequency,expected_day_of_month,expected_month, tolerance_amount_text,tolerance_percent,amount_tolerance_pct,date_tolerance_days, recurring_type,status,source,confidence,last_seen_date,next_expected_date,notes, candidate_evidence_json,created_at,updated_at,planning_cadence,planning_type, amount_min_text,amount_max_text,due_months_json,periodicity_status,data_version,user_override) VALUES (:recurring_id,:name,:merchant_name,NULL,:category_id,:account_id, :expected_amount_text,:currency,:frequency,:expected_day_of_month,:expected_month, :tolerance_amount_text,:tolerance_percent,:amount_tolerance_pct,:date_tolerance_days, :recurring_type,:status,:source,:confidence,:last_seen_date,NULL,NULL, :candidate_evidence_json,:created_at,:updated_at,:planning_cadence,:planning_type, :amount_min_text,:amount_max_text,:due_months_json,:periodicity_status,1,0)""", {"created_at": now(), **values}, ) return rid def detect_recurring_payment_candidates(conn: Connection, *, today: str | None = None) -> dict[str, Any]: rows = conn.execute( """SELECT t.*, c.name AS category_name FROM budget_transactions t LEFT JOIN budget_categories c ON c.category_id=t.category_id WHERE t.status='confirmed' AND t.transaction_type IN ('expense','fee') ORDER BY t.transaction_date""" ).fetchall() groups: dict[tuple[str, str], list[Any]] = defaultdict(list) for row in rows: name = row["payee"] or row["description"] if not row["category_id"] or _excluded(name, row["description"]): continue groups[(_similar_key(name), row["category_id"])].append(row) candidates: list[dict[str, Any]] = [] for (_key, cid), items in groups.items(): dates = [_date(r["transaction_date"]) for r in items] amounts = [_row_amount(r) for r in items] avg = sum(amounts, Decimal("0")) / Decimal(len(amounts)) name = str(items[-1]["payee"] or items[-1]["description"]) cat = str(items[-1]["category_name"] or "") text = f"{name} {cat}".lower() inferred = _infer_frequency(dates) semantic_yearly = len(items) == 1 and any(term in text for term in SEMANTIC_YEARLY_TERMS) if len(items) == 1 and not semantic_yearly and not any(term in text for term in SUBSCRIPTION_TERMS + FIXED_TERMS): continue max_delta = max(abs(a - avg) for a in amounts) if amounts else Decimal("0") if inferred == "irregular" and not (len(items) >= 3 and avg and max_delta <= avg * Decimal("0.15") and any(t in text for t in SUBSCRIPTION_TERMS + FIXED_TERMS)): continue cadence = "yearly" if semantic_yearly else inferred planning_type = "one_time" if len(items) == 1 else _infer_type(name, cat, avg, cadence) last = max(dates) due_months = [last.month] if semantic_yearly else [] rationale = ( "Semantischer Jahresvorschlag aus einer einzelnen Beobachtung; muss bestätigt werden." if semantic_yearly else ("Einmalige Beobachtung; Periodizität noch festzulegen." if len(items) == 1 else f"Vorschlag aus {len(items)} passenden bestätigten Buchungen.") ) item = { "name": name, "merchant_name": name, "category_id": cid, "account_id": items[-1]["account_id"], "expected_amount_text": _fmt(avg), "amount_min_text": _fmt(min(amounts)), "amount_max_text": _fmt(max(amounts)), "currency": items[-1]["currency_original"] or "CHF", "planning_cadence": cadence, "expected_day_of_month": int(round(sum(d.day for d in dates) / len(dates))), "expected_month": last.month if semantic_yearly else None, "due_months": due_months, "planning_type": planning_type, "periodicity_status": "semantic_proposal" if semantic_yearly else "evidence_proposal" if len(items) > 1 else "unconfirmed", "confidence": "0.86" if len(items) >= 3 else "0.50", "last_seen_date": last.isoformat(), "next_expected_date": None, "candidate_evidence_json": json.dumps({"confirmed_transaction_count": len(items), "sources": ["confirmed_transactions"], "rationale": rationale}, sort_keys=True), } item["candidate_id"] = _upsert_candidate(conn, item) item["frequency"] = cadence item["recurring_type"] = planning_type item["periodicity_rationale"] = rationale candidates.append(item) conn.commit() return {"purpose": "recurring_detection_v2", "today": today, "source_scope": "confirmed_budget_transactions_only", "candidates": sorted(candidates, key=lambda c: c["name"])} def _validate_payload(payload: dict[str, Any]) -> dict[str, Any]: name = str(payload.get("name") or "").strip() if not name: raise HTTPException(status_code=422, detail="name required") amount = require_decimal_text(str(payload.get("expected_amount_text") or payload.get("payment_amount_text") or ""), "expected_amount_text") cadence = str(payload.get("planning_cadence") or payload.get("frequency") or "undetermined") if cadence not in CADENCES: raise HTTPException(status_code=422, detail="invalid cadence") rtype = str(payload.get("planning_type") or payload.get("recurring_type") or "subscription") if rtype not in RECURRING_TYPES: raise HTTPException(status_code=422, detail="invalid recurring_type") due_months = payload.get("due_months") or [] if not isinstance(due_months, list) or any(not isinstance(m, int) or m < 1 or m > 12 for m in due_months): raise HTTPException(status_code=422, detail="due_months must contain month numbers 1..12") amount_min = require_decimal_text(str(payload.get("amount_min_text") or amount), "amount_min_text") amount_max = require_decimal_text(str(payload.get("amount_max_text") or amount), "amount_max_text") if _dec(amount_min) > _dec(amount_max): raise HTTPException(status_code=422, detail="amount range is invalid") return { "name": name, "merchant_id": payload.get("merchant_id"), "category_id": str(payload.get("category_id") or ""), "account_id": payload.get("account_id"), "expected_amount_text": amount, "amount_min_text": amount_min, "amount_max_text": amount_max, "merchant_name": payload.get("merchant_name") or name, "currency": validate_currency(str(payload.get("currency") or "CHF")), "frequency": _legacy_cadence(cadence), "planning_cadence": cadence, "expected_day_of_month": payload.get("expected_day_of_month"), "expected_month": payload.get("expected_month"), "due_months": sorted(set(due_months)), "due_months_json": json.dumps(sorted(set(due_months))), "tolerance_amount_text": require_decimal_text(str(payload["tolerance_amount_text"]), "tolerance_amount_text") if payload.get("tolerance_amount_text") not in (None, "") else None, "tolerance_percent": require_decimal_text(str(payload.get("tolerance_percent") or payload.get("amount_tolerance_pct") or "10"), "tolerance_percent"), "amount_tolerance_pct": require_decimal_text(str(payload.get("amount_tolerance_pct") or payload.get("tolerance_percent") or "10"), "amount_tolerance_pct"), "date_tolerance_days": int(payload.get("date_tolerance_days") or 5), "recurring_type": _legacy_type(rtype), "planning_type": rtype, "periodicity_status": "confirmed" if payload.get("periodicity_status") == "confirmed" else str(payload.get("periodicity_status") or "unconfirmed"), "notes": payload.get("notes"), } def preview_manual_recurring_payment(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: norm = _validate_payload(payload) if not conn.execute("SELECT 1 FROM budget_categories WHERE category_id=? AND category_type='expense'", (norm["category_id"],)).fetchone(): raise HTTPException(status_code=422, detail="expense category required") month, year = _expected_budget(_dec(norm["expected_amount_text"]), norm["planning_cadence"]) return {"preview_id": new_id("preview"), "summary": f"Fixkosten/Abo anlegen: {norm['name']}", "warnings": [], "payload": norm, "review": {"monthly_reserve_chf": _fmt(month), "annual_effect_chf": _fmt(year), "due_months": norm["due_months"]}, "requires_explicit_confirm": True} def confirm_manual_recurring_payment(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: norm = preview_manual_recurring_payment(conn, payload)["payload"] rid = new_id("rec") created = now() conn.execute( """INSERT INTO budget_recurring_payments( recurring_id,name,merchant_name,merchant_id,category_id,account_id,expected_amount_text, currency,frequency,expected_day_of_month,expected_month,tolerance_amount_text, tolerance_percent,amount_tolerance_pct,date_tolerance_days,recurring_type,status,source, confidence,last_seen_date,next_expected_date,notes,candidate_evidence_json,created_at,updated_at, planning_cadence,planning_type,amount_min_text,amount_max_text,due_months_json, periodicity_status,data_version,user_override) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', 'manual', '1.00', NULL, NULL, ?, '{}', ?, ?, ?, ?, ?, ?, ?, 'confirmed', 1, 1)""", (rid, norm["name"], norm["merchant_name"], norm["merchant_id"], norm["category_id"], norm["account_id"], norm["expected_amount_text"], norm["currency"], norm["frequency"], norm["expected_day_of_month"], norm["expected_month"], norm["tolerance_amount_text"], norm["tolerance_percent"], norm["amount_tolerance_pct"], norm["date_tolerance_days"], norm["recurring_type"], norm["notes"], created, created, norm["planning_cadence"], norm["planning_type"], norm["amount_min_text"], norm["amount_max_text"], norm["due_months_json"]), ) audit_id = record_audit_event(conn, source="vue_dashboard", action="recurring_manual_confirmed", entity_type="budget_recurring_payment", entity_id=rid, new_values=norm, created_by="user") conn.commit() return {"status": "confirmed", "entity_id": rid, "audit_id": audit_id, "message": "Fixkosten/Abo gespeichert"} def _recurring_row(conn: Connection, recurring_id: str): row = conn.execute("SELECT * FROM budget_recurring_payments WHERE recurring_id=?", (recurring_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="recurring payment not found") return row def preview_recurring_candidate_action(conn: Connection, candidate_id: str, payload: dict[str, Any]) -> dict[str, Any]: row = dict(_recurring_row(conn, candidate_id)) action = str(payload.get("action") or "") if action not in {"activate", "ignore", "edit", "pause", "archive"}: raise HTTPException(status_code=422, detail="invalid action") try: stored_months = json.loads(row.get("due_months_json") or "[]") except json.JSONDecodeError: stored_months = [] merged = { "name": row["name"], "merchant_name": row.get("merchant_name"), "category_id": row.get("category_id"), "account_id": row.get("account_id"), "expected_amount_text": row["expected_amount_text"], "amount_min_text": row.get("amount_min_text") or row["expected_amount_text"], "amount_max_text": row.get("amount_max_text") or row["expected_amount_text"], "currency": row["currency"], "planning_cadence": row.get("planning_cadence") or row["frequency"], "planning_type": row.get("planning_type") or row["recurring_type"], "periodicity_status": row.get("periodicity_status") or "unconfirmed", "expected_day_of_month": row.get("expected_day_of_month"), "expected_month": row.get("expected_month"), "due_months": stored_months, "tolerance_amount_text": row.get("tolerance_amount_text"), "tolerance_percent": row.get("tolerance_percent"), "amount_tolerance_pct": row.get("amount_tolerance_pct"), "date_tolerance_days": row.get("date_tolerance_days"), "notes": row.get("notes"), } editable = {"name", "merchant_name", "category_id", "account_id", "expected_amount_text", "payment_amount_text", "amount_min_text", "amount_max_text", "currency", "planning_cadence", "frequency", "planning_type", "recurring_type", "expected_day_of_month", "expected_month", "due_months", "tolerance_amount_text", "tolerance_percent", "amount_tolerance_pct", "date_tolerance_days", "notes"} merged.update({k: v for k, v in payload.items() if k in editable}) norm = _validate_payload(merged) if not conn.execute("SELECT 1 FROM budget_categories WHERE category_id=? AND category_type='expense' AND is_active=1", (norm["category_id"],)).fetchone(): raise HTTPException(status_code=422, detail="active expense category required") status_by_action = {"activate": "active", "ignore": "ignored", "edit": str(row["status"]), "pause": "paused", "archive": "archived"} status = status_by_action[action] if action == "activate" and norm["planning_cadence"] not in {"planned", "undetermined"}: norm["periodicity_status"] = "confirmed" month, year = _expected_budget(_dec(norm["expected_amount_text"]), norm["planning_cadence"]) binding = {"candidate_id": candidate_id, "action": action, "data_version": int(row.get("data_version") or 1), "status": status, "values": norm} fingerprint = _fingerprint(binding) return { "preview_id": new_id("preview"), "preview_fingerprint": fingerprint, "data_version": int(row.get("data_version") or 1), "summary": f"Laufende Zahlung {action}: {row['name']}", "payload": binding | {"preview_fingerprint": fingerprint}, "warnings": ["Periodizität bleibt unbestätigt; es wird keine sichere nächste Fälligkeit erzeugt."] if norm["planning_cadence"] in {"planned", "undetermined"} else [], "review": {"monthly_reserve_chf": _fmt(month), "annual_effect_chf": _fmt(year), "due_months": norm["due_months"], "cadence": norm["planning_cadence"], "calculation_basis": "Nutzerwahl im Preview"}, "requires_explicit_confirm": True, } def confirm_recurring_candidate_action(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: rid = str(payload.get("candidate_id") or "") old = dict(_recurring_row(conn, rid)) supplied_version = int(payload.get("data_version") or 0) if supplied_version != int(old.get("data_version") or 1): raise HTTPException(status_code=409, detail="recurring candidate changed; preview again") values = dict(payload.get("values") or {}) check = preview_recurring_candidate_action(conn, rid, {"action": payload.get("action"), **values}) if not payload.get("preview_fingerprint") or payload.get("preview_fingerprint") != check["preview_fingerprint"]: raise HTTPException(status_code=409, detail="recurring preview changed; preview again") norm = check["payload"]["values"] status = str(check["payload"]["status"]) cadence = str(norm["planning_cadence"]) last_seen = _safe_date(old.get("last_seen_date")) next_due = next_expected_date(last_seen, cadence, norm.get("expected_day_of_month"), norm.get("expected_month")) if status == "active" and last_seen and norm["periodicity_status"] == "confirmed" else None cursor = conn.execute( """UPDATE budget_recurring_payments SET name=?, merchant_name=?, category_id=?, account_id=?, expected_amount_text=?, amount_min_text=?, amount_max_text=?, currency=?, frequency=?, planning_cadence=?, expected_day_of_month=?, expected_month=?, due_months_json=?, tolerance_amount_text=?, tolerance_percent=?, amount_tolerance_pct=?, date_tolerance_days=?, recurring_type=?, planning_type=?, periodicity_status=?, status=?, next_expected_date=?, notes=?, user_override=1, data_version=data_version+1, updated_at=? WHERE recurring_id=? AND data_version=?""", (norm["name"], norm["merchant_name"], norm["category_id"], norm["account_id"], norm["expected_amount_text"], norm["amount_min_text"], norm["amount_max_text"], norm["currency"], norm["frequency"], cadence, norm["expected_day_of_month"], norm["expected_month"], norm["due_months_json"], norm["tolerance_amount_text"], norm["tolerance_percent"], norm["amount_tolerance_pct"], norm["date_tolerance_days"], norm["recurring_type"], norm["planning_type"], norm["periodicity_status"], status, next_due, norm["notes"], now(), rid, supplied_version), ) if cursor.rowcount != 1: conn.rollback() raise HTTPException(status_code=409, detail="recurring candidate changed; preview again") audit_values = norm | {"status": status, "preview_fingerprint": payload["preview_fingerprint"], "source_data_version": supplied_version, "monthly_reserve_chf": check["review"]["monthly_reserve_chf"], "annual_effect_chf": check["review"]["annual_effect_chf"], "next_expected_date": next_due} audit_id = record_audit_event(conn, source="vue_dashboard", action=f"recurring_{payload.get('action')}_confirmed", entity_type="budget_recurring_payment", entity_id=rid, old_values=old, new_values=audit_values, created_by="user") conn.commit() return {"status": "confirmed" if status == "active" or payload.get("action") == "edit" else status, "entity_id": rid, "audit_id": audit_id, "data_version": supplied_version + 1, "message": "Nutzerwahl für laufende Zahlung gespeichert"} def archive_recurring_payment(conn: Connection, recurring_id: str, payload: dict[str, Any] | None = None) -> dict[str, Any]: old = dict(_recurring_row(conn, recurring_id)) conn.execute("UPDATE budget_recurring_payments SET status='archived', notes=COALESCE(?, notes), updated_at=? WHERE recurring_id=?", ((payload or {}).get("notes"), now(), recurring_id)) audit_id = record_audit_event(conn, source="vue_dashboard", action="recurring_archived", entity_type="budget_recurring_payment", entity_id=recurring_id, old_values=old, new_values={"status": "archived"}, created_by="user") conn.commit() return {"status": "archived", "entity_id": recurring_id, "audit_id": audit_id} def _view(row: Any) -> dict[str, Any]: item = dict(row) cadence = str(item.get("planning_cadence") or item.get("frequency") or "undetermined") planning_type = str(item.get("planning_type") or item.get("recurring_type") or "fixed_cost") amount = _dec(item["expected_amount_text"]) month, year = _expected_budget(amount, cadence) try: due_months = json.loads(item.get("due_months_json") or "[]") except json.JSONDecodeError: due_months = [] item.update({ "frequency": cadence, "planning_cadence": cadence, "recurring_type": planning_type, "planning_type": planning_type, "due_months": due_months, "budget_month_chf": _fmt(month), "budget_year_chf": _fmt(year), "monthly_reserve_chf": _fmt(month), "annual_effect_chf": _fmt(year), "expected_amount_chf": _fmt(amount), "is_periodicity_confirmed": item.get("periodicity_status") == "confirmed", }) return item def list_recurring_warnings(conn: Connection, *, today: str | None = None) -> list[dict[str, Any]]: today_d = _date(today or date.today().isoformat()) data_row = conn.execute("SELECT MAX(transaction_date) AS data_through FROM budget_transactions WHERE status='confirmed'").fetchone() data_through = str(data_row["data_through"] or today_d.isoformat()) if data_row else today_d.isoformat() cards: list[dict[str, Any]] = [] for raw in conn.execute("SELECT * FROM budget_recurring_payments WHERE status IN ('active','candidate') ORDER BY name").fetchall(): r = _view(raw) expected = _dec(r["expected_amount_text"]) cadence = str(r["planning_cadence"]) reasons: list[dict[str, Any]] = [] expected_period = "Noch festzulegen" if cadence in {"planned", "undetermined"} else cadence recent = None if r["status"] == "active": recent = conn.execute( """SELECT * FROM budget_transactions WHERE status='confirmed' AND transaction_type IN ('expense','fee') AND category_id=? AND lower(COALESCE(payee, description,'')) LIKE lower(?) ORDER BY transaction_date DESC LIMIT 1""", (r["category_id"], f"%{str(r['name']).split()[0]}%"), ).fetchone() if recent and r["periodicity_status"] == "confirmed": seen = _date(recent["transaction_date"]) nex = next_expected_date(seen, cadence, r["expected_day_of_month"], r["expected_month"]) if nex and today_d > _date(nex) + timedelta(days=int(r["date_tolerance_days"] or 5)): reasons.append({"code": "missing", "observed": f"Seit {seen.isoformat()} keine passende Zahlung gefunden.", "expected": f"Erwartet um {nex}."}) actual = _row_amount(recent) tolerance = expected * _dec(r["amount_tolerance_pct"]) / Decimal("100") if expected and abs(actual - expected) > tolerance: _, annual_delta = _expected_budget(actual - expected, cadence) reasons.append({"code": "amount_changed", "observed": f"Zuletzt CHF {_fmt(actual)} statt CHF {_fmt(expected)}.", "expected": f"Bisheriger Vertragsbetrag CHF {_fmt(expected)}.", "annual_forecast_effect_chf": _fmt(annual_delta)}) elif not recent and r["periodicity_status"] == "confirmed": reasons.append({"code": "missing", "observed": "Keine passende bestätigte Zahlung gefunden.", "expected": f"Erwarteter Zeitraum: {expected_period}."}) else: reasons.append({"code": "new_candidate", "observed": "Eine mögliche laufende Zahlung wurde erkannt.", "expected": f"Vorgeschlagener Rhythmus: {expected_period}; noch nicht bestätigt."}) if not reasons: continue _, annual = _expected_budget(expected, cadence) cards.append({ "status": str(reasons[0]["code"]), "recurring_id": r["recurring_id"], "name": r["name"], "observed": " ".join(str(reason["observed"]) for reason in reasons), "expected_period": expected_period, "data_through": data_through, "relevance": "Kann Jahresplan, Reserven oder Forecast verändern.", "forecast_effect": "Noch nicht präzise eingerechnet." if r["periodicity_status"] != "confirmed" else f"Jahresplanung CHF {_fmt(annual)}; Abweichungen sind separat ausgewiesen.", "reasons": reasons, "actions": ["contract_review", "adjust_period", "mark_paused", "accept_new_amount", "ignore_exception", "open_transactions"], }) return cards def get_recurring_dashboard(conn: Connection, *, today: str | None = None) -> dict[str, Any]: rows = conn.execute("SELECT r.*, c.name AS category_name FROM budget_recurring_payments r LEFT JOIN budget_categories c ON c.category_id=r.category_id ORDER BY r.status, r.name").fetchall() candidates = [_view(r) for r in rows if r["status"] == "candidate"] active = [_view(r) for r in rows if r["status"] == "active"] warnings = list_recurring_warnings(conn, today=today) by_cat: dict[str, dict[str, Any]] = {} subscriptions_year = Decimal("0") fixed_year = Decimal("0") for item in active: month = _dec(item["budget_month_chf"]) year = _dec(item["budget_year_chf"]) if item["recurring_type"] == "subscription": subscriptions_year += year if item["recurring_type"] == "fixed_cost": fixed_year += year cid = item["category_id"] b = by_cat.setdefault(cid, {"category_id": cid, "category": item.get("category_name"), "known_recurring_month_chf": Decimal("0"), "known_recurring_year_chf": Decimal("0"), "subscription_count": 0, "fixed_cost_count": 0}) b["known_recurring_month_chf"] += month b["known_recurring_year_chf"] += year if item["recurring_type"] == "subscription": b["subscription_count"] += 1 if item["recurring_type"] == "fixed_cost": b["fixed_cost_count"] += 1 category_breakdown = [{**v, "known_recurring_month_chf": _fmt(v["known_recurring_month_chf"]), "known_recurring_year_chf": _fmt(v["known_recurring_year_chf"])} for v in by_cat.values()] total_expense = sum((_dec(r["amount_chf"] or r["amount_original"]) for r in conn.execute("SELECT amount_chf, amount_original FROM budget_transactions WHERE status='confirmed' AND transaction_type IN ('expense','fee')").fetchall()), Decimal("0")) total_month = sum((_dec(a["budget_month_chf"]) for a in active), Decimal("0")) total_year = sum((_dec(a["budget_year_chf"]) for a in active), Decimal("0")) fixed_quote = Decimal("0") if total_expense == 0 else total_month / total_expense * Decimal("100") largest = sorted(active, key=lambda a: _dec(a["budget_year_chf"]), reverse=True)[:5] next_due = sorted((a for a in active if a.get("next_expected_date")), key=lambda a: str(a.get("next_expected_date"))) kpis = { "monthly_fixed_costs_chf": _fmt(total_month), "annual_fixed_costs_chf": _fmt(total_year), "fixed_cost_quote_percent": _fmt(fixed_quote), "active_recurring_count": len(active), "open_candidate_count": len(candidates), "missing_payment_count": sum(1 for w in warnings if w["status"] == "missing"), "next_due": next_due[0] if next_due else None, "largest_fixed_costs": largest, } return {"purpose": "fixed_costs_subscriptions_v2", "candidates": candidates, "active": active, "warnings": warnings, "category_breakdown": category_breakdown, "kpis": kpis, "analytics": {"fixed_cost_quote_percent": _fmt(fixed_quote), "subscriptions_year_chf": _fmt(subscriptions_year), "fixed_costs_year_chf": _fmt(fixed_year), "top_subscriptions": [a for a in active if a["recurring_type"] == "subscription"][:10]}} def classify_transaction_recurring_type(conn: Connection, row: Any) -> str: if row["transaction_type"] not in {"expense", "fee"}: return "unknown" merchant = str(row["payee"] or row["description"] or "") for rec in conn.execute("SELECT * FROM budget_recurring_payments WHERE status='active'").fetchall(): if rec["category_id"] == row["category_id"] and _similar_key(rec["name"]) and _similar_key(rec["name"]) in _similar_key(merchant): return rec["recurring_type"] return "one_off"