===== src/jarvis_finance/services/modelled_wealth.py ===== 1 from __future__ import annotations 2 3 from calendar import monthrange 4 from datetime import date, timedelta 5 from decimal import Decimal 6 import re 7 from sqlite3 import Connection 8 from typing import Any 9 10 from jarvis_finance.services.cash_service import authoritative_cash_movements 11 from jarvis_finance.services.daily_valuations import SOURCE_KEY as CRYPTO_VALUATION_SOURCE 12 13 LEGACY_CRYPTO_VALUATION_SOURCE = "daily_crypto_current_valuation_v1" 14 15 MONEY = Decimal("0.01") 16 PERCENT = Decimal("0.0001") 17 MODEL_PERIODS = {"since_anchor", "1m", "3m", "ytd", "1y", "all"} 18 SNAPSHOT_PRECEDENCE = { 19 "reconciliation": 4, 20 "manual_balance": 3, 21 "csv_anchor_balance": 2, 22 "calculated_balance": 1, 23 } 24 COMPONENT_LABELS = { 25 "postfinance": "PostFinance", 26 "truewealth": "True Wealth", 27 "crypto": "Krypto", 28 "bank_cash": "Bankguthaben", 29 "other_assets": "Weitere Anlagen", 30 } 31 32 33 def _money(value: Decimal | None) -> str | None: 34 return None if value is None else format(value.quantize(MONEY), "f") 35 36 37 def _subtract_months(day: date, months: int) -> date: 38 absolute = day.year * 12 + day.month - 1 - months 39 year, month_index = divmod(absolute, 12) 40 month = month_index + 1 41 return date(year, month, min(day.day, monthrange(year, month)[1])) 42 43 44 def _authoritative_cash_movements( 45 conn: Connection, *, account_id: str, after: str | None, through: str 46 ) -> dict[str, Any]: 47 return authoritative_cash_movements( 48 conn, account_id=account_id, after=after, through=through 49 ) 50 51 52 def effective_cash_evidence( 53 conn: Connection, *, account_id: str, as_of: str 54 ) -> dict[str, Any]: 55 """Return one account's effective cash evidence without writing. 56 57 The newest business date wins. If several valid snapshots exist on that 58 date, the deterministic precedence is reconciliation > manual > CSV anchor. 59 Confirmed movements are then applied forward only. With no anchor, a 60 non-empty confirmed ledger may provide a calculated balance; empty evidence 61 is unavailable rather than a false CHF 0. 62 """ 63 64 snapshot = conn.execute( 65 """SELECT * FROM cash_account_snapshots 66 WHERE account_id=? AND balance_date<=? AND amount_chf IS NOT NULL 67 ORDER BY balance_date DESC, 68 CASE snapshot_type 69 WHEN 'reconciliation' THEN 4 70 WHEN 'manual_balance' THEN 3 71 WHEN 'csv_anchor_balance' THEN 2 72 WHEN 'calculated_balance' THEN 1 73 ELSE 0 END DESC, 74 created_at DESC,snapshot_id DESC LIMIT 1""", 75 (account_id, as_of), 76 ).fetchone() 77 if snapshot: 78 anchor_day = str(snapshot["balance_date"]) 79 anchor = Decimal(str(snapshot["amount_chf"])) 80 movement_evidence = _authoritative_cash_movements( 81 conn, account_id=account_id, after=anchor_day, through=as_of 82 ) 83 movement = movement_evidence["amount"] 84 return { 85 "account_id": account_id, 86 "anchor_type": str(snapshot["snapshot_type"]), 87 "anchor_date": anchor_day, 88 "anchor_value_chf": _money(anchor), 89 "movement_chf": _money(movement), 90 "value_chf": _money(anchor + movement), 91 "quality": "confirmed" if anchor_day == as_of and not movement else "carried", 92 "source_date": movement_evidence["last_date"] or anchor_day, 93 "movement_source": movement_evidence["source"], 94 } 95 96 movement_evidence = _authoritative_cash_movements( 97 conn, account_id=account_id, after=None, through=as_of 98 ) 99 if movement_evidence["count"]: 100 value = movement_evidence["amount"] 101 return { 102 "account_id": account_id, 103 "anchor_type": "calculated_balance", 104 "anchor_date": None, 105 "anchor_value_chf": None, 106 "movement_chf": _money(value), 107 "value_chf": _money(value), 108 "quality": "modelled", 109 "source_date": movement_evidence["last_date"], 110 "movement_source": movement_evidence["source"], 111 } 112 return { 113 "account_id": account_id, 114 "anchor_type": None, 115 "anchor_date": None, 116 "anchor_value_chf": None, 117 "movement_chf": None, 118 "value_chf": None, 119 "quality": "unavailable", 120 "source_date": None, 121 "movement_source": None, 122 } 123 124 125 def _role_account_ids(conn: Connection, role: str) -> list[str]: 126 return [ 127 str(row[0]) 128 for row in conn.execute( 129 """SELECT a.account_id FROM accounts a 130 JOIN performance_scope_classifications psc ON psc.account_id=a.account_id 131 WHERE a.is_active=1 AND psc.classification_role=? 132 AND psc.decision_version='investment_performance_scope_v1' 133 ORDER BY a.account_id""", 134 (role,), 135 ).fetchall() 136 ] 137 138 139 def _latest_model_rows( 140 conn: Connection, 141 *, 142 account_ids: list[str], 143 through: str, 144 source: str | None = None, 145 source_prefix: str | None = None, 146 ) -> list[dict[str, Any]]: 147 if not account_ids: 148 return [] 149 placeholders = ",".join("?" for _ in account_ids) 150 if source and source_prefix: 151 raise ValueError("source and source_prefix are mutually exclusive") 152 source_sql = " AND source=?" if source else " AND source LIKE ?" if source_prefix else "" 153 params: list[Any] = [*account_ids, through] 154 if source: 155 params.append(source) 156 elif source_prefix: 157 params.append(f"{source_prefix}%") 158 rows = conn.execute( 159 f"""WITH ranked AS ( 160 SELECT scope_id,substr(valuation_at,1,10) day,value_original,currency, 161 fx_rate_to_base,source,snapshot_version,captured_at,snapshot_id, 162 ROW_NUMBER() OVER( 163 PARTITION BY scope_id,substr(valuation_at,1,10) 164 ORDER BY snapshot_version DESC,captured_at DESC,snapshot_id DESC 165 ) rn 166 FROM portfolio_valuation_snapshots 167 WHERE scope_kind='account' AND scope_id IN ({placeholders}) 168 AND substr(valuation_at,1,10)<=? 169 AND quality_status IN ('complete','ok','partial') 170 {source_sql} 171 ) SELECT * FROM ranked WHERE rn=1 ORDER BY day,scope_id""", 172 tuple(params), 173 ).fetchall() 174 result = [] 175 for row in rows: 176 try: 177 observation_day = date.fromisoformat(str(row["day"])) 178 if str(row["currency"]) == "CHF": 179 value = Decimal(str(row["value_original"])) 180 elif row["fx_rate_to_base"] is not None: 181 value = Decimal(str(row["value_original"])) * Decimal( 182 str(row["fx_rate_to_base"]) 183 ) 184 else: 185 continue 186 except Exception: 187 continue 188 result.append( 189 { 190 "account_id": str(row["scope_id"]), 191 "date": observation_day.isoformat(), 192 "value": value, 193 "source": str(row["source"]), 194 "captured_at": str(row["captured_at"]), 195 } 196 ) 197 return result 198 199 200 def _official_rows( 201 conn: Connection, 202 *, 203 account_ids: list[str], 204 through: str, 205 allowed_sources: tuple[str, ...], 206 ) -> list[dict[str, Any]]: 207 if not account_ids: 208 return [] 209 placeholders = ",".join("?" for _ in account_ids) 210 source_placeholders = ",".join("?" for _ in allowed_sources) 211 rows = conn.execute( 212 f"""WITH ranked AS ( 213 SELECT account_id,valuation_date,total_value_chf,source_type, 214 COALESCE(valuation_at,created_at) captured_at, 215 ROW_NUMBER() OVER(PARTITION BY account_id,valuation_date 216 ORDER BY COALESCE(valuation_at,created_at) DESC,snapshot_id DESC) rn 217 FROM account_value_snapshots 218 WHERE account_id IN ({placeholders}) AND valuation_date<=? 219 AND COALESCE(is_active,1)=1 AND updated_at IS NULL 220 AND source_type IN ({source_placeholders}) 221 AND quality_status IN ('ok','complete') 222 ) SELECT * FROM ranked WHERE rn=1 ORDER BY valuation_date,account_id""", 223 (*account_ids, through, *allowed_sources), 224 ).fetchall() 225 result = [] 226 for row in rows: 227 try: 228 observation_day = date.fromisoformat(str(row["valuation_date"])) 229 value = Decimal(str(row["total_value_chf"])) 230 except Exception: 231 continue 232 if not value.is_finite() or value < Decimal("0"): 233 continue 234 result.append({ 235 "account_id": str(row["account_id"]), 236 "date": observation_day.isoformat(), 237 "value": value, 238 "source": str(row["source_type"]), 239 "captured_at": str(row["captured_at"]), 240 }) 241 return result 242 243 244 def _postfinance_events(conn: Connection, *, through: str) -> tuple[dict[str, dict[str, Any]], list[dict[str, Any]]]: 245 depot_ids = _role_account_ids(conn, "postfinance_etrading_depot") 246 cash_ids = _role_account_ids(conn, "postfinance_etrading_cash") 247 official = _official_rows( 248 conn, 249 account_ids=depot_ids, 250 through=through, 251 allowed_sources=("postfinance_official_import",), 252 ) 253 model_rows = _latest_model_rows( 254 conn, 255 account_ids=depot_ids + cash_ids, 256 through=through, 257 source_prefix="daily_market_fx_v", 258 ) 259 model_by_day: dict[str, dict[str, Decimal]] = {} 260 for row in model_rows: 261 model_by_day.setdefault(row["date"], {})[row["account_id"]] = row["value"] 262 models: dict[str, Decimal] = {} 263 model_observations: dict[str, dict[str, Any]] = {} 264 for day, values in model_by_day.items(): 265 if depot_ids and cash_ids and all(account in values for account in depot_ids + cash_ids): 266 models[day] = sum(values.values(), Decimal("0")) 267 model_observations[day] = { 268 "value": models[day], 269 "captured_at": max( 270 row["captured_at"] for row in model_rows if row["date"] == day 271 ), 272 } 273 events = { 274 day: {"value": value, "quality": "modelled", "source_date": day} 275 for day, value in models.items() 276 } 277 for row in official: 278 events[row["date"]] = { 279 "value": row["value"], 280 "quality": "confirmed", 281 "source_date": row["date"], 282 } 283 markers = _correction_markers( 284 source_key="postfinance", official=official, models=model_observations 285 ) 286 return events, markers 287 288 289 def _truewealth_events(conn: Connection, *, through: str) -> tuple[dict[str, dict[str, Any]], list[dict[str, Any]]]: 290 account_ids = _role_account_ids(conn, "canonical_truewealth_total_value") 291 official = _official_rows( 292 conn, 293 account_ids=account_ids, 294 through=through, 295 allowed_sources=("truewealth_official_import", "manual_total_value"), 296 ) 297 model_rows = _latest_model_rows( 298 conn, 299 account_ids=account_ids, 300 through=through, 301 source="truewealth_modelled_daily", 302 ) 303 models: dict[str, Decimal] = {} 304 model_observations: dict[str, dict[str, Any]] = {} 305 for row in model_rows: 306 models[row["date"]] = models.get(row["date"], Decimal("0")) + row["value"] 307 model_observations[row["date"]] = { 308 "value": models[row["date"]], 309 "captured_at": max( 310 str(row["captured_at"]), 311 str(model_observations.get(row["date"], {}).get("captured_at", "")), 312 ), 313 } 314 events = { 315 day: {"value": value, "quality": "modelled", "source_date": day} 316 for day, value in models.items() 317 } 318 for row in official: 319 events[row["date"]] = { 320 "value": row["value"], 321 "quality": "confirmed", 322 "source_date": row["date"], 323 } 324 return events, _correction_markers( 325 source_key="truewealth", official=official, models=model_observations 326 ) 327 328 329 def _crypto_events(conn: Connection, *, through: str) -> dict[str, dict[str, Any]]: 330 account_ids = _role_account_ids(conn, "crypto_portfolio") 331 if not account_ids: 332 account_ids = [ 333 str(row[0]) 334 for row in conn.execute( 335 """SELECT DISTINCT scope_id FROM portfolio_valuation_snapshots 336 WHERE scope_kind='account' AND source IN (?,?) 337 AND substr(valuation_at,1,10)<=?""", 338 ( 339 CRYPTO_VALUATION_SOURCE, 340 LEGACY_CRYPTO_VALUATION_SOURCE, 341 through, 342 ), 343 ).fetchall() 344 ] 345 rows = [] 346 for source in (CRYPTO_VALUATION_SOURCE, LEGACY_CRYPTO_VALUATION_SOURCE): 347 rows.extend( 348 _latest_model_rows( 349 conn, 350 account_ids=account_ids, 351 through=through, 352 source=source, 353 ) 354 ) 355 latest_by_account_day: dict[tuple[str, str], dict[str, Any]] = {} 356 for row in rows: 357 key = (row["account_id"], row["date"]) 358 if key not in latest_by_account_day or str(row["captured_at"]) > str( 359 latest_by_account_day[key]["captured_at"] 360 ): 361 latest_by_account_day[key] = row 362 values: dict[str, Decimal] = {} 363 for row in latest_by_account_day.values(): 364 values[row["date"]] = values.get(row["date"], Decimal("0")) + row["value"] 365 return { 366 day: {"value": value, "quality": "modelled", "source_date": day} 367 for day, value in values.items() 368 } 369 370 371 def _other_asset_events(conn: Connection, *, through: str) -> dict[str, dict[str, Any]]: 372 """Aggregate confirmed non-cash account values without inventing daily precision.""" 373 rows = conn.execute( 374 """WITH ranked AS ( 375 SELECT s.account_id,s.valuation_date,s.total_value_chf, 376 ROW_NUMBER() OVER( 377 PARTITION BY s.account_id,s.valuation_date 378 ORDER BY COALESCE(s.valuation_at,s.created_at) DESC,s.snapshot_id DESC 379 ) rn 380 FROM account_value_snapshots s 381 JOIN accounts a ON a.account_id=s.account_id 382 WHERE a.is_active=1 AND a.account_type IN ('other_asset','membership') 383 AND s.valuation_date<=? AND COALESCE(s.is_active,1)=1 384 AND s.quality_status IN ('confirmed','ok','complete') 385 ) SELECT account_id,valuation_date,total_value_chf 386 FROM ranked WHERE rn=1 ORDER BY valuation_date,account_id""", 387 (through,), 388 ).fetchall() 389 latest: dict[str, Decimal] = {} 390 events: dict[str, dict[str, Any]] = {} 391 for row in rows: 392 try: 393 value = Decimal(str(row["total_value_chf"])) 394 except Exception: 395 continue 396 if not value.is_finite() or value < Decimal("0"): 397 continue 398 latest[str(row["account_id"])] = value 399 day = str(row["valuation_date"]) 400 events[day] = { 401 "value": sum(latest.values(), Decimal("0")), 402 "quality": "confirmed", 403 "source_date": day, 404 } 405 return events 406 407 408 def _manual_cash_correction_markers(conn: Connection, *, through: str) -> list[dict[str, str]]: 409 rows = conn.execute( 410 """SELECT snapshot_id,account_id,balance_date,amount_chf,created_at 411 FROM cash_account_snapshots 412 WHERE source='manual_screenshot_snapshot' AND balance_date<=? 413 ORDER BY balance_date,created_at,snapshot_id""", 414 (through,), 415 ).fetchall() 416 grouped: dict[str, dict[str, Decimal]] = {} 417 for row in rows: 418 previous = conn.execute( 419 """SELECT amount_chf FROM cash_account_snapshots 420 WHERE account_id=? AND ( 421 balance_date list[dict[str, str]]: 451 markers = [] 452 model_days = sorted(models) 453 for row in official: 454 predecessor_days = [ 455 day 456 for day in model_days 457 if day < row["date"] 458 or ( 459 day == row["date"] 460 and str(models[day]["captured_at"]) < str(row["captured_at"]) 461 ) 462 ] 463 if not predecessor_days: 464 continue 465 predecessor_day = predecessor_days[-1] 466 predecessor = models[predecessor_day]["value"] 467 markers.append( 468 { 469 "date": row["date"], 470 "source_key": source_key, 471 "confirmed_value_chf": _money(row["value"]) or "0.00", 472 "predecessor_model_value_chf": _money(predecessor) or "0.00", 473 "difference_chf": _money(row["value"] - predecessor) or "0.00", 474 } 475 ) 476 return markers 477 478 479 def _event_at_or_before( 480 events: dict[str, dict[str, Any]], day: str 481 ) -> dict[str, Any] | None: 482 eligible = [event_day for event_day in events if event_day <= day] 483 if not eligible: 484 return None 485 source_day = max(eligible) 486 event = events[source_day] 487 return { 488 "value": event["value"], 489 "quality": event["quality"] if source_day == day else "carried", 490 "source_date": source_day, 491 } 492 493 494 def _bank_accounts(conn: Connection) -> list[dict[str, str]]: 495 rows = conn.execute( 496 """SELECT a.account_id,a.account_name,COALESCE(psc.classification_role,'') role 497 FROM accounts a 498 LEFT JOIN performance_scope_classifications psc ON psc.account_id=a.account_id 499 WHERE a.is_active=1 AND a.account_type='cash' 500 AND ( 501 EXISTS(SELECT 1 FROM cash_account_snapshots s WHERE s.account_id=a.account_id) 502 OR EXISTS(SELECT 1 FROM transactions t WHERE t.account_id=a.account_id) 503 OR EXISTS(SELECT 1 FROM cash_balances b WHERE b.account_id=a.account_id) 504 OR EXISTS( 505 SELECT 1 FROM budget_accounts ba 506 WHERE ba.linked_account_id=a.account_id AND ba.is_active=1 507 ) 508 OR EXISTS( 509 SELECT 1 FROM household_account_source_mappings m 510 WHERE m.canonical_account_id=a.account_id AND m.is_active=1 511 ) 512 ) 513 ORDER BY a.account_name,a.account_id""" 514 ).fetchall() 515 excluded_roles = {"postfinance_etrading_cash", "postfinance_efinance_control"} 516 return [ 517 {"account_id": str(row["account_id"]), "label": str(row["account_name"])} 518 for row in rows 519 if str(row["role"] or "") not in excluded_roles 520 ] 521 522 523 def _safe_bank_label(label: str) -> str: 524 """Expose only a generic label and an already-masked four-digit suffix.""" 525 526 suffix = re.search(r"(?:•{4}|\*{4}|x{4})\s*(\d{4})\b", label, re.IGNORECASE) 527 return f"Bankkonto •••• {suffix.group(1)}" if suffix else "Bankkonto" 528 529 530 def _earliest_evidence(conn: Connection, *, fallback: date) -> date: 531 rows = conn.execute( 532 """SELECT day FROM ( 533 SELECT valuation_date day FROM account_value_snapshots 534 WHERE COALESCE(is_active,1)=1 AND updated_at IS NULL 535 UNION ALL SELECT substr(valuation_at,1,10) FROM portfolio_valuation_snapshots 536 UNION ALL SELECT balance_date FROM cash_account_snapshots 537 )""" 538 ).fetchall() 539 valid_days: list[date] = [] 540 for row in rows: 541 try: 542 valid_days.append(date.fromisoformat(str(row[0]))) 543 except (TypeError, ValueError): 544 continue 545 return min(valid_days, default=fallback) 546 547 548 def _period_start( 549 conn: Connection, *, period: str, as_of: date, latest_anchor: date | None 550 ) -> date: 551 if period == "since_anchor": 552 return latest_anchor or as_of 553 if period == "1m": 554 return _subtract_months(as_of, 1) 555 if period == "3m": 556 return _subtract_months(as_of, 3) 557 if period == "ytd": 558 return date(as_of.year, 1, 1) 559 if period == "1y": 560 return _subtract_months(as_of, 12) 561 if period == "all": 562 return _earliest_evidence(conn, fallback=as_of) 563 raise ValueError("period must be since_anchor, 1m, 3m, ytd, 1y or all") 564 565 566 def build_modelled_wealth_development( 567 conn: Connection, *, period: str = "1m", as_of: str | None = None 568 ) -> dict[str, Any]: 569 """Compose existing immutable valuation/snapshot sources into one read model.""" 570 571 if period not in MODEL_PERIODS: 572 raise ValueError("period must be since_anchor, 1m, 3m, ytd, 1y or all") 573 reference = date.fromisoformat(as_of) if as_of else date.today() 574 through = reference.isoformat() 575 postfinance, pf_markers = _postfinance_events(conn, through=through) 576 truewealth, tw_markers = _truewealth_events(conn, through=through) 577 crypto = _crypto_events(conn, through=through) 578 other_assets = _other_asset_events(conn, through=through) 579 investment_events = { 580 "postfinance": postfinance, 581 "truewealth": truewealth, 582 "crypto": crypto, 583 "other_assets": other_assets, 584 } 585 expected_investment = { 586 "postfinance": bool( 587 _role_account_ids(conn, "postfinance_etrading_depot") 588 or _role_account_ids(conn, "postfinance_etrading_cash") 589 ), 590 "truewealth": bool( 591 _role_account_ids(conn, "canonical_truewealth_total_value") 592 ), 593 "crypto": bool(_role_account_ids(conn, "crypto_portfolio")), 594 "other_assets": bool(other_assets), 595 } 596 # Household anchors come from confirmed portfolio-import anchors. Component-only 597 # cash/membership corrections remain event markers and must not move the solid-line 598 # boundary or make mixed-date values look fully confirmed. 599 confirmed_days: list[date] = [] 600 for events in (postfinance, truewealth): 601 for day, event in events.items(): 602 if event["quality"] != "confirmed": 603 continue 604 try: 605 confirmed_days.append(date.fromisoformat(day)) 606 except ValueError: 607 continue 608 latest_anchor = max(confirmed_days, default=None) 609 start = _period_start( 610 conn, period=period, as_of=reference, latest_anchor=latest_anchor 611 ) 612 if start > reference: 613 start = reference 614 if (reference - start).days > 5000: 615 start = reference - timedelta(days=5000) 616 617 bank_accounts = _bank_accounts(conn) 618 points: list[dict[str, Any]] = [] 619 unknown_identity_by_day: dict[str, frozenset[str]] = {} 620 event_dates = { 621 day 622 for events in investment_events.values() 623 for day in events 624 if start.isoformat() <= day <= through 625 } 626 event_dates.update( 627 str(row[0]) 628 for row in conn.execute( 629 "SELECT DISTINCT balance_date FROM cash_account_snapshots WHERE balance_date BETWEEN ? AND ?", 630 (start.isoformat(), through), 631 ).fetchall() 632 ) 633 for account in bank_accounts: 634 movement_evidence = _authoritative_cash_movements( 635 conn, 636 account_id=account["account_id"], 637 after=start.isoformat(), 638 through=through, 639 ) 640 event_dates.update(movement_evidence["days"]) 641 cursor = start 642 while cursor <= reference: 643 day = cursor.isoformat() 644 components: list[dict[str, Any]] = [] 645 qualities: list[str] = [] 646 missing_investment: list[str] = [] 647 known_total = Decimal("0") 648 for key in ("postfinance", "truewealth", "crypto", "other_assets"): 649 selected = _event_at_or_before(investment_events[key], day) 650 value = selected["value"] if selected else None 651 quality = selected["quality"] if selected else "unavailable" 652 if value is not None: 653 known_total += value 654 qualities.append(quality) 655 elif expected_investment[key]: 656 missing_investment.append(key) 657 components.append( 658 { 659 "key": key, 660 "label": COMPONENT_LABELS[key], 661 "value_chf": _money(value), 662 "quality": quality, 663 "source_date": selected["source_date"] if selected else None, 664 } 665 ) 666 667 bank_total = Decimal("0") 668 bank_qualities: list[str] = [] 669 unknown_on_day: list[str] = [] 670 bank_source_days: list[str] = [] 671 for account in bank_accounts: 672 evidence = effective_cash_evidence( 673 conn, account_id=account["account_id"], as_of=day 674 ) 675 if evidence["value_chf"] is None: 676 unknown_on_day.append(account["account_id"]) 677 continue 678 bank_total += Decimal(evidence["value_chf"]) 679 bank_qualities.append(str(evidence["quality"])) 680 if evidence["source_date"]: 681 bank_source_days.append(str(evidence["source_date"])) 682 if bank_qualities: 683 bank_quality = ( 684 "modelled" 685 if "modelled" in bank_qualities 686 else "carried" 687 if "carried" in bank_qualities 688 else "confirmed" 689 ) 690 known_total += bank_total 691 qualities.append(bank_quality) 692 bank_value = _money(bank_total) 693 else: 694 bank_quality = "unavailable" 695 bank_value = None 696 components.append( 697 { 698 "key": "bank_cash", 699 "label": COMPONENT_LABELS["bank_cash"], 700 "value_chf": bank_value, 701 "quality": bank_quality, 702 "source_date": min(bank_source_days, default=None), 703 } 704 ) 705 if not qualities: 706 cursor += timedelta(days=1) 707 continue 708 has_confirmed_anchor = any( 709 component["quality"] == "confirmed" and component["source_date"] == day 710 for component in components 711 ) 712 has_modelled_value = (latest_anchor is None or cursor > latest_anchor) and any( 713 component["quality"] == "modelled" and component["source_date"] == day 714 for component in components 715 ) 716 point_quality = ( 717 "incomplete" 718 if unknown_on_day or missing_investment 719 else "modelled" 720 if "modelled" in qualities and (latest_anchor is None or cursor > latest_anchor) 721 else "carried" 722 if "carried" in qualities 723 else "confirmed" 724 ) 725 points.append( 726 { 727 "date": day, 728 "value_chf": _money(known_total) or "0.00", 729 "quality": point_quality, 730 "has_confirmed_anchor": has_confirmed_anchor, 731 "has_modelled_value": has_modelled_value, 732 "components": components, 733 "excluded_account_count": len(unknown_on_day) 734 + len(missing_investment), 735 } 736 ) 737 unknown_identity_by_day[day] = frozenset( 738 [f"bank:{account_id}" for account_id in unknown_on_day] 739 + [f"component:{key}" for key in missing_investment] 740 ) 741 cursor += timedelta(days=1) 742 743 current_point = points[-1] if points else None 744 current_unknown = [] 745 for index, account in enumerate(bank_accounts, start=1): 746 evidence = effective_cash_evidence( 747 conn, account_id=account["account_id"], as_of=through 748 ) 749 if evidence["value_chf"] is None: 750 current_unknown.append( 751 { 752 "key": f"unknown-bank-{index}", 753 "label": _safe_bank_label(account["label"]), 754 "reason_code": "confirmed_cash_evidence_missing", 755 } 756 ) 757 for key in ("postfinance", "truewealth", "crypto", "other_assets"): 758 if expected_investment[key] and not _event_at_or_before( 759 investment_events[key], through 760 ): 761 current_unknown.append( 762 { 763 "key": f"unknown-component-{key}", 764 "label": COMPONENT_LABELS[key], 765 "reason_code": "stored_valuation_evidence_missing", 766 } 767 ) 768 769 anchor_point = None 770 if latest_anchor: 771 anchor_point = next( 772 (point for point in points if point["date"] == latest_anchor.isoformat()), 773 None, 774 ) 775 anchor = ( 776 { 777 "date": anchor_point["date"], 778 "value_chf": anchor_point["value_chf"], 779 "quality": anchor_point["quality"], 780 } 781 if anchor_point 782 else None 783 ) 784 comparable_baseline = points[0] if points else None 785 if current_point: 786 current_unknown_identity = unknown_identity_by_day.get( 787 str(current_point["date"]), frozenset() 788 ) 789 current_known_keys = { 790 str(item["key"]) 791 for item in current_point["components"] 792 if item["value_chf"] is not None 793 } 794 comparable_baseline = next( 795 ( 796 point 797 for point in points 798 if { 799 str(item["key"]) 800 for item in point["components"] 801 if item["value_chf"] is not None 802 } 803 >= current_known_keys 804 and unknown_identity_by_day.get( 805 str(point["date"]), frozenset() 806 ) 807 == current_unknown_identity 808 ), 809 comparable_baseline, 810 ) 811 baseline_point = ( 812 anchor_point 813 if period == "since_anchor" and anchor_point is not None 814 else comparable_baseline 815 ) 816 baseline = ( 817 { 818 "date": baseline_point["date"], 819 "value_chf": baseline_point["value_chf"], 820 "quality": baseline_point["quality"], 821 } 822 if baseline_point 823 else None 824 ) 825 current = ( 826 { 827 "date": current_point["date"], 828 "value_chf": current_point["value_chf"], 829 "quality": current_point["quality"], 830 } 831 if current_point 832 else None 833 ) 834 baseline_value = Decimal(baseline["value_chf"]) if baseline else None 835 change = ( 836 Decimal(current["value_chf"]) - baseline_value 837 if current and baseline_value is not None 838 else None 839 ) 840 change_pct = None 841 if ( 842 change is not None 843 and baseline_value is not None 844 and baseline_value != Decimal("0") 845 ): 846 change_pct = change / baseline_value * Decimal("100") 847 chart_visible = len(event_dates) >= 2 and len(points) >= 2 848 component_summaries: list[dict[str, Any]] = [] 849 if current_point: 850 baseline_components = { 851 str(item["key"]): item for item in (baseline_point or {}).get("components", []) 852 } 853 for item in current_point["components"]: 854 key = str(item["key"]) 855 opening = baseline_components.get(key) 856 current_value = ( 857 Decimal(str(item["value_chf"])) 858 if item["value_chf"] is not None 859 else None 860 ) 861 opening_value = ( 862 Decimal(str(opening["value_chf"])) 863 if opening and opening["value_chf"] is not None 864 else None 865 ) 866 component_change = ( 867 current_value - opening_value 868 if current_value is not None and opening_value is not None 869 else None 870 ) 871 component_change_pct = ( 872 component_change / opening_value * Decimal("100") 873 if component_change is not None 874 and opening_value is not None 875 and opening_value != Decimal("0") 876 else None 877 ) 878 quality = str(item["quality"]) 879 if key == "bank_cash" and current_unknown: 880 quality = "incomplete" 881 component_summaries.append( 882 { 883 "key": key, 884 "label": str(item["label"]), 885 "current_value_chf": _money(current_value), 886 "change_chf": _money(component_change), 887 "change_pct": format(component_change_pct.quantize(PERCENT), "f") 888 if component_change_pct is not None 889 else None, 890 "quality": quality, 891 "as_of": item["source_date"], 892 "unknown_account_count": sum( 893 1 894 for unknown in current_unknown 895 if ( 896 str(unknown["key"]).startswith("unknown-bank-") 897 if key == "bank_cash" 898 else unknown["key"] == f"unknown-component-{key}" 899 ) 900 ), 901 } 902 ) 903 return { 904 "status": "available" if points else "unavailable", 905 "period": { 906 "preset": period, 907 "from": start.isoformat(), 908 "to": through, 909 }, 910 "last_confirmed_anchor_date": latest_anchor.isoformat() if latest_anchor else None, 911 "anchor": anchor, 912 "baseline": baseline, 913 "current": current, 914 "change_chf": _money(change), 915 "change_pct": format(change_pct.quantize(PERCENT), "f") 916 if change_pct is not None 917 else None, 918 "chart_visible": chart_visible, 919 "points": points, 920 "components": component_summaries, 921 "correction_markers": sorted( 922 [ 923 marker 924 for marker in pf_markers + tw_markers + _manual_cash_correction_markers(conn, through=through) 925 if start.isoformat() <= marker["date"] <= through 926 ], 927 key=lambda item: (item["date"], item["source_key"]), 928 ), 929 "unknown_accounts": current_unknown, 930 "method": "modelled_wealth_daily_v1", 931 "disclaimer": "Geschätzte Entwicklung aus bestätigten Ankern, gespeicherten Tagesbewertungen und fortgeschriebenen bekannten Salden; keine verifizierte TTWROR oder XIRR.", 932 } ===== src/jarvis_finance/services/wealth_cockpit.py ===== 1 from __future__ import annotations 2 3 from collections import defaultdict 4 from datetime import UTC, date, datetime, timedelta 5 from decimal import Decimal 6 from sqlite3 import Connection 7 from typing import Any, cast 8 9 from fastapi import HTTPException 10 11 from jarvis_finance.ledger.performance import effective_activities, external_cashflow 12 from jarvis_finance.quality.freshness import ( 13 FreshnessStatus, 14 assess_freshness, 15 combined_freshness, 16 ) 17 from jarvis_finance.services.budget_planning import get_annual_budget_assistant 18 from jarvis_finance.services.cash_service import get_cash_summary 19 from jarvis_finance.services.crypto_service import list_crypto_positions 20 from jarvis_finance.services.equity_service import get_equity_summary 21 from jarvis_finance.services.household_import import source_reference_hash 22 from jarvis_finance.services.modelled_wealth import build_modelled_wealth_development 23 from jarvis_finance.services.portfolio_analysis_v1 import build_portfolio_analysis_v1 24 from jarvis_finance.services.portfolio_performance import ( 25 build_performance_coverage, 26 build_portfolio_performance, 27 load_scope_activities, 28 ) 29 from jarvis_finance.services.portfolio_policy import active_policy 30 from jarvis_finance.services.reconciliation_snapshot import ( 31 build_reconciliation_snapshot, 32 safe_account_label, 33 ) 34 35 MONEY = Decimal("0.01") 36 PERIODS = { 37 "since_anchor", 38 "1m", 39 "3m", 40 "1y", 41 "ytd", 42 "previous_year", 43 "12m", 44 "all", 45 } 46 KNOWN_PERFORMANCE_ROLES = { 47 "postfinance_etrading_depot", 48 "postfinance_etrading_cash", 49 "canonical_truewealth_total_value", 50 "crypto_portfolio", 51 } 52 53 54 def _money(value: Decimal | None) -> str | None: 55 return None if value is None else format(value.quantize(MONEY), "f") 56 57 58 def _decimal(value: object) -> Decimal | None: 59 if value in (None, ""): 60 return None 61 return Decimal(str(value)) 62 63 64 def _latest_data_cutoff(conn: Connection) -> str: 65 candidates: list[str] = [] 66 for table, column in ( 67 ("transactions", "created_at"), 68 ("portfolio_valuation_snapshots", "captured_at"), 69 ("account_value_snapshots", "created_at"), 70 ("cash_account_snapshots", "created_at"), 71 ("crypto_prices", "fetched_at"), 72 ("portfolio_analysis_snapshots", "created_at"), 73 ): 74 row = conn.execute(f"SELECT MAX({column}) FROM {table}").fetchone() 75 if row and row[0]: 76 candidates.append(str(row[0])) 77 return max(candidates, default="1970-01-01T00:00:00Z") 78 79 80 def _latest_valuation_date(conn: Connection, fallback: date) -> date: 81 row = conn.execute( 82 """SELECT MAX(day) FROM ( 83 SELECT substr(valuation_at,1,10) day FROM portfolio_valuation_snapshots 84 UNION ALL SELECT valuation_date FROM account_value_snapshots 85 UNION ALL SELECT balance_date FROM cash_account_snapshots 86 )""" 87 ).fetchone() 88 if not row or not row[0]: 89 return fallback 90 return min(date.fromisoformat(str(row[0])[:10]), fallback) 91 92 93 def _earliest_evidence_date(conn: Connection, fallback: date) -> date: 94 row = conn.execute( 95 """SELECT MIN(day) FROM ( 96 SELECT substr(valuation_at,1,10) day FROM portfolio_valuation_snapshots 97 UNION ALL SELECT valuation_date FROM account_value_snapshots 98 UNION ALL SELECT balance_date FROM cash_account_snapshots 99 UNION ALL SELECT trade_date FROM transactions 100 )""" 101 ).fetchone() 102 return date.fromisoformat(str(row[0])[:10]) if row and row[0] else fallback 103 104 105 def period_bounds(conn: Connection, *, period: str, as_of: date) -> tuple[date, date]: 106 if period not in PERIODS: 107 raise ValueError( 108 "Zeitraum muss since_anchor, 1m, 3m, 1y, ytd, previous_year, 12m oder all sein" 109 ) 110 if period == "since_anchor": 111 row = conn.execute( 112 """SELECT MAX(day) FROM ( 113 SELECT valuation_date day FROM account_value_snapshots 114 WHERE COALESCE(is_active,1)=1 AND updated_at IS NULL 115 UNION ALL SELECT balance_date FROM cash_account_snapshots 116 ) WHERE day<=?""", 117 (as_of.isoformat(),), 118 ).fetchone() 119 return ( 120 date.fromisoformat(str(row[0])) if row and row[0] else as_of, 121 as_of, 122 ) 123 if period == "1m": 124 return as_of - timedelta(days=31), as_of 125 if period == "3m": 126 return as_of - timedelta(days=93), as_of 127 if period in {"1y", "12m"}: 128 try: 129 start = as_of.replace(year=as_of.year - 1) 130 except ValueError: 131 start = as_of.replace(year=as_of.year - 1, day=28) 132 return start, as_of 133 if period == "ytd": 134 return date(as_of.year, 1, 1), as_of 135 if period == "previous_year": 136 return date(as_of.year - 1, 1, 1), date(as_of.year - 1, 12, 31) 137 return _earliest_evidence_date(conn, as_of - timedelta(days=1)), as_of 138 139 140 def _account_ids(conn: Connection, *, investment_only: bool) -> list[str]: 141 if investment_only: 142 rows = conn.execute( 143 """SELECT a.account_id FROM accounts a 144 JOIN performance_scope_classifications psc ON psc.account_id=a.account_id 145 WHERE a.is_active=1 AND psc.included=1 146 AND psc.decision_version='investment_performance_scope_v1' 147 ORDER BY a.account_id""" 148 ).fetchall() 149 else: 150 rows = conn.execute( 151 """SELECT account_id FROM accounts 152 WHERE is_active=1 AND account_type<>'credit_card_liability' 153 ORDER BY account_id""" 154 ).fetchall() 155 return [str(row[0]) for row in rows] 156 157 158 def scope_cashflows( 159 conn: Connection, 160 *, 161 account_ids: list[str], 162 from_date: str, 163 to_date: str, 164 data_cutoff: str, 165 ) -> list[dict[str, str]]: 166 """Reuse canonical activity and transfer-boundary semantics for one ownership scope.""" 167 activities = load_scope_activities( 168 conn, 169 account_ids=account_ids, 170 to_date=to_date, 171 data_cutoff=data_cutoff, 172 base_currency="CHF", 173 ) 174 effective, _ = effective_activities(activities) 175 events: list[dict[str, str]] = [] 176 for activity in effective: 177 if not (from_date <= activity.occurred_at[:10] <= to_date): 178 continue 179 amount = external_cashflow(activity, "CHF") 180 if amount is None or activity.kind not in {"external_deposit", "external_withdrawal"}: 181 continue 182 events.append( 183 { 184 "at": activity.occurred_at, 185 "kind": activity.kind, 186 "amount_chf": _money(amount) or "0.00", 187 } 188 ) 189 return sorted(events, key=lambda item: (item["at"], item["kind"], item["amount_chf"])) 190 191 192 def _truewealth(conn: Connection) -> tuple[Decimal | None, str | None]: 193 row = conn.execute( 194 """SELECT avs.total_value_chf,avs.valuation_date 195 FROM account_value_snapshots avs 196 JOIN performance_scope_classifications psc ON psc.account_id=avs.account_id 197 WHERE psc.included=1 198 AND psc.classification_role='canonical_truewealth_total_value' 199 AND COALESCE(avs.is_active,1)=1 AND avs.updated_at IS NULL 200 AND avs.source_type<>'truewealth_manual_provisional' 201 ORDER BY avs.valuation_date DESC, 202 CASE avs.source_type WHEN 'truewealth_official_import' THEN 3 ELSE 1 END DESC, 203 COALESCE(avs.valuation_at,avs.created_at) DESC,avs.snapshot_id DESC LIMIT 1""" 204 ).fetchone() 205 return (Decimal(str(row[0])), str(row[1])) if row else (None, None) 206 207 208 def _unassigned_values(conn: Connection) -> tuple[Decimal, list[dict[str, str]]]: 209 rows = conn.execute( 210 """WITH ranked AS ( 211 SELECT avs.account_id,avs.total_value_chf,avs.valuation_date,a.account_name, 212 ROW_NUMBER() OVER (PARTITION BY avs.account_id ORDER BY avs.valuation_date DESC,avs.created_at DESC,avs.snapshot_id DESC) rn 213 FROM account_value_snapshots avs 214 JOIN accounts a ON a.account_id=avs.account_id AND a.is_active=1 215 LEFT JOIN performance_scope_classifications psc ON psc.account_id=a.account_id 216 WHERE COALESCE(avs.is_active,1)=1 AND avs.updated_at IS NULL 217 AND a.account_type NOT IN ('cash','credit_card_liability') 218 AND COALESCE(psc.classification_role,'') NOT IN ( 219 'postfinance_etrading_depot','postfinance_etrading_cash', 220 'canonical_truewealth_total_value','crypto_portfolio') 221 ) SELECT account_name,total_value_chf,valuation_date FROM ranked WHERE rn=1 ORDER BY account_name""" 222 ).fetchall() 223 items = [ 224 {"label": safe_account_label(str(row[0])), "value_chf": _money(Decimal(str(row[1]))) or "0.00", "as_of": str(row[2])} 225 for row in rows 226 ] 227 return sum((Decimal(item["value_chf"]) for item in items), Decimal("0")), items 228 229 230 def _household_import_meta( 231 conn: Connection, 232 profile: str, 233 *, 234 canonical_account_id: str | None = None, 235 ) -> dict[str, Any]: 236 source_type = { 237 "akb": "akb_bank", 238 "raiffeisen": "raiffeisen_bank", 239 "viseca_one": "visa_credit_card", 240 "migros_receipts": "migros_receipts", 241 }.get(profile, profile) 242 unavailable = { 243 "imported_at": None, 244 "coverage_from": None, 245 "coverage_to": None, 246 "coverage_status": "unavailable", 247 "new_rows": 0, 248 "duplicate_rows": 0, 249 "review_rows": 0, 250 } 251 if canonical_account_id: 252 budget_ids = { 253 str(row[0]) 254 for row in conn.execute( 255 """SELECT budget_account_id FROM budget_accounts 256 WHERE linked_account_id=? AND is_active=1""", 257 (canonical_account_id,), 258 ).fetchall() 259 } 260 mapping_hashes = { 261 str(row[0]) 262 for row in conn.execute( 263 """SELECT source_reference_hash 264 FROM household_account_source_mappings 265 WHERE canonical_account_id=? AND source_type=? AND is_active=1""", 266 (canonical_account_id, source_type), 267 ).fetchall() 268 } 269 candidates = conn.execute( 270 """SELECT c.transaction_candidate_id,c.transaction_date,c.status, 271 c.household_batch_id,c.account_source,c.confirmed_transaction_id, 272 bt.account_id confirmed_budget_account_id,b.confirmed_at 273 FROM budget_transaction_candidates c 274 LEFT JOIN budget_transactions bt 275 ON bt.budget_transaction_id=c.confirmed_transaction_id 276 LEFT JOIN household_import_batches b 277 ON b.batch_id=c.household_batch_id 278 WHERE c.source_type=? AND c.household_batch_id IS NOT NULL 279 ORDER BY COALESCE(b.confirmed_at,c.created_at),c.transaction_date, 280 c.transaction_candidate_id""", 281 (source_type,), 282 ).fetchall() 283 attributable = [] 284 for candidate in candidates: 285 bound_by_transaction = str( 286 candidate["confirmed_budget_account_id"] or "" 287 ) in budget_ids 288 bound_by_mapping = False 289 source_reference = str(candidate["account_source"] or "").strip() 290 if source_reference and mapping_hashes: 291 try: 292 bound_by_mapping = ( 293 source_reference_hash(source_reference) in mapping_hashes 294 ) 295 except HTTPException: 296 # API tests and offline fixtures may intentionally omit the 297 # private fingerprint key. Confirmed canonical lineage still 298 # remains usable; weak source text never becomes identity. 299 bound_by_mapping = False 300 if bound_by_transaction or bound_by_mapping: 301 attributable.append(candidate) 302 if not attributable: 303 return unavailable 304 latest_batch = max( 305 attributable, 306 key=lambda row: ( 307 str(row["confirmed_at"] or ""), str(row["household_batch_id"]) 308 ), 309 )["household_batch_id"] 310 selected = [ 311 row for row in attributable if row["household_batch_id"] == latest_batch 312 ] 313 batch = conn.execute( 314 """SELECT confirmed_at FROM household_import_batches WHERE batch_id=?""", 315 (latest_batch,), 316 ).fetchone() 317 days = [str(row["transaction_date"]) for row in selected] 318 review_rows = sum(str(row["status"]) == "needs_review" for row in selected) 319 return { 320 "imported_at": batch["confirmed_at"] if batch else None, 321 "coverage_from": min(days, default=None), 322 "coverage_to": max(days, default=None), 323 "coverage_status": "partial", 324 "new_rows": len(selected), 325 # Duplicates cannot be assigned to an account without durable 326 # source-row lineage. Report zero rather than inheriting a provider 327 # count from a sibling account. 328 "duplicate_rows": 0, 329 "review_rows": review_rows, 330 } 331 332 row = conn.execute( 333 """SELECT f.batch_id,f.row_count,f.period_start,f.period_end,f.created_at,b.confirmed_at 334 FROM household_import_files f 335 JOIN household_import_batches b ON b.batch_id=f.batch_id 336 WHERE f.source_type=? ORDER BY b.confirmed_at DESC,f.created_at DESC LIMIT 1""", 337 (source_type,), 338 ).fetchone() 339 if not row: 340 return unavailable 341 imported_rows = int( 342 conn.execute( 343 "SELECT COUNT(*) FROM household_import_items WHERE batch_id=? AND source_type=?", 344 (row["batch_id"], source_type), 345 ).fetchone()[0] 346 ) 347 review_rows = int( 348 conn.execute( 349 "SELECT COUNT(*) FROM budget_transaction_candidates WHERE household_batch_id=? AND source_type=? AND status='needs_review'", 350 (row["batch_id"], source_type), 351 ).fetchone()[0] 352 ) 353 row_count = int(row["row_count"]) 354 return { 355 "imported_at": row["confirmed_at"] or row["created_at"], 356 "coverage_from": row["period_start"], 357 "coverage_to": row["period_end"], 358 "coverage_status": "complete" 359 if row["period_start"] and row["period_end"] 360 else "partial", 361 "new_rows": imported_rows, 362 "duplicate_rows": max(row_count - imported_rows, 0), 363 "review_rows": review_rows, 364 } 365 366 367 def _postfinance_import_meta(conn: Connection) -> dict[str, Any]: 368 row = conn.execute( 369 """SELECT b.confirmed_at,b.activity_coverage_from,b.activity_coverage_to, 370 b.performance_coverage_complete,s.snapshot_at 371 FROM postfinance_import_batches b JOIN postfinance_snapshots s ON s.batch_id=b.batch_id 372 ORDER BY b.confirmed_at DESC LIMIT 1""" 373 ).fetchone() 374 if not row: 375 return {"imported_at": None, "coverage_from": None, "coverage_to": None, "coverage_status": "unavailable", "last_snapshot": None} 376 return { 377 "imported_at": row["confirmed_at"], "coverage_from": row["activity_coverage_from"], 378 "coverage_to": row["activity_coverage_to"], 379 "coverage_status": "complete" if int(row["performance_coverage_complete"]) else "partial", 380 "last_snapshot": str(row["snapshot_at"])[:10], 381 } 382 383 384 def _truewealth_import_meta(conn: Connection) -> dict[str, Any]: 385 row = conn.execute( 386 """SELECT confirmed_at,period_from,period_to,external_cashflows_complete,snapshot_date 387 FROM truewealth_import_batches ORDER BY confirmed_at DESC LIMIT 1""" 388 ).fetchone() 389 if not row: 390 return {"imported_at": None, "coverage_from": None, "coverage_to": None, "coverage_status": "unavailable", "last_snapshot": None} 391 return { 392 "imported_at": row["confirmed_at"], "coverage_from": row["period_from"], "coverage_to": row["period_to"], 393 "coverage_status": "complete" if int(row["external_cashflows_complete"]) else "partial", 394 "last_snapshot": row["snapshot_date"], 395 } 396 397 398 def _household_history( 399 conn: Connection, 400 *, 401 from_date: date, 402 to_date: date, 403 current: dict[str, Any], 404 as_of: date, 405 ) -> tuple[list[dict[str, str]], str]: 406 """Return only dates with a complete exact stored value for every known account. 407 408 No carry-forward or interpolation is allowed. Cash snapshots, canonical account 409 valuations and existing performance valuations remain separate source contracts. 410 """ 411 classified = conn.execute( 412 """SELECT a.account_id,a.account_type,psc.classification_role 413 FROM accounts a 414 LEFT JOIN performance_scope_classifications psc 415 ON psc.account_id=a.account_id AND psc.included=1 416 AND psc.decision_version='investment_performance_scope_v1' 417 WHERE a.is_active=1 AND a.account_type<>'credit_card_liability' 418 ORDER BY a.account_id""" 419 ).fetchall() 420 required: list[tuple[str, str]] = [] 421 roles: set[str] = set() 422 for row in classified: 423 account_id, account_type, role = str(row[0]), str(row[1]), str(row[2] or "") 424 if role: 425 roles.add(role) 426 if role in {"postfinance_etrading_depot", "postfinance_etrading_cash", "crypto_portfolio"}: 427 required.append((account_id, "performance")) 428 elif role == "canonical_truewealth_total_value": 429 required.append((account_id, "account_value")) 430 elif account_type == "cash": 431 required.append((account_id, "cash")) 432 elif not role and conn.execute( 433 "SELECT 1 FROM account_value_snapshots WHERE account_id=? AND COALESCE(is_active,1)=1 AND updated_at IS NULL LIMIT 1", 434 (account_id,), 435 ).fetchone(): 436 required.append((account_id, "account_value")) 437 438 distribution = {str(row["key"]): row["value_chf"] for row in current["distribution"]} 439 if Decimal(str(distribution.get("equity") or "0")) and "postfinance_etrading_depot" not in roles: 440 return [], "Für Aktien und ETFs fehlt eine kanonische historische Kontobewertung." 441 if Decimal(str(distribution.get("crypto") or "0")) and "crypto_portfolio" not in roles: 442 return [], "Für Kryptowährungen fehlt eine kanonische historische Portfoliobewertung." 443 if distribution.get("truewealth") is not None and "canonical_truewealth_total_value" not in roles: 444 return [], "Für True Wealth fehlt die kanonische historische Gesamtwertreihe." 445 if not required: 446 return [], "Es liegen noch keine gemeinsamen historischen Kontobewertungen vor." 447 448 series: list[dict[str, Decimal]] = [] 449 for account_id, source_kind in required: 450 if source_kind == "performance": 451 rows = conn.execute( 452 """WITH ranked AS ( 453 SELECT substr(valuation_at,1,10) day, 454 CASE 455 WHEN currency=base_currency THEN value_original 456 WHEN fx_rate_to_base IS NOT NULL 457 THEN CAST(value_original AS NUMERIC)*CAST(fx_rate_to_base AS NUMERIC) 458 END value_base, 459 ROW_NUMBER() OVER (PARTITION BY substr(valuation_at,1,10) 460 ORDER BY captured_at DESC,snapshot_id DESC) rn 461 FROM portfolio_valuation_snapshots 462 WHERE scope_kind='account' AND scope_id=? AND base_currency='CHF' 463 AND quality_status IN ('complete','ok') 464 AND substr(valuation_at,1,10) BETWEEN ? AND ? 465 ) SELECT day,value_base FROM ranked WHERE rn=1 AND value_base IS NOT NULL""", 466 (account_id, from_date.isoformat(), to_date.isoformat()), 467 ).fetchall() 468 elif source_kind == "account_value": 469 rows = conn.execute( 470 """WITH ranked AS ( 471 SELECT valuation_date day,total_value_chf, 472 ROW_NUMBER() OVER (PARTITION BY valuation_date ORDER BY 473 CASE source_type WHEN 'truewealth_official_import' THEN 3 ELSE 1 END DESC, 474 COALESCE(valuation_at,created_at) DESC,snapshot_id DESC) rn 475 FROM account_value_snapshots 476 WHERE account_id=? AND COALESCE(is_active,1)=1 AND updated_at IS NULL 477 AND source_type<>'truewealth_manual_provisional' 478 AND valuation_date BETWEEN ? AND ? 479 ) SELECT day,total_value_chf FROM ranked WHERE rn=1 AND total_value_chf IS NOT NULL""", 480 (account_id, from_date.isoformat(), to_date.isoformat()), 481 ).fetchall() 482 else: 483 rows = conn.execute( 484 """WITH ranked AS ( 485 SELECT balance_date day,amount_chf, 486 ROW_NUMBER() OVER (PARTITION BY balance_date ORDER BY 487 CASE snapshot_type WHEN 'reconciliation' THEN 4 WHEN 'manual_balance' THEN 3 488 WHEN 'csv_anchor_balance' THEN 2 ELSE 1 END DESC, 489 created_at DESC,snapshot_id DESC) rn 490 FROM cash_account_snapshots 491 WHERE account_id=? AND balance_date BETWEEN ? AND ? 492 ) SELECT day,amount_chf FROM ranked WHERE rn=1 AND amount_chf IS NOT NULL""", 493 (account_id, from_date.isoformat(), to_date.isoformat()), 494 ).fetchall() 495 series.append({str(row[0]): Decimal(str(row[1])) for row in rows}) 496 497 complete_dates = set(series[0]) 498 for values in series[1:]: 499 complete_dates.intersection_update(values) 500 points = [ 501 { 502 "at": day, 503 "value_chf": _money(sum((values[day] for values in series), Decimal("0"))) or "0.00", 504 } 505 for day in sorted(complete_dates) 506 ] 507 if to_date == as_of and current["complete"] and (not points or points[-1]["at"] != as_of.isoformat()): 508 points.append({"at": as_of.isoformat(), "value_chf": _money(current["total"]) or "0.00"}) 509 reason = ( 510 "Nur Stichtage mit vollständigen gespeicherten Bewertungen werden verbunden; Datenlücken werden nicht ergänzt." 511 if len(points) >= 2 512 else "Für einen Verlauf fehlen mindestens zwei gemeinsame vollständige Stichtage; Zwischenwerte werden nicht erfunden." 513 ) 514 return points, reason 515 516 517 def _current_values(conn: Connection, *, as_of: date) -> dict[str, Any]: 518 cash = get_cash_summary(conn) 519 equity = get_equity_summary(conn) 520 crypto_positions = list_crypto_positions(conn) 521 truewealth, truewealth_as_of = _truewealth(conn) 522 unassigned, unassigned_items = _unassigned_values(conn) 523 524 cash_known = [ 525 item 526 for item in cash.positions 527 if item.amount_chf is not None 528 and (item.last_manual_reconciliation or item.last_imported_booking) 529 ] 530 cash_value = sum((Decimal(item.amount_chf or "0") for item in cash_known), Decimal("0")) 531 equity_value = Decimal(equity.valued_partial_chf) 532 crypto_known = [item for item in crypto_positions if item.market_value_chf is not None] 533 crypto_value = sum((Decimal(item.market_value_chf or "0") for item in crypto_known), Decimal("0")) 534 complete = ( 535 equity.coverage_complete 536 and len(crypto_known) == len(crypto_positions) 537 and len(cash_known) == len(cash.positions) 538 and truewealth is not None 539 ) 540 known_equity_positions = int( 541 getattr(equity, "valued_positions", 1 if equity_value else 0) 542 ) 543 equity_positions = int( 544 getattr( 545 equity, 546 "total_positions", 547 known_equity_positions + int(equity.unvalued_positions), 548 ) 549 ) 550 equity_display = ( 551 None 552 if not equity_positions or (equity.unvalued_positions and not equity_value) 553 else _money(equity_value) 554 ) 555 crypto_display = ( 556 None if not crypto_positions or (not crypto_known and crypto_positions) else _money(crypto_value) 557 ) 558 distribution = [ 559 { 560 "key": "cash", 561 "label": "Bankguthaben", 562 "value_chf": _money(cash_value) if cash_known else None, 563 }, 564 {"key": "equity", "label": "Aktien und ETFs", "value_chf": equity_display}, 565 {"key": "truewealth", "label": "True Wealth", "value_chf": _money(truewealth)}, 566 {"key": "crypto", "label": "Kryptowährungen", "value_chf": crypto_display}, 567 ] 568 if unassigned or unassigned_items: 569 distribution.append( 570 {"key": "other", "label": "Nicht zugeordnet", "value_chf": _money(unassigned)} 571 ) 572 total = sum( 573 (Decimal(str(row["value_chf"])) for row in distribution if row["value_chf"] is not None), 574 Decimal("0"), 575 ) 576 577 grouped_cash: dict[tuple[str, str], dict[str, Any]] = defaultdict( 578 lambda: { 579 "value": Decimal("0"), 580 "known": 0, 581 "total": 0, 582 "dates": [], 583 "statuses": [], 584 "balance_modes": [], 585 "account_ids": set(), 586 } 587 ) 588 for item in cash.positions: 589 account_label = getattr(item, "account_label", item.platform) 590 group = grouped_cash[(item.platform, account_label)] 591 group["account_ids"].add(str(getattr(item, "account_id", ""))) 592 group["total"] += 1 593 has_value = bool( 594 item.amount_chf is not None 595 and (item.last_manual_reconciliation or item.last_imported_booking) 596 ) 597 if has_value: 598 group["value"] += Decimal(item.amount_chf or "0") 599 group["known"] += 1 600 group["dates"].extend( 601 value for value in (item.last_manual_reconciliation, item.last_imported_booking) if value 602 ) 603 group["statuses"].append(item.status) 604 group["balance_modes"].append(getattr(item, "balance_mode", "unknown")) 605 sources: list[dict[str, Any]] = [] 606 for index, ((platform, account_label), group) in enumerate( 607 sorted(grouped_cash.items()), start=1 608 ): 609 source_as_of = max(group["dates"], default=None) 610 freshness = assess_freshness( 611 available=bool(group["known"]), 612 as_of=source_as_of, 613 now=datetime.combine(as_of, datetime.max.time(), tzinfo=UTC), 614 source_kind="bank_balance", 615 ) 616 statuses = [str(status) for status in group["statuses"]] 617 sources.append( 618 { 619 "key": f"cash-{index}", 620 "_canonical_account_id": next(iter(group["account_ids"])) 621 if len(group["account_ids"]) == 1 and "" not in group["account_ids"] 622 else None, 623 "label": safe_account_label(account_label), 624 "provider_label": safe_account_label(platform), 625 "kind": "Bankguthaben", 626 "source_role": "account", 627 "performance_scope": ( 628 "postfinance" 629 if "official_components" in group["balance_modes"] 630 else None 631 ), 632 "current_value_chf": _money(group["value"]) 633 if group["known"] 634 else None, 635 "current_value_status": ( 636 "ready" 637 if group["total"] and group["known"] == group["total"] 638 else "partial" 639 if group["known"] 640 else "not_ready" 641 ), 642 "change_chf": None, 643 "net_contributions_chf": None, 644 "return_pct": None, 645 "as_of": source_as_of, 646 "freshness_status": freshness.status, 647 "freshness_reason_code": freshness.reason_code, 648 "expected_as_of": freshness.expected_as_of, 649 "reconciliation_status": ( 650 "not_assessable" 651 if group["known"] != group["total"] 652 else "difference" 653 if any(status == "Abgleich offen" for status in statuses) 654 else "reconciled" 655 if statuses 656 and all(status.startswith("Offiziell abgeglichen") for status in statuses) 657 else "not_assessable" 658 ), 659 "performance_status": "not_applicable", 660 } 661 ) 662 663 def investment_source( 664 *, 665 key: str, 666 label: str, 667 value: Decimal | None, 668 source_as_of: str | None, 669 source_kind: str, 670 source_complete: bool, 671 ) -> dict[str, Any]: 672 freshness = assess_freshness( 673 available=value is not None, 674 as_of=source_as_of, 675 now=datetime.combine(as_of, datetime.max.time(), tzinfo=UTC), 676 source_kind=source_kind, # type: ignore[arg-type] 677 ) 678 return { 679 "key": key, 680 "label": label, 681 "provider_label": label, 682 "kind": "Anlage", 683 "source_role": "canonical_value", 684 "performance_scope": { 685 "postfinance-investments": "postfinance", 686 "truewealth": "truewealth", 687 "crypto": "crypto", 688 }.get(key), 689 "current_value_chf": _money(value), 690 "current_value_status": ( 691 "ready" 692 if value is not None and source_complete 693 else "partial" 694 if value is not None 695 else "not_ready" 696 ), 697 "change_chf": None, 698 "net_contributions_chf": None, 699 "return_pct": None, 700 "as_of": source_as_of, 701 "freshness_status": freshness.status, 702 "freshness_reason_code": freshness.reason_code, 703 "expected_as_of": freshness.expected_as_of, 704 "reconciliation_status": "not_assessable", 705 "performance_status": "not_ready", 706 } 707 708 sources.extend( 709 [ 710 investment_source( 711 key="postfinance-investments", 712 label="PostFinance Aktien und ETFs", 713 value=equity_value, 714 source_as_of=equity.as_of, 715 source_kind="market", 716 source_complete=equity.coverage_complete, 717 ), 718 investment_source( 719 key="truewealth", 720 label="True Wealth Gesamtwert", 721 value=truewealth, 722 source_as_of=truewealth_as_of, 723 source_kind="managed_portfolio", 724 source_complete=truewealth is not None, 725 ), 726 investment_source( 727 key="crypto", 728 label="Kryptowährungen", 729 value=crypto_value if crypto_known else None, 730 source_as_of=max( 731 (item.last_price_update or "" for item in crypto_known), default="" 732 ) 733 or None, 734 source_kind="crypto_24_7", 735 source_complete=bool(crypto_positions) 736 and len(crypto_known) == len(crypto_positions), 737 ), 738 ] 739 ) 740 visa_account = conn.execute( 741 "SELECT account_id FROM accounts WHERE account_type='credit_card_liability' AND is_active=1 LIMIT 1" 742 ).fetchone() 743 if visa_account: 744 sources.append( 745 { 746 "key": "visa-liability", 747 "label": "VISA Kartenverbindlichkeit", 748 "provider_label": "VISA", 749 "kind": "Verbindlichkeit", 750 "source_role": "liability", 751 "performance_scope": None, 752 "current_value_chf": None, 753 "current_value_status": "partial", 754 "change_chf": None, 755 "net_contributions_chf": None, 756 "return_pct": None, 757 "as_of": None, 758 "freshness_status": "unknown", 759 "freshness_reason_code": "current_liability_snapshot_missing", 760 "expected_as_of": None, 761 "reconciliation_status": "not_assessable", 762 "performance_status": "not_applicable", 763 } 764 ) 765 if unassigned_items: 766 sources.append( 767 { 768 "key": "unassigned", 769 "label": "Weitere bestätigte Werte", 770 "kind": "Nicht zugeordnet", 771 "source_role": "canonical_value", 772 "performance_scope": None, 773 "current_value_chf": _money(unassigned), 774 "current_value_status": "ready", 775 "change_chf": None, 776 "net_contributions_chf": None, 777 "return_pct": None, 778 "as_of": max((item["as_of"] for item in unassigned_items), default=None), 779 "freshness_status": assess_freshness( 780 available=True, 781 as_of=max((item["as_of"] for item in unassigned_items), default=None), 782 now=datetime.combine(as_of, datetime.max.time(), tzinfo=UTC), 783 ).status, 784 "reconciliation_status": "not_assessable", 785 "performance_status": "not_ready", 786 } 787 ) 788 data_dates = [str(source["as_of"]) for source in sources if source["as_of"]] 789 return { 790 "total": total, 791 "investments": equity_value + (truewealth or Decimal("0")) + crypto_value + unassigned, 792 "cash": cash_value, 793 "complete": complete, 794 "distribution": distribution, 795 "sources": sources, 796 "data_as_of": max(data_dates, default=None), 797 "unpriced_count": equity.unvalued_positions + len(crypto_positions) - len(crypto_known), 798 "missing_cash_count": sum( 799 1 for group in grouped_cash.values() if group["known"] != group["total"] 800 ), 801 "unassigned_items": unassigned_items, 802 } 803 804 805 def _policy_comparison(conn: Connection, current: dict[str, Any]) -> dict[str, Any]: 806 loaded = active_policy(conn) 807 policy = loaded.get("policy") if loaded.get("configured") else None 808 if not policy: 809 return {"configured": False, "version": None, "rows": [], "contribution": None} 810 values = {str(row["key"]): Decimal(str(row["value_chf"] or "0")) for row in current["distribution"]} 811 by_policy = { 812 "cash": values.get("cash", Decimal("0")), 813 "equity": values.get("equity", Decimal("0")), 814 "crypto": values.get("crypto", Decimal("0")), 815 "other": values.get("truewealth", Decimal("0")) + values.get("other", Decimal("0")), 816 } 817 rows = [] 818 for allocation in policy["allocations"]: 819 asset = str(allocation["asset_class"]) 820 current_pct = by_policy.get(asset, Decimal("0")) / current["total"] * Decimal("100") if current["total"] and current["complete"] else None 821 lower = Decimal(str(allocation["lower_pct"])) 822 upper = Decimal(str(allocation["upper_pct"])) 823 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" 824 rows.append( 825 { 826 "asset_class": asset, 827 "current_pct": _money(current_pct), 828 "target_pct": str(allocation["target_pct"]), 829 "lower_pct": str(allocation["lower_pct"]), 830 "upper_pct": str(allocation["upper_pct"]), 831 "deviation_pct_points": _money(current_pct - Decimal(str(allocation["target_pct"]))) if current_pct is not None else None, 832 "status": status, 833 } 834 ) 835 monthly = _decimal(policy.get("monthly_contribution")) 836 return { 837 "configured": True, 838 "version": policy["version"], 839 "rows": rows, 840 "contribution": {"monthly_target_chf": _money(monthly), "annual_target_chf": _money(monthly * Decimal("12"))} if monthly is not None else None, 841 } 842 843 844 def _readiness_status(value: str) -> str: 845 return { 846 "complete": "ready", 847 "available": "ready", 848 "partial": "partial", 849 "unavailable": "not_ready", 850 "not_calculable": "not_ready", 851 }.get(value, "not_ready") 852 853 854 def _join_labels(labels: list[str]) -> str: 855 unique = list(dict.fromkeys(label for label in labels if label)) 856 if not unique: 857 return "keine Quelle" 858 if len(unique) == 1: 859 return unique[0] 860 return ", ".join(unique[:-1]) + " und " + unique[-1] 861 862 863 def _performance_scope_status(row: dict[str, Any] | None) -> str: 864 if not row: 865 return "not_ready" 866 statuses = { 867 str(row.get("ttwror_status")), 868 str(row.get("xirr_status")), 869 str(row.get("attribution_status")), 870 } 871 if statuses and all(status == "complete" for status in statuses): 872 return "ready" 873 if any(status in {"complete", "partial"} for status in statuses): 874 return "partial" 875 return "not_ready" 876 877 878 def _build_diagnostics( 879 *, 880 current: dict[str, Any], 881 coverage: dict[str, Any], 882 policy: dict[str, Any], 883 period: dict[str, str], 884 ) -> list[dict[str, Any]]: 885 rows = { 886 str(row["scope"]): row 887 for row in coverage.get("rows", []) 888 if str(row.get("scope")) != "portfolio" 889 } 890 scope_labels = { 891 "postfinance": "PostFinance", 892 "truewealth": "True Wealth", 893 "crypto": "Kryptowährungen", 894 } 895 diagnostics: list[dict[str, Any]] = [] 896 opening_sources = [ 897 scope_labels[scope] 898 for scope in ("postfinance", "truewealth") 899 if not rows.get(scope, {}).get("valuation_from") 900 or str(rows[scope]["valuation_from"]) > period["from"] 901 ] 902 if opening_sources: 903 diagnostics.append( 904 { 905 "dimension": "performance", 906 "affected_sources": opening_sources, 907 "message": ( 908 f"Für die Rendite {period['from'][:4]} fehlen Anfangsbewertungen bei " 909 f"{_join_labels(opening_sources)}." 910 ), 911 "action": "Bestätigte Gesamtwerte zum Periodenbeginn bereitstellen.", 912 "reason_code": "opening_valuation_missing_by_source", 913 "prominent": True, 914 } 915 ) 916 closing_sources = [ 917 scope_labels[scope] 918 for scope in ("postfinance", "truewealth") 919 if not rows.get(scope, {}).get("valuation_to") 920 or str(rows[scope]["valuation_to"]) < period["to"] 921 ] 922 if closing_sources: 923 diagnostics.append( 924 { 925 "dimension": "performance", 926 "affected_sources": closing_sources, 927 "message": ( 928 "Für den gewählten Periodenendstichtag fehlen kompatible Bewertungen bei " 929 f"{_join_labels(closing_sources)}." 930 ), 931 "action": "Bestätigte Endbewertungen für denselben fachlichen Stichtag bereitstellen.", 932 "reason_code": "closing_valuation_date_mismatch_by_source", 933 "prominent": False, 934 } 935 ) 936 crypto = rows.get("crypto", {}) 937 if not crypto.get("valuation_from"): 938 diagnostics.append( 939 { 940 "dimension": "performance", 941 "affected_sources": ["Kryptowährungen"], 942 "message": "Für Kryptowährungen ist noch keine historische Gesamtbewertung vorhanden.", 943 "action": "Belastbare historische Gesamtwerte und vollständige Aktivitäten bereitstellen.", 944 "reason_code": "crypto_portfolio_history_missing", 945 "prominent": True, 946 } 947 ) 948 missing_bank_sources = [ 949 ( 950 str(source["label"]) 951 if "E-Finance" in str(source["label"]) 952 else str(source.get("provider_label") or source["label"]) 953 ) 954 for source in current["sources"] 955 if source["kind"] == "Bankguthaben" and source["current_value_chf"] is None 956 ] 957 if missing_bank_sources: 958 preferred_order = {"AKB": 0, "Raiffeisen": 1, "PostFinance E-Finance": 2} 959 providers = sorted( 960 dict.fromkeys(missing_bank_sources), 961 key=lambda label: (preferred_order.get(label, 99), label), 962 ) 963 diagnostics.append( 964 { 965 "dimension": "current_value", 966 "affected_sources": providers, 967 "message": ( 968 f"{_join_labels(providers)} besitzen noch keinen bestätigten aktuellen Kontostand." 969 ), 970 "action": "Je echtes Konto einen bestätigten Saldo mit fachlichem Stichtag erfassen oder importieren.", 971 "reason_code": "current_bank_balance_missing", 972 "prominent": True, 973 } 974 ) 975 flow_sources = [ 976 scope_labels[scope] 977 for scope in ("postfinance", "truewealth", "crypto") 978 if { 979 "external_cashflow_history_missing", 980 "cashflow_classification_incomplete", 981 } 982 & set(rows.get(scope, {}).get("reason_codes", [])) 983 ] 984 if flow_sources: 985 diagnostics.append( 986 { 987 "dimension": "performance", 988 "affected_sources": flow_sources, 989 "message": ( 990 "Externe Ein- und Auszahlungen sind für " 991 f"{_join_labels(flow_sources)} noch nicht vollständig belegt." 992 ), 993 "action": "Vollständige Aktivitäten und Kapitalfluss-Coverage für den Zeitraum nachweisen.", 994 "reason_code": "external_cashflow_coverage_incomplete", 995 "prominent": False, 996 } 997 ) 998 stale_sources = [ 999 str(source["label"]) 1000 for source in current["sources"] 1001 if source["current_value_chf"] is not None 1002 and source["freshness_status"] == "stale" 1003 ] 1004 if stale_sources: 1005 diagnostics.append( 1006 { 1007 "dimension": "freshness", 1008 "affected_sources": stale_sources, 1009 "message": f"Der Datenstand von {_join_labels(stale_sources)} ist veraltet.", 1010 "action": "Bestehenden Aktualisierungsbereich der betroffenen Quelle verwenden.", 1011 "reason_code": "source_update_overdue", 1012 "prominent": False, 1013 } 1014 ) 1015 if not policy.get("configured"): 1016 diagnostics.append( 1017 { 1018 "dimension": "policy", 1019 "affected_sources": [], 1020 "message": "Es ist keine bestätigte Portfolioorientierung hinterlegt.", 1021 "action": "Optional eine Portfolioorientierung hinterlegen; Wert und Performance bleiben davon unabhängig.", 1022 "reason_code": "portfolio_policy_not_configured", 1023 "prominent": False, 1024 } 1025 ) 1026 unique: list[dict[str, Any]] = [] 1027 seen: set[tuple[str, tuple[str, ...], str]] = set() 1028 for item in diagnostics: 1029 identity = ( 1030 str(item["dimension"]), 1031 tuple(str(value) for value in item["affected_sources"]), 1032 str(item["reason_code"]), 1033 ) 1034 if identity not in seen: 1035 seen.add(identity) 1036 unique.append(item) 1037 return unique 1038 1039 1040 def _build_readiness( 1041 *, 1042 current: dict[str, Any], 1043 coverage: dict[str, Any], 1044 policy: dict[str, Any], 1045 period: dict[str, str], 1046 summary: dict[str, Any], 1047 ttwror_quality: dict[str, Any], 1048 household_change: str | None, 1049 history_points: list[dict[str, str]], 1050 reconciliation_status: str, 1051 ) -> dict[str, Any]: 1052 known_current = [ 1053 str(source["label"]) 1054 for source in current["sources"] 1055 if source["current_value_chf"] is not None 1056 ] 1057 missing_current = [ 1058 str(source["label"]) 1059 for source in current["sources"] 1060 if source.get("current_value_status") != "ready" 1061 ] 1062 coverage_rows = { 1063 str(row["scope"]): row 1064 for row in coverage.get("rows", []) 1065 if str(row.get("scope")) != "portfolio" 1066 } 1067 performance_scopes = ( 1068 ("postfinance", "PostFinance"), 1069 ("truewealth", "True Wealth"), 1070 ("crypto", "Kryptowährungen"), 1071 ) 1072 1073 def sources_for_status(field: str) -> tuple[list[str], list[str]]: 1074 included = [ 1075 label 1076 for scope, label in performance_scopes 1077 if coverage_rows.get(scope, {}).get(field) == "complete" 1078 ] 1079 missing = [label for _, label in performance_scopes if label not in included] 1080 return included, missing 1081 1082 ttwror_included, ttwror_missing = sources_for_status("ttwror_status") 1083 attribution_included, attribution_missing = sources_for_status( 1084 "attribution_status" 1085 ) 1086 flow_included = [ 1087 label 1088 for scope, label in performance_scopes 1089 if coverage_rows.get(scope, {}).get("cashflow_coverage_status") == "complete" 1090 and coverage_rows.get(scope, {}).get("scope_classification_status") == "complete" 1091 ] 1092 flow_missing = [label for _, label in performance_scopes if label not in flow_included] 1093 1094 def metric( 1095 key: str, 1096 label: str, 1097 status: str, 1098 *, 1099 included: list[str], 1100 missing: list[str], 1101 blocker: str | None, 1102 action: str | None, 1103 reason_code: str | None, 1104 as_of: str | None = None, 1105 ) -> dict[str, Any]: 1106 return { 1107 "key": key, 1108 "label": label, 1109 "status": status, 1110 "included_sources": included, 1111 "missing_sources": missing, 1112 "as_of": as_of, 1113 "period": period, 1114 "blocker": blocker, 1115 "action": action, 1116 "reason_code": reason_code, 1117 } 1118 1119 current_status = "ready" if current["complete"] else "partial" if known_current else "not_ready" 1120 metrics = [ 1121 metric( 1122 "captured_wealth", 1123 "Erfasstes Vermögen heute", 1124 current_status, 1125 included=known_current, 1126 missing=missing_current, 1127 blocker=None if current_status == "ready" else "Für einzelne Quellen fehlt ein bestätigter aktueller Wert.", 1128 action=None if current_status == "ready" else "Fehlende Kontostände mit fachlichem Stichtag bestätigen.", 1129 reason_code=None if current_status == "ready" else "current_values_incomplete", 1130 as_of=current.get("data_as_of"), 1131 ), 1132 metric( 1133 "wealth_change", 1134 "Vermögensveränderung im Zeitraum", 1135 "ready" if household_change is not None else "not_ready", 1136 included=known_current if household_change is not None else [], 1137 missing=[] if household_change is not None else [str(source["label"]) for source in current["sources"]], 1138 blocker=None if household_change is not None else "Gemeinsame vollständige Anfangs- und Endbewertungen fehlen.", 1139 action=None if household_change is not None else "Bestätigte Bewertungen am Periodenanfang und -ende bereitstellen.", 1140 reason_code=None if household_change is not None else "household_boundary_values_missing", 1141 ), 1142 metric( 1143 "net_contributions", 1144 "Nettoeinzahlungen", 1145 "ready" if summary.get("net_external_cashflows") is not None else "not_ready", 1146 included=flow_included, 1147 missing=flow_missing, 1148 blocker=None if summary.get("net_external_cashflows") is not None else "Externe Kapitalflüsse sind nicht für alle Anlagequellen vollständig belegt.", 1149 action=None if summary.get("net_external_cashflows") is not None else "Kapitalfluss-Coverage und Klassifikation vervollständigen.", 1150 reason_code=None if summary.get("net_external_cashflows") is not None else "external_cashflow_coverage_incomplete", 1151 ), 1152 metric( 1153 "investment_result", 1154 "Anlageergebnis ohne Einzahlungen", 1155 "ready" if summary.get("investment_result") is not None else "not_ready", 1156 included=attribution_included, 1157 missing=attribution_missing, 1158 blocker=None if summary.get("investment_result") is not None else "Anfang, Ende oder Nettoeinzahlungen sind nicht vollständig belegt.", 1159 action=None if summary.get("investment_result") is not None else "Bewertungen und externe Kapitalflüsse für denselben Zeitraum vervollständigen.", 1160 reason_code=None if summary.get("investment_result") is not None else "investment_result_inputs_missing", 1161 ), 1162 metric( 1163 "ttwror", 1164 "Zeitgewichtete Rendite", 1165 _readiness_status(str(ttwror_quality.get("status", "unavailable"))), 1166 included=ttwror_included, 1167 missing=ttwror_missing, 1168 blocker=None if ttwror_quality.get("status") == "complete" else "Bewertungs- oder Kapitalflussgrenzen der bestehenden TTWROR-Engine fehlen.", 1169 action=None if ttwror_quality.get("status") == "complete" else "Anfangs-, End- und Kapitalflussgrenzen mit kanonischen FX-Werten vervollständigen.", 1170 reason_code=None if ttwror_quality.get("status") == "complete" else "ttwror_prerequisites_incomplete", 1171 ), 1172 metric( 1173 "wealth_history", 1174 "Vermögensverlaufsreihe", 1175 "ready" if len(history_points) >= 2 else "not_ready", 1176 included=known_current if len(history_points) >= 2 else [], 1177 missing=[] if len(history_points) >= 2 else [str(source["label"]) for source in current["sources"]], 1178 blocker=None if len(history_points) >= 2 else "Mindestens zwei gemeinsame vollständige Stichtage fehlen.", 1179 action=None if len(history_points) >= 2 else "Keine Zwischenwerte schätzen; gemeinsame bestätigte Stichtage bereitstellen.", 1180 reason_code=None if len(history_points) >= 2 else "complete_history_points_missing", 1181 ), 1182 metric( 1183 "policy_allocation", 1184 "Aufteilung gegenüber Portfolioorientierung", 1185 "ready" if policy.get("configured") and current["complete"] else "partial" if policy.get("configured") else "not_applicable", 1186 included=known_current, 1187 missing=missing_current, 1188 blocker=None if policy.get("configured") else "Keine bestätigte Portfolioorientierung vorhanden; Performance und aktueller Wert bleiben unberührt.", 1189 action=None if policy.get("configured") else "Optional eine Portfolioorientierung hinterlegen.", 1190 reason_code=None if policy.get("configured") else "portfolio_policy_not_configured", 1191 ), 1192 ] 1193 freshness_status = combined_freshness( 1194 [ 1195 cast(FreshnessStatus, source["freshness_status"]) 1196 for source in current["sources"] 1197 ] 1198 ) 1199 dimensions = { 1200 "current_value": {"status": current_status, "reason_code": None if current_status == "ready" else "current_values_incomplete"}, 1201 "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"}, 1202 "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"}, 1203 "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")}, 1204 "policy": {"status": "ready" if policy.get("configured") else "not_applicable", "reason_code": None if policy.get("configured") else "portfolio_policy_not_configured"}, 1205 } 1206 return {"dimensions": dimensions, "metrics": metrics} 1207 1208 1209 def build_wealth_cockpit( 1210 conn: Connection, 1211 *, 1212 period: str = "ytd", 1213 as_of: str | None = None, 1214 data_cutoff: str | None = None, 1215 ) -> dict[str, Any]: 1216 reference = date.fromisoformat(as_of) if as_of else date.today() 1217 start, requested_end = period_bounds(conn, period=period, as_of=reference) 1218 cutoff = data_cutoff or _latest_data_cutoff(conn) 1219 model_period = ( 1220 period 1221 if period in {"since_anchor", "1m", "3m", "ytd", "1y", "all"} 1222 else "1y" 1223 if period in {"previous_year", "12m"} 1224 else "all" 1225 ) 1226 modelled_development = build_modelled_wealth_development( 1227 conn, period=model_period, as_of=reference.isoformat() 1228 ) 1229 portfolio_analysis = build_portfolio_analysis_v1( 1230 conn, as_of=reference.isoformat(), modelled=modelled_development 1231 ) 1232 current = _current_values(conn, as_of=reference) 1233 valuation_end = _latest_valuation_date(conn, requested_end) 1234 performance: dict[str, Any] | None = None 1235 if start < valuation_end: 1236 performance = build_portfolio_performance( 1237 conn, 1238 from_date=start.isoformat(), 1239 to_date=valuation_end.isoformat(), 1240 method="both", 1241 base_currency="CHF", 1242 data_cutoff=cutoff, 1243 ) 1244 summary = performance.get("summary", {}) if performance else {} 1245 quality = performance.get("quality", {}).get("ttwror", {}) if performance else {} 1246 xirr_quality = performance.get("quality", {}).get("xirr", {}) if performance else {} 1247 investment_events = performance.get("external_cashflows", []) if performance else [] 1248 household_events = scope_cashflows( 1249 conn, 1250 account_ids=_account_ids(conn, investment_only=False), 1251 from_date=start.isoformat(), 1252 to_date=requested_end.isoformat(), 1253 data_cutoff=cutoff, 1254 ) 1255 reconciliation = build_reconciliation_snapshot( 1256 conn, now=datetime.combine(reference, datetime.max.time(), tzinfo=UTC) 1257 ) 1258 history_points, history_reason = _household_history( 1259 conn, 1260 from_date=start, 1261 to_date=requested_end, 1262 current=current, 1263 as_of=reference, 1264 ) 1265 policy = _policy_comparison(conn, current) 1266 coverage = build_performance_coverage( 1267 conn, 1268 from_date=start.isoformat(), 1269 to_date=requested_end.isoformat(), 1270 ) 1271 coverage_items = coverage.get("rows") 1272 coverage_rows = { 1273 str(row["scope"]): row 1274 for row in coverage_items 1275 if str(row.get("scope")) != "portfolio" 1276 } if isinstance(coverage_items, list) else {} 1277 for source in current["sources"]: 1278 scope = source.get("performance_scope") 1279 if scope: 1280 source["performance_status"] = _performance_scope_status( 1281 coverage_rows.get(str(scope)) 1282 ) 1283 key = str(source.get("key", "")) 1284 provider = str(source.get("provider_label", "")).casefold() 1285 if key == "truewealth": 1286 meta = _truewealth_import_meta(conn) 1287 elif key == "postfinance-investments" or "postfinance" in provider: 1288 meta = _postfinance_import_meta(conn) 1289 elif key == "visa-liability": 1290 meta = _household_import_meta(conn, "viseca_one") 1291 elif "akb" in provider: 1292 meta = _household_import_meta( 1293 conn, 1294 "akb", 1295 canonical_account_id=source.get("_canonical_account_id"), 1296 ) 1297 elif "raiffeisen" in provider: 1298 meta = _household_import_meta( 1299 conn, 1300 "raiffeisen", 1301 canonical_account_id=source.get("_canonical_account_id"), 1302 ) 1303 else: 1304 meta = {"imported_at": None, "coverage_from": None, "coverage_to": None, "coverage_status": "unavailable", "new_rows": 0, "duplicate_rows": 0, "review_rows": 0} 1305 source.pop("_canonical_account_id", None) 1306 source.update({name: value for name, value in meta.items() if name != "last_snapshot"}) 1307 source["last_activity_day"] = meta.get("coverage_to") or source.get("as_of") 1308 source["last_confirmed_snapshot"] = meta.get("last_snapshot") or source.get("as_of") 1309 source["value_basis"] = ( 1310 "modelled" if key == "crypto" and source.get("current_value_chf") is not None 1311 else "confirmed" if source.get("current_value_chf") is not None 1312 else "unavailable" 1313 ) 1314 review_rows = int(meta.get("review_rows", 0) or 0) 1315 if key == "postfinance-investments" or "postfinance" in provider: 1316 source["performance_blocker"] = None if meta.get("coverage_status") == "complete" else ( 1317 "PostFinance E-Trading-Kontoauszug oder vollständiger Transaktionsreport vom 01.08.–26.08.2026 fehlt; bei null Aktivitäten ist ein offizieller Nachweis erforderlich." 1318 ) 1319 elif key == "truewealth": 1320 source["performance_blocker"] = None if meta.get("coverage_status") == "complete" else "Externe Ein- und Auszahlungen sind noch nicht vollständig belegt." 1321 elif key == "crypto": 1322 source["performance_blocker"] = "Mengen-, Aktivitäts-, Preis- oder Cashflow-Coverage ist weiterhin unvollständig." 1323 source["coverage_status"] = "partial" 1324 elif key == "visa-liability": 1325 source["performance_blocker"] = "Aktueller Abrechnungssaldo ist nicht vollständig belegt." 1326 else: 1327 source["performance_blocker"] = None 1328 source["next_action"] = ( 1329 f"{review_rows} prüfpflichtige Zeilen bearbeiten." 1330 if review_rows 1331 else source.get("performance_blocker") 1332 or "Keine offene Aktion." 1333 ) 1334 if policy.get("contribution"): 1335 invested = _decimal(summary.get("net_external_cashflows")) if period == "ytd" else None 1336 policy["contribution"].update( 1337 invested_ytd_chf=_money(invested), 1338 expected_year_end_chf=_money(invested / Decimal(max(reference.month, 1)) * Decimal("12")) if invested is not None and period == "ytd" else None, 1339 difference_to_target_chf=_money(invested - Decimal(policy["contribution"]["annual_target_chf"])) if invested is not None else None, 1340 status="available" if invested is not None else "not_assessable", 1341 ) 1342 planning = get_annual_budget_assistant(conn, year=str(reference.year), current_month=f"{reference.year:04d}-{reference.month:02d}") 1343 free_row = next((row for row in planning["summary_kpis"] if row["key"] == "free_after_special"), None) 1344 missing_areas = ["Verbindlichkeiten und Immobilienwerte sind nicht vollständig und aktuell erfasst."] 1345 if current["unpriced_count"]: 1346 missing_areas.append(f"{current['unpriced_count']} Positionen besitzen keinen belastbaren aktuellen Wert.") 1347 if current["missing_cash_count"]: 1348 missing_areas.append( 1349 f"{current['missing_cash_count']} Bankkonten besitzen keinen bestätigten aktuellen Saldo." 1350 ) 1351 if next((row for row in current["distribution"] if row["key"] == "truewealth"), {}).get("value_chf") is None: 1352 missing_areas.append("Für True Wealth fehlt ein bestätigter aktueller Gesamtwert.") 1353 1354 history_by_date = {point["at"]: Decimal(point["value_chf"]) for point in history_points} 1355 opening_household = history_by_date.get(start.isoformat()) 1356 closing_household = history_by_date.get(requested_end.isoformat()) 1357 household_change = _money(closing_household - opening_household) if opening_household is not None and closing_household is not None else None 1358 household_change_status = "available" if household_change is not None else "not_calculable" 1359 investment_result = summary.get("investment_result") 1360 net_contributions = summary.get("net_external_cashflows") 1361 reconciliation_rows = reconciliation.get("reconciliations", []) 1362 if not isinstance(reconciliation_rows, list): 1363 reconciliation_rows = [] 1364 reconciliation_statuses = [str(row["status"]) for row in reconciliation_rows] 1365 reconciliation_status = ( 1366 "difference" 1367 if "difference" in reconciliation_statuses 1368 else "reconciled" 1369 if reconciliation_statuses and all(status == "reconciled" for status in reconciliation_statuses) 1370 else "not_assessable" 1371 ) 1372 period_payload = { 1373 "preset": period, 1374 "from": start.isoformat(), 1375 "to": requested_end.isoformat(), 1376 } 1377 diagnostics = _build_diagnostics( 1378 current=current, 1379 coverage=coverage, 1380 policy=policy, 1381 period=period_payload, 1382 ) 1383 hints = [str(item["message"]) for item in diagnostics if item["prominent"]][:3] 1384 readiness = _build_readiness( 1385 current=current, 1386 coverage=coverage, 1387 policy=policy, 1388 period=period_payload, 1389 summary=summary, 1390 ttwror_quality=quality, 1391 household_change=household_change, 1392 history_points=history_points, 1393 reconciliation_status=reconciliation_status, 1394 ) 1395 current_freshness = combined_freshness( 1396 [ 1397 cast(FreshnessStatus, source["freshness_status"]) 1398 for source in current["sources"] 1399 ] 1400 ) 1401 ttwror_verified = ( 1402 quality.get("status") == "complete" 1403 and summary.get("ttwror_cumulative") is not None 1404 ) 1405 xirr_verified = ( 1406 xirr_quality.get("status") == "complete" 1407 and summary.get("xirr_annualized") is not None 1408 ) 1409 verified_performance = { 1410 "status": "verified" if ttwror_verified and xirr_verified else "not_verified", 1411 "label": "Verifiziert" if ttwror_verified and xirr_verified else "Noch nicht verifiziert", 1412 "ttwror_status": "ready" if ttwror_verified else "not_ready", 1413 "xirr_status": "ready" if xirr_verified else "not_ready", 1414 "ttwror_pct": summary.get("ttwror_cumulative") if ttwror_verified else None, 1415 "xirr_pct": summary.get("xirr_annualized") if xirr_verified else None, 1416 } 1417 return { 1418 "scope_label": "Erfasstes Vermögen", 1419 "not_net_worth": True, 1420 "period": period_payload, 1421 "data_cutoff": cutoff, 1422 "modelled_development": modelled_development, 1423 "portfolio_analysis": portfolio_analysis, 1424 "verified_performance": verified_performance, 1425 "kpis": [ 1426 {"key": "captured_wealth", "label": "Erfasstes Vermögen heute", "value_chf": _money(current["total"]), "status": "complete" if current["complete"] else "approximate"}, 1427 {"key": "wealth_change", "label": "Veränderung im Zeitraum", "value_chf": household_change, "status": household_change_status}, 1428 {"key": "investment_result", "label": "Anlageergebnis ohne Einzahlungen", "value_chf": investment_result, "status": "available" if investment_result is not None else "not_calculable"}, 1429 {"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"}, 1430 {"key": "net_contributions", "label": "Nettoeinzahlungen ins Anlageportfolio", "value_chf": net_contributions, "status": "available" if net_contributions is not None else "not_calculable"}, 1431 {"key": "data_as_of", "label": "Datenstand", "value_date": current["data_as_of"], "status": "available" if current["data_as_of"] else "unknown"}, 1432 ], 1433 "totals": {"captured_wealth_chf": _money(current["total"]), "investments_chf": _money(current["investments"]), "bank_cash_chf": _money(current["cash"]), "complete": current["complete"]}, 1434 "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}, 1435 "distribution": current["distribution"], 1436 "sources": current["sources"], 1437 "readiness": readiness, 1438 "diagnostics": diagnostics, 1439 "performance_coverage": coverage, 1440 "policy": policy, 1441 "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}, 1442 "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"]}, 1443 "hints": hints[:3], 1444 "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."}, 1445 } ===== src/jarvis_finance/services/portfolio_analysis_v1.py ===== 1 from __future__ import annotations 2 3 import json 4 from collections import defaultdict 5 from decimal import Decimal, InvalidOperation 6 from sqlite3 import Connection 7 from typing import Any 8 9 from jarvis_finance.services.modelled_wealth import ( 10 _bank_accounts, 11 build_modelled_wealth_development, 12 effective_cash_evidence, 13 ) 14 from jarvis_finance.services.portfolio_policy import active_policy 15 16 ZERO = Decimal("0") 17 HUNDRED = Decimal("100") 18 MONEY = Decimal("0.01") 19 PCT = Decimal("0.01") 20 21 22 def _decimal(value: object) -> Decimal: 23 try: 24 result = Decimal(str(value or "0")) 25 return result if result.is_finite() else ZERO 26 except (InvalidOperation, ValueError): 27 return ZERO 28 29 30 def _money(value: Decimal | None) -> str | None: 31 return None if value is None else format(value.quantize(MONEY), "f") 32 33 34 def _pct(value: Decimal | None) -> str | None: 35 return None if value is None else format(value.quantize(PCT), "f") 36 37 38 def _latest_positions(conn: Connection, as_of: str) -> tuple[list[dict[str, Any]], str]: 39 row = conn.execute( 40 """SELECT as_of,quality_status,summary_json FROM portfolio_analysis_snapshots 41 WHERE as_of<=? ORDER BY as_of DESC,created_at DESC,analysis_snapshot_id DESC LIMIT 1""", 42 (as_of,), 43 ).fetchone() 44 if not row: 45 return [], "unavailable" 46 try: 47 summary = json.loads(str(row["summary_json"] or "{}")) 48 positions = [item for item in summary.get("positions", []) if isinstance(item, dict)] 49 except (TypeError, ValueError, json.JSONDecodeError): 50 return [], "unavailable" 51 metadata = { 52 str(item["instrument_id"]): dict(item) 53 for item in conn.execute( 54 "SELECT instrument_id,country,sector,currency,asset_class FROM instruments WHERE is_active=1" 55 ).fetchall() 56 } 57 result: list[dict[str, Any]] = [] 58 for item in positions: 59 value = _decimal(item.get("value_chf")) 60 if value <= ZERO: 61 continue 62 instrument = metadata.get(str(item.get("instrument_id") or ""), {}) 63 result.append( 64 { 65 **item, 66 "value": value, 67 "asset_class": str(item.get("asset_class") or instrument.get("asset_class") or "").lower(), 68 "currency": str(item.get("currency") or instrument.get("currency") or "").upper(), 69 "country": str(instrument.get("country") or "").strip(), 70 "sector": str(instrument.get("sector") or "").strip(), 71 } 72 ) 73 quality = "complete" if str(row["quality_status"]) == "complete" else "partial" 74 return result, quality 75 76 77 def _current_components(modelled: dict[str, Any]) -> dict[str, Decimal]: 78 return { 79 str(item["key"]): _decimal(item.get("current_value_chf")) 80 for item in modelled.get("components", []) 81 if item.get("current_value_chf") is not None 82 } 83 84 85 def _policy_rows(conn: Connection) -> dict[str, dict[str, Any]]: 86 configured = active_policy(conn) 87 policy = configured.get("policy") if configured.get("configured") else None 88 if not policy: 89 return {} 90 return {str(item["asset_class"]): item for item in policy.get("allocations", [])} 91 92 93 def _allocation_row( 94 *, key: str, label: str, value: Decimal, total: Decimal, policy: dict[str, Any] | None 95 ) -> dict[str, Any]: 96 current_pct = value / total * HUNDRED if total > ZERO else None 97 if not policy or current_pct is None: 98 return { 99 "key": key, "label": label, "current_value_chf": _money(value), 100 "current_pct": _pct(current_pct), "target_pct": None, "lower_pct": None, 101 "upper_pct": None, "deviation_pp": None, "deviation_chf": None, 102 "status": "unavailable", 103 } 104 target = _decimal(policy.get("target_pct")) 105 lower = _decimal(policy.get("lower_pct")) 106 upper = _decimal(policy.get("upper_pct")) 107 deviation_pp = current_pct - target 108 deviation_chf = value - total * target / HUNDRED 109 status = "below_corridor" if current_pct < lower else "above_corridor" if current_pct > upper else "within_corridor" 110 return { 111 "key": key, "label": label, "current_value_chf": _money(value), 112 "current_pct": _pct(current_pct), "target_pct": _pct(target), 113 "lower_pct": _pct(lower), "upper_pct": _pct(upper), 114 "deviation_pp": _pct(deviation_pp), "deviation_chf": _money(deviation_chf), 115 "status": status, 116 } 117 118 119 def _dimension( 120 positions: list[dict[str, Any]], field: str, total: Decimal, *, include_chf: Decimal = ZERO 121 ) -> dict[str, Any]: 122 values: dict[str, Decimal] = defaultdict(lambda: ZERO) 123 assessed = ZERO 124 if include_chf > ZERO and field == "currency": 125 values["CHF"] += include_chf 126 assessed += include_chf 127 for item in positions: 128 label = str(item.get(field) or "").strip() 129 if not label: 130 continue 131 values[label] += item["value"] 132 assessed += item["value"] 133 rows = [ 134 {"label": label, "pct": _pct(value / total * HUNDRED) if total > ZERO else None} 135 for label, value in sorted(values.items(), key=lambda item: (-item[1], item[0])) 136 ] 137 if total <= ZERO or not rows: 138 status = "unavailable" 139 else: 140 status = "complete" if assessed >= total - Decimal("0.01") else "partial" 141 return {"status": status, "rows": rows} 142 143 144 def _concentrations(values: list[Decimal], total: Decimal) -> dict[str, str | None]: 145 ordered = sorted((value for value in values if value > ZERO), reverse=True) 146 def share(limit: int) -> str | None: 147 return _pct(sum(ordered[:limit], ZERO) / total * HUNDRED) if total > ZERO and ordered else None 148 return {"top1_pct": share(1), "top5_pct": share(5), "top10_pct": share(10)} 149 150 151 def _prioritized_hints(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: 152 priority = {"above_corridor": 1, "below_corridor": 2, "unavailable": 3, "within_corridor": 4} 153 prefix = { 154 "above_corridor": "Reduktion prüfen", 155 "below_corridor": "Erhöhung prüfen", 156 "unavailable": "Daten ergänzen", 157 "within_corridor": "Im Zielkorridor", 158 } 159 candidates = sorted(rows, key=lambda row: (priority.get(str(row["status"]), 9), str(row["label"]))) 160 return [ 161 {"priority": index, "text": f"{prefix.get(str(row['status']), 'Daten ergänzen')}: {row['label']}."} 162 for index, row in enumerate(candidates[:5], start=1) 163 ] 164 165 166 def build_portfolio_analysis_v1( 167 conn: Connection, *, as_of: str, modelled: dict[str, Any] | None = None 168 ) -> dict[str, Any]: 169 """Read-only v1 analysis over canonical stored valuations and active versioned policy.""" 170 model = modelled or build_modelled_wealth_development(conn, period="1m", as_of=as_of) 171 components = _current_components(model) 172 positions, position_quality = _latest_positions(conn, as_of) 173 truewealth_accounts = { 174 str(row["account_id"]) 175 for row in conn.execute( 176 "SELECT DISTINCT account_id FROM truewealth_portfolios WHERE is_active=1" 177 ).fetchall() 178 } 179 directly_classified = [ 180 item for item in positions 181 if str(item.get("account_id") or "") not in truewealth_accounts 182 ] 183 stock_value = sum( 184 (item["value"] for item in directly_classified if item["asset_class"] in {"equity", "stock"}), 185 ZERO, 186 ) 187 etf_value = sum( 188 (item["value"] for item in directly_classified if item["asset_class"] in {"etf", "fund"}), 189 ZERO, 190 ) 191 classified_pf = stock_value + etf_value 192 postfinance_total = components.get("postfinance", ZERO) 193 settlement_cash = max(ZERO, postfinance_total - classified_pf) 194 bank_cash = components.get("bank_cash", ZERO) 195 cash_value = bank_cash + settlement_cash 196 truewealth_value = components.get("truewealth", ZERO) 197 crypto_value = components.get("crypto", ZERO) 198 other_value = components.get("other_assets", ZERO) 199 total = cash_value + stock_value + etf_value + truewealth_value + crypto_value + other_value 200 policy = _policy_rows(conn) 201 202 requested = [ 203 ("cash", "Cash", cash_value, policy.get("cash")), 204 ("stocks", "Aktien", stock_value, None), 205 ("etf", "ETF", etf_value, None), 206 ("truewealth", "True Wealth", truewealth_value, None), 207 ("crypto", "Krypto", crypto_value, policy.get("crypto")), 208 ("other", "Weitere Anlagen", other_value, policy.get("other")), 209 ] 210 allocation = [ 211 _allocation_row(key=key, label=label, value=value, total=total, policy=target) 212 for key, label, value, target in requested 213 ] 214 if policy.get("equity"): 215 allocation.append( 216 _allocation_row( 217 key="equity_policy_group", 218 label="Policy-Gruppe Aktien / ETF / True Wealth", 219 value=stock_value + etf_value + truewealth_value, 220 total=total, 221 policy=policy["equity"], 222 ) 223 ) 224 225 concentration_values = [item["value"] for item in directly_classified] 226 concentration_values.extend(value for value in (truewealth_value, crypto_value, other_value) if value > ZERO) 227 for account in _bank_accounts(conn): 228 evidence = effective_cash_evidence(conn, account_id=account["account_id"], as_of=as_of) 229 if evidence["value_chf"] is not None: 230 concentration_values.append(_decimal(evidence["value_chf"])) 231 if settlement_cash > ZERO: 232 concentration_values.append(settlement_cash) 233 234 contributions = [] 235 for item in model.get("components", []): 236 value = item.get("change_chf") 237 if value is None or _decimal(value) == ZERO: 238 continue 239 contributions.append( 240 {"key": str(item["key"]), "label": str(item["label"]), "value_chf": _money(_decimal(value)), "status": "modelled"} 241 ) 242 positives = sorted((row for row in contributions if _decimal(row["value_chf"]) > ZERO), key=lambda row: _decimal(row["value_chf"]), reverse=True)[:2] 243 negatives = sorted((row for row in contributions if _decimal(row["value_chf"]) < ZERO), key=lambda row: _decimal(row["value_chf"]))[:2] 244 245 status = "unavailable" if total <= ZERO else "partial" if position_quality != "complete" or any(row["status"] == "unavailable" for row in allocation) else "complete" 246 return { 247 "status": status, 248 "allocation": allocation, 249 "concentrations": _concentrations(concentration_values, total), 250 "dimensions": { 251 "currency": _dimension(positions, "currency", total, include_chf=cash_value + other_value), 252 "region": _dimension(positions, "country", total), 253 "sector": _dimension(positions, "sector", total), 254 }, 255 "contributions": positives + negatives, 256 "hints": _prioritized_hints(allocation), 257 } ===== src/jarvis_finance/api/schemas/wealth_cockpit.py ===== 1 from __future__ import annotations 2 3 from typing import Any, Literal 4 5 from pydantic import BaseModel, ConfigDict, Field 6 7 from jarvis_finance.api.schemas.portfolio_performance import PerformanceCoverageResponse 8 9 ReadinessStatus = Literal["ready", "partial", "not_ready", "not_applicable"] 10 11 12 class WealthPeriod(BaseModel): 13 model_config = ConfigDict(extra="forbid", populate_by_name=True) 14 15 preset: Literal[ 16 "since_anchor", "1m", "3m", "1y", "ytd", "previous_year", "12m", "all" 17 ] 18 from_: str = Field(alias="from") 19 to: str 20 21 22 class ModelledWealthPeriod(BaseModel): 23 model_config = ConfigDict(extra="forbid", populate_by_name=True) 24 25 preset: Literal["since_anchor", "1m", "3m", "ytd", "1y", "all"] 26 from_: str = Field(alias="from") 27 to: str 28 29 30 ModelledValueQuality = Literal[ 31 "confirmed", "modelled", "carried", "incomplete", "unavailable" 32 ] 33 34 35 class ModelledValueSummary(BaseModel): 36 model_config = ConfigDict(extra="forbid") 37 38 date: str 39 value_chf: str 40 quality: ModelledValueQuality 41 42 43 class ModelledPointComponent(BaseModel): 44 model_config = ConfigDict(extra="forbid") 45 46 key: Literal["postfinance", "truewealth", "crypto", "bank_cash", "other_assets"] 47 label: str 48 value_chf: str | None 49 quality: ModelledValueQuality 50 source_date: str | None 51 52 53 class ModelledDailyPoint(BaseModel): 54 model_config = ConfigDict(extra="forbid") 55 56 date: str 57 value_chf: str 58 quality: ModelledValueQuality 59 has_confirmed_anchor: bool 60 has_modelled_value: bool 61 components: list[ModelledPointComponent] 62 excluded_account_count: int 63 64 65 class ModelledComponentSummary(BaseModel): 66 model_config = ConfigDict(extra="forbid") 67 68 key: Literal["postfinance", "truewealth", "crypto", "bank_cash", "other_assets"] 69 label: str 70 current_value_chf: str | None 71 change_chf: str | None 72 change_pct: str | None 73 quality: ModelledValueQuality 74 as_of: str | None 75 unknown_account_count: int 76 77 78 class ModelledCorrectionMarker(BaseModel): 79 model_config = ConfigDict(extra="forbid") 80 81 date: str 82 source_key: Literal["postfinance", "truewealth", "bank_cash"] 83 confirmed_value_chf: str 84 predecessor_model_value_chf: str 85 difference_chf: str 86 87 88 class ModelledUnknownAccount(BaseModel): 89 model_config = ConfigDict(extra="forbid") 90 91 key: str 92 label: str 93 reason_code: str 94 95 96 class ModelledWealthDevelopment(BaseModel): 97 model_config = ConfigDict(extra="forbid") 98 99 status: Literal["available", "unavailable"] 100 period: ModelledWealthPeriod 101 last_confirmed_anchor_date: str | None 102 anchor: ModelledValueSummary | None 103 baseline: ModelledValueSummary | None 104 current: ModelledValueSummary | None 105 change_chf: str | None 106 change_pct: str | None 107 chart_visible: bool 108 points: list[ModelledDailyPoint] 109 components: list[ModelledComponentSummary] 110 correction_markers: list[ModelledCorrectionMarker] 111 unknown_accounts: list[ModelledUnknownAccount] 112 method: Literal["modelled_wealth_daily_v1"] 113 disclaimer: str 114 115 116 class VerifiedPerformanceSummary(BaseModel): 117 model_config = ConfigDict(extra="forbid") 118 119 status: Literal["verified", "not_verified"] 120 label: str 121 ttwror_status: ReadinessStatus 122 xirr_status: ReadinessStatus 123 ttwror_pct: str | None 124 xirr_pct: str | None 125 126 127 class PortfolioAnalysisAllocation(BaseModel): 128 model_config = ConfigDict(extra="forbid") 129 key: str 130 label: str 131 current_value_chf: str | None 132 current_pct: str | None 133 target_pct: str | None 134 lower_pct: str | None 135 upper_pct: str | None 136 deviation_pp: str | None 137 deviation_chf: str | None 138 status: Literal["below_corridor", "within_corridor", "above_corridor", "unavailable"] 139 140 141 class PortfolioAnalysisDimensionRow(BaseModel): 142 model_config = ConfigDict(extra="forbid") 143 label: str 144 pct: str | None 145 146 147 class PortfolioAnalysisDimension(BaseModel): 148 model_config = ConfigDict(extra="forbid") 149 status: Literal["complete", "partial", "unavailable"] 150 rows: list[PortfolioAnalysisDimensionRow] 151 152 153 class PortfolioAnalysisContribution(BaseModel): 154 model_config = ConfigDict(extra="forbid") 155 key: str 156 label: str 157 value_chf: str | None 158 status: str | None = None 159 160 161 class PortfolioAnalysisHint(BaseModel): 162 model_config = ConfigDict(extra="forbid") 163 priority: int 164 text: str 165 166 167 class PortfolioAnalysisV1(BaseModel): 168 model_config = ConfigDict(extra="forbid") 169 status: Literal["complete", "partial", "unavailable"] 170 allocation: list[PortfolioAnalysisAllocation] 171 concentrations: dict[str, str | None] 172 dimensions: dict[str, PortfolioAnalysisDimension] 173 contributions: list[PortfolioAnalysisContribution] 174 hints: list[PortfolioAnalysisHint] 175 176 177 class WealthDimensionStatus(BaseModel): 178 model_config = ConfigDict(extra="forbid") 179 180 status: ReadinessStatus 181 reason_code: str | None 182 183 184 class WealthReadinessDimensions(BaseModel): 185 model_config = ConfigDict(extra="forbid") 186 187 current_value: WealthDimensionStatus 188 freshness: WealthDimensionStatus 189 reconciliation: WealthDimensionStatus 190 performance: WealthDimensionStatus 191 policy: WealthDimensionStatus 192 193 194 class WealthReadinessMetric(BaseModel): 195 model_config = ConfigDict(extra="forbid") 196 197 key: str 198 label: str 199 status: ReadinessStatus 200 included_sources: list[str] 201 missing_sources: list[str] 202 as_of: str | None 203 period: WealthPeriod 204 blocker: str | None 205 action: str | None 206 reason_code: str | None 207 208 209 class WealthReadiness(BaseModel): 210 model_config = ConfigDict(extra="forbid") 211 212 dimensions: WealthReadinessDimensions 213 metrics: list[WealthReadinessMetric] 214 215 216 class WealthDiagnostic(BaseModel): 217 model_config = ConfigDict(extra="forbid") 218 219 dimension: Literal[ 220 "current_value", "freshness", "reconciliation", "performance", "policy" 221 ] 222 affected_sources: list[str] 223 message: str 224 action: str 225 reason_code: str 226 prominent: bool 227 228 229 class WealthSource(BaseModel): 230 model_config = ConfigDict(extra="forbid") 231 232 key: str 233 label: str 234 provider_label: str | None = None 235 kind: str 236 source_role: Literal["account", "canonical_value", "liability"] | None = None 237 performance_scope: Literal["postfinance", "truewealth", "crypto"] | None = None 238 current_value_chf: str | None 239 current_value_status: ReadinessStatus 240 change_chf: str | None 241 net_contributions_chf: str | None 242 return_pct: str | None 243 as_of: str | None 244 freshness_status: Literal["fresh", "stale", "unavailable", "unknown"] 245 freshness_reason_code: str | None = None 246 expected_as_of: str | None = None 247 reconciliation_status: Literal["reconciled", "difference", "not_assessable"] 248 performance_status: ReadinessStatus 249 value_basis: Literal["confirmed", "modelled", "unavailable"] = "unavailable" 250 last_activity_day: str | None = None 251 last_confirmed_snapshot: str | None = None 252 imported_at: str | None = None 253 coverage_from: str | None = None 254 coverage_to: str | None = None 255 coverage_status: Literal["complete", "partial", "stale", "unavailable"] = "unavailable" 256 new_rows: int = 0 257 duplicate_rows: int = 0 258 review_rows: int = 0 259 next_action: str | None = None 260 performance_blocker: str | None = None 261 262 263 class WealthCockpitResponse(BaseModel): 264 """Runtime-validated contract for the read-only wealth cockpit.""" 265 266 model_config = ConfigDict(extra="forbid") 267 268 scope_label: str 269 not_net_worth: bool 270 period: WealthPeriod 271 data_cutoff: str 272 modelled_development: ModelledWealthDevelopment 273 portfolio_analysis: PortfolioAnalysisV1 274 verified_performance: VerifiedPerformanceSummary 275 kpis: list[dict[str, Any]] 276 totals: dict[str, Any] 277 history: dict[str, Any] 278 distribution: list[dict[str, Any]] 279 sources: list[WealthSource] 280 readiness: WealthReadiness 281 diagnostics: list[WealthDiagnostic] 282 performance_coverage: PerformanceCoverageResponse 283 policy: dict[str, Any] 284 planning: dict[str, Any] 285 data_quality: dict[str, Any] 286 hints: list[str] 287 method: dict[str, str] __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23__HERMES_CWD_8d46a20096ed__