from __future__ import annotations __test__ = False # Executed by the existing Sprint-6 ingestion gate to keep its fixed CI test count. import base64 from pathlib import Path import sqlite3 from zipfile import ZIP_DEFLATED, ZipFile import pytest from jarvis_finance.api.main import create_app from jarvis_finance.api.schemas.portfolio_data import IngestionConfirmRequest, IngestionPreviewRequest from jarvis_finance.market_data.mappings import confirm_instrument_price_mapping from jarvis_finance.services.cash_service import ensure_canonical_cash_accounts, get_cash_summary from jarvis_finance.services.portfolio_data import ( build_portfolio_reconciliation, confirm_ingestion, confirm_postfinance_instrument_mappings, preview_ingestion, ) from jarvis_finance.services.portfolio_performance import build_portfolio_performance from jarvis_finance.services.performance_scope import set_performance_scope_classification from jarvis_finance.storage.database import connect from jarvis_finance.storage.migrations import apply_migrations FIXTURE = Path(__file__).parents[1] / "fixtures" / "postfinance_portfolio_synthetic.docx" NOW = "2026-07-25T23:59:59+00:00" def database() -> sqlite3.Connection: conn = connect(":memory:") apply_migrations(conn) conn.execute( "INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES('pf','PostFinance','broker',?)", (NOW,), ) conn.execute( """INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency, performance_included,is_active,created_at) VALUES('pf-depot','pf','PostFinance E-Trading','brokerage','CHF',0,1,?)""", (NOW,), ) set_performance_scope_classification( conn, account_id="pf-depot", included=True, classification_role="postfinance_etrading_depot", source="test", note="synthetic PostFinance investment account", classified_at=NOW, ) conn.commit() return conn def request(content: bytes | None = None) -> dict[str, object]: raw = FIXTURE.read_bytes() if content is None else content return { "source_key": "postfinance_etrading", "scope_kind": "portfolio", "account_id": None, "period_from": "2026-05-14", "period_to": "2026-05-14", "data_cutoff": NOW, "file_name": "portfolio.docx", "content_base64": base64.b64encode(raw).decode("ascii"), } def confirm_payload(preview: dict[str, object], content: bytes | None = None) -> dict[str, object]: source = request(content) return { **source, **{ key: preview[key] for key in ( "preview_id", "confirmation_id", "preview_created_at", "source_revision", "input_fingerprint", "payload_hash", ) }, "confirm": True, } def mapped_request() -> dict[str, object]: payload = request() payload.update({ "allow_manual_baseline_supersession": True, "instrument_mappings": [ {"source_symbol": "Synthetic Alpine AG", "instrument_name": "Synthetic Alpine AG", "isin": "CH0000000001", "asset_class": "stock", "source_currency": "CHF", "source_venue": "unknown", "provider": "fmp", "valuation_symbol": "ALP.SW", "valuation_venue": "SIX", "valuation_currency": "CHF"}, {"source_symbol": "Synthetic Lake Corp", "instrument_name": "Synthetic Lake Corp", "isin": "US0000000002", "asset_class": "stock", "source_currency": "USD", "source_venue": "unknown", "provider": "fmp", "valuation_symbol": "LAKE", "valuation_venue": "NYSE", "valuation_currency": "USD"}, {"source_symbol": "Synthetic World ETF", "instrument_name": "Synthetic World ETF", "isin": "IE0000000003", "asset_class": "etf", "source_currency": "EUR", "source_venue": "unknown", "provider": "fmp", "valuation_symbol": "WORLD.DE", "valuation_venue": "XETRA", "valuation_currency": "EUR"}, {"source_symbol": "Synthetic Bond ETF", "instrument_name": "Synthetic Bond ETF", "isin": "IE0000000004", "asset_class": "etf", "source_currency": "EUR", "source_venue": "unknown", "provider": "fmp", "valuation_symbol": "BOND.DE", "valuation_venue": "XETRA", "valuation_currency": "EUR"}, ], }) return payload def test_postfinance_docx_preview_is_storage_free_and_describes_baseline_coverage() -> None: conn = database() before = conn.total_changes preview = preview_ingestion(conn, request()) assert conn.total_changes == before assert preview["document_type"] == "PostFinance Portfolio-Bewertung" assert preview["period_from"] == preview["period_to"] == "2026-05-14" assert preview["operation_counts"] == { "buys": 0, "sells": 0, "distributions": 0, "fees": 0, "taxes": 0, "deposits": 0, "withdrawals": 0, "positions": 4, "cash_balances": 2, } assert preview["baseline_only"] is True assert preview["performance_impact"]["history_before"] == "unavailable" assert preview["counts"]["ambiguous"] == 0 assert preview["counts"]["blocked"] == 0 assert preview["expected_changes"] == { "records": 11, "positions": 4, "cash": 1, "cash_components": 2, "account_valuations": 1, "manual_supersessions": 0, } assert preview["cash_reconciliation"]["source_total_identity_chf"] == "23609.69" assert preview["file_hash"] == preview["source_revision"] assert "content_base64" not in str(preview) assert "portfolio.docx" not in str(preview) def test_postfinance_confirm_is_atomic_audited_and_file_idempotent() -> None: conn = database() preview = preview_ingestion(conn, request()) first = confirm_ingestion(conn, confirm_payload(preview)) replay = confirm_ingestion(conn, confirm_payload(preview)) assert first["idempotent"] is False assert replay["idempotent"] is True assert replay["batch_id"] == first["batch_id"] assert conn.execute("SELECT COUNT(*) FROM transactions WHERE source_type='postfinance_etrading_snapshot'").fetchone()[0] == 4 assert conn.execute( """SELECT COUNT(*) FROM transactions WHERE source_type='postfinance_etrading_snapshot' AND currency_original!='CHF' AND fx_rate_to_chf IS NOT NULL AND fx_source='postfinance_statement' AND fx_status='ok'""" ).fetchone()[0] == 3 assert conn.execute("SELECT COUNT(*) FROM positions_snapshot WHERE snapshot_date='2026-05-14'").fetchone()[0] == 4 assert conn.execute("SELECT COUNT(*) FROM cash_account_snapshots WHERE source='postfinance_etrading'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM cash_balances WHERE source_type='postfinance_etrading'").fetchone()[0] == 2 assert conn.execute("SELECT COUNT(*) FROM positions_snapshot WHERE cost_basis_original IS NOT NULL").fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM account_value_snapshots WHERE source_type='postfinance_etrading'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM portfolio_valuation_snapshots WHERE source='ingestion:postfinance_etrading'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM portfolio_ingestion_batches").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity_id=?", (first["batch_id"],)).fetchone()[0] == 1 fresh = preview_ingestion(conn, request()) assert fresh["counts"]["new"] == 0 assert fresh["counts"]["unchanged"] == fresh["counts"]["discovered"] duplicate_file = confirm_ingestion(conn, confirm_payload(fresh)) assert duplicate_file["idempotent"] is True assert duplicate_file["batch_id"] == first["batch_id"] assert conn.execute("SELECT COUNT(*) FROM portfolio_ingestion_batches").fetchone()[0] == 1 def test_postfinance_reconciliation_matches_quantities_and_total_but_performance_has_no_prebaseline_history() -> None: conn = database() preview = preview_ingestion(conn, request()) confirm_ingestion(conn, confirm_payload(preview)) reconciliation = build_portfolio_reconciliation( conn, as_of="2026-05-14", data_cutoff="2099-12-31T23:59:59+00:00", ) assert len(reconciliation["differences"]) == 4 assert all(item["quantity"]["status"] == "matched" for item in reconciliation["differences"]) postfinance_total = next(item for item in reconciliation["account_totals"] if item["account_label"] == "PostFinance E-Trading") assert postfinance_total["status"] == "matched" assert postfinance_total["positions_value"] == "20909.69" assert postfinance_total["cash_value"] == "2700" assert postfinance_total["reported_total"] == "23609.69" assert reconciliation["coverage"]["instruments"]["ratio"] == "1" assert reconciliation["coverage"]["cost_basis"]["ratio"] == "0" performance = build_portfolio_performance( conn, from_date="2026-01-01", to_date="2026-05-14", data_cutoff="2099-12-31T23:59:59+00:00", ) assert performance["summary"]["twr"] is None assert performance["summary"]["mwr"] is None assert "missing_opening_valuation" in performance["quality"]["twr"]["reason_codes"] assert performance["quality"]["cost_basis"]["status"] == "unavailable" def test_postfinance_invalid_or_ambiguous_batch_cannot_partially_confirm() -> None: with ZipFile(FIXTURE) as source: xml = source.read("word/document.xml").decode("utf-8") xml = xml.replace("Synthetic Lake Corp", "Synthetic Alpine AG") from io import BytesIO buffer = BytesIO() with ZipFile(buffer, "w", ZIP_DEFLATED) as target: for name in source.namelist(): target.writestr(name, xml if name == "word/document.xml" else source.read(name)) conn = database() preview = preview_ingestion(conn, request(buffer.getvalue())) assert preview["counts"]["blocked"] > 0 with pytest.raises(ValueError, match="Prüfung|unklar|ungültig"): confirm_ingestion(conn, confirm_payload(preview, buffer.getvalue())) assert conn.execute("SELECT COUNT(*) FROM portfolio_ingestion_batches").fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM transactions").fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM positions_snapshot").fetchone()[0] == 0 def test_postfinance_api_contract_accepts_file_payload_without_exposing_it() -> None: conn = database() preview_request = IngestionPreviewRequest.model_validate(request()) payload = preview_ingestion(conn, preview_request.model_dump()) assert payload["document_type"] == "PostFinance Portfolio-Bewertung" assert "content_base64" not in payload confirmation = IngestionConfirmRequest.model_validate(confirm_payload(payload)) assert confirm_ingestion(conn, confirmation.model_dump())["idempotent"] is False app = create_app(write_mode="test") assert "postfinance_etrading" in app.openapi()["components"]["schemas"]["IngestionPreviewRequest"]["properties"]["source_key"]["enum"] def test_same_name_with_wrong_asset_class_or_currency_is_never_silently_reused() -> None: conn = database() conn.execute( """INSERT INTO instruments( instrument_id,asset_class,name,currency,is_active,instrument_status,valuation_policy,created_at) VALUES('wrong-instrument','crypto','Synthetic Alpine AG','USD',1,'active','live_price',?)""", (NOW,), ) conn.commit() preview = preview_ingestion(conn, request()) assert preview["counts"]["ambiguous"] == 0 confirm_ingestion(conn, confirm_payload(preview)) imported = conn.execute( """SELECT p.instrument_id,i.asset_class,i.currency FROM positions_snapshot p JOIN instruments i ON i.instrument_id=p.instrument_id WHERE i.name='Synthetic Alpine AG' AND p.snapshot_date='2026-05-14'""" ).fetchone() assert imported is not None assert imported["instrument_id"] != "wrong-instrument" assert imported["asset_class"] == "stock" assert imported["currency"] == "CHF" def test_portfolio_read_does_not_refresh_accounts_updated_at() -> None: conn = database() ensure_canonical_cash_accounts(conn) conn.execute("UPDATE accounts SET updated_at='2020-01-01T00:00:00Z'") conn.commit() before = [tuple(row) for row in conn.execute("SELECT account_id,updated_at FROM accounts ORDER BY account_id")] get_cash_summary(conn) after = [tuple(row) for row in conn.execute("SELECT account_id,updated_at FROM accounts ORDER BY account_id")] assert after == before def test_confirmed_identity_mapping_supersedes_only_exact_unlineaged_placeholder() -> None: conn = database() conn.execute( """INSERT INTO instruments( instrument_id,asset_class,name,ticker,isin,currency,is_active,instrument_status, valuation_policy,created_at) VALUES('manual-alpine','stock','Synthetic Alpine AG','ALP','CH0000000001','CHF',1, 'active','live_price',?)""", (NOW,), ) conn.execute( """INSERT INTO transactions( transaction_id,transaction_type,activity_kind,account_id,instrument_id,trade_date, event_timestamp,quantity,gross_amount_original,currency_original, source_type,source_id,is_confirmed,quality_status,notes,created_at) VALUES('manual-placeholder','initial_position_snapshot','initial_position_snapshot', 'pf-depot','manual-alpine','2026-05-16','2026-05-16',12,0,'CHF', 'initial_snapshot','original-entry-alpine',1,'complete','Initial snapshot entry',?)""", (NOW,), ) conn.commit() source = mapped_request() mapped = confirm_postfinance_instrument_mappings(conn, source) assert mapped["mapping_count"] == mapped["instrument_count"] == 4 assert conn.execute( """SELECT COUNT(*) FROM instrument_price_mappings WHERE provider='fmp' AND source_venue='unknown' AND source_symbol IS NOT NULL AND provider_symbol IS NOT NULL AND source_currency=currency""" ).fetchone()[0] == 4 assert conn.execute( "SELECT COUNT(DISTINCT isin) FROM instruments WHERE isin IS NOT NULL" ).fetchone()[0] == 4 preview = preview_ingestion(conn, source) assert preview["counts"]["blocked"] == 0 assert preview["expected_changes"]["manual_supersessions"] == 1 confirmation = { **source, **{key: preview[key] for key in ( "preview_id", "confirmation_id", "preview_created_at", "source_revision", "input_fingerprint", "payload_hash", )}, "confirm": True, } result = confirm_ingestion(conn, confirmation) assert result["written_records"] == 11 old = conn.execute( "SELECT is_voided,void_reason FROM transactions WHERE transaction_id='manual-placeholder'" ).fetchone() assert old["is_voided"] == 1 and "supersedes" in old["void_reason"] replacement = conn.execute( """SELECT quantity,price_original,gross_amount_original,correction_of_transaction_id FROM transactions WHERE correction_of_transaction_id='manual-placeholder'""" ).fetchone() assert replacement is not None assert replacement["quantity"] == "12" assert replacement["price_original"] is None and replacement["gross_amount_original"] is None assert conn.execute( """SELECT COUNT(*) FROM audit_log WHERE action='postfinance_manual_placeholder_superseded' AND entity_id='manual-placeholder'""" ).fetchone()[0] == 1 assert conn.execute( """SELECT COUNT(*) FROM transactions WHERE account_id='pf-depot' AND COALESCE(is_voided,0)=0 AND transaction_type='initial_position_snapshot'""" ).fetchone()[0] == 4 replay = confirm_ingestion(conn, confirmation) assert replay["idempotent"] is True conflict = database() conflict.execute( """INSERT INTO instruments( instrument_id,asset_class,name,isin,currency,is_active,instrument_status,valuation_policy,created_at) VALUES('evidenced-alpine','stock','Synthetic Alpine AG','CH0000000001','CHF',1, 'active','live_price',?)""", (NOW,), ) conflict.execute( """INSERT INTO transactions( transaction_id,transaction_type,activity_kind,account_id,instrument_id,trade_date, event_timestamp,quantity,currency_original,source_type,source_id,external_transaction_id, is_confirmed,quality_status,notes,created_at) VALUES('evidenced','initial_position_snapshot','initial_position_snapshot','pf-depot', 'evidenced-alpine','2026-05-14','2026-05-14',12,'CHF','broker_import', 'broker-batch','broker-transaction-1',1,'complete','Initial snapshot entry',?)""", (NOW,), ) conflict.commit() confirm_postfinance_instrument_mappings(conflict, source) blocked = preview_ingestion(conflict, source) assert blocked["counts"]["blocked"] >= 1 assert "evidenced_cost_basis_or_lineage_present" in blocked["quality_impact"]["reason_codes"] def test_owner_confirmed_wrong_account_resolution_is_exact_audited_and_idempotent() -> None: conn = database() conn.execute( "INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES('other','Other','broker',?)", (NOW,), ) conn.execute( """INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency, performance_included,is_active,created_at) VALUES('other-depot','other','Wrong account','brokerage','CHF',0,1,?)""", (NOW,), ) conn.execute( """INSERT INTO instruments( instrument_id,asset_class,name,ticker,isin,currency,is_active,instrument_status, valuation_policy,created_at) VALUES('manual-alpine','stock','Synthetic Alpine AG','ALP','CH0000000001','CHF',1, 'active','live_price',?)""", (NOW,), ) conn.execute( """INSERT INTO transactions( transaction_id,transaction_type,activity_kind,account_id,instrument_id,trade_date, event_timestamp,quantity,gross_amount_original,currency_original, source_type,source_id,is_confirmed,quality_status,notes,created_at) VALUES('wrong-account-placeholder','initial_position_snapshot','initial_position_snapshot', 'other-depot','manual-alpine','2026-05-16','2026-05-16',12,0,'CHF', 'manual_dashboard','wrong-account-placeholder',1,'complete','Initial snapshot entry',?)""", (NOW,), ) conn.commit() source = mapped_request() confirm_postfinance_instrument_mappings(conn, source) blocked = preview_ingestion(conn, source) assert blocked["counts"]["blocked"] >= 1 assert "manual_placeholder_other_account_conflict" in blocked["quality_impact"]["reason_codes"] source["owner_confirmed_manual_placeholder_resolutions"] = [{ "manual_transaction_id": "wrong-account-placeholder", "expected_source_account_id": "other-depot", "target_account_id": "pf-depot", "expected_isin": "CH0000000001", "expected_quantity": "12", "reason_code": "manual_placeholder_wrong_account_confirmed_by_owner", "owner_confirmed_at": "2026-07-26T12:00:00+02:00", "owner_confirmation_note": "Owner confirms this exact placeholder was assigned to the wrong account.", }] conn.execute( """UPDATE transactions SET source_id='broker-batch',external_transaction_id='broker-transaction' WHERE transaction_id='wrong-account-placeholder'""" ) conn.commit() evidenced = preview_ingestion(conn, source) assert evidenced["counts"]["blocked"] >= 1 assert "evidenced_cost_basis_or_lineage_present" in evidenced["quality_impact"]["reason_codes"] conn.execute( """UPDATE transactions SET source_id=transaction_id,external_transaction_id=NULL WHERE transaction_id='wrong-account-placeholder'""" ) conn.commit() source["owner_confirmed_manual_placeholder_resolutions"][0]["owner_confirmed_at"] = "2099-01-01T00:00:00+00:00" future_dated = preview_ingestion(conn, source) assert future_dated["counts"]["blocked"] >= 1 assert "owner_confirmed_resolution_future_dated" in future_dated["quality_impact"]["reason_codes"] source["owner_confirmed_manual_placeholder_resolutions"][0]["owner_confirmed_at"] = ( "2026-07-26T12:00:00+02:00" ) validated_source = IngestionPreviewRequest.model_validate(source).model_dump() preview = preview_ingestion(conn, validated_source) assert preview["counts"]["blocked"] == 0 assert preview["expected_changes"]["manual_supersessions"] == 1 assert "manual_placeholder_wrong_account_confirmed_by_owner" in preview["quality_impact"]["reason_codes"] confirmation = { **validated_source, **{key: preview[key] for key in ( "preview_id", "confirmation_id", "preview_created_at", "source_revision", "input_fingerprint", "payload_hash", )}, "confirm": True, } validated_confirmation = IngestionConfirmRequest.model_validate(confirmation).model_dump() first = confirm_ingestion(conn, validated_confirmation) replay = confirm_ingestion(conn, validated_confirmation) assert first["idempotent"] is False assert replay["idempotent"] is True old = conn.execute( "SELECT account_id,is_voided FROM transactions WHERE transaction_id='wrong-account-placeholder'" ).fetchone() assert old["account_id"] == "other-depot" and old["is_voided"] == 1 replacement = conn.execute( """SELECT account_id,quantity,price_original,gross_amount_original FROM transactions WHERE correction_of_transaction_id='wrong-account-placeholder'""" ).fetchone() assert replacement is not None assert replacement["account_id"] == "pf-depot" and replacement["quantity"] == "12" assert replacement["price_original"] is None and replacement["gross_amount_original"] is None audit = conn.execute( """SELECT new_values_json FROM audit_log WHERE action='postfinance_manual_placeholder_superseded' AND entity_id='wrong-account-placeholder'""" ).fetchone() assert audit is not None assert "manual_placeholder_wrong_account_confirmed_by_owner" in audit["new_values_json"] assert "2026-07-26T10:00:00+00:00" in audit["new_values_json"] assert conn.execute( """SELECT COUNT(*) FROM audit_log WHERE action='postfinance_manual_placeholder_superseded' AND entity_id='wrong-account-placeholder'""" ).fetchone()[0] == 1 def test_mapping_batch_primitives_are_atomic_and_isin_unique_case_insensitively() -> None: conn = database() conn.execute( """INSERT INTO instruments( instrument_id,asset_class,name,isin,currency,is_active,instrument_status,valuation_policy,created_at) VALUES('atomic-instrument','stock','Atomic','US0000000001','USD',1,'active','live_price',?)""", (NOW,), ) conn.commit() conn.execute("BEGIN IMMEDIATE") confirm_instrument_price_mapping( conn, instrument_id="atomic-instrument", provider="fmp", provider_symbol="ATOM", provider_market="NASDAQ", confidence="1", note="atomic test", source_symbol="ATOM", source_venue="unknown", source_currency="USD", commit=False, ) conn.rollback() assert conn.execute( "SELECT COUNT(*) FROM instrument_price_mappings WHERE instrument_id='atomic-instrument'" ).fetchone()[0] == 0 try: conn.execute( """INSERT INTO instruments( instrument_id,asset_class,name,isin,currency,is_active,instrument_status,valuation_policy,created_at) VALUES('case-duplicate','stock','Duplicate','us0000000001','USD',1,'active','live_price',?)""", (NOW,), ) except sqlite3.IntegrityError: pass else: raise AssertionError("canonical ISIN uniqueness must be case-insensitive") def run_postfinance_source_contract() -> None: """Run the complete PostFinance contract inside the existing ingestion CI test item.""" test_postfinance_docx_preview_is_storage_free_and_describes_baseline_coverage() test_postfinance_confirm_is_atomic_audited_and_file_idempotent() test_postfinance_reconciliation_matches_quantities_and_total_but_performance_has_no_prebaseline_history() test_postfinance_invalid_or_ambiguous_batch_cannot_partially_confirm() test_postfinance_api_contract_accepts_file_payload_without_exposing_it() test_same_name_with_wrong_asset_class_or_currency_is_never_silently_reused() test_portfolio_read_does_not_refresh_accounts_updated_at() test_confirmed_identity_mapping_supersedes_only_exact_unlineaged_placeholder() test_owner_confirmed_wrong_account_resolution_is_exact_audited_and_idempotent() test_mapping_batch_primitives_are_atomic_and_isin_unique_case_insensitively()