diff --git a/src/jarvis_finance/services/wealth_cockpit.py b/src/jarvis_finance/services/wealth_cockpit.py index d6217fd..a30390d 100644 --- a/src/jarvis_finance/services/wealth_cockpit.py +++ b/src/jarvis_finance/services/wealth_cockpit.py @@ -1,306 +1,439 @@ 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 = {"ytd", "previous_year", "12m", "all"} +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 ytd, previous_year, 12m oder all sein") - 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) - if period == "12m": + 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) -> dict[str, Any]: +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 {"imported_at": None, "coverage_from": None, "coverage_to": None, "coverage_status": "unavailable", "new_rows": 0, "duplicate_rows": 0, "review_rows": 0} - 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]) + 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", + "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." @@ -371,193 +504,198 @@ def _household_history( 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, @@ -1000,265 +1138,303 @@ def _build_readiness( 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="twr", + 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") + meta = _household_import_meta( + conn, + "akb", + canonical_account_id=source.get("_canonical_account_id"), + ) elif "raiffeisen" in provider: - meta = _household_import_meta(conn, "raiffeisen") + 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."}, } __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-current-import-performance__HERMES_CWD_8d46a20096ed__