#!/usr/bin/env python3 """Sprint-9 PostFinance mapping/baseline activation with deterministic gates.""" from __future__ import annotations import argparse import base64 from datetime import UTC, datetime import json from pathlib import Path from typing import Any from jarvis_finance.services.portfolio_data import ( confirm_ingestion, confirm_postfinance_instrument_mappings, preview_ingestion, ) from jarvis_finance.storage.database import connect from jarvis_finance.storage.migrations import apply_migrations AS_OF = "2026-05-14" MAPPINGS = [ ("ADBE", "Adobe Inc.", "US00724F1012", "stock", "USD", "ADBE", "NASDAQ"), ("ANET", "Arista Networks Inc.", "US0404132054", "stock", "USD", "ANET", "NYSE"), ("APC", "Apple Inc.", "US0378331005", "stock", "EUR", "APC.F", "FSX"), ("APP", "AppLovin Corp.", "US03831W1080", "stock", "USD", "APP", "NASDAQ"), ("AVGO", "Broadcom Inc.", "US11135F1012", "stock", "USD", "AVGO", "NASDAQ"), ("BABA", "Alibaba Group Holding ADR", "US01609W1027", "stock", "USD", "BABA", "NYSE"), ("CRWD", "CrowdStrike Holdings Inc.", "US22788C1053", "stock", "USD", "CRWD", "NASDAQ"), ("GOOG", "Alphabet Inc.", "US02079K1079", "stock", "USD", "GOOG", "NASDAQ"), ("MCD", "McDonald's Corp.", "US5801351017", "stock", "USD", "MCD", "NYSE"), ("MSFT", "Microsoft Corp.", "US5949181045", "stock", "USD", "MSFT", "NASDAQ"), ("MU", "Micron Technology Inc.", "US5951121038", "stock", "USD", "MU", "NASDAQ"), ("NOW", "ServiceNow Inc.", "US81762P1021", "stock", "USD", "NOW", "NYSE"), ("NVDA", "NVIDIA Corp.", "US67066G1040", "stock", "USD", "NVDA", "NASDAQ"), ("OPK", "OPKO Health Inc.", "US68375N1037", "stock", "USD", "OPK", "NASDAQ"), ("PANW", "Palo Alto Networks Inc.", "US6974351057", "stock", "USD", "PANW", "NASDAQ"), ("TPL", "Texas Pacific Land Corp.", "US88262P1021", "stock", "USD", "TPL", "NYSE"), ("TSLA", "Tesla Inc.", "US88160R1014", "stock", "USD", "TSLA", "NASDAQ"), ("UNH", "UnitedHealth Group Inc.", "US91324P1021", "stock", "USD", "UNH", "NYSE"), ("ARKG", "ARK Genomic Revolution ETF", "US00214Q3020", "etf", "USD", "ARKG", "CBOE"), ("BCHE", "Invesco CoinShares Global Blockchain UCITS ETF", "IE00BGBN6P67", "etf", "USD", "BCHN.L", "LSE"), ("IUIT", "iShares S&P 500 Information Technology Sector UCITS ETF", "IE00B3WJKG14", "etf", "USD", "IUIT.L", "LSE"), ("WTAI", "WisdomTree Artificial Intelligence UCITS ETF USD Acc", "IE00BDVPNG13", "etf", "USD", "WTAI.L", "LSE"), ] def decisions() -> list[dict[str, str]]: return [ { "source_symbol": source_symbol, "instrument_name": name, "isin": isin, "asset_class": asset_class, "source_currency": currency, "source_venue": "unknown", "provider": "fmp", "valuation_symbol": valuation_symbol, "valuation_venue": valuation_venue, "valuation_currency": currency, } for source_symbol, name, isin, asset_class, currency, valuation_symbol, valuation_venue in MAPPINGS ] def request(raw: bytes, owner_resolutions: list[dict[str, str]] | None = None) -> dict[str, object]: return { "source_key": "postfinance_etrading", "scope_kind": "portfolio", "account_id": None, "period_from": AS_OF, "period_to": AS_OF, "data_cutoff": datetime.now(UTC).isoformat(), "file_name": "postfinance-portfolio-2026-05-14.docx", "content_base64": base64.b64encode(raw).decode("ascii"), "instrument_mappings": decisions(), "allow_manual_baseline_supersession": True, "owner_confirmed_manual_placeholder_resolutions": owner_resolutions or [], } def preview_gates(preview: dict[str, Any]) -> None: counts = preview["counts"] expected = preview["expected_changes"] reconciliation = preview["cash_reconciliation"] if counts["blocked"] or counts["ambiguous"]: raise RuntimeError(f"Baseline preview blocked: {counts}") if counts["new"] != 47 or expected["records"] != 47: raise RuntimeError(f"Expected exactly 47 new baseline records: {counts}, {expected}") if expected["positions"] != 22 or expected["manual_supersessions"] != 2: raise RuntimeError(f"Expected 22 positions and two supersessions: {expected}") if reconciliation["source_total_identity_chf"] != reconciliation["reported_document_total_chf"]: raise RuntimeError(f"Source aggregates do not reconcile exactly: {reconciliation}") def summary(preview: dict[str, Any]) -> dict[str, object]: return { "preview_id": preview["preview_id"], "source_revision": preview["source_revision"], "counts": preview["counts"], "expected_changes": preview["expected_changes"], "cash_reconciliation": preview["cash_reconciliation"], "quality_impact": preview["quality_impact"], } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--db", required=True) parser.add_argument("--document", required=True) parser.add_argument("--mode", choices=("mappings", "preview", "confirm"), required=True) parser.add_argument( "--owner-resolution", help="Path outside the repository to an owner-confirmed candidate resolution JSON array", ) args = parser.parse_args() raw = Path(args.document).read_bytes() conn = connect(args.db) apply_migrations(conn) owner_resolutions = json.loads(Path(args.owner_resolution).read_text()) if args.owner_resolution else [] if not isinstance(owner_resolutions, list): raise ValueError("Owner resolution must be a JSON array") payload = request(raw, owner_resolutions) if args.mode == "mappings": result = confirm_postfinance_instrument_mappings(conn, payload) if result["mapping_count"] != 22 or result["instrument_count"] != 22: raise RuntimeError(f"Mapping gate failed: {result}") else: mapped = conn.execute( """SELECT COUNT(DISTINCT i.isin) FROM instrument_price_mappings m JOIN instruments i ON i.instrument_id=m.instrument_id WHERE m.provider='fmp' AND m.mapping_status='mapped' AND i.isin IN ({})""".format(",".join("?" for _ in MAPPINGS)), tuple(value[2] for value in MAPPINGS), ).fetchone()[0] if mapped != 22: raise RuntimeError(f"Expected 22 approved mappings before baseline preview, found {mapped}") preview = preview_ingestion(conn, payload) if args.mode == "mappings": print(json.dumps({"mappings": 22, "preview": summary(preview)}, indent=2, sort_keys=True)) return 0 preview_gates(preview) if args.mode == "preview": print(json.dumps(summary(preview), indent=2, sort_keys=True)) return 0 confirmation = { **payload, **{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) replay = confirm_ingestion(conn, confirmation) if not replay.get("idempotent") or replay.get("batch_id") != result.get("batch_id"): raise RuntimeError(f"Immediate confirmation replay was not idempotent: {result}, {replay}") output = {"preview": summary(preview), "confirm": result, "confirm_replay": replay} print(json.dumps(output, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())