from __future__ import annotations import base64 import binascii import json import os 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_aggregation import ( latest_official_postfinance_cash, latest_official_postfinance_positions, ) TOLERANCE = Decimal("0.05") 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]: return ( _decode(str(request.get("zip_content_base64") or ""), "ZIP"), _decode(str(request.get("overview_content_base64") or ""), "overview PDF"), ) 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] = [] 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"])) return _hash(state) 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 _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) for event in bundle.events: event_counts[event.event_type] += 1 quality_counts: dict[str, int] = defaultdict(int) for event in bundle.events: quality_counts[event.quality_status] += 1 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, "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, "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), "warnings": [ "Calculated cost basis remains partial where no documented acquisition exists in this source bundle" ], } 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) fingerprint = _hash(payload, revision) 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() position_collision = conn.execute( "SELECT COUNT(*) FROM positions_snapshot WHERE account_id=? AND snapshot_date=?", (accounts["etrading_depot"], bundle.snapshot.valuation_at[:10]), ).fetchone()[0] cash_collision = conn.execute( """SELECT COUNT(*) FROM cash_balances WHERE account_id=? AND balance_date=? AND source_type='postfinance_official_import'""", (accounts["etrading_cash"], bundle.snapshot.valuation_at[:10]), ).fetchone()[0] if not duplicate and (position_collision or cash_collision): 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( "A different immutable PostFinance source has the same business identity" ) payload.update( { "preview_id": f"pfpreview_{fingerprint[:24]}", "confirmation_id": f"pfconfirm_{fingerprint[24:48]}", "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 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 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") _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, ), ) 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, }