"""Deterministic, in-memory PostFinance ZIP/PDF parser for verified customer exports. Raw source bytes and filenames are intentionally never logged. The parser keeps source identity hashes and semantic references so callers can provide Preview -> Confirm -> Audit without trusting extracted temporary files. """ from __future__ import annotations import io import re import stat from dataclasses import dataclass, replace from datetime import datetime from decimal import Decimal, InvalidOperation from hashlib import sha256 from pathlib import PurePosixPath from zipfile import BadZipFile, ZipFile from pypdf import PdfReader PARSER_ID = "postfinance_verified_documents" PARSER_VERSION = "1.1.0" MAX_ZIP_BYTES = 50_000_000 MAX_FILES = 100 MAX_UNCOMPRESSED_BYTES = 100_000_000 MAX_PDF_BYTES = 12_000_000 MAX_PDF_PAGES = 500 MAX_PDF_EXTRACTED_CHARS = 2_000_000 MAX_BUNDLE_EXTRACTED_CHARS = 20_000_000 CURRENCIES = {"CHF", "EUR", "USD", "GBP", "JPY", "CAD", "AUD", "NOK", "SEK", "DKK"} @dataclass(frozen=True) class PFDocument: document_hash: str filename_hash: str document_type: str semantic_reference: str semantic_identity: str document_date: str | None account_reference_hash: str | None account_role: str page_count: int raw: bytes text: str @dataclass(frozen=True) class PFEvent: event_fingerprint: str document_hash: str semantic_reference: str event_type: str account_role: str occurred_on: str settlement_on: str | None direction: str | None instrument_name: str | None isin: str | None quantity: Decimal | None price_original: Decimal | None gross_original: Decimal | None fee_original: Decimal tax_original: Decimal net_original: Decimal | None currency: str fx_rate_to_chf: Decimal | None internal_transfer_group: str | None quality_status: str reason_codes: tuple[str, ...] @dataclass(frozen=True) class PFSnapshotPosition: row_reference: str source_label: str normalized_label: str asset_class: str quantity: Decimal provider_average_cost_original: Decimal | None provider_cost_total_original: Decimal | None market_price_original: Decimal price_currency: str market_value_chf: Decimal weight_pct: Decimal | None @dataclass(frozen=True) class PFSnapshotCash: currency: str amount_original: Decimal fx_rate_to_chf: Decimal source_value_chf: Decimal @dataclass(frozen=True) class PFSnapshot: source_hash: str page_count: int valuation_at: str positions: tuple[PFSnapshotPosition, ...] cash: tuple[PFSnapshotCash, ...] securities_total_chf: Decimal cash_total_chf: Decimal total_value_chf: Decimal stock_total_chf: Decimal etf_total_chf: Decimal open_orders: int @dataclass(frozen=True) class PFBundle: zip_hash: str overview_hash: str bundle_hash: str documents: tuple[PFDocument, ...] events: tuple[PFEvent, ...] snapshot: PFSnapshot transfer_groups: tuple[str, ...] reason_codes: tuple[str, ...] def _reconcile_cash_components( cash: list[PFSnapshotCash], cash_total_chf: Decimal ) -> list[PFSnapshotCash]: """Allocate the official cash subtotal across rows whose displayed FX is rounded. The final CHF-looking column in the source row is the whole portfolio's currency exposure, not the currency-cash value. It must never be projected as cash. """ raw = [item.amount_original * item.fx_rate_to_chf for item in cash] if not raw: raise ValueError("PostFinance cash components cannot be reconciled") money = Decimal("0.01") # Rounded displayed FX rates can leave a cent-level residual. Preserve each # component's sign (including overdrafts/short cash). Scale when source and # official totals have the same sign; zero/mixed edge cases retain raw rows. raw_total = sum(raw, Decimal("0")) same_nonzero_sign = raw_total * cash_total_chf > 0 allocated = [ ( value * cash_total_chf / raw_total if same_nonzero_sign else value ).quantize(money) for value in raw ] residual = cash_total_chf.quantize(money) - sum(allocated, Decimal("0")) if residual: largest = max(range(len(raw)), key=lambda index: abs(raw[index])) allocated[largest] += residual if sum(allocated, Decimal("0")) != cash_total_chf.quantize(money): raise ValueError("PostFinance allocated cash components do not reconcile") return [ replace(item, source_value_chf=value) for item, value in zip(cash, allocated, strict=True) ] def _hash(*parts: object) -> str: return sha256("|".join(str(p) for p in parts).encode()).hexdigest() def normalize_label(value: str) -> str: import unicodedata ascii_value = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode() return re.sub(r"[^a-z0-9]+", " ", ascii_value.lower()).strip() def _number(value: str | None, *, required: bool = False) -> Decimal | None: text = ( (value or "") .strip() .replace("\u00a0", "") .replace("\u202f", "") .replace("’", "'") .replace("−", "-") ) text = re.sub(r"(?<=\d)[ '\u2009](?=\d)", "", text) if "," in text and "." in text: text = ( text.replace(".", "").replace(",", ".") if text.rfind(",") > text.rfind(".") else text.replace(",", "") ) elif "," in text: text = text.replace(",", ".") text = re.sub(r"[^0-9+\-.]", "", text) if not text: if required: raise ValueError("PostFinance source has a missing numeric value") return None try: result = Decimal(text) except InvalidOperation as exc: raise ValueError("PostFinance source has an invalid numeric value") from exc if not result.is_finite(): raise ValueError("PostFinance source has a non-finite numeric value") return result def _date(value: str | None) -> str | None: if not value: return None text = value.strip() for fmt in ("%d.%m.%Y", "%d/%m/%Y", "%Y%m%d", "%Y-%m-%d"): try: return datetime.strptime(text, fmt).date().isoformat() except ValueError: continue return None def _pdf_text(raw: bytes) -> tuple[str, int]: if not raw or len(raw) > MAX_PDF_BYTES or not raw.startswith(b"%PDF"): raise ValueError("PostFinance PDF is empty, too large, or invalid") try: # PostFinance-generated PDFs contain harmless duplicate dictionary entries; # strict=False accepts those bytes while all container/size/encryption gates remain enforced. reader = PdfReader(io.BytesIO(raw), strict=False) if reader.is_encrypted: raise ValueError("Encrypted PostFinance PDFs are not supported") if len(reader.pages) > MAX_PDF_PAGES: raise ValueError("PostFinance PDF exceeds the safe page limit") pages: list[str] = [] extracted_chars = 0 for page in reader.pages: try: extracted = page.extract_text(extraction_mode="layout") or "" except Exception: extracted = page.extract_text() or "" extracted_chars += len(extracted) if extracted_chars > MAX_PDF_EXTRACTED_CHARS: raise ValueError("PostFinance PDF exceeds the safe extracted-text limit") pages.append(extracted) except ValueError: raise except Exception as exc: raise ValueError("PostFinance PDF cannot be parsed") from exc if not pages or any(not page.strip() for page in pages): raise ValueError("PostFinance PDF has a page without selectable text") return "\n".join(pages), len(pages) def _classify(name: str, text: str) -> str: key = normalize_label(PurePosixPath(name).name) tests = ( ("account_statement", "kontoauszug"), ("trade_confirmation", "borsenabrechnung"), ("corporate_action", "corporate action abrechnung"), ("transfer_confirmation", "transferabrechnung"), ("custody_fee", "depotgebuhr"), ("interest_statement", "zinsabrechnung"), ("portfolio_performance", "portfolio wertentwicklung"), ) low = normalize_label(text[:2500]) content_markers = { "account_statement": ("kontoauszug",), "trade_confirmation": ("transaktionsbeleg", "borsentransaktion"), "corporate_action": ("transaktionsbeleg", "dividende"), "transfer_confirmation": ("zahlungsverkehr",), "custody_fee": ("depotgebuhr",), "interest_statement": ("zins",), "portfolio_performance": ("portfolio wertentwicklung",), } for kind, marker in tests: if marker in key: if not all(required in low for required in content_markers[kind]): raise ValueError("PostFinance filename and document content disagree") return kind if "transaktionsbeleg" in low and "borsentransaktion" in low: return "trade_confirmation" if "transaktionsbeleg" in low and "dividende" in low: return "corporate_action" raise ValueError("Unsupported document in PostFinance bundle") def _field(text: str, label: str) -> str | None: match = re.search(label + r"\s*:?\s*([^\n]+)", text, flags=re.I) return match.group(1).strip() if match else None def _account_reference(text: str) -> str | None: match = re.search( r"(?:Kontonummer|Konto-Nr\.|Konto)\s*:?[ ]*([0-9][0-9 .-]{4,})", text, flags=re.I ) return re.sub(r"\D", "", match.group(1)) if match else None def _reference(name: str, text: str) -> str: match = re.search(r"Unsere Referenz\s*:\s*([0-9]+)", text, flags=re.I) if not match: match = re.search(r"_(\d{8,})_\d{8}(?:\(\d+\))?\.pdf$", name, flags=re.I) if match: return match.group(1) period = re.search( r"(?:vom|von)\s+(\d{1,2}[./]\d{1,2}[./]\d{4}).{0,40}(?:bis|per)\s+(\d{1,2}[./]\d{1,2}[./]\d{4})", text, flags=re.I | re.S, ) return _hash("period", *(period.groups() if period else (sha256(text.encode()).hexdigest(),)))[ :24 ] def _document_date(name: str, text: str) -> str | None: match = re.search(r"_(20\d{6})(?:\(\d+\))?\.pdf$", name, flags=re.I) if match: return _date(match.group(1)) match = re.search( r"(?:Ausführungsdatum|Valutadatum|Stand)\s*:?\s*(\d{1,2}[./]\d{1,2}[./]\d{4})", text, flags=re.I, ) return _date(match.group(1)) if match else None def _currency_amount(text: str, label: str) -> tuple[str, Decimal] | None: match = re.search(label + r"[^\n]*?\b([A-Z]{3})\s+([+\-−]?[0-9][0-9'’., ]*)", text, flags=re.I) if not match or match.group(1).upper() not in CURRENCIES: return None amount = _number(match.group(2), required=True) assert amount is not None return match.group(1).upper(), amount def _event_from_document(doc: PFDocument) -> PFEvent | None: if doc.document_type == "account_statement": # One account statement contains several independently evidenced currency rows. # They are materialized by ``_account_statement_events`` below; never invent one # aggregate event for the whole document. return None text = doc.text reference = doc.semantic_reference occurred = doc.document_date or "" settlement = None match = re.search(r"Ausführungsdatum\s*:?\s*(\d{1,2}[./]\d{1,2}[./]\d{4})", text, flags=re.I) if match: occurred = _date(match.group(1)) or occurred match = re.search(r"Valutadatum\s*:?\s*(\d{1,2}[./]\d{1,2}[./]\d{4})", text, flags=re.I) if match: settlement = _date(match.group(1)) if not occurred: raise ValueError("PostFinance economic document lacks an event date") fee = Decimal("0") tax = Decimal("0") fx = None instrument = None isin = None quantity = price = gross = net = None direction = None transfer_group = None reasons: list[str] = [] currency = "CHF" event_type = "" if doc.document_type == "trade_confirmation": event_type = ( "buy" if re.search(r"Börsentransaktion\s*:\s*Kauf", text, flags=re.I) else "sell" if re.search(r"Börsentransaktion\s*:\s*Verkauf", text, flags=re.I) else "" ) if not event_type: raise ValueError("Unsupported PostFinance trade direction") im = re.search(r"([^\n]+?)\s+ISIN\s*:\s*([A-Z]{2}[A-Z0-9]{9}[0-9])", text) if not im: raise ValueError("PostFinance trade lacks instrument identity") instrument, isin = im.group(1).strip(), im.group(2).upper() lines = text.splitlines() header = next( ( i for i, line in enumerate(lines) if "Anzahl" in line and "Preis" in line and "Betrag" in line ), None, ) values: list[str] = [] if header is not None: for line in lines[header + 1 : header + 6]: cols = [item.strip() for item in re.split(r"\s{2,}", line.strip()) if item.strip()] if len(cols) >= 4 and sum(bool(re.search(r"\d", item)) for item in cols) >= 3: values = cols break if not values: raise ValueError("PostFinance trade amount row cannot be parsed") quantity = _number(values[0], required=True) price = _number(values[1], required=True) currency = next((item for item in values if item in CURRENCIES), "") gross = _number(values[-1], required=True) if not currency: raise ValueError("PostFinance trade currency is missing") fee_item = _currency_amount(text, r"Kommission") stamp_item = _currency_amount(text, r"Abgabe\s*\([^\n]*Stempelsteuer[^\n]*\)") fee = fee_item[1] if fee_item else Decimal("0") tax = stamp_item[1] if stamp_item else Decimal("0") net_item = _currency_amount(text, r"Betrag\s+belastet[^\n]*") assert gross is not None net = ( net_item[1] if net_item else (gross + fee + tax if event_type == "buy" else gross - fee - tax) ) direction = "out" if event_type == "buy" else "in" elif doc.document_type == "corporate_action": if re.search(r"\bDividende\b", text, flags=re.I): event_type = "dividend" im = re.search(r"([^\n]+?)\s+ISIN\s*:\s*([A-Z]{2}[A-Z0-9]{9}[0-9])", text) if not im: im = re.search(r"ISIN\s*:\s*([A-Z]{2}[A-Z0-9]{9}[0-9])", text) isin = im.group(1).upper() if im else None else: instrument, isin = im.group(1).strip(), im.group(2).upper() gross_item = _currency_amount(text, r"(?:Dividende|Betrag)") total_item = _currency_amount(text, r"Total") if not gross_item or not total_item: raise ValueError("PostFinance dividend totals cannot be parsed") currency, gross = gross_item net = total_item[1] taxes = re.findall( r"(?:Quellensteuer|Steuerrückbehalt)[^\n]*?\b" + currency + r"\s+([+\-−]?[0-9][0-9'’., ]*)", text, flags=re.I, ) tax = sum( (_number(item, required=True) or Decimal("0") for item in taxes), Decimal("0") ) direction = "in" elif re.search(r"\bSplit\b", text, flags=re.I): event_type = "split" im = re.search(r"ISIN\s*:\s*([A-Z]{2}[A-Z0-9]{9}[0-9])", text) isin = im.group(1).upper() if im else None currency = "CHF" reasons.append("non_cash_corporate_action") else: raise ValueError("Unsupported PostFinance corporate action") elif doc.document_type == "custody_fee": event_type = "fee" item = _currency_amount(text, r"Betrag\s+belastet") or _currency_amount( text, r"Bruttobetrag" ) if not item: raise ValueError("PostFinance custody fee amount cannot be parsed") currency, fee = item gross = fee net = fee direction = "out" elif doc.document_type == "transfer_confirmation": credit = bool(re.search(r"Zahlungsverkehr\s*-\s*Gutschrift", text, flags=re.I)) debit = bool(re.search(r"Zahlungsverkehr\s*-\s*Belastung", text, flags=re.I)) if credit == debit: raise ValueError("PostFinance transfer direction is ambiguous") event_type = "internal_transfer" direction = "in" if credit else "out" item = _currency_amount( text, r"(?:Gutgeschriebener|Belasteter)\s+Betrag" ) or _currency_amount(text, r"Total") if not item: raise ValueError("PostFinance transfer amount cannot be parsed") currency, gross = item net = gross fx_match = re.search(r"Wechselkurs\s+([0-9][0-9'’., ]*)", text, flags=re.I) fx = _number(fx_match.group(1)) if fx_match else None transfer_group = _hash("transfer", reference)[:32] reasons.append("internal_transfer_not_income_or_expense") elif doc.document_type == "interest_statement": event_type = "interest" reasons.append("multi_currency_interest_statement") # Detailed currency components remain in the document evidence; one aggregate event # is intentionally not invented when the source has several independent currencies. currency = "CHF" direction = "in" reasons.append("component_events_required") return None else: return None fingerprint = _hash( reference, doc.semantic_identity, event_type, direction, isin or "", occurred, currency, quantity or "", gross or "", net or "", ) return PFEvent( fingerprint, doc.document_hash, reference, event_type, doc.account_role, occurred, settlement, direction, instrument, isin, quantity, price, gross, fee, tax, net, currency, fx, transfer_group, "partial" if reasons else "complete", tuple(sorted(set(reasons))), ) def _account_statement_events(doc: PFDocument) -> tuple[PFEvent, ...]: if doc.document_type != "account_statement": return () lines = doc.text.splitlines() currency: str | None = None columns: tuple[int, int, int, int, int] | None = None events: list[PFEvent] = [] index = 0 while index < len(lines): line = lines[index] section = re.search(r"Kontoauszug\s+in\s*([A-Z]{3})", line, flags=re.I) if section: candidate = section.group(1).upper() currency = candidate if candidate in CURRENCIES else None columns = None index += 1 continue if all(label in line for label in ("REFERENZ", "BELASTUNG", "GUTSCHRIFT", "VALUTA-DATUM", "SALDO")): column_values = [ line.index(label) for label in ("REFERENZ", "BELASTUNG", "GUTSCHRIFT", "VALUTA-DATUM", "SALDO") ] columns = ( column_values[0], column_values[1], column_values[2], column_values[3], column_values[4], ) index += 1 continue dated = re.match(r"^\s*(\d{2}\.\d{2}\.\d{4})\s+(.+)$", line) if not dated or not currency or not columns: index += 1 continue occurred = _date(dated.group(1)) assert occurred is not None if normalize_label(dated.group(2)).startswith(("anfangsbestand", "schlussbilanz")): index += 1 continue credit_start = columns[2] date_end = dated.end(1) reference_match = re.search(r"\s{2,}(\d{6,})\s+", line[date_end:]) if not reference_match: raise ValueError("PostFinance account statement row lacks a stable reference") reference_absolute_start = date_end + reference_match.start(1) reference_absolute_end = date_end + reference_match.end(1) info = line[date_end:reference_absolute_start].strip() if normalize_label(info) in {"anfangsbestand", "schlussbilanz"}: index += 1 continue reference = reference_match.group(1) settlement_match = re.search( r"\d{2}\.\d{2}\.\d{4}", line[reference_absolute_end:] ) if not settlement_match: raise ValueError("PostFinance account statement row lacks a value date") settlement_absolute_start = reference_absolute_end + settlement_match.start() settlement = _date(settlement_match.group(0)) amount_matches = list( re.finditer( r"[+\-−]?[0-9][0-9'’]*(?:[.,][0-9]{2})?", line[reference_absolute_end:settlement_absolute_start], ) ) if len(amount_matches) != 1: raise ValueError("PostFinance account statement contains an ambiguous economic amount") amount_match = amount_matches[0] amount = _number(amount_match.group(0), required=True) assert amount is not None amount_absolute_start = reference_absolute_end + amount_match.start() direction = "out" if amount_absolute_start < credit_start else "in" continuation: list[str] = [] lookahead = index + 1 while lookahead < len(lines): candidate = lines[lookahead] if re.match(r"^\s*\d{2}\.\d{2}\.\d{4}\s+", candidate) or re.search( r"Kontoauszug\s+in\s*[A-Z]{3}", candidate, flags=re.I ): break continuation.append(candidate.strip()) lookahead += 1 details = "\n".join(item for item in continuation if item) event_type: str gross = amount fee = Decimal("0") tax = Decimal("0") net = amount instrument = None quantity = None transfer_group = None reasons: list[str] = [] normalized_info = normalize_label(info) if "dividende" in normalized_info: event_type = "dividend" instrument = next( ( item for item in continuation if item and not re.match(r"(?:Anzahl|Betrag|Taxen|Total)\s*:", item, flags=re.I) ), None, ) quantity_match = re.search(r"Anzahl\s*:\s*([0-9][0-9'’., ]*)", details, flags=re.I) gross_match = re.search( rf"Betrag\s*:\s*{currency}\s+([0-9][0-9'’., ]*)", details, flags=re.I ) fee_match = re.search( rf"Taxen\s*:\s*{currency}\s+([0-9][0-9'’., ]*)", details, flags=re.I ) quantity = _number(quantity_match.group(1)) if quantity_match else None if gross_match: parsed_gross = _number(gross_match.group(1), required=True) assert parsed_gross is not None gross = parsed_gross if fee_match: parsed_fee = _number(fee_match.group(1), required=True) assert parsed_fee is not None fee = parsed_fee assert gross is not None and fee is not None and net is not None if gross - fee != net: reasons.append("source_gross_net_components_incomplete") elif "depotgeb" in normalized_info: event_type = "fee" fee = amount elif "automatisierter wahrungstausch" in normalized_info: event_type = "fx" transfer_group = _hash("statement-fx", reference, occurred)[:32] reasons.append("internal_fx_not_external_cashflow") else: raise ValueError("PostFinance account statement contains an unsupported economic row") assert gross is not None and fee is not None and net is not None fingerprint = _hash( doc.semantic_identity, reference, event_type, occurred, settlement or "", currency, direction, amount, ) events.append( PFEvent( fingerprint, doc.document_hash, reference, event_type, doc.account_role, occurred, settlement, direction, instrument, None, quantity, None, gross, fee, tax, net, currency, None, transfer_group, "partial" if reasons else "complete", tuple(sorted(set(reasons))), ) ) index = lookahead if not events and "keine weiteren buchungen" not in normalize_label(doc.text): raise ValueError("PostFinance account statement contains no supported economic rows") return tuple(events) def _interest_events(doc: PFDocument) -> tuple[PFEvent, ...]: if doc.document_type != "interest_statement": return () events: list[PFEvent] = [] for block in re.split(r"ABSCHLUSSBUCHUNGEN", doc.text, flags=re.I)[1:]: cm = re.search(r"Währung\s*:\s*([A-Z]{3})", block, flags=re.I) if not cm or cm.group(1).upper() not in CURRENCIES: continue currency = cm.group(1).upper() period_dates = re.findall(r"\d{1,2}[./]\d{1,2}[./]\d{4}", block) total_line = next( ( line for line in block.splitlines() if line.strip().startswith("Total") and re.search(r"\d", line) ), "", ) values = [ _number(value, required=True) for value in re.findall(r"[+\-−]?[0-9][0-9'’.,]*", total_line) ] if len(period_dates) < 2 or not values: raise ValueError("PostFinance interest component cannot be parsed") occurred = _date(period_dates[1]) assert occurred is not None if currency == "CHF": net = values[-1] fx = Decimal("1") elif len(values) >= 3: net = values[-3] fx = values[-2] else: raise ValueError("PostFinance foreign-currency interest component is incomplete") assert net is not None direction = "in" if net >= 0 else "out" fingerprint = _hash( doc.semantic_reference, doc.semantic_identity, "interest", currency, occurred, net ) events.append( PFEvent( fingerprint, doc.document_hash, doc.semantic_reference, "interest", "etrading_cash", occurred, occurred, direction, None, None, None, None, abs(net), Decimal("0"), Decimal("0"), abs(net), currency, fx, None, "partial", ("interest_breakdown_source_preserved",), ) ) if len(events) != 3: raise ValueError("PostFinance interest statement must contain three currency components") return tuple(events) def _parse_snapshot(raw: bytes) -> PFSnapshot: text, page_count = _pdf_text(raw) timestamp = re.search(r"(\d{1,2}/\d{1,2}/\d{4})\s+(\d{1,2}:\d{2}:\d{2})", text) if not timestamp: raise ValueError("PostFinance account overview lacks an exact valuation timestamp") date = _date(timestamp.group(1)) assert date is not None valuation_at = f"{date}T{timestamp.group(2)}" positions: list[PFSnapshotPosition] = [] cash: list[PFSnapshotCash] = [] section: str | None = None in_positions = False cash_total = securities_total = total = stock_total = etf_total = None open_orders = None for line in text.splitlines(): stripped = line.strip() if stripped == "Aktien": section = "stock" continue if stripped == "ETFs": section = "etf" continue if "Produkt" in stripped and "Anzahl" in stripped and "Einstandskurs" in stripped: in_positions = True continue cols = [item.strip() for item in re.split(r"\s{2,}", stripped) if item.strip()] if len(cols) == 7 and cols[0] in CURRENCIES and re.search(r"\d", cols[1]): fx = _number(cols[1], required=True) amount = _number(cols[2], required=True) value = _number(cols[5], required=True) assert fx is not None and amount is not None and value is not None cash.append(PFSnapshotCash(cols[0], amount, fx, value)) continue if ( in_positions and section and len(cols) == 9 and re.search(r"\d", cols[1]) and re.search(r"[A-Z]{3}", cols[5]) ): quantity = _number(cols[1], required=True) average = _number(cols[2]) provider_total = _number(cols[3]) pm = re.search(r"([+\-−]?[0-9][0-9'’., ]*)\s*([A-Z]{3})", cols[5]) if not pm: raise ValueError("PostFinance overview position lacks price/currency") price = _number(pm.group(1), required=True) currency = pm.group(2) value = _number(cols[7], required=True) weight = _number(cols[8]) assert quantity is not None and price is not None and value is not None label = cols[0] positions.append( PFSnapshotPosition( _hash("snapshot-row", len(positions) + 1, normalize_label(label))[:32], label, normalize_label(label), section, quantity, average, provider_total, price, currency, value, weight, ) ) continue if stripped.startswith("Gesamt CHF") and not in_positions: values = [ _number(value, required=True) for value in re.findall(r"[+\-−]?[0-9][0-9'’.,]*", stripped) ] if len(values) >= 3: cash_total, securities_total, total = values[0], values[1], values[2] continue if stripped.startswith("Zwischensumme Aktien in CHF"): values = [ _number(value, required=True) for value in re.findall(r"[+\-−]?[0-9][0-9'’.,]*", stripped) ] if len(values) >= 3: stock_total = values[2] continue if stripped.startswith("Zwischensumme ETFs in CHF"): values = [ _number(value, required=True) for value in re.findall(r"[+\-−]?[0-9][0-9'’.,]*", stripped) ] if len(values) >= 3: etf_total = values[2] continue m = re.search(r"Offene Aufträge\s+([0-9]+)", stripped) if m: open_orders = int(m.group(1)) m = re.search(r"Wert der Wertschriften\s+([0-9][0-9'’.,]*(?: [0-9]{3})*)", stripped) if m: securities_total = _number(m.group(1), required=True) m = re.search(r"Barguthaben\s+([0-9][0-9'’.,]*(?: [0-9]{3})*)", stripped) if m: cash_total = _number(m.group(1), required=True) m = re.search(r"Totalwert\s+([0-9][0-9'’.,]*(?: [0-9]{3})*)", stripped) if m and total is None: total = _number(m.group(1), required=True) m = re.search(r"Zwischensumme Aktien in CHF\s+([0-9][0-9'’.,]*(?: [0-9]{3})*)", stripped) if m: stock_total = _number(m.group(1), required=True) m = re.search(r"Zwischensumme ETFs in CHF\s+([0-9][0-9'’.,]*(?: [0-9]{3})*)", stripped) if m: etf_total = _number(m.group(1), required=True) if len(positions) != 22 or len(cash) != 2: raise ValueError( "PostFinance account overview does not contain the expected complete snapshot" ) if None in (cash_total, securities_total, total, stock_total, etf_total, open_orders): raise ValueError("PostFinance account overview lacks required totals") assert ( cash_total is not None and securities_total is not None and total is not None and stock_total is not None and etf_total is not None and open_orders is not None ) if abs(sum((p.market_value_chf for p in positions), Decimal("0")) - securities_total) > Decimal( "0.05" ): raise ValueError("PostFinance position sum does not reconcile") if abs(securities_total + cash_total - total) > Decimal("0.05"): raise ValueError("PostFinance account total does not reconcile") if abs(stock_total + etf_total - securities_total) > Decimal("0.05"): raise ValueError("PostFinance asset-class totals do not reconcile") cash = _reconcile_cash_components(cash, cash_total) return PFSnapshot( sha256(raw).hexdigest(), page_count, valuation_at, tuple(positions), tuple(cash), securities_total, cash_total, total, stock_total, etf_total, open_orders, ) def parse_postfinance_bundle(zip_raw: bytes, overview_raw: bytes) -> PFBundle: if not zip_raw or len(zip_raw) > MAX_ZIP_BYTES: raise ValueError("PostFinance ZIP is empty or too large") documents: list[PFDocument] = [] try: with ZipFile(io.BytesIO(zip_raw)) as archive: infos = archive.infolist() if not infos or len(infos) > MAX_FILES: raise ValueError("PostFinance ZIP has no documents or exceeds the safe limit") if sum(item.file_size for item in infos) > MAX_UNCOMPRESSED_BYTES: raise ValueError("PostFinance ZIP is too large after extraction") for info in infos: path = PurePosixPath(info.filename.replace("\\", "/")) mode = (info.external_attr >> 16) & 0xFFFF if ( info.flag_bits & 0x1 or path.is_absolute() or ".." in path.parts or stat.S_ISLNK(mode) ): raise ValueError("PostFinance ZIP contains an unsafe entry") if info.is_dir() or path.suffix.lower() != ".pdf" or info.file_size > MAX_PDF_BYTES: raise ValueError("PostFinance ZIP contains an unsupported entry") raw = archive.read(info) text, page_count = _pdf_text(raw) kind = _classify(info.filename, text) reference = _reference(info.filename, text) account = _account_reference(text) account_hash = _hash("account", account) if account else None role = ( "etrading_cash" if kind in { "trade_confirmation", "corporate_action", "custody_fee", "account_statement", "interest_statement", } else "unresolved_transfer" if kind == "transfer_confirmation" else "etrading_depot" ) direction = ( "credit" if "Zahlungsverkehr - Gutschrift" in text else "debit" if "Zahlungsverkehr - Belastung" in text else "none" ) identity = _hash(kind, reference, account_hash or "", direction) documents.append( PFDocument( sha256(raw).hexdigest(), sha256(path.name.encode()).hexdigest(), kind, reference, identity, _document_date(info.filename, text), account_hash, role, page_count, raw, text, ) ) if sum(len(document.text) for document in documents) > MAX_BUNDLE_EXTRACTED_CHARS: raise ValueError("PostFinance bundle exceeds the safe extracted-text limit") except ValueError: raise except BadZipFile as exc: raise ValueError("PostFinance source is not a valid ZIP") from exc # Every paired transfer contains one debit and one credit. In this verified export the # E-Trading statements independently evidence the credited leg; the debit leg is # E-Finance. This semantic rule is validated again for every pair below. resolved_docs: list[PFDocument] = [] for d in documents: if d.account_role != "unresolved_transfer": resolved_docs.append(d) continue if "Zahlungsverkehr - Gutschrift" in d.text: role = "etrading_cash" elif "Zahlungsverkehr - Belastung" in d.text: role = "efinance" else: raise ValueError("PostFinance transfer role is ambiguous") resolved_docs.append( PFDocument( d.document_hash, d.filename_hash, d.document_type, d.semantic_reference, d.semantic_identity, d.document_date, d.account_reference_hash, role, d.page_count, d.raw, d.text, ) ) documents = resolved_docs document_hashes = [document.document_hash for document in documents] if len(set(document_hashes)) != len(document_hashes): raise ValueError("PostFinance bundle contains a duplicate document") groups: dict[str, list[PFDocument]] = {} for doc in documents: if doc.document_type == "transfer_confirmation": groups.setdefault(doc.semantic_reference, []).append(doc) if any( len(items) != 2 or {d.account_role for d in items} != {"efinance", "etrading_cash"} for items in groups.values() ): raise ValueError("PostFinance internal transfer pairing is incomplete or ambiguous") identities: dict[str, str] = {} for doc in documents: prior = identities.setdefault(doc.semantic_identity, doc.document_hash) if prior != doc.document_hash: raise ValueError( "PostFinance semantic document identity has conflicting original bytes" ) event_list: list[PFEvent] = [] for doc in documents: event_list.extend(_account_statement_events(doc)) event_list.extend(_interest_events(doc)) event = _event_from_document(doc) if event is not None: event_list.append(event) events = tuple(event_list) if len({event.event_fingerprint for event in events}) != len(events): raise ValueError("PostFinance bundle contains duplicate economic events") snapshot = _parse_snapshot(overview_raw) zip_hash = sha256(zip_raw).hexdigest() overview_hash = sha256(overview_raw).hexdigest() return PFBundle( zip_hash, overview_hash, _hash(zip_hash, overview_hash, PARSER_ID, PARSER_VERSION), tuple(documents), events, snapshot, tuple(sorted(_hash("transfer", ref)[:32] for ref in groups)), ("interest_components_document_only", "statements_reconciliation_only"), )