"""Deterministic portfolio ingestion and reconciliation over existing normalized data. No function in this module reads raw files or calls external providers. Preview and reconciliation are pure database projections. Confirm is the only write boundary. """ from __future__ import annotations import base64 import binascii import hashlib import json import re import uuid from dataclasses import dataclass from datetime import UTC, date, datetime, timedelta from decimal import Decimal, InvalidOperation from sqlite3 import Connection, IntegrityError from typing import Any from jarvis_finance.audit.log import record_audit_event from jarvis_finance.imports.common import stable_id from jarvis_finance.imports.postfinance_etrading import ( PostFinanceDocument, normalize_label, parse_postfinance_portfolio, ) from jarvis_finance.imports.postfinance_etrading import ( decimal_text as postfinance_decimal_text, ) from jarvis_finance.market_data.mappings import confirm_instrument_price_mapping from jarvis_finance.services.portfolio_aggregation import ( latest_official_postfinance_cash, postfinance_account_roles, ) INGESTION_VERSION = "portfolio_ingestion_v1" RECONCILIATION_VERSION = "portfolio_reconciliation_v1" TOLERANCE_VERSION = "portfolio_reconciliation_tolerance_v1" PREVIEW_TTL_MINUTES = 15 SOURCE_KEYS = {"canonical_transactions", "legacy_account_values", "cash_account_snapshots", "postfinance_etrading"} SUPPORTED_ACTIVITIES = { "external_deposit", "deposit", "cash_deposit", "external_withdrawal", "withdrawal", "cash_withdrawal", "internal_transfer", "transfer", "buy", "partial_sell", "full_sell", "sell", "dividend", "etf_distribution", "distribution", "fee", "tax", "withholding_tax", "reversal", "correction", } POSITION_ACTIVITY_TYPES = {"buy", "partial_sell", "full_sell", "sell", "initial_position_snapshot"} @dataclass(frozen=True) class SourceRecord: source_record_fingerprint: str source_record_ref: str record_kind: str disposition: str logical_key: str target_type: str | None target_id: str | None lineage_hash: str account_id: str account_ref: str account_label: str as_of: str summary: dict[str, Any] write_payload: dict[str, Any] | None = None def _now() -> datetime: return datetime.now(UTC) def _instant(value: str, field: str) -> datetime: try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise ValueError(f"{field} ist kein gültiger ISO-8601-Zeitstempel") from exc if parsed.tzinfo is None: raise ValueError(f"{field} benötigt eine eindeutige Zeitzone") return parsed.astimezone(UTC) def _date(value: str, field: str) -> date: try: return date.fromisoformat(value) except ValueError as exc: raise ValueError(f"{field} ist kein gültiges ISO-Datum") from exc def _decimal(value: object, field: str, *, non_negative: bool = False, positive: bool = False) -> Decimal: if isinstance(value, float): raise ValueError(f"{field} muss als exakter Decimal-Text angegeben werden") try: result = Decimal(str(value).strip()) except (InvalidOperation, ValueError) as exc: raise ValueError(f"{field} ist keine gültige Dezimalzahl") from exc if not result.is_finite(): raise ValueError(f"{field} muss endlich sein") if positive and result <= 0: raise ValueError(f"{field} muss grösser als null sein") if non_negative and result < 0: raise ValueError(f"{field} darf nicht negativ sein") return result def _stored_decimal(value: object, field: str, *, non_negative: bool = False, positive: bool = False) -> Decimal: """Convert a database scalar to Decimal; SQLite NUMERIC affinity may return int or float.""" return _decimal(str(value), field, non_negative=non_negative, positive=positive) def _decimal_text(value: Decimal | None) -> str | None: if value is None: return None text = format(value, "f") return text.rstrip("0").rstrip(".") if "." in text else text def _hash(payload: object) -> str: return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode()).hexdigest() def _safe_ref(prefix: str, value: str) -> str: return f"{prefix}-" + hashlib.sha256(value.encode()).hexdigest()[:12] def _safe_label(value: str) -> str: sensitive = r"\b[A-Z]{2}\s?\d{2}(?:\s?[A-Z0-9]){11,30}\b|\b\d{10,}\b|\b0x[a-f0-9]{20,}\b|(?:^|\s)[~.]?[/\\]|\.(?:csv|xlsx|sqlite|db|json)\b" return "Konto (redigiert)" if re.search(sensitive, value, flags=re.IGNORECASE) else value def _validate_scope(conn: Connection, scope_kind: str, account_id: str | None) -> list[dict[str, str]]: if scope_kind not in {"portfolio", "account"}: raise ValueError("Scope muss portfolio oder account sein") if scope_kind == "account" and not account_id: raise ValueError("Für den Account-Scope ist ein Konto erforderlich") params: tuple[object, ...] = () where = "WHERE is_active=1 AND performance_included=1" if account_id: where += " AND account_id=?" params = (account_id,) rows = conn.execute(f"SELECT account_id, account_name, account_type, currency FROM accounts {where} ORDER BY account_id", params).fetchall() if scope_kind == "account" and not rows: raise ValueError("Portfolio-Konto wurde nicht gefunden") return [dict(row) for row in rows] def _validate_request(conn: Connection, request: dict[str, Any]) -> tuple[list[dict[str, str]], str, str, str]: source_key = str(request.get("source_key") or "") if source_key not in SOURCE_KEYS: raise ValueError("Datenquelle ist nicht vorhanden oder nicht für Ingestion freigegeben") period_from = str(request.get("period_from") or "") period_to = str(request.get("period_to") or "") start, end = _date(period_from, "Von"), _date(period_to, "Bis") if start > end: raise ValueError("Zeitraum: 'von' darf nicht nach 'bis' liegen") cutoff = str(request.get("data_cutoff") or _now().isoformat()) cutoff = _instant(cutoff, "Data-Cutoff").isoformat() scope_kind = str(request.get("scope_kind") or "portfolio") account_id = str(request.get("account_id") or "") or None accounts = _validate_scope(conn, scope_kind, account_id) return accounts, period_from, period_to, cutoff def _record_ref(source_key: str, raw_id: str) -> str: return _safe_ref("record", f"{source_key}:{raw_id}") def _existing_lineage(conn: Connection, lineage_hash: str) -> tuple[str, str] | None: row = conn.execute( """SELECT target_type, target_id FROM portfolio_ingestion_items WHERE lineage_hash=? AND disposition IN ('new','versioned','unchanged') AND target_id IS NOT NULL ORDER BY created_at DESC LIMIT 1""", (lineage_hash,), ).fetchone() return (str(row["target_type"]), str(row["target_id"])) if row else None def _valuation_snapshot_id(source_key: str, lineage_hash: str, version: int) -> str: return "valuation-" + _hash([source_key, lineage_hash, version])[:24] def _valuation_disposition( conn: Connection, *, source_key: str, account_id: str, valuation_at: str, lineage_hash: str, planned_versions: dict[tuple[str, str], tuple[int, str | None]], ) -> tuple[str, str | None, int, str | None]: existing = _existing_lineage(conn, lineage_hash) if existing: return "unchanged", existing[1], 0, None key = (account_id, valuation_at[:10]) planned = planned_versions.get(key) if planned is None: row = conn.execute( """SELECT snapshot_id,snapshot_version FROM portfolio_valuation_snapshots WHERE scope_kind='account' AND scope_id=? AND substr(valuation_at,1,10)=? ORDER BY snapshot_version DESC,captured_at DESC LIMIT 1""", key, ).fetchone() version = int(row["snapshot_version"]) + 1 if row else 1 supersedes = str(row["snapshot_id"]) if row else None else: version, supersedes = planned snapshot_id = _valuation_snapshot_id(source_key, lineage_hash, version) planned_versions[key] = (version + 1, snapshot_id) return ("versioned" if supersedes else "new"), None, version, supersedes def _legacy_account_records( conn: Connection, accounts: list[dict[str, str]], period_from: str, period_to: str, cutoff: str ) -> list[SourceRecord]: if not accounts: return [] account_map = {row["account_id"]: row for row in accounts} placeholders = ",".join("?" for _ in account_map) rows = conn.execute( f"""SELECT * FROM account_value_snapshots WHERE account_id IN ({placeholders}) AND valuation_date BETWEEN ? AND ? AND created_at <= ? ORDER BY valuation_date, account_id, created_at, snapshot_id""", (*account_map, period_from, period_to, cutoff), ).fetchall() records: list[SourceRecord] = [] seen: set[str] = set() planned_versions: dict[tuple[str, str], tuple[int, str | None]] = {} for row in rows: account_id = str(row["account_id"]) ref = _record_ref("legacy_account_values", str(row["snapshot_id"])) base_payload = { "account": account_id, "at": str(row["valuation_date"]), "value": str(row["total_value_chf"]), "currency": "CHF", "source": str(row["source_type"]), "quality": str(row["quality_status"]), "created": str(row["created_at"]), } fingerprint = _hash(base_payload) quality = str(row["quality_status"] or "").lower() reasons: list[str] = [] status = "complete" try: value = _stored_decimal(row["total_value_chf"], "Kontototal", non_negative=True) except ValueError: value = None status, reasons = "unavailable", ["source_error"] if row["updated_at"] is not None: status, reasons = "unavailable", ["source_error"] if quality in {"stale", "partial"}: status, reasons = "partial", ["stale_snapshot"] elif quality not in {"ok", "complete", "stale", "partial"}: status, reasons = "unavailable", ["source_error"] disposition = "blocked" if value is None or status == "unavailable" else "new" target_id: str | None = None version, supersedes = 0, None if fingerprint in seen: disposition = "duplicate" elif disposition != "blocked": disposition, target_id, version, supersedes = _valuation_disposition( conn, source_key="legacy_account_values", account_id=account_id, valuation_at=str(row["valuation_date"]), lineage_hash=fingerprint, planned_versions=planned_versions, ) seen.add(fingerprint) write_payload = None if disposition in {"new", "versioned"} and value is not None: write_payload = { "account_id": account_id, "value": _decimal_text(value), "currency": "CHF", "base_currency": "CHF", "fx": "1", "valuation_at": str(row["valuation_date"]), "version": version, "supersedes": supersedes, "record_ref": ref, "quality_status": status, "reason_codes": reasons, } records.append(SourceRecord( fingerprint, ref, "valuation", disposition, f"account:{account_id}:{row['valuation_date']}", "portfolio_valuation_snapshot", target_id, fingerprint, account_id, _safe_ref("account", account_id), _safe_label(account_map[account_id]["account_name"]), str(row["valuation_date"]), {"kind": "account_total", "value": _decimal_text(value), "currency": "CHF", "quality_status": status, "reason_codes": reasons}, write_payload, )) return records def _cash_snapshot_records( conn: Connection, accounts: list[dict[str, str]], period_from: str, period_to: str, cutoff: str ) -> list[SourceRecord]: if not accounts: return [] account_map = {row["account_id"]: row for row in accounts} placeholders = ",".join("?" for _ in account_map) rows = conn.execute( f"""SELECT * FROM cash_account_snapshots WHERE account_id IN ({placeholders}) AND balance_date BETWEEN ? AND ? AND created_at <= ? ORDER BY balance_date, account_id, created_at, snapshot_id""", (*account_map, period_from, period_to, cutoff), ).fetchall() records: list[SourceRecord] = [] seen: set[str] = set() planned_versions: dict[tuple[str, str], tuple[int, str | None]] = {} for row in rows: account_id = str(row["account_id"]) ref = _record_ref("cash_account_snapshots", str(row["snapshot_id"])) payload = { "account": account_id, "at": str(row["balance_date"]), "amount": str(row["amount_original"]), "amount_chf": str(row["amount_chf"]), "currency": str(row["currency"]), "source": str(row["source"]), "type": str(row["snapshot_type"]), "created": str(row["created_at"]), } fingerprint = _hash(payload) reasons: list[str] = [] try: value = _stored_decimal(row["amount_chf"], "Cash-Bewertung", non_negative=True) except ValueError: value = None reasons = ["source_error"] disposition = "blocked" if value is None else "new" target_id: str | None = None version, supersedes = 0, None if fingerprint in seen: disposition = "duplicate" elif value is not None: disposition, target_id, version, supersedes = _valuation_disposition( conn, source_key="cash_account_snapshots", account_id=account_id, valuation_at=str(row["balance_date"]), lineage_hash=fingerprint, planned_versions=planned_versions, ) seen.add(fingerprint) write_payload = None if disposition in {"new", "versioned"} and value is not None: write_payload = { "account_id": account_id, "value": _decimal_text(value), "currency": "CHF", "base_currency": "CHF", "fx": "1", "valuation_at": str(row["balance_date"]), "version": version, "supersedes": supersedes, "record_ref": ref, "quality_status": "complete", "reason_codes": reasons, } records.append(SourceRecord( fingerprint, ref, "valuation", disposition, f"account:{account_id}:{row['balance_date']}", "portfolio_valuation_snapshot", target_id, fingerprint, account_id, _safe_ref("account", account_id), _safe_label(account_map[account_id]["account_name"]), str(row["balance_date"]), {"kind": "cash_total", "value": _decimal_text(value), "currency": "CHF", "quality_status": "complete" if value is not None else "unavailable", "reason_codes": reasons}, write_payload, )) return records def _canonical_transaction_records( conn: Connection, accounts: list[dict[str, str]], period_from: str, period_to: str, cutoff: str ) -> list[SourceRecord]: if not accounts: return [] account_map = {row["account_id"]: row for row in accounts} placeholders = ",".join("?" for _ in account_map) rows = conn.execute( f"""SELECT * FROM transactions WHERE account_id IN ({placeholders}) AND trade_date BETWEEN ? AND ? AND created_at <= ? AND COALESCE(is_voided,0)=0 AND COALESCE(is_confirmed,1)=1 ORDER BY trade_date, created_at, transaction_id""", (*account_map, period_from, period_to, cutoff), ).fetchall() records: list[SourceRecord] = [] seen: set[str] = set() for row in rows: account_id = str(row["account_id"]) ref = _record_ref("canonical_transactions", str(row["transaction_id"])) raw_kind = str(row["activity_kind"] or row["transaction_type"] or "").lower() payload = { "transaction": str(row["transaction_id"]), "kind": raw_kind, "account": account_id, "instrument": str(row["instrument_id"] or ""), "at": str(row["event_timestamp"] or row["trade_date"]), "quantity": str(row["quantity"] or ""), "gross": str(row["gross_amount_original"] or ""), "fee": str(row["fee_original"] or "0"), "tax": str(row["tax_original"] or "0"), "net": str(row["net_amount_original"] or ""), "currency": str(row["currency_original"] or ""), "fx": str(row["fx_rate_to_chf"] or ""), "row_hash": str(row["row_hash"] or ""), "source_reference": str(row["source_reference"] or row["external_transaction_id"] or ""), } fingerprint = str(row["row_hash"] or "") or _hash(payload) reasons: list[str] = [] disposition = "unchanged" if fingerprint in seen: disposition, reasons = "duplicate", ["duplicate_source_record"] elif raw_kind not in SUPPORTED_ACTIVITIES or raw_kind in {"initial_position_snapshot"}: disposition, reasons = "blocked", ["unsupported_activity"] elif raw_kind in POSITION_ACTIVITY_TYPES and not row["instrument_id"]: disposition, reasons = "ambiguous", ["ambiguous_instrument"] seen.add(fingerprint) records.append(SourceRecord( fingerprint, ref, "activity", disposition, f"activity:{fingerprint}", "transaction", ref, fingerprint, account_id, _safe_ref("account", account_id), _safe_label(account_map[account_id]["account_name"]), str(row["event_timestamp"] or row["trade_date"]), {"kind": raw_kind, "quantity": str(row["quantity"]) if row["quantity"] is not None else None, "amount": str(row["net_amount_original"] or row["gross_amount_original"]) if row["net_amount_original"] is not None or row["gross_amount_original"] is not None else None, "currency": str(row["currency_original"] or ""), "reason_codes": reasons}, None, )) return records def _postfinance_document(request: dict[str, Any]) -> PostFinanceDocument: encoded = str(request.get("content_base64") or "") if not encoded: raise ValueError("PostFinance-Datei fehlt") try: raw = base64.b64decode(encoded, validate=True) except (binascii.Error, ValueError) as exc: raise ValueError("PostFinance-Datei ist nicht gültig kodiert") from exc return parse_postfinance_portfolio(raw) def _postfinance_account(conn: Connection) -> dict[str, str]: rows = conn.execute( """SELECT a.account_id,a.account_name,a.account_type,a.currency,p.platform_id FROM accounts a JOIN platforms p ON p.platform_id=a.platform_id WHERE a.is_active=1 AND a.performance_included=1 AND lower(p.name || ' ' || a.account_name) LIKE '%postfinance%' AND a.account_type NOT IN ('cash','robo_advisor','total_value') ORDER BY a.account_id""" ).fetchall() if len(rows) != 1: raise ValueError("PostFinance E-Trading-Konto ist nicht eindeutig zugeordnet") return dict(rows[0]) def _postfinance_mapping_decisions(request: dict[str, Any], document: PostFinanceDocument) -> dict[str, dict[str, str]]: raw = request.get("instrument_mappings") or [] if not raw: return {} decisions: dict[str, dict[str, str]] = {} seen_isins: set[str] = set() for value in raw: item = {key: str(part or "").strip() for key, part in dict(value).items()} symbol = item.get("source_symbol", "").upper() isin = item.get("isin", "").upper() if not symbol or symbol in decisions or not re.fullmatch(r"[A-Z]{2}[A-Z0-9]{9}[0-9]", isin): raise ValueError("PostFinance-Instrumentzuordnung ist unvollständig oder doppelt") if isin in seen_isins: raise ValueError("PostFinance-Instrumentzuordnung enthält eine doppelte ISIN") if item.get("source_venue") != "unknown": raise ValueError("PostFinance-Quellhandelsplatz muss ohne Beleg unknown bleiben") if ( item.get("provider") != "fmp" or not item.get("instrument_name") or not item.get("valuation_symbol") or not item.get("valuation_venue") ): raise ValueError("PostFinance-Bewertungszuordnung ist unvollständig") decisions[symbol] = item seen_isins.add(isin) source_symbols = {item.source_label.upper() for item in document.positions} if set(decisions) != source_symbols: raise ValueError("PostFinance-Instrumentzuordnungen müssen alle und nur die Quellpositionen abdecken") for position in document.positions: decision = decisions[position.source_label.upper()] if decision.get("asset_class") != position.asset_class: raise ValueError("PostFinance-Instrumentzuordnung widerspricht der Anlageklasse") if decision.get("source_currency", "").upper() != position.currency: raise ValueError("PostFinance-Instrumentzuordnung widerspricht der Quellwährung") if decision.get("valuation_currency", "").upper() != position.currency: raise ValueError("Bewertungswährung widerspricht der bestätigten Handelslinie") return decisions def confirm_postfinance_instrument_mappings( conn: Connection, request: dict[str, Any], *, created_by: str = "user", ) -> dict[str, Any]: """Apply the explicitly approved PostFinance identity and valuation mappings atomically.""" document = _postfinance_document(request) decisions = _postfinance_mapping_decisions(request, document) if not decisions: raise ValueError("PostFinance mapping decisions are required") now = _now().isoformat() mapping_ids: list[str] = [] instrument_ids: list[str] = [] try: conn.execute("BEGIN IMMEDIATE") for item in document.positions: decision = decisions[item.source_label.upper()] rows = conn.execute( "SELECT * FROM instruments WHERE upper(isin)=? ORDER BY instrument_id", (decision["isin"],), ).fetchall() if len(rows) > 1: raise ValueError(f"Duplicate canonical ISIN exists: {decision['isin']}") if rows: instrument_id = str(rows[0]["instrument_id"]) else: instrument_id = stable_id("instrument", decision["isin"]) conn.execute( """INSERT INTO instruments( instrument_id,asset_class,name,ticker,isin,currency,trading_currency,is_active, position_category,instrument_status,valuation_policy,notes,created_at) VALUES(?,?,?,?,?,?,?,1,'postfinance_baseline','active','live_price',?,?)""", (instrument_id, item.asset_class, decision["instrument_name"], decision["source_symbol"], decision["isin"], item.currency, item.currency, "Canonical identity from explicitly confirmed PostFinance mapping decision", now), ) mapping_id = confirm_instrument_price_mapping( conn, instrument_id=instrument_id, provider=decision["provider"], provider_symbol=decision["valuation_symbol"], provider_market=decision["valuation_venue"], exchange=decision["valuation_venue"], currency=decision["valuation_currency"], confidence="1", note="Sprint 9 explicitly approved PostFinance mapping decision", trading_currency=decision["valuation_currency"], canonical_isin=decision["isin"], instrument_name=decision["instrument_name"], source_symbol=decision["source_symbol"], source_venue=decision["source_venue"], source_currency=item.currency, created_by=created_by, commit=False, ) mapping_ids.append(mapping_id) instrument_ids.append(instrument_id) record_audit_event( conn, source="postfinance_mapping_set", action="confirm_postfinance_instrument_mappings", entity_type="instrument_mapping_set", entity_id="postfinance-2026-05-14", new_values={ "mapping_count": len(mapping_ids), "instrument_count": len(set(instrument_ids)), "decisions": [ {key: decisions[symbol][key] for key in ( "isin", "source_symbol", "source_venue", "valuation_symbol", "valuation_currency", "provider", )} for symbol in sorted(decisions) ], }, user_text_note="Explicitly approved Sprint 9 PostFinance instrument mappings", confirmed=True, created_by=created_by, ) conn.commit() except Exception: conn.rollback() raise return { "mapping_count": len(mapping_ids), "instrument_count": len(set(instrument_ids)), "mapping_ids": mapping_ids, } def _instrument_for_source( conn: Connection, label: str, asset_class: str, currency: str, decision: dict[str, str] | None = None, ) -> tuple[str, bool]: if decision: isin = decision["isin"].upper() matches = conn.execute( "SELECT instrument_id,asset_class FROM instruments WHERE is_active=1 AND UPPER(isin)=?", (isin,), ).fetchall() if len(matches) > 1: raise ValueError("Mehrere kanonische Instrumente verwenden dieselbe ISIN") if matches: if str(matches[0]["asset_class"]).lower() != asset_class: raise ValueError("Kanonisches Instrument widerspricht der bestätigten Anlageklasse") return str(matches[0]["instrument_id"]), False source_id = "pf-instrument-" + _hash([isin])[:20] existing = conn.execute("SELECT instrument_id,isin FROM instruments WHERE instrument_id=?", (source_id,)).fetchone() if existing and str(existing["isin"] or "").upper() != isin: raise ValueError("Deterministische Instrumentidentität kollidiert") return source_id, existing is None normalized = normalize_label(label) matches = [ row for row in conn.execute( "SELECT instrument_id,name,asset_class,currency FROM instruments WHERE is_active=1" ).fetchall() if normalize_label(str(row["name"])) == normalized and str(row["asset_class"]).lower() == asset_class and str(row["currency"]).upper() == currency ] if len(matches) == 1: return str(matches[0]["instrument_id"]), False source_id = "pf-instrument-" + _hash([normalized, asset_class, currency])[:20] existing = conn.execute("SELECT instrument_id FROM instruments WHERE instrument_id=?", (source_id,)).fetchone() return source_id, existing is None def _postfinance_disposition(conn: Connection, fingerprint: str) -> tuple[str, str | None]: existing = _existing_lineage(conn, fingerprint) return ("unchanged", existing[1]) if existing else ("new", None) def _manual_placeholder_for_supersession( conn: Connection, *, account_id: str, instrument_id: str, quantity: Decimal, enabled: bool, owner_resolutions: list[dict[str, Any]], ) -> tuple[str | None, str | None, dict[str, Any] | None]: rows = conn.execute( """SELECT transaction_id,transaction_type,quantity,source_type,source_id, external_transaction_id,row_hash,notes,price_original,gross_amount_original FROM transactions WHERE account_id=? AND instrument_id=? AND COALESCE(is_voided,0)=0 AND COALESCE(is_confirmed,1)=1 ORDER BY trade_date,transaction_id""", (account_id, instrument_id), ).fetchall() if not rows: cross_account = conn.execute( """SELECT t.transaction_id,t.account_id,t.transaction_type,t.quantity,t.source_type,t.source_id, t.external_transaction_id,t.row_hash,t.price_original,t.gross_amount_original,i.isin FROM transactions t JOIN instruments i ON i.instrument_id=t.instrument_id WHERE t.account_id<>? AND t.instrument_id=? AND COALESCE(t.is_voided,0)=0 AND COALESCE(t.is_confirmed,1)=1 ORDER BY t.account_id,t.trade_date,t.transaction_id""", (account_id, instrument_id), ).fetchall() if not cross_account: return None, None, None if len(cross_account) != 1: return None, "relevant_transaction_history_present", None candidate = cross_account[0] try: candidate_quantity = Decimal(str(candidate["quantity"])) except (InvalidOperation, TypeError): return None, "manual_placeholder_quantity_invalid", None unlineaged_placeholder = ( str(candidate["transaction_type"]) == "initial_position_snapshot" and str(candidate["source_type"]) in {"initial_snapshot", "manual_dashboard", "vue_manual_position"} and not candidate["external_transaction_id"] and not candidate["row_hash"] and candidate["price_original"] is None and Decimal(str(candidate["gross_amount_original"] or "0")) == 0 and ( not candidate["source_id"] or str(candidate["source_id"]) == str(candidate["transaction_id"]) or str(candidate["source_id"]).startswith("original-entry-") ) ) if candidate_quantity != quantity: return None, "manual_placeholder_quantity_conflict", None if not unlineaged_placeholder: return None, "evidenced_cost_basis_or_lineage_present", None matches: list[tuple[Any, dict[str, Any]]] = [] for row in cross_account: for resolution in owner_resolutions: try: expected_quantity = Decimal(str(resolution.get("expected_quantity") or "0")) except (InvalidOperation, TypeError): continue if ( str(resolution.get("manual_transaction_id") or "") == str(row["transaction_id"]) and str(resolution.get("expected_source_account_id") or "") == str(row["account_id"]) and str(resolution.get("target_account_id") or "") == account_id and str(resolution.get("expected_isin") or "").upper() == str(row["isin"] or "").upper() and expected_quantity == quantity and resolution.get("reason_code") == "manual_placeholder_wrong_account_confirmed_by_owner" ): matches.append((row, resolution)) if len(matches) != 1: return None, "manual_placeholder_other_account_conflict", None row, resolution = matches[0] try: owner_confirmed_at = _instant(str(resolution.get("owner_confirmed_at") or ""), "Eigentümerbestätigung") except ValueError: return None, "owner_confirmed_resolution_invalid", None if owner_confirmed_at > _now(): return None, "owner_confirmed_resolution_future_dated", None note = str(resolution.get("owner_confirmation_note") or "").strip() if len(note) < 20: return None, "owner_confirmed_resolution_invalid", None decision = { "reason_code": "manual_placeholder_wrong_account_confirmed_by_owner", "owner_confirmed_at": owner_confirmed_at.isoformat(), "owner_confirmation_note": note, "expected_source_account_id": str(row["account_id"]), "target_account_id": account_id, "expected_isin": str(row["isin"]), "expected_quantity": postfinance_decimal_text(quantity), } return str(row["transaction_id"]), None, decision if not enabled: return None, "manual_position_overlap_requires_confirmation", None if len(rows) != 1: return None, "relevant_transaction_history_present", None row = rows[0] try: stored_quantity = Decimal(str(row["quantity"])) except (InvalidOperation, TypeError): return None, "manual_placeholder_quantity_invalid", None unlineaged_placeholder = ( str(row["transaction_type"]) == "initial_position_snapshot" and str(row["source_type"]) in {"initial_snapshot", "manual_dashboard", "vue_manual_position"} and not row["external_transaction_id"] and not row["row_hash"] and row["price_original"] is None and Decimal(str(row["gross_amount_original"] or "0")) == 0 and ( not row["source_id"] or str(row["source_id"]) == str(row["transaction_id"]) or str(row["source_id"]).startswith("original-entry-") ) ) if stored_quantity != quantity: return None, "manual_placeholder_quantity_conflict", None if not unlineaged_placeholder: return None, "evidenced_cost_basis_or_lineage_present", None return str(row["transaction_id"]), None, None def _postfinance_records( conn: Connection, document: PostFinanceDocument, request: dict[str, Any], ) -> list[SourceRecord]: account = _postfinance_account(conn) decisions = _postfinance_mapping_decisions(request, document) allow_supersession = bool(request.get("allow_manual_baseline_supersession")) owner_resolutions = list(request.get("owner_confirmed_manual_placeholder_resolutions") or []) account_id = str(account["account_id"]) account_ref = _safe_ref("account", account_id) account_label = _safe_label(str(account["account_name"])) platform_id = str(account["platform_id"]) fx_rates = {item.currency: item.fx_rate_to_chf for item in document.cash} fx_rates["CHF"] = Decimal("1") records: list[SourceRecord] = [] duplicate_labels = { item.normalized_label for item in document.positions if sum(other.normalized_label == item.normalized_label for other in document.positions) > 1 } for item in document.positions: decision = decisions.get(item.source_label.upper()) instrument_id, create_instrument = _instrument_for_source( conn, item.source_label, item.asset_class, item.currency, decision ) base_reasons = ["baseline_only", "missing_transaction_history"] if not decision: base_reasons.append("missing_instrument_identifier") blocked = item.normalized_label in duplicate_labels activity_fingerprint = _hash([document.file_hash, item.row_hash, "initial_position"]) disposition, target_id = _postfinance_disposition(conn, activity_fingerprint) supersede_transaction_id: str | None = None overlap_reason: str | None = None supersession_resolution: dict[str, Any] | None = None if disposition == "new" and not create_instrument: supersede_transaction_id, overlap_reason, supersession_resolution = _manual_placeholder_for_supersession( conn, account_id=account_id, instrument_id=instrument_id, quantity=item.quantity, enabled=allow_supersession, owner_resolutions=owner_resolutions, ) if overlap_reason: blocked = True base_reasons.append(overlap_reason) if blocked: disposition, target_id = "blocked", None transaction_id = "pf-tx-" + activity_fingerprint[:20] activity_summary = { "kind": "initial_position_snapshot", "instrument_label": item.source_label, "canonical_isin": decision.get("isin") if decision else None, "source_symbol": item.source_label, "source_venue": "unknown", "asset_class": item.asset_class, "quantity": postfinance_decimal_text(item.quantity), "currency": item.currency, "fx_rate_to_chf": postfinance_decimal_text(fx_rates.get(item.currency)), "supersedes_transaction_ref": _safe_ref("transaction", supersede_transaction_id) if supersede_transaction_id else None, "quality_status": "partial", "reason_codes": sorted(set(base_reasons + ([ supersession_resolution["reason_code"] if supersession_resolution else "manual_placeholder_supersession" ] if supersede_transaction_id else []))), } records.append(SourceRecord( activity_fingerprint, _safe_ref("record", activity_fingerprint), "activity", disposition, f"postfinance:initial_position:{item.row_hash}", "transaction", target_id, activity_fingerprint, account_id, account_ref, account_label, document.as_of, activity_summary, None if disposition != "new" else { "write_kind": "postfinance_initial_position", "transaction_id": transaction_id, "account_id": account_id, "instrument_id": instrument_id, "platform_id": platform_id, "instrument_label": item.source_label, "instrument_name": decision.get("instrument_name") if decision else item.source_label, "isin": decision.get("isin") if decision else None, "source_symbol": item.source_label, "source_venue": "unknown", "asset_class": item.asset_class, "quantity": postfinance_decimal_text(item.quantity), "currency": item.currency, "fx_rate_to_chf": postfinance_decimal_text(fx_rates.get(item.currency)), "row_hash": activity_fingerprint, "as_of": document.as_of, "create_instrument": create_instrument, "supersede_transaction_id": supersede_transaction_id, "supersession_resolution": supersession_resolution, }, )) snapshot_fingerprint = _hash([document.file_hash, item.row_hash, "reported_position"]) disposition, target_id = _postfinance_disposition(conn, snapshot_fingerprint) snapshot_id = "pf-position-" + snapshot_fingerprint[:20] existing_snapshot = conn.execute( """SELECT position_snapshot_id,quantity,market_value_chf FROM positions_snapshot WHERE snapshot_date=? AND account_id=? AND instrument_id=?""", (document.as_of, account_id, instrument_id), ).fetchone() snapshot_reasons = list(base_reasons) if blocked: disposition, target_id = "ambiguous", None snapshot_reasons.append("ambiguous_instrument") elif existing_snapshot and disposition != "unchanged": same = ( _stored_decimal(existing_snapshot["quantity"], "Menge") == item.quantity and _stored_decimal(existing_snapshot["market_value_chf"], "Positionswert") == item.market_value_chf ) disposition = "unchanged" if same else "blocked" target_id = str(existing_snapshot["position_snapshot_id"]) if same else None if not same: snapshot_reasons.append("conflicting_snapshot") records.append(SourceRecord( snapshot_fingerprint, _safe_ref("record", snapshot_fingerprint), "valuation", disposition, f"postfinance:position:{item.row_hash}", "positions_snapshot", target_id, snapshot_fingerprint, account_id, account_ref, account_label, document.as_of, {"kind": "position_snapshot", "instrument_label": item.source_label, "canonical_isin": decision.get("isin") if decision else None, "asset_class": item.asset_class, "quantity": postfinance_decimal_text(item.quantity), "market_value_chf": postfinance_decimal_text(item.market_value_chf), "market_price_original": postfinance_decimal_text(item.market_price), "source_reported_average_cost": postfinance_decimal_text(item.average_cost), "source_reported_cost_total": postfinance_decimal_text(item.cost_total), "cost_basis_semantics": "unverified_source_display_only", "currency": item.currency, "quality_status": "partial", "reason_codes": sorted(set(snapshot_reasons))}, None if disposition != "new" else { "write_kind": "postfinance_position_snapshot", "snapshot_id": snapshot_id, "account_id": account_id, "platform_id": platform_id, "instrument_id": instrument_id, "quantity": postfinance_decimal_text(item.quantity), "average_cost": None, "cost_total": None, "market_price": postfinance_decimal_text(item.market_price), "market_value_chf": postfinance_decimal_text(item.market_value_chf), "pnl_chf": None, "weight_pct": postfinance_decimal_text(item.weight_pct), "as_of": document.as_of, }, )) cash_fingerprint = _hash([document.file_hash, "cash", [item.row_hash for item in document.cash]]) cash_disposition, cash_target = _postfinance_disposition(conn, cash_fingerprint) computed_cash_total = sum((item.amount_chf for item in document.cash), Decimal("0")) cash_components = [ {"source_label": f"Kontosaldo {item.currency}", "currency": item.currency, "amount_original": postfinance_decimal_text(item.amount_original), "displayed_fx_rate_to_chf": postfinance_decimal_text(item.fx_rate_to_chf), "fx_as_of": document.as_of, "computed_amount_chf_unrounded": postfinance_decimal_text(item.amount_chf), "computed_amount_chf_display": postfinance_decimal_text(item.amount_chf.quantize(Decimal("0.01"))), "reported_currency_account_value_chf": postfinance_decimal_text(item.reported_account_value_chf)} for item in document.cash ] records.append(SourceRecord( cash_fingerprint, _safe_ref("record", cash_fingerprint), "valuation", cash_disposition, f"postfinance:cash:{document.as_of}", "cash_account_snapshot", cash_target, cash_fingerprint, account_id, account_ref, account_label, document.as_of, {"kind": "cash_snapshot", "value": postfinance_decimal_text(document.reported_cash_value_chf), "currency": "CHF", "components": cash_components, "computed_from_displayed_fx_chf": postfinance_decimal_text(computed_cash_total), "displayed_fx_difference_chf": postfinance_decimal_text(computed_cash_total - document.reported_cash_value_chf), "quality_status": "complete", "reason_codes": ["displayed_fx_rate_is_rounded"]}, None if cash_disposition != "new" else { "write_kind": "postfinance_cash_snapshot", "account_id": account_id, "value": postfinance_decimal_text(document.reported_cash_value_chf), "as_of": document.as_of, "components": cash_components, }, )) for target_kind in ("reported_total", "performance_baseline"): fingerprint = _hash([document.file_hash, target_kind, postfinance_decimal_text(document.total_value_chf)]) disposition, target_id = _postfinance_disposition(conn, fingerprint) write_kind = "postfinance_account_total" if target_kind == "reported_total" else "postfinance_performance_baseline" reasons = ["baseline_only", "missing_transaction_history"] records.append(SourceRecord( fingerprint, _safe_ref("record", fingerprint), "valuation", disposition, f"postfinance:{target_kind}:{document.as_of}", "account_value_snapshot" if target_kind == "reported_total" else "portfolio_valuation_snapshot", target_id, fingerprint, account_id, account_ref, account_label, document.as_of, {"kind": target_kind, "value": postfinance_decimal_text(document.total_value_chf), "currency": "CHF", "quality_status": "partial", "reason_codes": reasons}, None if disposition != "new" else { "write_kind": write_kind, "account_id": account_id, "value": postfinance_decimal_text(document.total_value_chf), "as_of": document.as_of, "record_ref": _safe_ref("record", fingerprint), }, )) if "unsupported_row" in document.reason_codes or "portfolio_total_mismatch" in document.reason_codes: reason = "unsupported_activity" if "unsupported_row" in document.reason_codes else "source_error" fingerprint = _hash([document.file_hash, reason]) records.append(SourceRecord( fingerprint, _safe_ref("record", fingerprint), "valuation", "blocked", f"postfinance:review:{reason}", None, None, fingerprint, account_id, account_ref, account_label, document.as_of, {"kind": "review_required", "quality_status": "unavailable", "reason_codes": [reason]}, None, )) return records def _records(conn: Connection, source_key: str, accounts: list[dict[str, str]], period_from: str, period_to: str, cutoff: str, request: dict[str, Any] | None = None) -> list[SourceRecord]: if source_key == "postfinance_etrading": if request is None: raise ValueError("PostFinance-Datei fehlt") return _postfinance_records(conn, _postfinance_document(request), request) if source_key == "legacy_account_values": return _legacy_account_records(conn, accounts, period_from, period_to, cutoff) if source_key == "cash_account_snapshots": return _cash_snapshot_records(conn, accounts, period_from, period_to, cutoff) return _canonical_transaction_records(conn, accounts, period_from, period_to, cutoff) def _preview_payload(request: dict[str, Any], records: list[SourceRecord], source_revision: str) -> dict[str, Any]: return { "version": INGESTION_VERSION, "source_key": request["source_key"], "scope_kind": request.get("scope_kind") or "portfolio", "account_id": request.get("account_id"), "period_from": request["period_from"], "period_to": request["period_to"], "data_cutoff": request["data_cutoff"], "source_revision": source_revision, "instrument_mappings": request.get("instrument_mappings") or [], "allow_manual_baseline_supersession": bool(request.get("allow_manual_baseline_supersession")), "owner_confirmed_manual_placeholder_resolutions": request.get( "owner_confirmed_manual_placeholder_resolutions" ) or [], "records": [[r.source_record_fingerprint, r.record_kind, r.disposition, r.logical_key, r.lineage_hash] for r in records], } def preview_ingestion( conn: Connection, request: dict[str, Any], *, preview_created_at: str | None = None, confirmation_id: str | None = None, ) -> dict[str, Any]: effective_request = dict(request) postfinance_document: PostFinanceDocument | None = None if str(request.get("source_key") or "") == "postfinance_etrading": postfinance_document = _postfinance_document(request) effective_request["period_from"] = postfinance_document.as_of effective_request["period_to"] = postfinance_document.as_of accounts, period_from, period_to, cutoff = _validate_request(conn, effective_request) normalized_request = { "source_key": str(effective_request["source_key"]), "scope_kind": str(effective_request.get("scope_kind") or "portfolio"), "account_id": str(effective_request.get("account_id") or "") or None, "period_from": period_from, "period_to": period_to, "data_cutoff": cutoff, } records = _records(conn, normalized_request["source_key"], accounts, period_from, period_to, cutoff, effective_request) source_revision = postfinance_document.file_hash if postfinance_document else _hash([[r.source_record_fingerprint, r.disposition, r.logical_key] for r in records]) canonical_request = dict(normalized_request) if postfinance_document: canonical_request["instrument_mappings"] = effective_request.get("instrument_mappings") or [] canonical_request["allow_manual_baseline_supersession"] = bool( effective_request.get("allow_manual_baseline_supersession") ) canonical_request["owner_confirmed_manual_placeholder_resolutions"] = effective_request.get( "owner_confirmed_manual_placeholder_resolutions" ) or [] canonical = _preview_payload(canonical_request, records, source_revision) fingerprint = _hash(canonical) created = _instant(preview_created_at, "Preview-Zeitpunkt") if preview_created_at else _now() expires = created + timedelta(minutes=PREVIEW_TTL_MINUTES) counts = {key: sum(r.disposition == key for r in records) for key in ("new", "unchanged", "duplicate", "ambiguous", "blocked", "versioned")} counts["discovered"] = len(records) preview_id = "portfolio-preview-" + fingerprint[:24] active_confirmation_id = confirmation_id or "portfolio-confirm-" + uuid.uuid4().hex payload_hash = _hash({ "preview_id": preview_id, "confirmation_id": active_confirmation_id, "input_fingerprint": fingerprint, "preview_created_at": created.isoformat(), }) response = { **normalized_request, "preview_id": preview_id, "confirmation_id": active_confirmation_id, "preview_created_at": created.isoformat(), "expires_at": expires.isoformat(), "source_revision": source_revision, "input_fingerprint": fingerprint, "payload_hash": payload_hash, "counts": counts, "planned_activities": [r.summary for r in records if r.record_kind == "activity" and r.disposition in {"new", "versioned"}], "planned_valuations": [r.summary for r in records if r.record_kind == "valuation" and r.disposition in {"new", "versioned"}], "planned_corrections": [ r.summary for r in records if r.disposition == "versioned" or r.summary.get("supersedes_transaction_ref") ], "quality_impact": { "status": "unavailable" if counts["blocked"] and counts["discovered"] == counts["blocked"] else "partial" if counts["blocked"] or counts["ambiguous"] or any(r.summary.get("reason_codes") for r in records) else "complete", "reason_codes": sorted({code for r in records for code in r.summary.get("reason_codes", [])}), }, "items": [ {"source_record_ref": r.source_record_ref, "record_kind": r.record_kind, "disposition": r.disposition, "account_ref": r.account_ref, "account_label": r.account_label, "as_of": r.as_of, "summary": r.summary} for r in records[:200] ], "truncated": len(records) > 200, } if postfinance_document: operation_counts = { "buys": 0, "sells": 0, "distributions": 0, "fees": 0, "taxes": 0, "deposits": 0, "withdrawals": 0, "positions": len(postfinance_document.positions), "cash_balances": len(postfinance_document.cash), } computed_cash = sum((item.amount_chf for item in postfinance_document.cash), Decimal("0")) row_positions = sum((item.market_value_chf for item in postfinance_document.positions), Decimal("0")) cash_components = [ {"source_label": f"Kontosaldo {item.currency}", "source_currency": item.currency, "source_amount": postfinance_decimal_text(item.amount_original), "displayed_fx_rate_to_chf": postfinance_decimal_text(item.fx_rate_to_chf), "fx_as_of": postfinance_document.as_of, "computed_chf_unrounded": postfinance_decimal_text(item.amount_chf), "computed_chf_display": postfinance_decimal_text(item.amount_chf.quantize(Decimal("0.01"))), "reported_currency_account_value_chf": postfinance_decimal_text(item.reported_account_value_chf)} for item in postfinance_document.cash ] response.update({ "document_type": "PostFinance Portfolio-Bewertung", "file_hash": postfinance_document.file_hash, "operation_counts": operation_counts, "baseline_only": True, "expected_changes": { "records": len(records), "positions": sum(r.target_type == "positions_snapshot" and r.disposition == "new" for r in records), "cash": sum(r.target_type == "cash_account_snapshot" and r.disposition == "new" for r in records), "cash_components": len(postfinance_document.cash), "account_valuations": sum(r.target_type == "portfolio_valuation_snapshot" and r.disposition == "new" for r in records), "manual_supersessions": sum(bool(r.summary.get("supersedes_transaction_ref")) for r in records), }, "cash_reconciliation": { "components": cash_components, "computed_cash_from_displayed_fx_chf": postfinance_decimal_text(computed_cash), "reported_cash_chf": postfinance_decimal_text(postfinance_document.reported_cash_value_chf), "cash_difference_chf": postfinance_decimal_text(computed_cash - postfinance_document.reported_cash_value_chf), "sum_position_rows_chf": postfinance_decimal_text(row_positions), "reported_positions_chf": postfinance_decimal_text(postfinance_document.reported_positions_value_chf), "position_display_difference_chf": postfinance_decimal_text(row_positions - postfinance_document.reported_positions_value_chf), "calculated_from_displayed_rows_chf": postfinance_decimal_text(row_positions + computed_cash), "reported_document_total_chf": postfinance_decimal_text(postfinance_document.total_value_chf), "total_difference_chf": postfinance_decimal_text(row_positions + computed_cash - postfinance_document.total_value_chf), "source_total_identity_chf": postfinance_decimal_text( postfinance_document.reported_positions_value_chf + postfinance_document.reported_cash_value_chf ), "explanation": "Displayed FX rates and row CHF values are rounded; source aggregate positions plus source aggregate cash equals the document total exactly. No balancing entry is created.", }, "performance_impact": { "history_before": "unavailable", "twr": "restricted", "mwr": "restricted", "reason_codes": ["baseline_only", "missing_transaction_history"], }, }) return response def _insert_valuation(conn: Connection, record: SourceRecord, source_key: str, captured_at: str) -> str: payload = record.write_payload if payload is None: raise ValueError("Vorschau enthält keine schreibbare Bewertung") snapshot_id = _valuation_snapshot_id(source_key, record.lineage_hash, payload["version"]) conn.execute( """INSERT INTO portfolio_valuation_snapshots( snapshot_id, scope_kind, scope_id, account_id, value_original, currency, base_currency, fx_rate_to_base, fx_direction, valuation_at, source, captured_at, snapshot_version, supersedes_snapshot_id, source_reference, quality_status, reason_codes_json ) VALUES(?, 'account', ?, ?, ?, ?, ?, ?, 'original_to_base', ?, ?, ?, ?, ?, ?, ?, ?)""", (snapshot_id, payload["account_id"], payload["account_id"], payload["value"], payload["currency"], payload["base_currency"], payload["fx"], payload["valuation_at"], f"ingestion:{source_key}", captured_at, payload["version"], payload["supersedes"], payload["record_ref"], payload["quality_status"], json.dumps(payload["reason_codes"], sort_keys=True)), ) return snapshot_id def _insert_postfinance_record(conn: Connection, record: SourceRecord, captured_at: str) -> str: payload = record.write_payload if payload is None: raise ValueError("PostFinance-Vorschau enthält keinen schreibbaren Datensatz") kind = str(payload["write_kind"]) if kind == "postfinance_initial_position": instrument_id = str(payload["instrument_id"]) if payload.get("create_instrument"): conn.execute( """INSERT INTO instruments( instrument_id,asset_class,name,ticker,isin,currency,trading_currency,is_active, position_category,instrument_status,valuation_policy,notes,created_at) VALUES(?,?,?,?,?,?,?,1,'postfinance_baseline','active','live_price',?,?)""", (instrument_id, payload["asset_class"], payload["instrument_name"], payload["source_symbol"], payload.get("isin"), payload["currency"], payload["currency"], "Canonical PostFinance baseline identity; source venue unknown", captured_at), ) supersede_transaction_id = payload.get("supersede_transaction_id") resolution = payload.get("supersession_resolution") supersession_reason = ( "Owner-confirmed wrong-account manual placeholder superseded by authoritative PostFinance baseline" if resolution else "Authoritative PostFinance baseline supersedes unlineaged manual placeholder" ) if supersede_transaction_id: old = conn.execute( "SELECT transaction_id,is_voided,source_type,quantity FROM transactions WHERE transaction_id=?", (supersede_transaction_id,), ).fetchone() if old is None or int(old["is_voided"] or 0) != 0: raise ValueError("Manueller Platzhalter kann nicht mehr sicher supersediert werden") conn.execute( """UPDATE transactions SET is_voided=1,voided_at=?,void_reason=?,voided_by='postfinance_confirm',updated_at=? WHERE transaction_id=?""", (captured_at, supersession_reason, captured_at, supersede_transaction_id), ) record_audit_event( conn, source="portfolio_ingestion_api", action="postfinance_manual_placeholder_superseded", entity_type="transaction", entity_id=str(supersede_transaction_id), old_values={"is_voided": 0, "source_type": old["source_type"], "quantity": old["quantity"]}, new_values={ "is_voided": 1, "superseded_by_transaction_id": payload["transaction_id"], "superseded_by_source_record_ref": record.source_record_ref, "reason": supersession_reason, "reason_code": ( resolution["reason_code"] if resolution else "manual_placeholder_same_account_supersession" ), "owner_confirmation": resolution, }, user_text_note=supersession_reason, confirmed=True, created_by="user", ) fx = payload.get("fx_rate_to_chf") conn.execute( """INSERT INTO transactions( transaction_id,transaction_type,activity_kind,account_id,instrument_id,trade_date, event_timestamp,quantity,currency_original,fx_rate_to_chf,fx_source,fx_status, source_type,source_id,external_transaction_id,row_hash,is_confirmed,quality_status, correction_of_transaction_id,correction_reason,notes,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,1,'incomplete',?,?,?,?)""", (payload["transaction_id"], "initial_position_snapshot", "initial_position_snapshot", payload["account_id"], instrument_id, payload["as_of"], payload["as_of"], payload["quantity"], payload["currency"], fx, "document_identity" if payload["currency"] == "CHF" else "postfinance_statement" if fx else "missing_in_source", "ok" if fx else "missing", "postfinance_etrading_snapshot", record.source_record_ref, record.source_record_ref, payload["row_hash"], supersede_transaction_id, supersession_reason if supersede_transaction_id else None, "Authoritative initial-position baseline; purchase price and historical cost basis unknown", captured_at), ) return str(payload["transaction_id"]) if kind == "postfinance_position_snapshot": conn.execute( """INSERT INTO positions_snapshot( position_snapshot_id,snapshot_date,account_id,platform_id,instrument_id,quantity, average_cost_original,cost_basis_original,market_price_original,market_value_chf, unrealized_price_pnl_chf,portfolio_weight_pct,data_quality_status,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?, 'incomplete',?)""", (payload["snapshot_id"], payload["as_of"], payload["account_id"], payload["platform_id"], payload["instrument_id"], payload["quantity"], payload["average_cost"], payload["cost_total"], payload["market_price"], payload["market_value_chf"], payload["pnl_chf"], payload["weight_pct"], captured_at), ) return str(payload["snapshot_id"]) if kind == "postfinance_cash_snapshot": for component in payload["components"]: component_id = "pf-cash-component-" + _hash([ payload["account_id"], payload["as_of"], component["currency"], component["amount_original"] ])[:20] conn.execute( """INSERT INTO cash_balances( cash_balance_id,account_id,balance_date,currency,amount_original,fx_rate_to_chf, amount_chf,source_type,quality_status,notes,created_at) VALUES(?,?,?,?,?,?,?,'postfinance_etrading','partial',?,?)""", (component_id, payload["account_id"], payload["as_of"], component["currency"], component["amount_original"], component["displayed_fx_rate_to_chf"], component["computed_amount_chf_unrounded"], "Original currency retained; CHF component is calculated from the statement's displayed rounded FX rate", captured_at), ) snapshot_id = "pf-cash-" + record.lineage_hash[:20] conn.execute( """INSERT INTO cash_account_snapshots( snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency, amount_chf,source,note,created_at,created_by) VALUES(?,?,'reconciliation',?,?,'CHF',?,'postfinance_etrading',?,?,'user')""", (snapshot_id, payload["account_id"], payload["as_of"], payload["value"], payload["value"], f"{len(payload['components'])} original-currency components retained separately; aggregate CHF is source-reported", captured_at), ) return snapshot_id if kind == "postfinance_account_total": snapshot_id = "pf-account-total-" + record.lineage_hash[:20] conn.execute( """INSERT INTO account_value_snapshots( snapshot_id,account_id,valuation_date,total_value_chf,currency,source_type, quality_status,created_at) VALUES(?,?,?,?,'CHF','postfinance_etrading','partial',?)""", (snapshot_id, payload["account_id"], payload["as_of"], payload["value"], captured_at), ) return snapshot_id if kind == "postfinance_performance_baseline": row = conn.execute( """SELECT snapshot_id,snapshot_version FROM portfolio_valuation_snapshots WHERE scope_kind='account' AND scope_id=? AND substr(valuation_at,1,10)=? ORDER BY snapshot_version DESC LIMIT 1""", (payload["account_id"], payload["as_of"]), ).fetchone() version = int(row["snapshot_version"]) + 1 if row else 1 snapshot_id = "pf-valuation-" + record.lineage_hash[:20] conn.execute( """INSERT INTO portfolio_valuation_snapshots( snapshot_id,scope_kind,scope_id,account_id,value_original,currency,base_currency, fx_rate_to_base,fx_direction,valuation_at,source,captured_at,snapshot_version, supersedes_snapshot_id,source_reference,quality_status,reason_codes_json) VALUES(?,'account',?,?,?,'CHF','CHF','1','original_to_base',?, 'ingestion:postfinance_etrading',?,?,?,?, 'partial',?)""", (snapshot_id, payload["account_id"], payload["account_id"], payload["value"], payload["as_of"], captured_at, version, row["snapshot_id"] if row else None, payload["record_ref"], json.dumps(["baseline_only", "missing_transaction_history"])), ) return snapshot_id raise ValueError("Unbekannter PostFinance-Schreibdatensatz") def confirm_ingestion(conn: Connection, request: dict[str, Any]) -> dict[str, Any]: if request.get("confirm") is not True: raise ValueError("Explizite Bestätigung ist erforderlich") preview_created_at = str(request.get("preview_created_at") or "") created = _instant(preview_created_at, "Preview-Zeitpunkt") confirmation_id = str(request.get("confirmation_id") or "") if not confirmation_id: raise ValueError("Bestätigungskennung fehlt") base_request = {key: request.get(key) for key in ( "source_key", "scope_kind", "account_id", "period_from", "period_to", "data_cutoff", "file_name", "content_base64", "instrument_mappings", "allow_manual_baseline_supersession", "owner_confirmed_manual_placeholder_resolutions", )} expected_existing_hash = _hash({ "preview_id": str(request.get("preview_id") or ""), "confirmation_id": confirmation_id, "input_fingerprint": str(request.get("input_fingerprint") or ""), "preview_created_at": created.isoformat(), }) try: conn.execute("BEGIN IMMEDIATE") existing = conn.execute( """SELECT batch_id, payload_hash, audit_id, preview_id, source_key, scope_kind, scope_id, period_from, period_to, data_cutoff, source_revision, input_fingerprint FROM portfolio_ingestion_batches WHERE confirmation_id=?""", (confirmation_id,), ).fetchone() if existing: request_identity = ( str(request.get("preview_id") or ""), str(request.get("source_key") or ""), str(request.get("scope_kind") or "portfolio"), str(request.get("account_id") or "") or None, str(request.get("period_from") or ""), str(request.get("period_to") or ""), _instant(str(request.get("data_cutoff") or ""), "Data-Cutoff").isoformat(), str(request.get("source_revision") or ""), str(request.get("input_fingerprint") or ""), ) stored_identity = ( existing["preview_id"], existing["source_key"], existing["scope_kind"], existing["scope_id"], existing["period_from"], existing["period_to"], existing["data_cutoff"], existing["source_revision"], existing["input_fingerprint"], ) if request_identity != stored_identity or existing["payload_hash"] != expected_existing_hash or request.get("payload_hash") != expected_existing_hash: raise ValueError("Bestätigungskennung wurde mit verändertem Inhalt wiederverwendet") conn.rollback() return {"batch_id": existing["batch_id"], "audit_id": existing["audit_id"], "idempotent": True} if _now() > created + timedelta(minutes=PREVIEW_TTL_MINUTES): raise ValueError("Die Vorschau ist abgelaufen; bitte eine neue Vorschau erstellen") preview = preview_ingestion(conn, base_request, preview_created_at=preview_created_at, confirmation_id=confirmation_id) if str(request.get("preview_id") or "") != preview["preview_id"]: raise ValueError("Die Vorschau ist veraltet oder die Quelldaten haben sich geändert") if str(request.get("source_revision") or "") != preview["source_revision"]: raise ValueError("Die Ausgangsrevision der Quelle hat sich geändert") payload_hash = _hash({ "preview_id": preview["preview_id"], "confirmation_id": confirmation_id, "input_fingerprint": preview["input_fingerprint"], "preview_created_at": preview["preview_created_at"], }) supplied_hash = str(request.get("payload_hash") or "") if supplied_hash != payload_hash: raise ValueError("Bestätigungsinhalt stimmt nicht mit der Vorschau überein") if preview["source_key"] == "postfinance_etrading": prior = conn.execute( """SELECT batch_id,audit_id FROM portfolio_ingestion_batches WHERE source_key='postfinance_etrading' AND source_revision=? ORDER BY confirmed_at LIMIT 1""", (preview["source_revision"],), ).fetchone() if prior: conn.rollback() return {"batch_id": prior["batch_id"], "audit_id": prior["audit_id"], "idempotent": True} batch_id = "ingestion-" + uuid.uuid4().hex confirmed_at = _now().isoformat() writable = 0 item_rows: list[tuple[SourceRecord, str | None]] = [] records = _records( conn, preview["source_key"], _validate_scope(conn, preview["scope_kind"], preview["account_id"]), preview["period_from"], preview["period_to"], preview["data_cutoff"], base_request, ) if preview["source_key"] == "postfinance_etrading" and any( record.disposition in {"ambiguous", "blocked"} for record in records ): raise ValueError("PostFinance-Batch enthält unklare oder ungültige Zeilen; Prüfung erforderlich") for record in records: target_id = record.target_id if preview["source_key"] == "postfinance_etrading" and record.disposition == "new": target_id = _insert_postfinance_record(conn, record, confirmed_at) writable += 1 elif record.record_kind == "valuation" and record.disposition in {"new", "versioned"}: target_id = _insert_valuation(conn, record, preview["source_key"], confirmed_at) writable += 1 item_rows.append((record, target_id)) audit_id = record_audit_event( conn, source="portfolio_ingestion_api", action="portfolio_ingestion_confirmed", entity_type="portfolio_ingestion_batch", entity_id=batch_id, new_values={ "source_key": preview["source_key"], "written_records": writable, "version": INGESTION_VERSION, "reason_codes": preview["quality_impact"]["reason_codes"], }, confirmed=True, created_by="user", ) conn.execute( """INSERT INTO portfolio_ingestion_batches( batch_id, source_key, scope_kind, scope_id, period_from, period_to, data_cutoff, source_revision, input_fingerprint, preview_id, confirmation_id, payload_hash, status, counts_json, audit_id, confirmed_at, confirmed_by ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?, 'confirmed', ?,?,?, 'user')""", (batch_id, preview["source_key"], preview["scope_kind"], preview["account_id"], preview["period_from"], preview["period_to"], preview["data_cutoff"], preview["source_revision"], preview["input_fingerprint"], preview["preview_id"], confirmation_id, payload_hash, json.dumps(preview["counts"], sort_keys=True), audit_id, confirmed_at), ) for record, target_id in item_rows: conn.execute( """INSERT INTO portfolio_ingestion_items( ingestion_item_id, batch_id, source_record_fingerprint, source_record_ref, record_kind, disposition, target_type, target_id, lineage_hash, summary_json, created_at ) VALUES(?,?,?,?,?,?,?,?,?,?,?)""", ("ingestion-item-" + uuid.uuid4().hex, batch_id, record.source_record_fingerprint, record.source_record_ref, record.record_kind, record.disposition, record.target_type, target_id, record.lineage_hash, json.dumps(record.summary, sort_keys=True), confirmed_at), ) conn.commit() return {"batch_id": batch_id, "audit_id": audit_id, "idempotent": False, "written_records": writable, "payload_hash": payload_hash} except IntegrityError as exc: conn.rollback() existing = conn.execute("SELECT batch_id, payload_hash, audit_id FROM portfolio_ingestion_batches WHERE confirmation_id=?", (confirmation_id,)).fetchone() if existing and existing["payload_hash"] == expected_existing_hash and str(request.get("payload_hash") or "") == expected_existing_hash: return {"batch_id": existing["batch_id"], "audit_id": existing["audit_id"], "idempotent": True} if existing: raise ValueError("Bestätigungskennung wurde mit verändertem Inhalt wiederverwendet") from exc raise ValueError("Ingestion konnte nicht atomar bestätigt werden") from exc except Exception: conn.rollback() raise def list_data_sources(conn: Connection) -> dict[str, Any]: definitions = [ ("postfinance_etrading", "PostFinance E-Trading Portfolio", "portfolio_ingestion_batches", True, True, True, ["Positionsbestand", "Cash", "FX", "Performance-Baseline"]), ("canonical_transactions", "Kanonisches Transaktionsledger", "transactions", True, True, False, ["Kauf", "Verkauf", "Dividende", "Gebühr", "Steuer", "Transfer"]), ("legacy_account_values", "Bestehende Account-Totalwerte", "account_value_snapshots", True, False, True, ["Account-Totalwert"]), ("cash_account_snapshots", "Bestehende Cash-Snapshots", "cash_account_snapshots", True, False, True, ["Cash-Totalwert"]), ("reported_positions", "Gespeicherte Positionssnapshots", "positions_snapshot", False, False, False, ["Menge", "Positionswert"]), ("market_prices", "Gespeicherte Marktpreise", "market_prices", False, False, False, ["Preis"]), ("fx_rates", "Gespeicherte FX-Kurse", "fx_rates", False, False, False, ["FX"]), ] rows: list[dict[str, Any]] = [] for key, label, table, ingestible, activities, valuations, capabilities in definitions: if key == "postfinance_etrading": count = int(conn.execute( "SELECT COUNT(*) AS count FROM portfolio_ingestion_batches WHERE source_key=?", (key,) ).fetchone()["count"]) else: count = int(conn.execute(f"SELECT COUNT(*) AS count FROM {table}").fetchone()["count"]) batch = conn.execute("SELECT confirmed_at, status FROM portfolio_ingestion_batches WHERE source_key=? ORDER BY confirmed_at DESC LIMIT 1", (key,)).fetchone() attempt = None if key == "canonical_transactions": attempt = conn.execute("SELECT finished_at, status FROM import_sessions WHERE import_type IN ('transactions','crypto_transactions') ORDER BY COALESCE(finished_at,started_at) DESC LIMIT 1").fetchone() elif key == "cash_account_snapshots": attempt = conn.execute("SELECT finished_at, status FROM import_sessions WHERE import_type='cash_balances' ORDER BY COALESCE(finished_at,started_at) DESC LIMIT 1").fetchone() rows.append({ "source_key": key, "label": label, "available": count > 0, "configured": True, "ingestion_supported": ingestible, "activity_support": activities, "valuation_support": valuations, "capabilities": capabilities, "record_count": count, "last_successful_confirmed_at": batch["confirmed_at"] if batch else None, "last_attempt_at": (attempt["finished_at"] if attempt else None) or (batch["confirmed_at"] if batch else None), "last_attempt_status": (attempt["status"] if attempt else None) or (batch["status"] if batch else None), "known_gaps": ["Keine neue externe Provider-Abfrage", "Keine automatische Ingestion beim Start"], }) fingerprint = _hash([[row["source_key"], row["record_count"], row["last_successful_confirmed_at"]] for row in rows]) return {"sources": rows, "data_cutoff": _now().isoformat(), "input_fingerprint": fingerprint} def ingestion_history(conn: Connection, *, limit: int = 50, offset: int = 0) -> dict[str, Any]: if limit < 1 or limit > 100 or offset < 0: raise ValueError("Pagination ist ungültig") total = int(conn.execute("SELECT COUNT(*) AS count FROM portfolio_ingestion_batches").fetchone()["count"]) rows = conn.execute( """SELECT batch_id, source_key, scope_kind, scope_id, period_from, period_to, data_cutoff, input_fingerprint, status, counts_json, confirmed_at FROM portfolio_ingestion_batches ORDER BY confirmed_at DESC LIMIT ? OFFSET ?""", (limit, offset) ).fetchall() return { "items": [{key: value for key, value in { **dict(row), "batch_id": _safe_ref("batch", str(row["batch_id"])), "scope_id": _safe_ref("account", str(row["scope_id"])) if row["scope_id"] else None, "counts": json.loads(row["counts_json"]), }.items() if key != "counts_json"} for row in rows], "limit": limit, "offset": offset, "total": total, } def _status(diff: Decimal, base: Decimal | None, absolute: Decimal, relative: Decimal) -> str: if diff == 0: return "matched" if abs(diff) <= absolute: return "within_tolerance" if base is not None and base != 0 and abs(diff / base) <= relative: return "within_tolerance" return "mismatch" def _latest_fx(conn: Connection, currency: str, as_of: str, cutoff: str) -> tuple[Decimal | None, str | None, str | None]: if currency == "CHF": return Decimal("1"), as_of, "identity" row = conn.execute( """SELECT rate, rate_date, provider FROM fx_rates WHERE base_currency=? AND quote_currency='CHF' AND rate_date<=? AND datetime(created_at)<=datetime(?) AND rate IS NOT NULL AND quality_status IN ('ok','fresh','complete') ORDER BY rate_date DESC, created_at DESC, fx_rate_id DESC LIMIT 1""", (currency, as_of, cutoff) ).fetchone() if not row: return None, None, None return _stored_decimal(row["rate"], "FX", positive=True), str(row["rate_date"]), str(row["provider"]) def build_portfolio_reconciliation( conn: Connection, *, as_of: str, data_cutoff: str | None = None, account_id: str | None = None, base_currency: str = "CHF", absolute_tolerance: str = "0.01", relative_tolerance: str = "0.001", limit: int = 200, offset: int = 0, ) -> dict[str, Any]: _date(as_of, "Stichtag") cutoff = _instant(data_cutoff or _now().isoformat(), "Data-Cutoff").isoformat() if base_currency.upper() != "CHF": raise ValueError("Reconciliation v1 unterstützt als Basiswährung ausschliesslich CHF") absolute = _decimal(absolute_tolerance, "Absolute Toleranz", non_negative=True) relative = _decimal(relative_tolerance, "Relative Toleranz", non_negative=True) if limit < 1 or limit > 500 or offset < 0: raise ValueError("Pagination ist ungültig") accounts = _validate_scope(conn, "account" if account_id else "portfolio", account_id) account_map = {row["account_id"]: row for row in accounts} if not account_map: empty_coverage = {key: {"covered": 0, "total": 0, "ratio": None} for key in ("accounts", "instruments", "activities", "opening_valuations", "closing_valuations", "prices", "fx", "cost_basis", "reconciliation")} return {"as_of": as_of, "data_cutoff": cutoff, "base_currency": "CHF", "status": "unavailable", "reason_codes": ["missing_account_valuation"], "tolerance": {"version": TOLERANCE_VERSION, "absolute": str(absolute), "relative": str(relative)}, "coverage": empty_coverage, "differences": [], "account_totals": [], "input_fingerprint": _hash([as_of, cutoff, []]), "sources": [], "limit": limit, "offset": offset, "total": 0} placeholders = ",".join("?" for _ in account_map) transaction_rows = conn.execute( f"""SELECT * FROM transactions WHERE account_id IN ({placeholders}) AND trade_date<=? AND datetime(created_at)<=datetime(?) AND COALESCE(is_voided,0)=0 AND COALESCE(is_confirmed,1)=1 ORDER BY trade_date, created_at, transaction_id""", (*account_map, as_of, cutoff) ).fetchall() quantities: dict[tuple[str, str], Decimal] = {} activity_supported = 0 cost_basis_keys: set[tuple[str, str]] = set() for row in transaction_rows: kind = str(row["activity_kind"] or row["transaction_type"] or "").lower() if kind in SUPPORTED_ACTIVITIES: activity_supported += 1 if not row["instrument_id"] or row["quantity"] in (None, ""): continue key = (str(row["account_id"]), str(row["instrument_id"])) try: quantity = _stored_decimal(row["quantity"], "Menge", non_negative=True) except ValueError: continue if kind in {"buy", "initial_position_snapshot"}: quantities[key] = quantities.get(key, Decimal("0")) + quantity if row["gross_amount_original"] not in (None, ""): cost_basis_keys.add(key) elif kind in {"partial_sell", "full_sell", "sell"}: quantities[key] = quantities.get(key, Decimal("0")) - quantity reported_rows = conn.execute( f"""WITH ranked AS ( SELECT p.*, ROW_NUMBER() OVER(PARTITION BY account_id,instrument_id ORDER BY snapshot_date DESC,created_at DESC,position_snapshot_id DESC) rn FROM positions_snapshot p WHERE account_id IN ({placeholders}) AND snapshot_date<=? AND datetime(created_at)<=datetime(?) ) SELECT * FROM ranked WHERE rn=1""", (*account_map, as_of, cutoff) ).fetchall() reported = {(str(row["account_id"]), str(row["instrument_id"])): row for row in reported_rows} keys = sorted(set(quantities) | set(reported)) differences: list[dict[str, Any]] = [] price_covered = fx_covered = recon_covered = 0 account_position_values: dict[str, Decimal] = {key: Decimal("0") for key in account_map} account_position_dates: dict[str, set[str]] = {key: set() for key in account_map} for account, instrument in keys: ledger_qty = quantities.get((account, instrument)) source = reported.get((account, instrument)) source_qty = _stored_decimal(source["quantity"], "Gemeldete Menge") if source and source["quantity"] not in (None, "") else None reasons: list[str] = [] if source and str(source["snapshot_date"]) != as_of: reasons.append("stale_snapshot") if source and str(source["data_quality_status"] or "").lower() not in {"ok", "fresh", "complete"}: reasons.append("source_error") quantity_status = "unavailable" quantity_diff: Decimal | None = None if ledger_qty is None or source_qty is None: reasons.append("missing_position_snapshot") else: quantity_diff = ledger_qty - source_qty if source and str(source["snapshot_date"]) != as_of: quantity_status = "not_comparable" else: quantity_status = _status(quantity_diff, source_qty, absolute, relative) if quantity_status == "mismatch": reasons.append("quantity_mismatch") derived_value: Decimal | None = None reported_value = _stored_decimal(source["market_value_chf"], "Gemeldeter Positionswert") if source and source["market_value_chf"] not in (None, "") else None price_row = conn.execute( """SELECT close,currency,price_date,provider FROM market_prices WHERE instrument_id=? AND price_date<=? AND datetime(created_at)<=datetime(?) AND quality_status IN ('ok','fresh','complete') ORDER BY price_date DESC,created_at DESC,market_price_id DESC LIMIT 1""", (instrument, as_of, cutoff) ).fetchone() price_date = str(price_row["price_date"]) if price_row else None price_source = str(price_row["provider"]) if price_row else None fx_date = fx_source = None if source and price_row and str(source["snapshot_date"]) != price_date: reasons.append("cutoff_mismatch") if ledger_qty is None: reasons.append("missing_position_snapshot") elif not price_row: reasons.append("missing_price") else: if price_date != as_of: reasons.append("stale_price") else: price_covered += 1 price = _stored_decimal(price_row["close"], "Marktpreis", non_negative=True) fx, fx_date, fx_source = _latest_fx(conn, str(price_row["currency"]).upper(), as_of, cutoff) if fx is None: reasons.append("missing_fx") else: if fx_date != as_of: reasons.append("stale_fx") else: fx_covered += 1 derived_value = ledger_qty * price * fx value_status = "unavailable" value_diff: Decimal | None = None relative_diff: Decimal | None = None if derived_value is not None and reported_value is not None: if source and ( "cutoff_mismatch" in reasons or "stale_snapshot" in reasons or "stale_price" in reasons or "stale_fx" in reasons or (fx_date and fx_date != price_date) ): value_status = "not_comparable" if "cutoff_mismatch" not in reasons: reasons.append("cutoff_mismatch") else: value_diff = derived_value - reported_value relative_diff = value_diff / reported_value if reported_value != 0 else None value_status = _status(value_diff, reported_value, absolute, relative) if value_status == "mismatch": reasons.append("value_mismatch") recon_covered += 1 elif reported_value is None: reasons.append("missing_position_snapshot") if source and reported_value is not None: account_position_values[account] += reported_value account_position_dates[account].add(str(source["snapshot_date"])) instrument_row = conn.execute("SELECT name FROM instruments WHERE instrument_id=?", (instrument,)).fetchone() differences.append({ "difference_ref": _safe_ref("difference", f"{account}:{instrument}:{as_of}"), "account_ref": _safe_ref("account", account), "account_label": _safe_label(account_map[account]["account_name"]), "instrument_ref": _safe_ref("instrument", instrument), "instrument_label": str(instrument_row["name"]) if instrument_row else "Instrument (unbekannt)", "quantity": {"ledger": _decimal_text(ledger_qty), "reported": _decimal_text(source_qty), "difference": _decimal_text(quantity_diff), "status": quantity_status}, "valuation": {"derived": _decimal_text(derived_value), "reported": _decimal_text(reported_value), "difference": _decimal_text(value_diff), "relative_difference": _decimal_text(relative_diff), "status": value_status, "position_as_of": str(source["snapshot_date"]) if source else None, "price_as_of": price_date, "fx_as_of": fx_date, "price_source": price_source, "fx_source": fx_source}, "reason_codes": sorted(set(reasons)), }) pf_roles = postfinance_account_roles(conn) official_pf_cash = latest_official_postfinance_cash( conn, as_of=as_of, data_cutoff=cutoff, ) official_pf_cash_total = sum( (component.amount_chf for component in official_pf_cash), Decimal("0") ) official_pf_cash_date = max( (component.snapshot_date for component in official_pf_cash), default=None ) account_totals: list[dict[str, Any]] = [] account_covered = 0 for account, meta in account_map.items(): cash = conn.execute( """SELECT amount_chf,balance_date,source FROM cash_account_snapshots WHERE account_id=? AND balance_date<=? AND datetime(created_at)<=datetime(?) ORDER BY balance_date DESC,created_at DESC,snapshot_id DESC LIMIT 1""", (account, as_of, cutoff) ).fetchone() cash_value = _stored_decimal(cash["amount_chf"], "Cashbestand") if cash else None cash_date = str(cash["balance_date"]) if cash else None if account == pf_roles.get("etrading_depot") and official_pf_cash: # The provider account total spans the distinct depot and trading-cash # roles. This is a reconciliation-only join, not an additive holding. cash_value = official_pf_cash_total cash_date = official_pf_cash_date legacy = conn.execute( """SELECT total_value_chf,valuation_date,source_type FROM account_value_snapshots WHERE account_id=? AND valuation_date<=? AND datetime(created_at)<=datetime(?) AND COALESCE(is_active,1)=1 ORDER BY valuation_date DESC, CASE source_type WHEN 'truewealth_official_import' THEN 3 WHEN 'truewealth_manual_provisional' THEN 2 ELSE 1 END DESC, COALESCE(valuation_at,created_at) DESC,created_at DESC,snapshot_id DESC LIMIT 1""", (account, as_of, cutoff) ).fetchone() reported_total = _stored_decimal(legacy["total_value_chf"], "Kontototal") if legacy else None total_date = str(legacy["valuation_date"]) if legacy else None has_positions = any(key[0] == account for key in reported) derived_total = account_position_values[account] + (cash_value or Decimal("0")) if has_positions or cash else None reasons: list[str] = [] total_status = "unavailable" diff: Decimal | None = None total_value_only = meta["account_type"] in {"robo_advisor", "total_value", "pension", "truewealth"} or (legacy is not None and not any(key[0] == account for key in keys)) canonical_date: str | None = None if total_value_only: canonical = conn.execute( """SELECT value_original,valuation_at FROM portfolio_valuation_snapshots WHERE scope_kind='account' AND scope_id=? AND substr(valuation_at,1,10)<=? AND datetime(captured_at)<=datetime(?) ORDER BY valuation_at DESC,snapshot_version DESC,captured_at DESC,snapshot_id DESC LIMIT 1""", (account, as_of, cutoff) ).fetchone() derived_total = _stored_decimal(canonical["value_original"], "Kanonischer Totalwert") if canonical else None canonical_date = str(canonical["valuation_at"])[:10] if canonical else None if canonical and total_date != canonical_date: reasons.append("cutoff_mismatch") dates = set(account_position_dates[account]) | ({cash_date} if cash_date else set()) | ({total_date} if total_date else set()) | ({canonical_date} if canonical_date else set()) if derived_total is None: reasons.append("missing_position_snapshot" if not total_value_only else "missing_account_valuation") if reported_total is None: reasons.append("missing_account_valuation") if derived_total is not None and reported_total is not None: stale_dates = {item for item in dates if item and item != as_of} if len({item for item in dates if item}) > 1 or stale_dates or "cutoff_mismatch" in reasons: total_status = "not_comparable" if stale_dates: reasons.append("stale_snapshot") if "cutoff_mismatch" not in reasons: reasons.append("cutoff_mismatch") else: diff = derived_total - reported_total total_status = _status(diff, reported_total, absolute, relative) if total_status == "mismatch": reasons.append("cash_gap" if cash_value is None and not total_value_only else "value_mismatch") account_covered += 1 account_totals.append({ "account_ref": _safe_ref("account", account), "account_label": _safe_label(meta["account_name"]), "total_value_only": total_value_only, "positions_value": None if total_value_only else _decimal_text(account_position_values[account]), "cash_value": None if total_value_only else _decimal_text(cash_value), "derived_total": _decimal_text(derived_total), "reported_total": _decimal_text(reported_total), "difference": _decimal_text(diff), "status": total_status, "as_of": as_of if {item for item in dates if item} == {as_of} else None, "reason_codes": sorted(set(reasons)), }) def coverage(covered: int, total: int) -> dict[str, Any]: return {"covered": covered, "total": total, "ratio": _decimal_text(Decimal(covered) / Decimal(total)) if total else None} instrument_total = len(keys) opening_rows = conn.execute( f"SELECT COUNT(DISTINCT account_id) AS count FROM portfolio_valuation_snapshots WHERE account_id IN ({placeholders}) AND substr(valuation_at,1,10)