from __future__ import annotations from collections import defaultdict from datetime import UTC, date, datetime, timedelta from decimal import Decimal from sqlite3 import Connection from typing import Any, cast from fastapi import HTTPException from jarvis_finance.ledger.performance import effective_activities, external_cashflow from jarvis_finance.quality.freshness import ( FreshnessStatus, assess_freshness, combined_freshness, ) from jarvis_finance.services.budget_planning import get_annual_budget_assistant from jarvis_finance.services.cash_service import get_cash_summary from jarvis_finance.services.crypto_service import list_crypto_positions from jarvis_finance.services.equity_service import get_equity_summary from jarvis_finance.services.household_import import source_reference_hash from jarvis_finance.services.modelled_wealth import build_modelled_wealth_development from jarvis_finance.services.portfolio_performance import ( build_performance_coverage, build_portfolio_performance, load_scope_activities, ) from jarvis_finance.services.portfolio_policy import active_policy from jarvis_finance.services.reconciliation_snapshot import ( build_reconciliation_snapshot, safe_account_label, ) MONEY = Decimal("0.01") PERIODS = { "since_anchor", "1m", "3m", "1y", "ytd", "previous_year", "12m", "all", } KNOWN_PERFORMANCE_ROLES = { "postfinance_etrading_depot", "postfinance_etrading_cash", "canonical_truewealth_total_value", "crypto_portfolio", } def _money(value: Decimal | None) -> str | None: return None if value is None else format(value.quantize(MONEY), "f") def _decimal(value: object) -> Decimal | None: if value in (None, ""): return None return Decimal(str(value)) def _latest_data_cutoff(conn: Connection) -> str: candidates: list[str] = [] for table, column in ( ("transactions", "created_at"), ("portfolio_valuation_snapshots", "captured_at"), ("account_value_snapshots", "created_at"), ("cash_account_snapshots", "created_at"), ("crypto_prices", "fetched_at"), ("portfolio_analysis_snapshots", "created_at"), ): row = conn.execute(f"SELECT MAX({column}) FROM {table}").fetchone() if row and row[0]: candidates.append(str(row[0])) return max(candidates, default="1970-01-01T00:00:00Z") def _latest_valuation_date(conn: Connection, fallback: date) -> date: row = conn.execute( """SELECT MAX(day) FROM ( SELECT substr(valuation_at,1,10) day FROM portfolio_valuation_snapshots UNION ALL SELECT valuation_date FROM account_value_snapshots UNION ALL SELECT balance_date FROM cash_account_snapshots )""" ).fetchone() if not row or not row[0]: return fallback return min(date.fromisoformat(str(row[0])[:10]), fallback) def _earliest_evidence_date(conn: Connection, fallback: date) -> date: row = conn.execute( """SELECT MIN(day) FROM ( SELECT substr(valuation_at,1,10) day FROM portfolio_valuation_snapshots UNION ALL SELECT valuation_date FROM account_value_snapshots UNION ALL SELECT balance_date FROM cash_account_snapshots UNION ALL SELECT trade_date FROM transactions )""" ).fetchone() return date.fromisoformat(str(row[0])[:10]) if row and row[0] else fallback def period_bounds(conn: Connection, *, period: str, as_of: date) -> tuple[date, date]: if period not in PERIODS: raise ValueError( "Zeitraum muss since_anchor, 1m, 3m, 1y, ytd, previous_year, 12m oder all sein" ) if period == "since_anchor": row = conn.execute( """SELECT MAX(day) FROM ( SELECT valuation_date day FROM account_value_snapshots WHERE COALESCE(is_active,1)=1 AND updated_at IS NULL UNION ALL SELECT balance_date FROM cash_account_snapshots ) WHERE day<=?""", (as_of.isoformat(),), ).fetchone() return ( date.fromisoformat(str(row[0])) if row and row[0] else as_of, as_of, ) if period == "1m": return as_of - timedelta(days=31), as_of if period == "3m": return as_of - timedelta(days=93), as_of if period in {"1y", "12m"}: try: start = as_of.replace(year=as_of.year - 1) except ValueError: start = as_of.replace(year=as_of.year - 1, day=28) return start, as_of if period == "ytd": return date(as_of.year, 1, 1), as_of if period == "previous_year": return date(as_of.year - 1, 1, 1), date(as_of.year - 1, 12, 31) return _earliest_evidence_date(conn, as_of - timedelta(days=1)), as_of def _account_ids(conn: Connection, *, investment_only: bool) -> list[str]: if investment_only: rows = conn.execute( """SELECT a.account_id FROM accounts a JOIN performance_scope_classifications psc ON psc.account_id=a.account_id WHERE a.is_active=1 AND psc.included=1 AND psc.decision_version='investment_performance_scope_v1' ORDER BY a.account_id""" ).fetchall() else: rows = conn.execute( """SELECT account_id FROM accounts WHERE is_active=1 AND account_type<>'credit_card_liability' ORDER BY account_id""" ).fetchall() return [str(row[0]) for row in rows] def scope_cashflows( conn: Connection, *, account_ids: list[str], from_date: str, to_date: str, data_cutoff: str, ) -> list[dict[str, str]]: """Reuse canonical activity and transfer-boundary semantics for one ownership scope.""" activities = load_scope_activities( conn, account_ids=account_ids, to_date=to_date, data_cutoff=data_cutoff, base_currency="CHF", ) effective, _ = effective_activities(activities) events: list[dict[str, str]] = [] for activity in effective: if not (from_date <= activity.occurred_at[:10] <= to_date): continue amount = external_cashflow(activity, "CHF") if amount is None or activity.kind not in {"external_deposit", "external_withdrawal"}: continue events.append( { "at": activity.occurred_at, "kind": activity.kind, "amount_chf": _money(amount) or "0.00", } ) return sorted(events, key=lambda item: (item["at"], item["kind"], item["amount_chf"])) def _truewealth(conn: Connection) -> tuple[Decimal | None, str | None]: row = conn.execute( """SELECT avs.total_value_chf,avs.valuation_date FROM account_value_snapshots avs JOIN performance_scope_classifications psc ON psc.account_id=avs.account_id WHERE psc.included=1 AND psc.classification_role='canonical_truewealth_total_value' AND COALESCE(avs.is_active,1)=1 AND avs.updated_at IS NULL AND avs.source_type<>'truewealth_manual_provisional' ORDER BY avs.valuation_date DESC, CASE avs.source_type WHEN 'truewealth_official_import' THEN 3 ELSE 1 END DESC, COALESCE(avs.valuation_at,avs.created_at) DESC,avs.snapshot_id DESC LIMIT 1""" ).fetchone() return (Decimal(str(row[0])), str(row[1])) if row else (None, None) def _unassigned_values(conn: Connection) -> tuple[Decimal, list[dict[str, str]]]: rows = conn.execute( """WITH ranked AS ( SELECT avs.account_id,avs.total_value_chf,avs.valuation_date,a.account_name, ROW_NUMBER() OVER (PARTITION BY avs.account_id ORDER BY avs.valuation_date DESC,avs.created_at DESC,avs.snapshot_id DESC) rn FROM account_value_snapshots avs JOIN accounts a ON a.account_id=avs.account_id AND a.is_active=1 LEFT JOIN performance_scope_classifications psc ON psc.account_id=a.account_id WHERE COALESCE(avs.is_active,1)=1 AND avs.updated_at IS NULL AND a.account_type NOT IN ('cash','credit_card_liability') AND COALESCE(psc.classification_role,'') NOT IN ( 'postfinance_etrading_depot','postfinance_etrading_cash', 'canonical_truewealth_total_value','crypto_portfolio') ) SELECT account_name,total_value_chf,valuation_date FROM ranked WHERE rn=1 ORDER BY account_name""" ).fetchall() items = [ {"label": safe_account_label(str(row[0])), "value_chf": _money(Decimal(str(row[1]))) or "0.00", "as_of": str(row[2])} for row in rows ] return sum((Decimal(item["value_chf"]) for item in items), Decimal("0")), items def _household_import_meta( conn: Connection, profile: str, *, canonical_account_id: str | None = None, ) -> dict[str, Any]: source_type = { "akb": "akb_bank", "raiffeisen": "raiffeisen_bank", "viseca_one": "visa_credit_card", "migros_receipts": "migros_receipts", }.get(profile, profile) unavailable = { "imported_at": None, "coverage_from": None, "coverage_to": None, "coverage_status": "unavailable", "new_rows": 0, "duplicate_rows": 0, "review_rows": 0, } if canonical_account_id: budget_ids = { str(row[0]) for row in conn.execute( """SELECT budget_account_id FROM budget_accounts WHERE linked_account_id=? AND is_active=1""", (canonical_account_id,), ).fetchall() } mapping_hashes = { str(row[0]) for row in conn.execute( """SELECT source_reference_hash FROM household_account_source_mappings WHERE canonical_account_id=? AND source_type=? AND is_active=1""", (canonical_account_id, source_type), ).fetchall() } candidates = conn.execute( """SELECT c.transaction_candidate_id,c.transaction_date,c.status, c.household_batch_id,c.account_source,c.confirmed_transaction_id, bt.account_id confirmed_budget_account_id,b.confirmed_at FROM budget_transaction_candidates c LEFT JOIN budget_transactions bt ON bt.budget_transaction_id=c.confirmed_transaction_id LEFT JOIN household_import_batches b ON b.batch_id=c.household_batch_id WHERE c.source_type=? AND c.household_batch_id IS NOT NULL ORDER BY COALESCE(b.confirmed_at,c.created_at),c.transaction_date, c.transaction_candidate_id""", (source_type,), ).fetchall() attributable = [] for candidate in candidates: bound_by_transaction = str( candidate["confirmed_budget_account_id"] or "" ) in budget_ids bound_by_mapping = False source_reference = str(candidate["account_source"] or "").strip() if source_reference and mapping_hashes: try: bound_by_mapping = ( source_reference_hash(source_reference) in mapping_hashes ) except HTTPException: # API tests and offline fixtures may intentionally omit the # private fingerprint key. Confirmed canonical lineage still # remains usable; weak source text never becomes identity. bound_by_mapping = False if bound_by_transaction or bound_by_mapping: attributable.append(candidate) if not attributable: return unavailable latest_batch = max( attributable, key=lambda row: ( str(row["confirmed_at"] or ""), str(row["household_batch_id"]) ), )["household_batch_id"] selected = [ row for row in attributable if row["household_batch_id"] == latest_batch ] batch = conn.execute( """SELECT confirmed_at FROM household_import_batches WHERE batch_id=?""", (latest_batch,), ).fetchone() days = [str(row["transaction_date"]) for row in selected] review_rows = sum(str(row["status"]) == "needs_review" for row in selected) return { "imported_at": batch["confirmed_at"] if batch else None, "coverage_from": min(days, default=None), "coverage_to": max(days, default=None), "coverage_status": "partial", "new_rows": len(selected), # Duplicates cannot be assigned to an account without durable # source-row lineage. Report zero rather than inheriting a provider # count from a sibling account. "duplicate_rows": 0, "review_rows": review_rows, } row = conn.execute( """SELECT f.batch_id,f.row_count,f.period_start,f.period_end,f.created_at,b.confirmed_at FROM household_import_files f JOIN household_import_batches b ON b.batch_id=f.batch_id WHERE f.source_type=? ORDER BY b.confirmed_at DESC,f.created_at DESC LIMIT 1""", (source_type,), ).fetchone() if not row: return unavailable imported_rows = int( conn.execute( "SELECT COUNT(*) FROM household_import_items WHERE batch_id=? AND source_type=?", (row["batch_id"], source_type), ).fetchone()[0] ) review_rows = int( conn.execute( "SELECT COUNT(*) FROM budget_transaction_candidates WHERE household_batch_id=? AND source_type=? AND status='needs_review'", (row["batch_id"], source_type), ).fetchone()[0] ) row_count = int(row["row_count"]) return { "imported_at": row["confirmed_at"] or row["created_at"], "coverage_from": row["period_start"], "coverage_to": row["period_end"], "coverage_status": "complete" if row["period_start"] and row["period_end"] else "partial", "new_rows": imported_rows, "duplicate_rows": max(row_count - imported_rows, 0), "review_rows": review_rows, } def _postfinance_import_meta(conn: Connection) -> dict[str, Any]: row = conn.execute( """SELECT b.confirmed_at,b.activity_coverage_from,b.activity_coverage_to, b.performance_coverage_complete,s.snapshot_at FROM postfinance_import_batches b JOIN postfinance_snapshots s ON s.batch_id=b.batch_id ORDER BY b.confirmed_at DESC LIMIT 1""" ).fetchone() if not row: return {"imported_at": None, "coverage_from": None, "coverage_to": None, "coverage_status": "unavailable", "last_snapshot": None} return { "imported_at": row["confirmed_at"], "coverage_from": row["activity_coverage_from"], "coverage_to": row["activity_coverage_to"], "coverage_status": "complete" if int(row["performance_coverage_complete"]) else "partial", "last_snapshot": str(row["snapshot_at"])[:10], } def _truewealth_import_meta(conn: Connection) -> dict[str, Any]: row = conn.execute( """SELECT confirmed_at,period_from,period_to,external_cashflows_complete,snapshot_date FROM truewealth_import_batches ORDER BY confirmed_at DESC LIMIT 1""" ).fetchone() if not row: return {"imported_at": None, "coverage_from": None, "coverage_to": None, "coverage_status": "unavailable", "last_snapshot": None} return { "imported_at": row["confirmed_at"], "coverage_from": row["period_from"], "coverage_to": row["period_to"], "coverage_status": "complete" if int(row["external_cashflows_complete"]) else "partial", "last_snapshot": row["snapshot_date"], } def _household_history( conn: Connection, *, from_date: date, to_date: date, current: dict[str, Any], as_of: date, ) -> tuple[list[dict[str, str]], str]: """Return only dates with a complete exact stored value for every known account. No carry-forward or interpolation is allowed. Cash snapshots, canonical account valuations and existing performance valuations remain separate source contracts. """ classified = conn.execute( """SELECT a.account_id,a.account_type,psc.classification_role FROM accounts a LEFT JOIN performance_scope_classifications psc ON psc.account_id=a.account_id AND psc.included=1 AND psc.decision_version='investment_performance_scope_v1' WHERE a.is_active=1 AND a.account_type<>'credit_card_liability' ORDER BY a.account_id""" ).fetchall() required: list[tuple[str, str]] = [] roles: set[str] = set() for row in classified: account_id, account_type, role = str(row[0]), str(row[1]), str(row[2] or "") if role: roles.add(role) if role in {"postfinance_etrading_depot", "postfinance_etrading_cash", "crypto_portfolio"}: required.append((account_id, "performance")) elif role == "canonical_truewealth_total_value": required.append((account_id, "account_value")) elif account_type == "cash": required.append((account_id, "cash")) elif not role and conn.execute( "SELECT 1 FROM account_value_snapshots WHERE account_id=? AND COALESCE(is_active,1)=1 AND updated_at IS NULL LIMIT 1", (account_id,), ).fetchone(): required.append((account_id, "account_value")) distribution = {str(row["key"]): row["value_chf"] for row in current["distribution"]} if Decimal(str(distribution.get("equity") or "0")) and "postfinance_etrading_depot" not in roles: return [], "Für Aktien und ETFs fehlt eine kanonische historische Kontobewertung." if Decimal(str(distribution.get("crypto") or "0")) and "crypto_portfolio" not in roles: return [], "Für Kryptowährungen fehlt eine kanonische historische Portfoliobewertung." if distribution.get("truewealth") is not None and "canonical_truewealth_total_value" not in roles: return [], "Für True Wealth fehlt die kanonische historische Gesamtwertreihe." if not required: return [], "Es liegen noch keine gemeinsamen historischen Kontobewertungen vor." series: list[dict[str, Decimal]] = [] for account_id, source_kind in required: if source_kind == "performance": rows = conn.execute( """WITH ranked AS ( SELECT substr(valuation_at,1,10) day, CASE WHEN currency=base_currency THEN value_original WHEN fx_rate_to_base IS NOT NULL THEN CAST(value_original AS NUMERIC)*CAST(fx_rate_to_base AS NUMERIC) END value_base, ROW_NUMBER() OVER (PARTITION BY substr(valuation_at,1,10) ORDER BY captured_at DESC,snapshot_id DESC) rn FROM portfolio_valuation_snapshots WHERE scope_kind='account' AND scope_id=? AND base_currency='CHF' AND quality_status IN ('complete','ok') AND substr(valuation_at,1,10) BETWEEN ? AND ? ) SELECT day,value_base FROM ranked WHERE rn=1 AND value_base IS NOT NULL""", (account_id, from_date.isoformat(), to_date.isoformat()), ).fetchall() elif source_kind == "account_value": rows = conn.execute( """WITH ranked AS ( SELECT valuation_date day,total_value_chf, ROW_NUMBER() OVER (PARTITION BY valuation_date ORDER BY CASE source_type WHEN 'truewealth_official_import' THEN 3 ELSE 1 END DESC, COALESCE(valuation_at,created_at) DESC,snapshot_id DESC) rn FROM account_value_snapshots WHERE account_id=? AND COALESCE(is_active,1)=1 AND updated_at IS NULL AND source_type<>'truewealth_manual_provisional' AND valuation_date BETWEEN ? AND ? ) SELECT day,total_value_chf FROM ranked WHERE rn=1 AND total_value_chf IS NOT NULL""", (account_id, from_date.isoformat(), to_date.isoformat()), ).fetchall() else: rows = conn.execute( """WITH ranked AS ( SELECT balance_date day,amount_chf, ROW_NUMBER() OVER (PARTITION BY balance_date ORDER BY CASE snapshot_type WHEN 'reconciliation' THEN 4 WHEN 'manual_balance' THEN 3 WHEN 'csv_anchor_balance' THEN 2 ELSE 1 END DESC, created_at DESC,snapshot_id DESC) rn FROM cash_account_snapshots WHERE account_id=? AND balance_date BETWEEN ? AND ? ) SELECT day,amount_chf FROM ranked WHERE rn=1 AND amount_chf IS NOT NULL""", (account_id, from_date.isoformat(), to_date.isoformat()), ).fetchall() series.append({str(row[0]): Decimal(str(row[1])) for row in rows}) complete_dates = set(series[0]) for values in series[1:]: complete_dates.intersection_update(values) points = [ { "at": day, "value_chf": _money(sum((values[day] for values in series), Decimal("0"))) or "0.00", } for day in sorted(complete_dates) ] if to_date == as_of and current["complete"] and (not points or points[-1]["at"] != as_of.isoformat()): points.append({"at": as_of.isoformat(), "value_chf": _money(current["total"]) or "0.00"}) reason = ( "Nur Stichtage mit vollständigen gespeicherten Bewertungen werden verbunden; Datenlücken werden nicht ergänzt." if len(points) >= 2 else "Für einen Verlauf fehlen mindestens zwei gemeinsame vollständige Stichtage; Zwischenwerte werden nicht erfunden." ) return points, reason def _current_values(conn: Connection, *, as_of: date) -> dict[str, Any]: cash = get_cash_summary(conn) equity = get_equity_summary(conn) crypto_positions = list_crypto_positions(conn) truewealth, truewealth_as_of = _truewealth(conn) unassigned, unassigned_items = _unassigned_values(conn) cash_known = [ item for item in cash.positions if item.amount_chf is not None and (item.last_manual_reconciliation or item.last_imported_booking) ] cash_value = sum((Decimal(item.amount_chf or "0") for item in cash_known), Decimal("0")) equity_value = Decimal(equity.valued_partial_chf) crypto_known = [item for item in crypto_positions if item.market_value_chf is not None] crypto_value = sum((Decimal(item.market_value_chf or "0") for item in crypto_known), Decimal("0")) complete = ( equity.coverage_complete and len(crypto_known) == len(crypto_positions) and len(cash_known) == len(cash.positions) and truewealth is not None ) known_equity_positions = int( getattr(equity, "valued_positions", 1 if equity_value else 0) ) equity_positions = int( getattr( equity, "total_positions", known_equity_positions + int(equity.unvalued_positions), ) ) equity_display = ( None if not equity_positions or (equity.unvalued_positions and not equity_value) else _money(equity_value) ) crypto_display = ( None if not crypto_positions or (not crypto_known and crypto_positions) else _money(crypto_value) ) distribution = [ { "key": "cash", "label": "Bankguthaben", "value_chf": _money(cash_value) if cash_known else None, }, {"key": "equity", "label": "Aktien und ETFs", "value_chf": equity_display}, {"key": "truewealth", "label": "True Wealth", "value_chf": _money(truewealth)}, {"key": "crypto", "label": "Kryptowährungen", "value_chf": crypto_display}, ] if unassigned or unassigned_items: distribution.append( {"key": "other", "label": "Nicht zugeordnet", "value_chf": _money(unassigned)} ) total = sum( (Decimal(str(row["value_chf"])) for row in distribution if row["value_chf"] is not None), Decimal("0"), ) grouped_cash: dict[tuple[str, str], dict[str, Any]] = defaultdict( lambda: { "value": Decimal("0"), "known": 0, "total": 0, "dates": [], "statuses": [], "balance_modes": [], "account_ids": set(), } ) for item in cash.positions: account_label = getattr(item, "account_label", item.platform) group = grouped_cash[(item.platform, account_label)] group["account_ids"].add(str(getattr(item, "account_id", ""))) group["total"] += 1 has_value = bool( item.amount_chf is not None and (item.last_manual_reconciliation or item.last_imported_booking) ) if has_value: group["value"] += Decimal(item.amount_chf or "0") group["known"] += 1 group["dates"].extend( value for value in (item.last_manual_reconciliation, item.last_imported_booking) if value ) group["statuses"].append(item.status) group["balance_modes"].append(getattr(item, "balance_mode", "unknown")) sources: list[dict[str, Any]] = [] for index, ((platform, account_label), group) in enumerate( sorted(grouped_cash.items()), start=1 ): source_as_of = max(group["dates"], default=None) freshness = assess_freshness( available=bool(group["known"]), as_of=source_as_of, now=datetime.combine(as_of, datetime.max.time(), tzinfo=UTC), source_kind="bank_balance", ) statuses = [str(status) for status in group["statuses"]] sources.append( { "key": f"cash-{index}", "_canonical_account_id": next(iter(group["account_ids"])) if len(group["account_ids"]) == 1 and "" not in group["account_ids"] else None, "label": safe_account_label(account_label), "provider_label": safe_account_label(platform), "kind": "Bankguthaben", "source_role": "account", "performance_scope": ( "postfinance" if "official_components" in group["balance_modes"] else None ), "current_value_chf": _money(group["value"]) if group["known"] else None, "current_value_status": ( "ready" if group["total"] and group["known"] == group["total"] else "partial" if group["known"] else "not_ready" ), "change_chf": None, "net_contributions_chf": None, "return_pct": None, "as_of": source_as_of, "freshness_status": freshness.status, "freshness_reason_code": freshness.reason_code, "expected_as_of": freshness.expected_as_of, "reconciliation_status": ( "not_assessable" if group["known"] != group["total"] else "difference" if any(status == "Abgleich offen" for status in statuses) else "reconciled" if statuses and all(status.startswith("Offiziell abgeglichen") for status in statuses) else "not_assessable" ), "performance_status": "not_applicable", } ) def investment_source( *, key: str, label: str, value: Decimal | None, source_as_of: str | None, source_kind: str, source_complete: bool, ) -> dict[str, Any]: freshness = assess_freshness( available=value is not None, as_of=source_as_of, now=datetime.combine(as_of, datetime.max.time(), tzinfo=UTC), source_kind=source_kind, # type: ignore[arg-type] ) return { "key": key, "label": label, "provider_label": label, "kind": "Anlage", "source_role": "canonical_value", "performance_scope": { "postfinance-investments": "postfinance", "truewealth": "truewealth", "crypto": "crypto", }.get(key), "current_value_chf": _money(value), "current_value_status": ( "ready" if value is not None and source_complete else "partial" if value is not None else "not_ready" ), "change_chf": None, "net_contributions_chf": None, "return_pct": None, "as_of": source_as_of, "freshness_status": freshness.status, "freshness_reason_code": freshness.reason_code, "expected_as_of": freshness.expected_as_of, "reconciliation_status": "not_assessable", "performance_status": "not_ready", } sources.extend( [ investment_source( key="postfinance-investments", label="PostFinance Aktien und ETFs", value=equity_value, source_as_of=equity.as_of, source_kind="market", source_complete=equity.coverage_complete, ), investment_source( key="truewealth", label="True Wealth Gesamtwert", value=truewealth, source_as_of=truewealth_as_of, source_kind="managed_portfolio", source_complete=truewealth is not None, ), investment_source( key="crypto", label="Kryptowährungen", value=crypto_value if crypto_known else None, source_as_of=max( (item.last_price_update or "" for item in crypto_known), default="" ) or None, source_kind="crypto_24_7", source_complete=bool(crypto_positions) and len(crypto_known) == len(crypto_positions), ), ] ) visa_account = conn.execute( "SELECT account_id FROM accounts WHERE account_type='credit_card_liability' AND is_active=1 LIMIT 1" ).fetchone() if visa_account: sources.append( { "key": "visa-liability", "label": "VISA Kartenverbindlichkeit", "provider_label": "VISA", "kind": "Verbindlichkeit", "source_role": "liability", "performance_scope": None, "current_value_chf": None, "current_value_status": "partial", "change_chf": None, "net_contributions_chf": None, "return_pct": None, "as_of": None, "freshness_status": "unknown", "freshness_reason_code": "current_liability_snapshot_missing", "expected_as_of": None, "reconciliation_status": "not_assessable", "performance_status": "not_applicable", } ) if unassigned_items: sources.append( { "key": "unassigned", "label": "Weitere bestätigte Werte", "kind": "Nicht zugeordnet", "source_role": "canonical_value", "performance_scope": None, "current_value_chf": _money(unassigned), "current_value_status": "ready", "change_chf": None, "net_contributions_chf": None, "return_pct": None, "as_of": max((item["as_of"] for item in unassigned_items), default=None), "freshness_status": assess_freshness( available=True, as_of=max((item["as_of"] for item in unassigned_items), default=None), now=datetime.combine(as_of, datetime.max.time(), tzinfo=UTC), ).status, "reconciliation_status": "not_assessable", "performance_status": "not_ready", } ) data_dates = [str(source["as_of"]) for source in sources if source["as_of"]] return { "total": total, "investments": equity_value + (truewealth or Decimal("0")) + crypto_value + unassigned, "cash": cash_value, "complete": complete, "distribution": distribution, "sources": sources, "data_as_of": max(data_dates, default=None), "unpriced_count": equity.unvalued_positions + len(crypto_positions) - len(crypto_known), "missing_cash_count": sum( 1 for group in grouped_cash.values() if group["known"] != group["total"] ), "unassigned_items": unassigned_items, } def _policy_comparison(conn: Connection, current: dict[str, Any]) -> dict[str, Any]: loaded = active_policy(conn) policy = loaded.get("policy") if loaded.get("configured") else None if not policy: return {"configured": False, "version": None, "rows": [], "contribution": None} values = {str(row["key"]): Decimal(str(row["value_chf"] or "0")) for row in current["distribution"]} by_policy = { "cash": values.get("cash", Decimal("0")), "equity": values.get("equity", Decimal("0")), "crypto": values.get("crypto", Decimal("0")), "other": values.get("truewealth", Decimal("0")) + values.get("other", Decimal("0")), } rows = [] for allocation in policy["allocations"]: asset = str(allocation["asset_class"]) current_pct = by_policy.get(asset, Decimal("0")) / current["total"] * Decimal("100") if current["total"] and current["complete"] else None lower = Decimal(str(allocation["lower_pct"])) upper = Decimal(str(allocation["upper_pct"])) status = "not_assessable" if current_pct is None else "below_range" if current_pct < lower else "above_range" if current_pct > upper else "within_range" rows.append( { "asset_class": asset, "current_pct": _money(current_pct), "target_pct": str(allocation["target_pct"]), "lower_pct": str(allocation["lower_pct"]), "upper_pct": str(allocation["upper_pct"]), "deviation_pct_points": _money(current_pct - Decimal(str(allocation["target_pct"]))) if current_pct is not None else None, "status": status, } ) monthly = _decimal(policy.get("monthly_contribution")) return { "configured": True, "version": policy["version"], "rows": rows, "contribution": {"monthly_target_chf": _money(monthly), "annual_target_chf": _money(monthly * Decimal("12"))} if monthly is not None else None, } def _readiness_status(value: str) -> str: return { "complete": "ready", "available": "ready", "partial": "partial", "unavailable": "not_ready", "not_calculable": "not_ready", }.get(value, "not_ready") def _join_labels(labels: list[str]) -> str: unique = list(dict.fromkeys(label for label in labels if label)) if not unique: return "keine Quelle" if len(unique) == 1: return unique[0] return ", ".join(unique[:-1]) + " und " + unique[-1] def _performance_scope_status(row: dict[str, Any] | None) -> str: if not row: return "not_ready" statuses = { str(row.get("ttwror_status")), str(row.get("xirr_status")), str(row.get("attribution_status")), } if statuses and all(status == "complete" for status in statuses): return "ready" if any(status in {"complete", "partial"} for status in statuses): return "partial" return "not_ready" def _build_diagnostics( *, current: dict[str, Any], coverage: dict[str, Any], policy: dict[str, Any], period: dict[str, str], ) -> list[dict[str, Any]]: rows = { str(row["scope"]): row for row in coverage.get("rows", []) if str(row.get("scope")) != "portfolio" } scope_labels = { "postfinance": "PostFinance", "truewealth": "True Wealth", "crypto": "Kryptowährungen", } diagnostics: list[dict[str, Any]] = [] opening_sources = [ scope_labels[scope] for scope in ("postfinance", "truewealth") if not rows.get(scope, {}).get("valuation_from") or str(rows[scope]["valuation_from"]) > period["from"] ] if opening_sources: diagnostics.append( { "dimension": "performance", "affected_sources": opening_sources, "message": ( f"Für die Rendite {period['from'][:4]} fehlen Anfangsbewertungen bei " f"{_join_labels(opening_sources)}." ), "action": "Bestätigte Gesamtwerte zum Periodenbeginn bereitstellen.", "reason_code": "opening_valuation_missing_by_source", "prominent": True, } ) closing_sources = [ scope_labels[scope] for scope in ("postfinance", "truewealth") if not rows.get(scope, {}).get("valuation_to") or str(rows[scope]["valuation_to"]) < period["to"] ] if closing_sources: diagnostics.append( { "dimension": "performance", "affected_sources": closing_sources, "message": ( "Für den gewählten Periodenendstichtag fehlen kompatible Bewertungen bei " f"{_join_labels(closing_sources)}." ), "action": "Bestätigte Endbewertungen für denselben fachlichen Stichtag bereitstellen.", "reason_code": "closing_valuation_date_mismatch_by_source", "prominent": False, } ) crypto = rows.get("crypto", {}) if not crypto.get("valuation_from"): diagnostics.append( { "dimension": "performance", "affected_sources": ["Kryptowährungen"], "message": "Für Kryptowährungen ist noch keine historische Gesamtbewertung vorhanden.", "action": "Belastbare historische Gesamtwerte und vollständige Aktivitäten bereitstellen.", "reason_code": "crypto_portfolio_history_missing", "prominent": True, } ) missing_bank_sources = [ ( str(source["label"]) if "E-Finance" in str(source["label"]) else str(source.get("provider_label") or source["label"]) ) for source in current["sources"] if source["kind"] == "Bankguthaben" and source["current_value_chf"] is None ] if missing_bank_sources: preferred_order = {"AKB": 0, "Raiffeisen": 1, "PostFinance E-Finance": 2} providers = sorted( dict.fromkeys(missing_bank_sources), key=lambda label: (preferred_order.get(label, 99), label), ) diagnostics.append( { "dimension": "current_value", "affected_sources": providers, "message": ( f"{_join_labels(providers)} besitzen noch keinen bestätigten aktuellen Kontostand." ), "action": "Je echtes Konto einen bestätigten Saldo mit fachlichem Stichtag erfassen oder importieren.", "reason_code": "current_bank_balance_missing", "prominent": True, } ) flow_sources = [ scope_labels[scope] for scope in ("postfinance", "truewealth", "crypto") if { "external_cashflow_history_missing", "cashflow_classification_incomplete", } & set(rows.get(scope, {}).get("reason_codes", [])) ] if flow_sources: diagnostics.append( { "dimension": "performance", "affected_sources": flow_sources, "message": ( "Externe Ein- und Auszahlungen sind für " f"{_join_labels(flow_sources)} noch nicht vollständig belegt." ), "action": "Vollständige Aktivitäten und Kapitalfluss-Coverage für den Zeitraum nachweisen.", "reason_code": "external_cashflow_coverage_incomplete", "prominent": False, } ) stale_sources = [ str(source["label"]) for source in current["sources"] if source["current_value_chf"] is not None and source["freshness_status"] == "stale" ] if stale_sources: diagnostics.append( { "dimension": "freshness", "affected_sources": stale_sources, "message": f"Der Datenstand von {_join_labels(stale_sources)} ist veraltet.", "action": "Bestehenden Aktualisierungsbereich der betroffenen Quelle verwenden.", "reason_code": "source_update_overdue", "prominent": False, } ) if not policy.get("configured"): diagnostics.append( { "dimension": "policy", "affected_sources": [], "message": "Es ist keine bestätigte Portfolioorientierung hinterlegt.", "action": "Optional eine Portfolioorientierung hinterlegen; Wert und Performance bleiben davon unabhängig.", "reason_code": "portfolio_policy_not_configured", "prominent": False, } ) unique: list[dict[str, Any]] = [] seen: set[tuple[str, tuple[str, ...], str]] = set() for item in diagnostics: identity = ( str(item["dimension"]), tuple(str(value) for value in item["affected_sources"]), str(item["reason_code"]), ) if identity not in seen: seen.add(identity) unique.append(item) return unique def _build_readiness( *, current: dict[str, Any], coverage: dict[str, Any], policy: dict[str, Any], period: dict[str, str], summary: dict[str, Any], ttwror_quality: dict[str, Any], household_change: str | None, history_points: list[dict[str, str]], reconciliation_status: str, ) -> dict[str, Any]: known_current = [ str(source["label"]) for source in current["sources"] if source["current_value_chf"] is not None ] missing_current = [ str(source["label"]) for source in current["sources"] if source.get("current_value_status") != "ready" ] coverage_rows = { str(row["scope"]): row for row in coverage.get("rows", []) if str(row.get("scope")) != "portfolio" } performance_scopes = ( ("postfinance", "PostFinance"), ("truewealth", "True Wealth"), ("crypto", "Kryptowährungen"), ) def sources_for_status(field: str) -> tuple[list[str], list[str]]: included = [ label for scope, label in performance_scopes if coverage_rows.get(scope, {}).get(field) == "complete" ] missing = [label for _, label in performance_scopes if label not in included] return included, missing ttwror_included, ttwror_missing = sources_for_status("ttwror_status") attribution_included, attribution_missing = sources_for_status( "attribution_status" ) flow_included = [ label for scope, label in performance_scopes if coverage_rows.get(scope, {}).get("cashflow_coverage_status") == "complete" and coverage_rows.get(scope, {}).get("scope_classification_status") == "complete" ] flow_missing = [label for _, label in performance_scopes if label not in flow_included] def metric( key: str, label: str, status: str, *, included: list[str], missing: list[str], blocker: str | None, action: str | None, reason_code: str | None, as_of: str | None = None, ) -> dict[str, Any]: return { "key": key, "label": label, "status": status, "included_sources": included, "missing_sources": missing, "as_of": as_of, "period": period, "blocker": blocker, "action": action, "reason_code": reason_code, } current_status = "ready" if current["complete"] else "partial" if known_current else "not_ready" metrics = [ metric( "captured_wealth", "Erfasstes Vermögen heute", current_status, included=known_current, missing=missing_current, blocker=None if current_status == "ready" else "Für einzelne Quellen fehlt ein bestätigter aktueller Wert.", action=None if current_status == "ready" else "Fehlende Kontostände mit fachlichem Stichtag bestätigen.", reason_code=None if current_status == "ready" else "current_values_incomplete", as_of=current.get("data_as_of"), ), metric( "wealth_change", "Vermögensveränderung im Zeitraum", "ready" if household_change is not None else "not_ready", included=known_current if household_change is not None else [], missing=[] if household_change is not None else [str(source["label"]) for source in current["sources"]], blocker=None if household_change is not None else "Gemeinsame vollständige Anfangs- und Endbewertungen fehlen.", action=None if household_change is not None else "Bestätigte Bewertungen am Periodenanfang und -ende bereitstellen.", reason_code=None if household_change is not None else "household_boundary_values_missing", ), metric( "net_contributions", "Nettoeinzahlungen", "ready" if summary.get("net_external_cashflows") is not None else "not_ready", included=flow_included, missing=flow_missing, blocker=None if summary.get("net_external_cashflows") is not None else "Externe Kapitalflüsse sind nicht für alle Anlagequellen vollständig belegt.", action=None if summary.get("net_external_cashflows") is not None else "Kapitalfluss-Coverage und Klassifikation vervollständigen.", reason_code=None if summary.get("net_external_cashflows") is not None else "external_cashflow_coverage_incomplete", ), metric( "investment_result", "Anlageergebnis ohne Einzahlungen", "ready" if summary.get("investment_result") is not None else "not_ready", included=attribution_included, missing=attribution_missing, blocker=None if summary.get("investment_result") is not None else "Anfang, Ende oder Nettoeinzahlungen sind nicht vollständig belegt.", action=None if summary.get("investment_result") is not None else "Bewertungen und externe Kapitalflüsse für denselben Zeitraum vervollständigen.", reason_code=None if summary.get("investment_result") is not None else "investment_result_inputs_missing", ), metric( "ttwror", "Zeitgewichtete Rendite", _readiness_status(str(ttwror_quality.get("status", "unavailable"))), included=ttwror_included, missing=ttwror_missing, blocker=None if ttwror_quality.get("status") == "complete" else "Bewertungs- oder Kapitalflussgrenzen der bestehenden TTWROR-Engine fehlen.", action=None if ttwror_quality.get("status") == "complete" else "Anfangs-, End- und Kapitalflussgrenzen mit kanonischen FX-Werten vervollständigen.", reason_code=None if ttwror_quality.get("status") == "complete" else "ttwror_prerequisites_incomplete", ), metric( "wealth_history", "Vermögensverlaufsreihe", "ready" if len(history_points) >= 2 else "not_ready", included=known_current if len(history_points) >= 2 else [], missing=[] if len(history_points) >= 2 else [str(source["label"]) for source in current["sources"]], blocker=None if len(history_points) >= 2 else "Mindestens zwei gemeinsame vollständige Stichtage fehlen.", action=None if len(history_points) >= 2 else "Keine Zwischenwerte schätzen; gemeinsame bestätigte Stichtage bereitstellen.", reason_code=None if len(history_points) >= 2 else "complete_history_points_missing", ), metric( "policy_allocation", "Aufteilung gegenüber Portfolioorientierung", "ready" if policy.get("configured") and current["complete"] else "partial" if policy.get("configured") else "not_applicable", included=known_current, missing=missing_current, blocker=None if policy.get("configured") else "Keine bestätigte Portfolioorientierung vorhanden; Performance und aktueller Wert bleiben unberührt.", action=None if policy.get("configured") else "Optional eine Portfolioorientierung hinterlegen.", reason_code=None if policy.get("configured") else "portfolio_policy_not_configured", ), ] freshness_status = combined_freshness( [ cast(FreshnessStatus, source["freshness_status"]) for source in current["sources"] ] ) dimensions = { "current_value": {"status": current_status, "reason_code": None if current_status == "ready" else "current_values_incomplete"}, "freshness": {"status": "ready" if freshness_status == "fresh" else "partial" if known_current else "not_ready", "reason_code": None if freshness_status == "fresh" else "source_freshness_mixed"}, "reconciliation": {"status": "ready" if reconciliation_status == "reconciled" else "not_ready" if reconciliation_status == "difference" else "partial", "reason_code": None if reconciliation_status == "reconciled" else "reconciliation_not_fully_assessable"}, "performance": {"status": next(item["status"] for item in metrics if item["key"] == "ttwror"), "reason_code": next(item["reason_code"] for item in metrics if item["key"] == "ttwror")}, "policy": {"status": "ready" if policy.get("configured") else "not_applicable", "reason_code": None if policy.get("configured") else "portfolio_policy_not_configured"}, } return {"dimensions": dimensions, "metrics": metrics} def build_wealth_cockpit( conn: Connection, *, period: str = "ytd", as_of: str | None = None, data_cutoff: str | None = None, ) -> dict[str, Any]: reference = date.fromisoformat(as_of) if as_of else date.today() start, requested_end = period_bounds(conn, period=period, as_of=reference) cutoff = data_cutoff or _latest_data_cutoff(conn) model_period = ( period if period in {"since_anchor", "1m", "3m", "1y", "all"} else "1y" if period in {"ytd", "previous_year", "12m"} else "all" ) modelled_development = build_modelled_wealth_development( conn, period=model_period, as_of=reference.isoformat() ) current = _current_values(conn, as_of=reference) valuation_end = _latest_valuation_date(conn, requested_end) performance: dict[str, Any] | None = None if start < valuation_end: performance = build_portfolio_performance( conn, from_date=start.isoformat(), to_date=valuation_end.isoformat(), method="both", base_currency="CHF", data_cutoff=cutoff, ) summary = performance.get("summary", {}) if performance else {} quality = performance.get("quality", {}).get("ttwror", {}) if performance else {} xirr_quality = performance.get("quality", {}).get("xirr", {}) if performance else {} investment_events = performance.get("external_cashflows", []) if performance else [] household_events = scope_cashflows( conn, account_ids=_account_ids(conn, investment_only=False), from_date=start.isoformat(), to_date=requested_end.isoformat(), data_cutoff=cutoff, ) reconciliation = build_reconciliation_snapshot( conn, now=datetime.combine(reference, datetime.max.time(), tzinfo=UTC) ) history_points, history_reason = _household_history( conn, from_date=start, to_date=requested_end, current=current, as_of=reference, ) policy = _policy_comparison(conn, current) coverage = build_performance_coverage( conn, from_date=start.isoformat(), to_date=requested_end.isoformat(), ) coverage_items = coverage.get("rows") coverage_rows = { str(row["scope"]): row for row in coverage_items if str(row.get("scope")) != "portfolio" } if isinstance(coverage_items, list) else {} for source in current["sources"]: scope = source.get("performance_scope") if scope: source["performance_status"] = _performance_scope_status( coverage_rows.get(str(scope)) ) key = str(source.get("key", "")) provider = str(source.get("provider_label", "")).casefold() if key == "truewealth": meta = _truewealth_import_meta(conn) elif key == "postfinance-investments" or "postfinance" in provider: meta = _postfinance_import_meta(conn) elif key == "visa-liability": meta = _household_import_meta(conn, "viseca_one") elif "akb" in provider: meta = _household_import_meta( conn, "akb", canonical_account_id=source.get("_canonical_account_id"), ) elif "raiffeisen" in provider: meta = _household_import_meta( conn, "raiffeisen", canonical_account_id=source.get("_canonical_account_id"), ) else: meta = {"imported_at": None, "coverage_from": None, "coverage_to": None, "coverage_status": "unavailable", "new_rows": 0, "duplicate_rows": 0, "review_rows": 0} source.pop("_canonical_account_id", None) source.update({name: value for name, value in meta.items() if name != "last_snapshot"}) source["last_activity_day"] = meta.get("coverage_to") or source.get("as_of") source["last_confirmed_snapshot"] = meta.get("last_snapshot") or source.get("as_of") source["value_basis"] = ( "modelled" if key == "crypto" and source.get("current_value_chf") is not None else "confirmed" if source.get("current_value_chf") is not None else "unavailable" ) review_rows = int(meta.get("review_rows", 0) or 0) if key == "postfinance-investments" or "postfinance" in provider: source["performance_blocker"] = None if meta.get("coverage_status") == "complete" else ( "PostFinance E-Trading-Kontoauszug oder vollständiger Transaktionsreport vom 01.08.–26.08.2026 fehlt; bei null Aktivitäten ist ein offizieller Nachweis erforderlich." ) elif key == "truewealth": source["performance_blocker"] = None if meta.get("coverage_status") == "complete" else "Externe Ein- und Auszahlungen sind noch nicht vollständig belegt." elif key == "crypto": source["performance_blocker"] = "Mengen-, Aktivitäts-, Preis- oder Cashflow-Coverage ist weiterhin unvollständig." source["coverage_status"] = "partial" elif key == "visa-liability": source["performance_blocker"] = "Aktueller Abrechnungssaldo ist nicht vollständig belegt." else: source["performance_blocker"] = None source["next_action"] = ( f"{review_rows} prüfpflichtige Zeilen bearbeiten." if review_rows else source.get("performance_blocker") or "Keine offene Aktion." ) if policy.get("contribution"): invested = _decimal(summary.get("net_external_cashflows")) if period == "ytd" else None policy["contribution"].update( invested_ytd_chf=_money(invested), expected_year_end_chf=_money(invested / Decimal(max(reference.month, 1)) * Decimal("12")) if invested is not None and period == "ytd" else None, difference_to_target_chf=_money(invested - Decimal(policy["contribution"]["annual_target_chf"])) if invested is not None else None, status="available" if invested is not None else "not_assessable", ) planning = get_annual_budget_assistant(conn, year=str(reference.year), current_month=f"{reference.year:04d}-{reference.month:02d}") free_row = next((row for row in planning["summary_kpis"] if row["key"] == "free_after_special"), None) missing_areas = ["Verbindlichkeiten und Immobilienwerte sind nicht vollständig und aktuell erfasst."] if current["unpriced_count"]: missing_areas.append(f"{current['unpriced_count']} Positionen besitzen keinen belastbaren aktuellen Wert.") if current["missing_cash_count"]: missing_areas.append( f"{current['missing_cash_count']} Bankkonten besitzen keinen bestätigten aktuellen Saldo." ) if next((row for row in current["distribution"] if row["key"] == "truewealth"), {}).get("value_chf") is None: missing_areas.append("Für True Wealth fehlt ein bestätigter aktueller Gesamtwert.") history_by_date = {point["at"]: Decimal(point["value_chf"]) for point in history_points} opening_household = history_by_date.get(start.isoformat()) closing_household = history_by_date.get(requested_end.isoformat()) household_change = _money(closing_household - opening_household) if opening_household is not None and closing_household is not None else None household_change_status = "available" if household_change is not None else "not_calculable" investment_result = summary.get("investment_result") net_contributions = summary.get("net_external_cashflows") reconciliation_rows = reconciliation.get("reconciliations", []) if not isinstance(reconciliation_rows, list): reconciliation_rows = [] reconciliation_statuses = [str(row["status"]) for row in reconciliation_rows] reconciliation_status = ( "difference" if "difference" in reconciliation_statuses else "reconciled" if reconciliation_statuses and all(status == "reconciled" for status in reconciliation_statuses) else "not_assessable" ) period_payload = { "preset": period, "from": start.isoformat(), "to": requested_end.isoformat(), } diagnostics = _build_diagnostics( current=current, coverage=coverage, policy=policy, period=period_payload, ) hints = [str(item["message"]) for item in diagnostics if item["prominent"]][:3] readiness = _build_readiness( current=current, coverage=coverage, policy=policy, period=period_payload, summary=summary, ttwror_quality=quality, household_change=household_change, history_points=history_points, reconciliation_status=reconciliation_status, ) current_freshness = combined_freshness( [ cast(FreshnessStatus, source["freshness_status"]) for source in current["sources"] ] ) ttwror_verified = ( quality.get("status") == "complete" and summary.get("ttwror_cumulative") is not None ) xirr_verified = ( xirr_quality.get("status") == "complete" and summary.get("xirr_annualized") is not None ) verified_performance = { "status": "verified" if ttwror_verified and xirr_verified else "not_verified", "label": "Verifiziert" if ttwror_verified and xirr_verified else "Noch nicht verifiziert", "ttwror_status": "ready" if ttwror_verified else "not_ready", "xirr_status": "ready" if xirr_verified else "not_ready", "ttwror_pct": summary.get("ttwror_cumulative") if ttwror_verified else None, "xirr_pct": summary.get("xirr_annualized") if xirr_verified else None, } return { "scope_label": "Erfasstes Vermögen", "not_net_worth": True, "period": period_payload, "data_cutoff": cutoff, "modelled_development": modelled_development, "verified_performance": verified_performance, "kpis": [ {"key": "captured_wealth", "label": "Erfasstes Vermögen heute", "value_chf": _money(current["total"]), "status": "complete" if current["complete"] else "approximate"}, {"key": "wealth_change", "label": "Veränderung im Zeitraum", "value_chf": household_change, "status": household_change_status}, {"key": "investment_result", "label": "Anlageergebnis ohne Einzahlungen", "value_chf": investment_result, "status": "available" if investment_result is not None else "not_calculable"}, {"key": "return", "label": "Zeitgewichtete Rendite", "value_pct": summary.get("ttwror_cumulative"), "status": "available" if quality.get("status") == "complete" and summary.get("ttwror_cumulative") is not None else "not_calculable"}, {"key": "net_contributions", "label": "Nettoeinzahlungen ins Anlageportfolio", "value_chf": net_contributions, "status": "available" if net_contributions is not None else "not_calculable"}, {"key": "data_as_of", "label": "Datenstand", "value_date": current["data_as_of"], "status": "available" if current["data_as_of"] else "unknown"}, ], "totals": {"captured_wealth_chf": _money(current["total"]), "investments_chf": _money(current["investments"]), "bank_cash_chf": _money(current["cash"]), "complete": current["complete"]}, "history": {"status": "available" if len(history_points) >= 2 else "not_calculable", "points": history_points, "household_cashflow_events": household_events, "investment_cashflow_events": investment_events, "reason": history_reason}, "distribution": current["distribution"], "sources": current["sources"], "readiness": readiness, "diagnostics": diagnostics, "performance_coverage": coverage, "policy": policy, "planning": {"free_plannable_chf": free_row.get("value_chf") if free_row else None, "available": bool(free_row and free_row.get("value_chf") is not None), "link": "/planning/budget/planning", "included_in_wealth": False}, "data_quality": {"freshness_status": current_freshness, "reconciliation_status": reconciliation_status, "performance_status": quality.get("status", "unavailable"), "performance_reasons": quality.get("reason_codes", ["historical_portfolio_valuations_missing"]), "missing_areas": missing_areas, "unassigned": current["unassigned_items"]}, "hints": hints[:3], "method": {"wealth_change": "Endwert minus Anfangswert innerhalb des gesamten Haushalts; interne Transfers neutral.", "investment_result": "Endwert minus Anfangswert minus Nettoeinzahlungen innerhalb des Anlageportfolios.", "return": "Bestehende TTWROR-Engine; nur bei vollständigen Bewertungen und klassifizierten Kapitalflüssen."}, }