from __future__ import annotations import base64 import binascii import json import os import re from collections import defaultdict from datetime import UTC, datetime from decimal import Decimal from hashlib import sha256 from pathlib import Path from sqlite3 import Connection, IntegrityError from typing import Any from uuid import uuid4 from jarvis_finance.config.paths import DEFAULT_RUNTIME_DIR from jarvis_finance.imports.postfinance_documents import ( PARSER_ID, PARSER_VERSION, PFBundle, PFEvent, normalize_label, parse_postfinance_bundle, ) from jarvis_finance.services.performance_scope import set_performance_scope_classification from jarvis_finance.services.portfolio_analytics import store_valuation_snapshot from jarvis_finance.services.portfolio_aggregation import ( latest_official_postfinance_cash, latest_official_postfinance_positions, ) TOLERANCE = Decimal("0.05") ALLOWED_ZIP_MIME_TYPES = {"application/zip", "application/x-zip-compressed"} ALLOWED_OVERVIEW_MIME_TYPES = {"application/pdf"} def _now() -> str: return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") def _id(prefix: str) -> str: return f"{prefix}_{uuid4().hex[:20]}" def _hash(*values: object) -> str: return sha256( json.dumps(values, sort_keys=True, separators=(",", ":"), default=str).encode() ).hexdigest() def _text(value: Decimal | None) -> str | None: return format(value, "f") if value is not None else None def _money(value: Decimal) -> str: return format(value.quantize(Decimal("0.01")), "f") def _decode(value: str, label: str) -> bytes: try: return base64.b64decode(value, validate=True) except (binascii.Error, ValueError) as exc: raise ValueError(f"PostFinance {label} is not valid base64") from exc def _source_bytes(request: dict[str, Any]) -> tuple[bytes, bytes]: zip_name = str(request.get("zip_file_name") or "") overview_name = str(request.get("overview_file_name") or "") if ( not zip_name or Path(zip_name).name != zip_name or not zip_name.lower().endswith(".zip") or not overview_name or Path(overview_name).name != overview_name or not overview_name.lower().endswith(".pdf") ): raise ValueError("Bitte ein ZIP und eine separate PDF-Portfolioübersicht auswählen.") if request.get("zip_mime_type") not in ALLOWED_ZIP_MIME_TYPES: raise ValueError("Die Dokumentensammlung ist keine unterstützte ZIP-Datei.") if request.get("overview_mime_type") not in ALLOWED_OVERVIEW_MIME_TYPES: raise ValueError("Die Portfolioübersicht ist keine unterstützte PDF-Datei.") zip_raw = _decode(str(request.get("zip_content_base64") or ""), "ZIP") overview_raw = _decode( str(request.get("overview_content_base64") or ""), "overview PDF" ) if not zip_raw or not overview_raw: raise ValueError("Eine ausgewählte PostFinance-Datei ist leer.") if len(zip_raw) != int(request.get("zip_size_bytes") or 0): raise ValueError("Die ZIP-Dateigrösse stimmt nicht mit dem Upload überein.") if len(overview_raw) != int(request.get("overview_size_bytes") or 0): raise ValueError("Die PDF-Dateigrösse stimmt nicht mit dem Upload überein.") if not zip_raw.startswith((b"PK\x03\x04", b"PK\x05\x06", b"PK\x07\x08")): raise ValueError("Die Dokumentensammlung enthält kein gültiges ZIP.") if not overview_raw.startswith(b"%PDF-"): raise ValueError("Die Portfolioübersicht enthält kein gültiges PDF.") return zip_raw, overview_raw def _resolve_accounts(conn: Connection, request: dict[str, Any] | None = None) -> dict[str, str]: request = request or {} explicit = request.get("account_roles") or {} prospective_efinance = False if explicit: if set(explicit) != {"efinance", "etrading_depot", "etrading_cash"}: raise ValueError("All three PostFinance account roles are required") result = {key: str(value) for key, value in explicit.items()} else: rows = conn.execute( """SELECT a.account_id,a.account_name,a.account_type,a.platform_id FROM accounts a JOIN platforms p ON p.platform_id=a.platform_id WHERE a.is_active=1 AND replace(replace(lower(p.name),' ',''),'-','') LIKE '%postfinance%' ORDER BY a.account_id""" ).fetchall() depot = [ row for row in rows if "trading" in str(row["account_name"]).lower() and str(row["account_type"]).lower() in {"brokerage", "depot"} or str(row["account_type"]).lower() in {"brokerage", "depot"} ] cash = [ row for row in rows if row not in depot and ( "cash" in str(row["account_name"]).lower() or "finance" in str(row["account_name"]).lower() or str(row["account_type"]).lower() in {"cash", "bank", "checking"} ) ] efinance = [ row for row in cash if "efinance" in str(row["account_name"]) .lower() .replace("postfinance", "") .replace("-", "") .replace(" ", "") ] trading_cash = [row for row in cash if row not in efinance] if len(depot) != 1 or len(trading_cash) != 1 or len(efinance) > 1: raise ValueError( "PostFinance E-Finance, E-Trading depot and E-Trading cash cannot be resolved unambiguously" ) if efinance: efinance_id = str(efinance[0]["account_id"]) else: efinance_id = f"account_pf_efinance_{_hash(depot[0]['platform_id'])[:16]}" prospective_efinance = True result = { "efinance": efinance_id, "etrading_depot": str(depot[0]["account_id"]), "etrading_cash": str(trading_cash[0]["account_id"]), } if len(set(result.values())) != 3: raise ValueError("PostFinance account roles must map to three distinct canonical accounts") for role, account_id in result.items(): row = conn.execute( "SELECT is_active FROM accounts WHERE account_id=?", (account_id,) ).fetchone() if not row and role == "efinance" and prospective_efinance: continue if not row or not row["is_active"]: raise ValueError(f"PostFinance account role {role} is unavailable") return result def _ensure_efinance_account(conn: Connection, accounts: dict[str, str], now: str) -> None: if conn.execute( "SELECT 1 FROM accounts WHERE account_id=?", (accounts["efinance"],) ).fetchone(): return depot = conn.execute( "SELECT platform_id FROM accounts WHERE account_id=? AND is_active=1", (accounts["etrading_depot"],), ).fetchone() if not depot: raise ValueError("PostFinance depot account is unavailable") conn.execute( """INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency, performance_included,is_active,created_at,portfolio_bucket,notes) VALUES(?,?,'PostFinance E-Finance','cash','CHF',0,1,?,'cash', 'Created by verified PostFinance source-role separation')""", (accounts["efinance"], depot["platform_id"], now), ) def _revision(conn: Connection, accounts: dict[str, str]) -> str: account_ids = sorted(set(accounts.values())) marks = ",".join("?" for _ in account_ids) tables = [ "account_value_snapshots", "positions_snapshot", "cash_balances", "transactions", "postfinance_import_batches", ] state: list[object] = [ ("resolved_account_roles", sorted(accounts.items())), ] for table in tables: if table == "postfinance_import_batches": row = conn.execute( f"SELECT COUNT(*) c,COALESCE(MAX(confirmed_at),'') m FROM {table}" ).fetchone() else: row = conn.execute( f"SELECT COUNT(*) c,COALESCE(MAX(created_at),'') m FROM {table} WHERE account_id IN ({marks})", account_ids, ).fetchone() state.extend((table, row["c"], row["m"])) account_rows = conn.execute( f"""SELECT account_id,platform_id,account_name,account_type,currency,is_active FROM accounts WHERE account_id IN ({marks}) ORDER BY account_id""", account_ids, ).fetchall() state.extend(("accounts", [tuple(row) for row in account_rows])) role_rows = conn.execute( "SELECT role,account_id FROM postfinance_account_roles ORDER BY role" ).fetchall() state.extend(("postfinance_account_roles", [tuple(row) for row in role_rows])) instrument_rows = conn.execute( """SELECT DISTINCT i.instrument_id,i.name,i.ticker,i.provider_symbol,i.isin,i.is_active FROM positions_snapshot p JOIN instruments i ON i.instrument_id=p.instrument_id WHERE p.account_id=? ORDER BY i.instrument_id""", (accounts["etrading_depot"],), ).fetchall() state.extend(("instruments", [tuple(row) for row in instrument_rows])) return _hash(state) def _projection_collision( conn: Connection, accounts: dict[str, str], snapshot_day: str ) -> bool: position_collision = conn.execute( "SELECT COUNT(*) FROM positions_snapshot WHERE account_id=? AND snapshot_date=?", (accounts["etrading_depot"], snapshot_day), ).fetchone()[0] cash_collision = conn.execute( "SELECT COUNT(*) FROM cash_balances WHERE account_id=? AND balance_date=?", (accounts["etrading_cash"], snapshot_day), ).fetchone()[0] return bool(position_collision or cash_collision) def _instrument_rows(conn: Connection, account_id: str) -> list[Any]: rows = conn.execute( """SELECT DISTINCT i.instrument_id,i.name,i.ticker,i.provider_symbol,i.isin FROM positions_snapshot p JOIN instruments i ON i.instrument_id=p.instrument_id WHERE p.account_id=? AND i.is_active=1 ORDER BY i.instrument_id""", (account_id,), ).fetchall() if len(rows) != 22: raise ValueError("Confirmed PostFinance depot baseline must contain exactly 22 instruments") return rows def _match_instruments(conn: Connection, bundle: PFBundle, depot_id: str) -> dict[str, str]: rows = _instrument_rows(conn, depot_id) by_isin = {str(row["isin"]).upper(): str(row["instrument_id"]) for row in rows if row["isin"]} aliases: dict[str, set[str]] = defaultdict(set) for row in rows: for key in (row["name"], row["ticker"], row["provider_symbol"]): if key: aliases[normalize_label(str(key))].add(str(row["instrument_id"])) mapping: dict[str, str] = {} used: set[str] = set() for position in bundle.snapshot.positions: exact = aliases.get(position.normalized_label, set()) candidates = set(exact) if not candidates: tokens = set(position.normalized_label.split()) for alias, ids in aliases.items(): other = set(alias.split()) if ( tokens and other and ( tokens <= other or other <= tokens or len(tokens & other) >= min(2, len(tokens), len(other)) ) ): candidates.update(ids) candidates -= used if len(candidates) != 1: raise ValueError( "PostFinance snapshot position cannot be mapped unambiguously to the confirmed depot baseline" ) instrument_id = next(iter(candidates)) mapping[position.row_reference] = instrument_id used.add(instrument_id) if len(mapping) != 22 or len(used) != 22: raise ValueError("PostFinance snapshot-to-depot mapping is incomplete") # Trade and income ISINs independently verify that no document points to another instrument. for event in bundle.events: if event.isin and event.isin.upper() not in by_isin: raise ValueError( "PostFinance ledger document references an instrument outside the confirmed depot" ) return mapping def _document_period(text: str) -> tuple[str, str] | None: match = re.search( r"(?:vom|von)\s+(\d{1,2}[./]\d{1,2}[./]\d{4}).{0,80}(?:bis|per)\s+" r"(\d{1,2}[./]\d{1,2}[./]\d{4})", text, flags=re.I | re.S, ) if not match: return None parsed: list[str] = [] for value in match.groups(): for fmt in ("%d.%m.%Y", "%d/%m/%Y"): try: parsed.append(datetime.strptime(value, fmt).date().isoformat()) break except ValueError: continue return (parsed[0], parsed[1]) if len(parsed) == 2 else None def _cash_projection_issue(conn: Connection, accounts: dict[str, str]) -> bool: snapshot = conn.execute( """SELECT cash_chf,substr(snapshot_at,1,10) snapshot_day FROM postfinance_snapshots ORDER BY snapshot_at DESC,created_at DESC LIMIT 1""" ).fetchone() if not snapshot: return False cash = conn.execute( """SELECT COUNT(*) row_count, COALESCE(SUM(CAST(amount_chf AS NUMERIC)),0) total_chf FROM cash_balances WHERE account_id=? AND balance_date=? AND source_type='postfinance_official_import'""", (accounts["etrading_cash"], snapshot["snapshot_day"]), ).fetchone() return bool( cash and int(cash["row_count"] or 0) > 0 and abs(Decimal(str(cash["total_chf"])) - Decimal(str(snapshot["cash_chf"]))) > TOLERANCE ) def _event_instrument( event: PFEvent, bundle: PFBundle, position_map: dict[str, str], conn: Connection ) -> str | None: if event.isin: row = conn.execute( "SELECT instrument_id FROM instruments WHERE upper(isin)=upper(?) AND is_active=1", (event.isin,), ).fetchall() if len(row) != 1: raise ValueError("PostFinance ledger ISIN mapping is ambiguous") return str(row[0]["instrument_id"]) if event.instrument_name: key = normalize_label(event.instrument_name) for position in bundle.snapshot.positions: if ( position.normalized_label == key or key in position.normalized_label or position.normalized_label in key ): return position_map[position.row_reference] return None def _preview_payload( bundle: PFBundle, accounts: dict[str, str], mapping: dict[str, str] ) -> dict[str, Any]: snapshot = bundle.snapshot component_total = snapshot.securities_total_chf + snapshot.cash_total_chf difference = snapshot.total_value_chf - component_total event_counts: dict[str, int] = defaultdict(int) quality_counts: dict[str, int] = defaultdict(int) document_types: dict[str, int] = defaultdict(int) periods: list[dict[str, str]] = [] document_dates = sorted( {document.document_date for document in bundle.documents if document.document_date} ) for event in bundle.events: event_counts[event.event_type] += 1 quality_counts[event.quality_status] += 1 for document in bundle.documents: document_types[document.document_type] += 1 period = _document_period(document.text) if period: periods.append( { "document_type": document.document_type, "from": period[0], "to": period[1], } ) activity_dates = sorted(event.occurred_on for event in bundle.events) return { "bundle_sha256": bundle.bundle_hash, "zip_sha256": bundle.zip_hash, "overview_sha256": bundle.overview_hash, "parser_id": PARSER_ID, "parser_version": PARSER_VERSION, "snapshot_at": snapshot.valuation_at, "valuation_date": snapshot.valuation_at[:10], "document_count": len(bundle.documents) + 1, "zip_document_count": len(bundle.documents), "page_count": sum(doc.page_count for doc in bundle.documents) + snapshot.page_count, "document_types": dict(sorted(document_types.items())), "covered_periods": periods, "document_date_from": document_dates[0] if document_dates else None, "document_date_to": document_dates[-1] if document_dates else None, "activity_period_from": activity_dates[0] if activity_dates else None, "activity_period_to": activity_dates[-1] if activity_dates else None, "position_count": len(snapshot.positions), "cash_count": len(snapshot.cash), "event_count": len(bundle.events), "event_counts": dict(sorted(event_counts.items())), "quality_counts": dict(sorted(quality_counts.items())), "transfer_group_count": len(bundle.transfer_groups), "total_chf": _money(snapshot.total_value_chf), "securities_chf": _money(snapshot.securities_total_chf), "cash_chf": _money(snapshot.cash_total_chf), "stocks_chf": _money(snapshot.stock_total_chf), "etfs_chf": _money(snapshot.etf_total_chf), "component_total_chf": _money(component_total), "reconciliation_difference_chf": _money(difference), "reconciliation_status": "matched" if difference == 0 else "within_tolerance", "open_orders": snapshot.open_orders, "account_roles_resolved": sorted(accounts), "mapped_position_count": len(mapping), "mappings": { "secure": len(mapping), "review_required": 0, "blocked": 0, }, "components": [ { "kind": "depot", "status": "detected", "count": len(snapshot.positions), "value_chf": _money(snapshot.securities_total_chf), }, { "kind": "settlement_cash", "status": "detected", "count": len(snapshot.cash), "value_chf": _money(snapshot.cash_total_chf), }, ], "warnings": [ "Die Einstandskosten bleiben unvollständig, wenn im Bundle kein Erwerbsbeleg vorhanden ist." ], } def preview_postfinance_import(conn: Connection, request: dict[str, Any]) -> dict[str, Any]: accounts = _resolve_accounts(conn, request) zip_raw, overview_raw = _source_bytes(request) bundle = parse_postfinance_bundle(zip_raw, overview_raw) mapping = _match_instruments(conn, bundle, accounts["etrading_depot"]) payload = _preview_payload(bundle, accounts, mapping) revision = _revision(conn, accounts) duplicate = conn.execute( "SELECT batch_id FROM postfinance_import_batches WHERE bundle_sha256=?", (bundle.bundle_hash,), ).fetchone() conflict = conn.execute( "SELECT batch_id FROM postfinance_import_batches WHERE substr(snapshot_at,1,10)=? AND bundle_sha256<>?", (bundle.snapshot.valuation_at[:10], bundle.bundle_hash), ).fetchone() if not duplicate and _projection_collision( conn, accounts, bundle.snapshot.valuation_at[:10] ): conflict = conflict or {"batch_id": "canonical_projection_date_conflict"} semantic_conflicts = conn.execute( """SELECT COUNT(*) FROM postfinance_documents d WHERE EXISTS (SELECT 1 FROM json_each(?) j WHERE j.value=d.semantic_identity) AND d.document_hash NOT IN (SELECT value FROM json_each(?))""", ( json.dumps([d.semantic_identity for d in bundle.documents]), json.dumps([d.document_hash for d in bundle.documents]), ), ).fetchone()[0] if semantic_conflicts: conflict = conflict or {"batch_id": "semantic_document_conflict"} if conflict: payload["warnings"].append( "Eine andere unveränderliche Quelle besitzt bereits dieselbe fachliche Identität." ) snapshot_day = bundle.snapshot.valuation_at[:10] statement_periods = [ item for item in payload["covered_periods"] if item["document_type"] == "account_statement" ] blocker_codes: list[str] = [] data_gaps: list[str] = [] if not statement_periods: blocker_codes.append("account_statement_period_missing") data_gaps.append("Vollständiger E-Trading-Kontoauszug oder Transaktionsreport fehlt.") elif not any(item["to"] == snapshot_day for item in statement_periods): blocker_codes.append("portfolio_overview_end_date_mismatch") data_gaps.append("Depot- und Cashstichtag stimmen nicht überein.") if int(payload["quality_counts"].get("partial", 0)): blocker_codes.append("ambiguous_activities_present") data_gaps.append("Mindestens eine Aktivität oder Zuordnung ist noch prüfpflichtig.") if payload["reconciliation_status"] != "matched": blocker_codes.append("portfolio_components_not_reconciled") data_gaps.append("Wertschriften, Settlement-Cash und Gesamtwert stimmen nicht überein.") if conflict: blocker_codes.append("immutable_source_conflict") data_gaps.append("Ein Dokument kollidiert mit einer bereits bestätigten Quelle.") if _cash_projection_issue(conn, accounts): blocker_codes.append("known_cash_projection_requires_correction") data_gaps.append("Die bekannte Cashprojektion muss vor einem Confirm korrigiert werden.") payload.update( { "blocker_codes": sorted(set(blocker_codes)), "data_gaps": data_gaps, "confirm_allowed": not blocker_codes, "next_action": ( "Unterlagen prüfen und danach bewusst bestätigen." if not blocker_codes else data_gaps[0] ), } ) fingerprint = _hash(payload, revision) payload.update( { "preview_id": f"pfpreview_{fingerprint[:24]}", "confirmation_id": f"pfconfirm_{fingerprint[24:48]}", "source_references": {"upload": f"PF-{fingerprint[-8:].upper()}"}, "db_revision": revision, "duplicate": bool(duplicate), "existing_batch_id": str(duplicate["batch_id"]) if duplicate else None, "conflict": bool(conflict), "conflicting_batch_id": str(conflict["batch_id"]) if conflict else None, } ) return payload def _archive(raw: bytes, digest: str, suffix: str) -> tuple[str, bool]: root = Path(os.environ.get("JARVIS_FINANCE_RUNTIME_DIR", str(DEFAULT_RUNTIME_DIR))).expanduser() folder = root / "imports" / "postfinance" / "archive" folder.mkdir(mode=0o700, parents=True, exist_ok=True) folder.chmod(0o700) target = folder / f"{digest}.{suffix}" if target.exists(): if sha256(target.read_bytes()).hexdigest() != digest: raise ValueError("Archived PostFinance source hash mismatch") return f"postfinance/archive/{target.name}", False tmp = folder / f".{digest}.{uuid4().hex}.tmp" fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: with os.fdopen(fd, "wb") as handle: handle.write(raw) handle.flush() os.fsync(handle.fileno()) os.replace(tmp, target) target.chmod(0o600) finally: tmp.unlink(missing_ok=True) return f"postfinance/archive/{target.name}", True def _remove_archive(reference: str) -> None: root = Path(os.environ.get("JARVIS_FINANCE_RUNTIME_DIR", str(DEFAULT_RUNTIME_DIR))).expanduser() (root / "imports" / reference).unlink(missing_ok=True) def _audit(conn: Connection, batch_id: str, bundle: PFBundle, now: str) -> str: audit_id = _id("audit") details = { "bundle_sha256": bundle.bundle_hash, "parser_id": PARSER_ID, "parser_version": PARSER_VERSION, "document_count": len(bundle.documents) + 1, "event_count": len(bundle.events), "snapshot_at": bundle.snapshot.valuation_at, } conn.execute( """INSERT INTO audit_log(audit_id,timestamp,source,action,entity_type,entity_id,old_values_json,new_values_json,user_text_note,confirmed,confirmation_timestamp,auto_parsed,parse_confidence,created_by,quality_status,created_at) VALUES(?,?,'postfinance_sprint13','postfinance_import_confirm','postfinance_import_batch',?,NULL,?,'Verified official source bundle',1,?,0,NULL,'user','ok',?)""", (audit_id, now, batch_id, json.dumps(details, sort_keys=True), now, now), ) return audit_id def _canonical_transaction( conn: Connection, event_id: str, event: PFEvent, account_id: str, instrument_id: str | None, now: str, ) -> None: if event.event_type in {"split", "corporate_action"}: return tx_type = { "buy": "buy", "sell": "sell", "dividend": "dividend", "interest": "interest", "fee": "fee", "internal_transfer": "transfer", "fx": "fx", }[event.event_type] activity = ( "internal_transfer" if event.event_type == "internal_transfer" else "external_cashflow" if event.event_type in {"dividend", "interest", "fee"} else "trade" ) columns = ( "transaction_id,transaction_type,account_id,instrument_id,trade_date,settlement_date,quantity," "price_original,gross_amount_original,fee_original,tax_original,net_amount_original,currency_original," "fx_rate_to_chf,fx_source,fx_status,gross_amount_chf,fee_chf,tax_chf,net_amount_chf,source_type,source_id," "external_transaction_id,row_hash,is_confirmed,quality_status,notes,created_at,updated_at,is_voided," "activity_kind,booking_date,base_currency,internal_transfer_group_id,source_reference" ) values = ( event_id, tx_type, account_id, instrument_id, event.occurred_on, event.settlement_on, _text(event.quantity), _text(event.price_original), _text(event.gross_original), _text(event.fee_original) or "0", _text(event.tax_original) or "0", _text(event.net_original), event.currency, _text(event.fx_rate_to_chf), None, "unavailable", None, None, None, None, "postfinance_official_import", event_id, event.event_fingerprint, event.event_fingerprint, 1, event.quality_status, "Verified PostFinance document event", now, None, 0, activity, event.settlement_on or event.occurred_on, "CHF", event.internal_transfer_group, _hash(event.semantic_reference), ) conn.execute( f"INSERT INTO transactions({columns}) VALUES({','.join('?' for _ in values)})", values ) def store_postfinance_component_valuations( conn: Connection, *, depot_account_id: str, cash_account_id: str, securities_value_chf: Decimal, cash_value_chf: Decimal, valuation_at: str, batch_id: str, captured_at: str, ) -> int: """Write exactly the two additive PostFinance scope components.""" stored = 0 for component_account, component_value, component_name in ( (depot_account_id, securities_value_chf, "securities"), (cash_account_id, cash_value_chf, "settlement_cash"), ): stored += int( store_valuation_snapshot( conn, id_prefix="postfinance-component-valuation", run_id=f"{batch_id}:{component_name}", scope_kind="account", scope_id=component_account, account_id=component_account, value_original=_money(component_value), currency="CHF", fx_rate_to_base="1", valuation_at=valuation_at, source="postfinance_official_import_component_v1", captured_at=captured_at, quality_status="complete", reason_codes=[], ) ) return stored def confirm_postfinance_import(conn: Connection, request: dict[str, Any]) -> dict[str, Any]: if request.get("confirm") is not True: raise ValueError("Explicit PostFinance import confirmation is required") preview = preview_postfinance_import(conn, request) if ( request.get("preview_id") != preview["preview_id"] or request.get("confirmation_id") != preview["confirmation_id"] ): raise ValueError("PostFinance import preview is stale or changed") if preview["conflict"]: raise ValueError("PostFinance source conflicts with existing immutable business identity") if not preview["confirm_allowed"]: raise ValueError("PostFinance Confirm ist gesperrt, bis alle Datenlücken behoben sind.") if preview["duplicate"]: row = conn.execute( "SELECT batch_id,audit_id FROM postfinance_import_batches WHERE bundle_sha256=?", (preview["bundle_sha256"],), ).fetchone() return { "status": "confirmed", "batch_id": row["batch_id"], "audit_id": row["audit_id"], "idempotent": True, } accounts = _resolve_accounts(conn, request) zip_raw, overview_raw = _source_bytes(request) bundle = parse_postfinance_bundle(zip_raw, overview_raw) # mandatory byte reparse on confirm if ( bundle.bundle_hash != preview["bundle_sha256"] or _revision(conn, accounts) != preview["db_revision"] ): raise ValueError("PostFinance source bytes or database revision changed after preview") position_map = _match_instruments(conn, bundle, accounts["etrading_depot"]) event_instruments = { event.event_fingerprint: _event_instrument(event, bundle, position_map, conn) for event in bundle.events } archive_refs: list[tuple[str, bool]] = [] zip_ref, created = _archive(zip_raw, bundle.zip_hash, "zip") archive_refs.append((zip_ref, created)) overview_ref, created = _archive(overview_raw, bundle.overview_hash, "pdf") archive_refs.append((overview_ref, created)) for document in bundle.documents: reference, created = _archive(document.raw, document.document_hash, "pdf") archive_refs.append((reference, created)) now = _now() batch_id = _id("pfbatch") snapshot_id = _id("pfsnapshot") try: conn.execute("BEGIN IMMEDIATE") if _revision(conn, accounts) != preview["db_revision"]: raise ValueError("PostFinance database revision changed before persistence") if _projection_collision(conn, accounts, bundle.snapshot.valuation_at[:10]): raise ValueError("PostFinance canonical projection date conflicts with existing data") _ensure_efinance_account(conn, accounts, now) for role, account_id in accounts.items(): conn.execute( "INSERT OR IGNORE INTO postfinance_account_roles(role,account_id,source_reference_hash,created_at) VALUES(?,?,NULL,?)", (role, account_id, now), ) mapped = conn.execute( "SELECT account_id FROM postfinance_account_roles WHERE role=?", (role,) ).fetchone() if not mapped or mapped["account_id"] != account_id: raise ValueError( "PostFinance account-role mapping conflicts with confirmed mapping" ) set_performance_scope_classification( conn, account_id=account_id, included=role in {"etrading_depot", "etrading_cash"}, classification_role=( f"postfinance_{role}" if role in {"etrading_depot", "etrading_cash"} else "postfinance_efinance_control" ), source="postfinance_confirmed_role_mapping", note="Confirmed PostFinance account-role mapping", classified_at=now, ) audit_id = _audit(conn, batch_id, bundle, now) conn.execute( """INSERT INTO postfinance_import_batches(batch_id,bundle_sha256,zip_sha256,overview_sha256,parser_id,parser_version,db_revision,archive_reference,snapshot_at,status,audit_id,confirmed_at,confirmed_by) VALUES(?,?,?,?,?,?,?,?,?,'confirmed',?,?,'user')""", ( batch_id, bundle.bundle_hash, bundle.zip_hash, bundle.overview_hash, PARSER_ID, PARSER_VERSION, preview["db_revision"], zip_ref, bundle.snapshot.valuation_at, audit_id, now, ), ) for document in bundle.documents: reference = f"postfinance/archive/{document.document_hash}.pdf" ref_hash = _hash(document.semantic_reference) existing = conn.execute( "SELECT document_hash FROM postfinance_documents WHERE semantic_identity=?", (document.semantic_identity,), ).fetchone() if existing and existing["document_hash"] != document.document_hash: raise ValueError("PostFinance semantic document conflict") conn.execute( """INSERT OR IGNORE INTO postfinance_documents(document_hash,filename_hash,semantic_identity,semantic_reference_hash,document_type,document_date,account_reference_hash,account_role,page_count,archive_reference,parser_version,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""", ( document.document_hash, document.filename_hash, document.semantic_identity, ref_hash, document.document_type, document.document_date, document.account_reference_hash, document.account_role, document.page_count, reference, PARSER_VERSION, now, ), ) conn.execute( "INSERT INTO postfinance_batch_documents(batch_id,document_hash) VALUES(?,?)", (batch_id, document.document_hash), ) snapshot = bundle.snapshot component_total = snapshot.securities_total_chf + snapshot.cash_total_chf difference = snapshot.total_value_chf - component_total conn.execute( """INSERT INTO postfinance_snapshots(snapshot_id,batch_id,account_id,snapshot_at,total_chf,securities_chf,cash_chf,stocks_chf,etfs_chf,component_total_chf,difference_chf,tolerance_chf,position_count,cash_count,open_orders,reconciliation_status,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,? ,?,?,?,?,?)""", ( snapshot_id, batch_id, accounts["etrading_depot"], snapshot.valuation_at, _money(snapshot.total_value_chf), _money(snapshot.securities_total_chf), _money(snapshot.cash_total_chf), _money(snapshot.stock_total_chf), _money(snapshot.etf_total_chf), _money(component_total), _money(difference), _money(TOLERANCE), len(snapshot.positions), len(snapshot.cash), snapshot.open_orders, "matched" if difference == 0 else "within_tolerance", now, ), ) for event in bundle.events: instrument_id = event_instruments[event.event_fingerprint] event_id = f"pfevent_{event.event_fingerprint[:24]}" account_id = accounts[event.account_role] conn.execute( """INSERT INTO postfinance_ledger_events(event_id,event_fingerprint,document_hash,semantic_reference_hash,event_type,account_role,account_id,occurred_on,settlement_on,direction,instrument_id,quantity,price_original,gross_original,fee_original,tax_original,net_original,currency,fx_rate_to_chf,internal_transfer_group,quality_status,reason_codes_json,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( event_id, event.event_fingerprint, event.document_hash, _hash(event.semantic_reference), event.event_type, event.account_role, account_id, event.occurred_on, event.settlement_on, event.direction or "neutral", instrument_id, _text(event.quantity), _text(event.price_original), _text(event.gross_original), _text(event.fee_original) or "0", _text(event.tax_original) or "0", _text(event.net_original), event.currency, _text(event.fx_rate_to_chf), event.internal_transfer_group, event.quality_status, json.dumps(event.reason_codes), now, ), ) for kind, amount in ( ("gross", event.gross_original), ("fee", event.fee_original), ("tax", event.tax_original), ("net", event.net_original), ): if amount is not None and event.currency: conn.execute( "INSERT INTO postfinance_event_components(component_id,event_id,component_type,amount_original,currency,provenance_document_hash,created_at) VALUES(?,?,?,?,?,?,?)", ( _id("pfcomponent"), event_id, kind, _text(amount), event.currency, event.document_hash, now, ), ) _canonical_transaction(conn, event_id, event, account_id, instrument_id, now) lots: dict[str, list[dict[str, Any]]] = defaultdict(list) for event in sorted( bundle.events, key=lambda item: ( item.occurred_on, 0 if item.event_type == "buy" else 1, item.event_fingerprint, ), ): instrument_id = event_instruments[event.event_fingerprint] if ( event.event_type == "buy" and instrument_id and event.quantity and event.price_original and event.currency ): lots[instrument_id].append({"event": event, "remaining": event.quantity}) elif event.event_type == "sell" and instrument_id and event.quantity: to_consume = event.quantity for lot in lots[instrument_id]: if to_consume <= 0: break consumed = min(lot["remaining"], to_consume) lot["remaining"] -= consumed to_consume -= consumed for instrument_id, instrument_lots in lots.items(): for lot in instrument_lots: event = lot["event"] conn.execute( """INSERT INTO postfinance_cost_basis_lots(lot_id,event_id,instrument_id,acquired_on,quantity_acquired,quantity_remaining,unit_cost_original,fees_original,currency,method,provenance_document_hash,created_at) VALUES(?,?,?,?,?,?,?,?,?,'documented_trade_fifo',?,?)""", ( _id("pflot"), f"pfevent_{event.event_fingerprint[:24]}", instrument_id, event.occurred_on, _text(event.quantity), _text(lot["remaining"]), _text(event.price_original), _text(event.fee_original) or "0", event.currency, event.document_hash, now, ), ) for position in snapshot.positions: instrument_id = position_map[position.row_reference] matched_lots = lots.get(instrument_id, []) quantity = sum((lot["remaining"] for lot in matched_lots), Decimal("0")) complete = ( bool(matched_lots) and quantity == position.quantity and len({lot["event"].currency for lot in matched_lots}) == 1 ) calculated = ( sum( ( lot["remaining"] * (lot["event"].price_original or 0) + (lot["event"].fee_original + lot["event"].tax_original) * (lot["remaining"] / lot["event"].quantity) for lot in matched_lots if lot["event"].quantity ), Decimal("0"), ) if complete else None ) status = "complete" if complete else "partial" if matched_lots else "unavailable" conn.execute( """INSERT INTO postfinance_snapshot_positions(snapshot_position_id,snapshot_id,source_row_reference,instrument_id,asset_class,quantity,provider_average_cost_original,provider_cost_total_original,price_original,price_currency,value_chf,weight_pct,computed_cost_basis_original,computed_cost_basis_status,provenance_json,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( _id("pfposition"), snapshot_id, position.row_reference, instrument_id, position.asset_class, _text(position.quantity), _text(position.provider_average_cost_original), _text(position.provider_cost_total_original), _text(position.market_price_original), position.price_currency, _money(position.market_value_chf), _text(position.weight_pct), _text(calculated), status, json.dumps( { "provider": "official_account_overview", "calculated": "documented_trade_fifo", "estimated": False, }, sort_keys=True, ), now, ), ) 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_original,market_fx_rate_to_chf,market_value_chf,portfolio_weight_pct,category,data_quality_status,source_type,source_reference,created_at) SELECT ?,?,?,a.platform_id,?,?,?,?,?,?,?,?,?,?,?,?,?,? FROM accounts a WHERE a.account_id=?""", ( _id("position"), snapshot.valuation_at[:10], accounts["etrading_depot"], instrument_id, _text(position.quantity), _text(position.provider_average_cost_original), _text(position.provider_cost_total_original), _text(position.market_price_original), _text(position.quantity * position.market_price_original), None, _money(position.market_value_chf), _text(position.weight_pct), position.asset_class, "verified", "postfinance_official_import", snapshot_id, now, accounts["etrading_depot"], ), ) for cash in snapshot.cash: conn.execute( "INSERT INTO postfinance_snapshot_cash(snapshot_cash_id,snapshot_id,currency,amount_original,fx_rate_to_chf,value_chf,created_at) VALUES(?,?,?,?,?,?,?)", ( _id("pfcash"), snapshot_id, cash.currency, _text(cash.amount_original), _text(cash.fx_rate_to_chf), _money(cash.source_value_chf), now, ), ) 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(?,?,?,?,?,?,?,?,?,?,?)", ( _id("cash"), accounts["etrading_cash"], snapshot.valuation_at[:10], cash.currency, _text(cash.amount_original), _text(cash.fx_rate_to_chf), _money(cash.source_value_chf), "postfinance_official_import", "verified", "Official PostFinance account overview", now, ), ) conn.execute( "INSERT INTO account_value_snapshots(snapshot_id,account_id,valuation_date,total_value_chf,currency,source_type,quality_status,notes,created_at,valuation_at,source_reference,is_active) VALUES(?,?,?,?,'CHF','postfinance_official_import','ok','Official PostFinance account overview',?,?,?,1)", ( _id("acctval"), accounts["etrading_depot"], snapshot.valuation_at[:10], _money(snapshot.total_value_chf), now, snapshot.valuation_at, snapshot_id, ), ) # The source total above is retained solely for reconciliation. Canonical # performance gets exactly two additive components, never a third total. store_postfinance_component_valuations( conn, depot_account_id=accounts["etrading_depot"], cash_account_id=accounts["etrading_cash"], securities_value_chf=snapshot.securities_total_chf, cash_value_chf=snapshot.cash_total_chf, valuation_at=snapshot.valuation_at, batch_id=batch_id, captured_at=now, ) conn.commit() except IntegrityError as exc: conn.rollback() duplicate = conn.execute( "SELECT batch_id,audit_id FROM postfinance_import_batches WHERE bundle_sha256=?", (bundle.bundle_hash,), ).fetchone() if duplicate: return { "status": "confirmed", "batch_id": duplicate["batch_id"], "audit_id": duplicate["audit_id"], "idempotent": True, } for ref, created in archive_refs: if created: _remove_archive(ref) raise ValueError("PostFinance import conflicted with immutable data") from exc except Exception: conn.rollback() for ref, created in archive_refs: if created: _remove_archive(ref) raise return { "status": "confirmed", "batch_id": batch_id, "snapshot_id": snapshot_id, "audit_id": audit_id, "idempotent": False, } def get_postfinance_summary(conn: Connection) -> dict[str, Any]: latest = conn.execute( """SELECT snapshot_id,snapshot_at,total_chf,securities_chf,cash_chf,stocks_chf,etfs_chf, position_count,cash_count,open_orders,difference_chf,reconciliation_status,created_at FROM postfinance_snapshots ORDER BY snapshot_at DESC,created_at DESC LIMIT 1""" ).fetchone() if not latest: return { "latest_snapshot": None, "positions": [], "cash": [], "event_counts": {}, "imports": [], } official_position_values = { override.instrument_id: override.market_value_chf for override in latest_official_postfinance_positions(conn).values() } positions = [ { "row_key": _hash("position", row["instrument_name"], row["quantity"])[:16], "instrument_name": row["instrument_name"], "ticker": row["ticker"], "asset_class": row["asset_class"], "quantity": row["quantity"], "provider_average_cost_original": row["provider_average_cost_original"], "provider_cost_total_original": row["provider_cost_total_original"], "price_original": row["price_original"], "price_currency": row["price_currency"], "value_chf": _money(official_position_values.get(row["instrument_id"], Decimal(str(row["value_chf"])))), "weight_pct": row["weight_pct"], "computed_cost_basis_original": row["computed_cost_basis_original"], "computed_cost_basis_status": row["computed_cost_basis_status"], } for row in conn.execute( """SELECT p.instrument_id,p.asset_class,p.quantity,p.provider_average_cost_original, p.provider_cost_total_original,p.price_original,p.price_currency,p.value_chf, p.weight_pct,p.computed_cost_basis_original,p.computed_cost_basis_status, i.name AS instrument_name,i.ticker FROM postfinance_snapshot_positions p JOIN instruments i ON i.instrument_id=p.instrument_id WHERE p.snapshot_id=? ORDER BY CAST(p.value_chf AS REAL) DESC""", (latest["snapshot_id"],), ).fetchall() ] cash = [ { "row_key": _hash("cash", component.currency)[:16], "currency": component.currency, "amount_original": _text(component.amount_original), "fx_rate_to_chf": _text(component.source_fx_rate_to_chf), "value_chf": _money(component.amount_chf), } for component in latest_official_postfinance_cash(conn) ] events = { str(row["event_type"]): int(row["c"]) for row in conn.execute( "SELECT event_type,COUNT(*) c FROM postfinance_ledger_events GROUP BY event_type ORDER BY event_type" ).fetchall() } imports = [ dict(row) for row in conn.execute( """SELECT b.snapshot_at,b.parser_id,b.parser_version,b.confirmed_at,b.status, s.total_chf,s.securities_chf,s.cash_chf,s.position_count,s.cash_count, s.difference_chf,s.reconciliation_status, (SELECT COUNT(*) FROM postfinance_batch_documents d WHERE d.batch_id=b.batch_id) AS document_count FROM postfinance_import_batches b JOIN postfinance_snapshots s ON s.batch_id=b.batch_id ORDER BY b.snapshot_at DESC,b.confirmed_at DESC""" ).fetchall() ] safe_latest = {key: latest[key] for key in latest.keys() if key != "snapshot_id"} return { "latest_snapshot": safe_latest, "positions": positions, "cash": cash, "event_counts": events, "imports": imports, }