from __future__ import annotations from datetime import date from decimal import Decimal from sqlite3 import Connection from typing import Any from fastapi import HTTPException FINANCIAL_SEMANTICS_VERSION = "household_financial_semantics_v1" def validate_budget_account_ids( conn: Connection, account_ids: list[str] | None ) -> list[str]: """Validate canonical, opaque account identifiers without fail-open cleanup.""" if not account_ids: return [] clean = [str(value).strip() for value in account_ids] if any(not value for value in clean): raise HTTPException(status_code=422, detail="account_id values must not be empty") if len(clean) != len(set(clean)): raise HTTPException(status_code=422, detail="account_id values must be unique") if len(clean) > 20: raise HTTPException(status_code=422, detail="at most 20 accounts may be selected") placeholders = ",".join("?" for _ in clean) active = { str(row[0]) for row in conn.execute( f"SELECT budget_account_id FROM budget_accounts " f"WHERE is_active=1 AND budget_account_id IN ({placeholders})", clean, ).fetchall() } if active != set(clean): raise HTTPException(status_code=422, detail="unknown or inactive account_id") return sorted(clean) def _fmt(value: Any) -> str: return format(Decimal(str(value or "0")).quantize(Decimal("0.01")), ".2f") def financial_quality_metadata( *, unavailable_chf_count: int = 0, unlinked_refund_count: int = 0, transfer_membership_conflict_count: int = 0, ) -> dict[str, Any]: warnings: list[dict[str, Any]] = [] if unavailable_chf_count: warnings.append({ "code": "missing_chf_conversion", "count": unavailable_chf_count, "message": "CHF-Summen sind wegen fehlender belastbarer Umrechnung teilweise verfügbar.", }) if unlinked_refund_count: warnings.append({ "code": "unlinked_refund", "count": unlinked_refund_count, "message": "Ausgabensummen sind wegen nicht belastbar verknüpfter Rückerstattungen teilweise verfügbar.", }) if transfer_membership_conflict_count: warnings.append({ "code": "transfer_membership_conflict", "count": transfer_membership_conflict_count, "message": "Transfer-Volumen ist wegen widersprüchlicher Transferverknüpfungen nicht verfügbar.", }) return { "data_status": "partial" if warnings else "current", "unavailable_chf_count": unavailable_chf_count, "unlinked_refund_count": unlinked_refund_count, "transfer_membership_conflict_count": transfer_membership_conflict_count, "warnings": warnings, } def _filters( *, period: str | None = None, date_from: str | None = None, date_to: str | None = None, account: str | None = None, accounts: list[str] | None = None, ) -> tuple[list[str], list[Any]]: where = ["t.status='confirmed'"] params: list[Any] = [] if period: try: year, month = period.split("-", 1) if len(year) != 4 or len(month) != 2 or not 1 <= int(month) <= 12: raise ValueError except ValueError: raise HTTPException(status_code=422, detail="period must use YYYY-MM") from None where.append("substr(t.transaction_date,1,7)=?") params.append(period) for label, value, operator in (("date_from", date_from, ">="), ("date_to", date_to, "<=")): if value: try: date.fromisoformat(value) except ValueError: raise HTTPException(status_code=422, detail=f"{label} must use YYYY-MM-DD") from None where.append(f"t.transaction_date {operator} ?") params.append(value) if date_from and date_to and date_from > date_to: raise HTTPException(status_code=422, detail="date_from must not be after date_to") if account: where.append("lower(coalesce(a.name,'')) LIKE ?") params.append("%" + account.casefold() + "%") if accounts: where.append("t.account_id IN (" + ",".join("?" for _ in accounts) + ")") params.extend(accounts) return where, params def _effects_cte(where: list[str]) -> str: chf_amount = ( "CASE WHEN t.amount_chf IS NOT NULL THEN CAST(t.amount_chf AS REAL) " "WHEN upper(t.currency_original)='CHF' THEN CAST(t.amount_original AS REAL) " "ELSE NULL END" ) refund_amount = ( "CASE WHEN r.amount_chf IS NOT NULL THEN CAST(r.amount_chf AS REAL) " "WHEN upper(r.currency_original)='CHF' THEN CAST(r.amount_original AS REAL) " "ELSE NULL END" ) origin_amount = ( "CASE WHEN origin.amount_chf IS NOT NULL THEN CAST(origin.amount_chf AS REAL) " "WHEN upper(origin.currency_original)='CHF' THEN CAST(origin.amount_original AS REAL) " "ELSE NULL END" ) refund_target = ( "COALESCE(r.reversal_of_transaction_id,CASE WHEN json_valid(r.notes) " "THEN json_extract(r.notes,'$.reversal_of_transaction_id') END)" ) valid_refund = ( "t.transaction_type='refund' AND origin.budget_transaction_id IS NOT NULL " "AND origin.transaction_type IN ('expense','fee') " f"AND {origin_amount} IS NOT NULL AND rr.refund_amount_chf IS NOT NULL " f"AND rr.running_refund_chf<=abs({origin_amount})" ) return f""" WITH transfer_membership_rows AS ( SELECT transfer_id,from_transaction_id AS budget_transaction_id FROM budget_transfers UNION ALL SELECT transfer_id,to_transaction_id AS budget_transaction_id FROM budget_transfers ), transfer_memberships AS ( SELECT budget_transaction_id,MIN(transfer_id) AS transfer_id, COUNT(DISTINCT transfer_id) AS membership_count FROM transfer_membership_rows GROUP BY budget_transaction_id ), refund_base AS ( SELECT r.budget_transaction_id,r.transaction_date,r.created_at, {refund_target} AS origin_id,{refund_amount} AS refund_amount_chf FROM budget_transactions r WHERE r.status='confirmed' AND r.transaction_type='refund' ), refund_ranked AS ( SELECT refund_base.*, SUM(CASE WHEN refund_amount_chf IS NOT NULL THEN abs(refund_amount_chf) ELSE 0 END) OVER (PARTITION BY origin_id ORDER BY transaction_date,created_at,budget_transaction_id ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running_refund_chf FROM refund_base ), base AS ( SELECT t.budget_transaction_id,t.transaction_date,t.transaction_type, t.account_id,t.amount_chf,t.amount_original,t.currency_original,t.description,t.payee, t.source_type,t.source_candidate_id,t.created_at,t.category_id, CASE WHEN COALESCE(tm.membership_count,0)>1 THEN 'conflict:' || t.budget_transaction_id ELSE COALESCE(tm.transfer_id,'transaction:' || t.budget_transaction_id) END AS transfer_unit_id, CASE WHEN COALESCE(tm.membership_count,0)>1 THEN 1 ELSE 0 END AS transfer_membership_conflict, {chf_amount} AS effective_amount_chf, origin.budget_transaction_id AS origin_id,origin.category_id AS origin_category_id, rr.refund_amount_chf,rr.running_refund_chf,{origin_amount} AS origin_amount_chf, CASE WHEN {valid_refund} THEN 1 ELSE 0 END AS valid_refund FROM budget_transactions t LEFT JOIN refund_ranked rr ON rr.budget_transaction_id=t.budget_transaction_id LEFT JOIN budget_transactions origin ON origin.budget_transaction_id=rr.origin_id AND origin.status='confirmed' AND origin.budget_transaction_id<>t.budget_transaction_id LEFT JOIN transfer_memberships tm ON tm.budget_transaction_id=t.budget_transaction_id LEFT JOIN budget_accounts a ON a.budget_account_id=t.account_id WHERE {' AND '.join(where)} ), effects AS ( SELECT base.*, CASE WHEN valid_refund=1 THEN origin_category_id ELSE category_id END AS effective_category_id, CASE WHEN transaction_type='income' THEN COALESCE(abs(effective_amount_chf),0) ELSE 0 END AS income_effect, CASE WHEN transaction_type IN ('expense','fee') THEN COALESCE(abs(effective_amount_chf),0) WHEN valid_refund=1 THEN -abs(effective_amount_chf) ELSE 0 END AS expense_effect, CASE WHEN transaction_type IN ('income','expense','fee','refund','transfer') AND effective_amount_chf IS NULL THEN 1 ELSE 0 END AS unavailable_chf, CASE WHEN transaction_type='refund' AND valid_refund=0 THEN 1 ELSE 0 END AS unlinked_refund, CASE WHEN transaction_type='transfer' THEN 1 ELSE 0 END AS neutral_transfer, CASE WHEN transaction_type='transfer' AND (effective_amount_chf IS NULL OR transfer_membership_conflict=1) THEN 1 ELSE 0 END AS unavailable_transfer FROM base ) """ def get_household_financial_summary( conn: Connection, *, period: str | None = None, date_from: str | None = None, date_to: str | None = None, account: str | None = None, accounts: list[str] | None = None, ) -> dict[str, Any]: where, params = _filters( period=period, date_from=date_from, date_to=date_to, account=account, accounts=accounts, ) cte = _effects_cte(where) totals = conn.execute( cte + """ , transfer_units AS ( SELECT transfer_unit_id, MAX(unavailable_transfer) AS unavailable_transfer, MAX(CASE WHEN effective_amount_chf IS NOT NULL THEN abs(effective_amount_chf) ELSE 0 END) AS volume_chf FROM effects WHERE neutral_transfer=1 GROUP BY transfer_unit_id ) SELECT COALESCE(SUM(income_effect),0) AS income, COALESCE(SUM(expense_effect),0) AS expense, COALESCE(SUM(income_effect-expense_effect),0) AS net, SUM(unlinked_refund) AS unlinked_refund_count, SUM(unavailable_chf) AS unavailable_chf_count, SUM(transfer_membership_conflict) AS transfer_membership_conflict_count, COALESCE((SELECT SUM(unavailable_transfer) FROM transfer_units),0) AS unavailable_transfer_count, SUM(neutral_transfer) AS neutral_transfer_count, COALESCE((SELECT SUM(volume_chf) FROM transfer_units),0) AS neutral_transfer_volume, COUNT(*) AS confirmed_transaction_count FROM effects """, params, ).fetchone() categories = conn.execute( cte + """ SELECT COALESCE(c.category_id,'uncategorized') AS category_id, COALESCE(c.name,'Unkategorisiert') AS category_name, COALESCE(c.category_type, CASE WHEN SUM(income_effect)>0 AND SUM(expense_effect)=0 THEN 'income' ELSE 'expense' END ) AS category_type, COALESCE(SUM(income_effect),0) AS income, COALESCE(SUM(expense_effect),0) AS expense, SUM(CASE WHEN income_effect<>0 OR expense_effect<>0 THEN 1 ELSE 0 END) AS transaction_count FROM effects e LEFT JOIN budget_categories c ON c.category_id=e.effective_category_id WHERE income_effect<>0 OR expense_effect<>0 GROUP BY COALESCE(c.category_id,'uncategorized'),COALESCE(c.name,'Unkategorisiert'),c.category_type ORDER BY expense DESC,income DESC,category_name """, params, ).fetchall() quality = financial_quality_metadata( unavailable_chf_count=int(totals["unavailable_chf_count"] or 0), unlinked_refund_count=int(totals["unlinked_refund_count"] or 0), transfer_membership_conflict_count=int( totals["transfer_membership_conflict_count"] or 0 ), ) transfer_conflict_count = int(totals["transfer_membership_conflict_count"] or 0) unavailable_transfer_count = max( int(totals["unavailable_transfer_count"] or 0), transfer_conflict_count, ) return { "semantics_version": FINANCIAL_SEMANTICS_VERSION, "income_chf": _fmt(totals["income"]), "expense_chf": _fmt(totals["expense"]), "net_chf": _fmt(totals["net"]), **quality, "neutral_transfer_count": int(totals["neutral_transfer_count"] or 0), "unavailable_transfer_count": unavailable_transfer_count, "neutral_transfer_volume_chf": ( None if unavailable_transfer_count else _fmt(totals["neutral_transfer_volume"]) ), "confirmed_transaction_count": int(totals["confirmed_transaction_count"] or 0), "categories": [ { "category_id": str(row["category_id"]), "category_name": str(row["category_name"]), "category_type": str(row["category_type"]), "income_chf": _fmt(row["income"]), "expense_chf": _fmt(row["expense"]), "transaction_count": int(row["transaction_count"] or 0), } for row in categories ], } def list_household_financial_effects( conn: Connection, *, period: str | None = None, date_from: str | None = None, date_to: str | None = None, account: str | None = None, accounts: list[str] | None = None, ) -> list[dict[str, Any]]: """Return transaction-level effects under the canonical monetary semantics.""" where, params = _filters( period=period, date_from=date_from, date_to=date_to, account=account, accounts=accounts, ) rows = conn.execute( _effects_cte(where) + """ SELECT e.*, c.name AS effective_category_name, c.category_type AS effective_category_type, c.color AS effective_category_color, c.icon AS effective_category_icon, COALESCE(e.payee,e.description,e.source_type,'Quelle') AS merchant_display_name FROM effects e LEFT JOIN budget_categories c ON c.category_id=e.effective_category_id ORDER BY e.transaction_date DESC, e.budget_transaction_id DESC """, params, ).fetchall() return [dict(row) for row in rows]