from __future__ import annotations import hashlib import json from dataclasses import dataclass from datetime import date, datetime, timezone from decimal import Decimal from sqlite3 import Connection from typing import Any from jarvis_finance.audit.log import record_audit_event from jarvis_finance.imports.common import stable_id from jarvis_finance.services.modelled_wealth import ( build_modelled_wealth_development, effective_cash_evidence, ) ZERO = Decimal("0") SOURCE_KIND = "dated_manual_screenshot" SOURCE_DB = "manual_screenshot_snapshot" PRIVATE_SUFFIX = "5632" SAVINGS_SUFFIX = "5031" def _safe_label(account_name: str) -> str: compact = account_name.replace(" ", "") for suffix in (PRIVATE_SUFFIX, SAVINGS_SUFFIX): if compact.endswith(suffix): return f"Bankkonto ••••{suffix}" return "Bankkonto" @dataclass(frozen=True) class _Target: role: str account_id: str | None label: str asset_kind: str previous: Decimal | None previous_status: str new_value: Decimal platform_id: str def _now() -> datetime: return datetime.now(timezone.utc) def _fmt(value: Decimal) -> str: return str(value.quantize(Decimal("0.01"))) def _decimal(value: object) -> Decimal: try: return Decimal(str(value or "0")) except Exception: return ZERO def _hash(payload: object) -> str: return hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") ).hexdigest() def _platform_id(conn: Connection) -> str: rows = conn.execute( "SELECT platform_id FROM platforms WHERE lower(name) LIKE '%raiffeisen%' ORDER BY platform_id" ).fetchall() if len(rows) != 1: raise ValueError("raiffeisen_platform_not_uniquely_mapped") return str(rows[0]["platform_id"]) def _cash_account_for_suffix(conn: Connection, platform_id: str, suffix: str) -> Any: rows = conn.execute( """SELECT account_id,account_name FROM accounts WHERE platform_id=? AND account_type='cash' AND is_active=1 ORDER BY account_id""", (platform_id,), ).fetchall() matched = [row for row in rows if str(row["account_name"] or "").replace(" ", "").endswith(suffix)] if len(matched) != 1: raise ValueError(f"raiffeisen_cash_target_{suffix}_not_uniquely_mapped") return matched[0] def _membership_account(conn: Connection, platform_id: str) -> Any | None: rows = conn.execute( """SELECT account_id,account_name FROM accounts WHERE platform_id=? AND is_active=1 AND (account_type IN ('other_asset','membership') OR portfolio_bucket='other') AND lower(account_name) LIKE '%genossenschaft%' ORDER BY account_id""", (platform_id,), ).fetchall() if len(rows) > 1: raise ValueError("raiffeisen_membership_target_not_uniquely_mapped") return rows[0] if rows else None def _latest_cash_value(conn: Connection, account_id: str, as_of: str) -> Decimal | None: evidence = effective_cash_evidence(conn, account_id=account_id, as_of=as_of) value = evidence.get("value_chf") return _decimal(value) if value is not None else None def _latest_asset_value(conn: Connection, account_id: str, as_of: str) -> Decimal | None: row = conn.execute( """SELECT total_value_chf FROM account_value_snapshots WHERE account_id=? AND valuation_date<=? AND is_active=1 ORDER BY valuation_date DESC,created_at DESC,snapshot_id DESC LIMIT 1""", (account_id, as_of), ).fetchone() return _decimal(row["total_value_chf"]) if row else None def _input_values(payload: dict[str, Any]) -> dict[str, Decimal]: return { "private": _decimal(payload["private_account_value_chf"]), "savings": _decimal(payload["savings_account_value_chf"]), "membership": _decimal(payload["membership_value_chf"]), } def _targets(conn: Connection, payload: dict[str, Any]) -> list[_Target]: snapshot_date = str(payload["snapshot_date"]) platform_id = _platform_id(conn) private = _cash_account_for_suffix(conn, platform_id, PRIVATE_SUFFIX) savings = _cash_account_for_suffix(conn, platform_id, SAVINGS_SUFFIX) membership = _membership_account(conn, platform_id) values = _input_values(payload) targets = [ _Target( role="private", account_id=str(private["account_id"]), label=_safe_label(str(private["account_name"])), asset_kind="bank_cash", previous=_latest_cash_value(conn, str(private["account_id"]), snapshot_date), previous_status="confirmed" if _latest_cash_value(conn, str(private["account_id"]), snapshot_date) is not None else "unknown", new_value=values["private"], platform_id=platform_id, ), _Target( role="savings", account_id=str(savings["account_id"]), label=_safe_label(str(savings["account_name"])), asset_kind="bank_cash", previous=_latest_cash_value(conn, str(savings["account_id"]), snapshot_date), previous_status="confirmed" if _latest_cash_value(conn, str(savings["account_id"]), snapshot_date) is not None else "unknown", new_value=values["savings"], platform_id=platform_id, ), _Target( role="membership", account_id=str(membership["account_id"]) if membership else None, label="Raiffeisen Genossenschaftsanteil", asset_kind="membership_asset", previous=_latest_asset_value(conn, str(membership["account_id"]), snapshot_date) if membership else None, previous_status=("confirmed" if membership and _latest_asset_value(conn, str(membership["account_id"]), snapshot_date) is not None else "unknown" if membership else "not_created"), new_value=values["membership"], platform_id=platform_id, ), ] if len({target.account_id for target in targets if target.account_id}) != len([target for target in targets if target.account_id]): raise ValueError("raiffeisen_targets_not_distinct") return targets def _baseline( payload: dict[str, Any], targets: list[_Target], *, wealth_projection: dict[str, Any], ) -> str: details: list[dict[str, object]] = [] snapshot_date = str(payload["snapshot_date"]) for target in targets: details.append( { "role": target.role, "account_id": target.account_id, "previous": _fmt(target.previous) if target.previous is not None else None, "previous_status": target.previous_status, "new_value": _fmt(target.new_value), } ) return _hash( { "snapshot_date": snapshot_date, "targets": details, "wealth_projection": wealth_projection, } ) def _wealth_model(conn: Connection, as_of: str) -> dict[str, Any]: return build_modelled_wealth_development(conn, as_of=as_of, period="all") def _preview_payload(conn: Connection, payload: dict[str, Any]) -> tuple[dict[str, Any], list[_Target]]: snapshot_date = str(payload["snapshot_date"]) parsed_date = date.fromisoformat(snapshot_date) if parsed_date > _now().date(): raise ValueError("manual_snapshot_date_in_future") targets = _targets(conn, payload) model = _wealth_model(conn, snapshot_date) baseline_point = model.get("anchor") or model.get("current") known_before = _decimal(baseline_point.get("value_chf")) if isinstance(baseline_point, dict) else ZERO bank_before = next( ( _decimal(component.get("current_value_chf")) for component in model.get("components") or [] if component.get("key") == "bank_cash" ), ZERO, ) input_fingerprint = _baseline( payload, targets, wealth_projection={ "anchor": model.get("anchor"), "current": model.get("current"), "components": model.get("components"), "correction_markers": model.get("correction_markers"), "known_before_chf": _fmt(known_before), "bank_before_chf": _fmt(bank_before), }, ) delta = sum((target.new_value - (target.previous or ZERO) for target in targets), ZERO) cash_delta = sum( ( target.new_value - (target.previous or ZERO) for target in targets if target.asset_kind == "bank_cash" ), ZERO, ) bank_after = bank_before + cash_delta membership_after = sum((target.new_value for target in targets if target.asset_kind == "membership_asset"), ZERO) token = _hash( { "contract": "raiffeisen_manual_snapshot_preview_v1", "input_fingerprint": input_fingerprint, } )[:32] preview = { "preview_id": f"raiffeisen-preview-{token}", "confirmation_id": f"raiffeisen-confirm-{token}", "input_fingerprint": input_fingerprint, "source_kind": SOURCE_KIND, "snapshot_date": snapshot_date, "affected_accounts": [ { "account_label": target.label, "asset_kind": target.asset_kind, "previous_value_chf": _fmt(target.previous) if target.previous is not None else None, "new_value_chf": _fmt(target.new_value), "change_chf": _fmt(target.new_value - (target.previous or ZERO)), "previous_status": target.previous_status, } for target in targets ], "bank_cash_after_chf": _fmt(bank_after), "separate_membership_asset_after_chf": _fmt(membership_after), "known_wealth_before_chf": _fmt(known_before), "expected_known_wealth_after_chf": _fmt(known_before + delta), "expected_total_wealth_change_chf": _fmt(delta), "creates_transactions": False, "append_only": True, } return preview, targets def preview_raiffeisen_manual_snapshot(conn: Connection, **payload: Any) -> dict[str, Any]: """Pure preview over stored canonical baselines; this function never writes.""" preview, _ = _preview_payload(conn, payload) return preview def _existing_confirmation( conn: Connection, confirmation_id: str, input_fingerprint: str, payload_hash: str, ) -> dict[str, Any] | None: row = conn.execute( "SELECT * FROM manual_snapshot_confirmations WHERE confirmation_id=?", (confirmation_id,), ).fetchone() if not row: return None if ( str(row["input_fingerprint"]) != input_fingerprint or str(row["payload_hash"]) != payload_hash ): raise ValueError("confirmation_id_reused_with_different_input") return { "status": "already_applied", "confirmation_id": confirmation_id, "snapshot_date": str(row["snapshot_date"]), "created_snapshot_count": int(row["created_snapshot_count"]), "created_transaction_count": 0, "bank_cash_after_chf": str(row["bank_cash_after_chf"]), "separate_membership_asset_after_chf": str(row["separate_membership_asset_after_chf"]), "known_wealth_after_chf": str(row["known_wealth_after_chf"]), "audit_recorded": True, } def _confirm_raiffeisen_manual_snapshot_locked( conn: Connection, *, preview_id: str, confirmation_id: str, input_fingerprint: str, snapshot_date: date, private_account_value_chf: Decimal, savings_account_value_chf: Decimal, membership_value_chf: Decimal, ) -> dict[str, Any]: payload = { "snapshot_date": snapshot_date.isoformat(), "private_account_value_chf": private_account_value_chf, "savings_account_value_chf": savings_account_value_chf, "membership_value_chf": membership_value_chf, } payload_hash = _hash(payload) existing = _existing_confirmation( conn, confirmation_id, input_fingerprint, payload_hash, ) if existing: return existing preview, targets = _preview_payload(conn, payload) if preview["input_fingerprint"] != input_fingerprint: raise ValueError("manual_snapshot_baseline_changed") if ( preview_id != preview["preview_id"] or confirmation_id != preview["confirmation_id"] ): raise ValueError("manual_snapshot_confirmation_token_mismatch") now = _now().isoformat() membership = next(target for target in targets if target.asset_kind == "membership_asset") membership_account_id = membership.account_id or stable_id("account", "raiffeisen", "membership-share") created = 0 with conn: if membership.account_id is None: conn.execute( """INSERT INTO accounts( account_id,platform_id,account_name,account_type,currency,performance_included, is_active,notes,created_at,updated_at,balance_mode,portfolio_bucket ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""", ( membership_account_id, membership.platform_id, "Raiffeisen Genossenschaftsanteil", "other_asset", "CHF", 0, 1, "Separates Mitgliedschaftsvermögen; kein frei verfügbares Bankguthaben.", now, now, "snapshot", "other", ), ) audit_id = record_audit_event( conn, source=SOURCE_DB, action="confirm_manual_source_snapshot", entity_type="manual_source_snapshot", entity_id=confirmation_id, old_values={"input_fingerprint": input_fingerprint}, new_values={ "snapshot_date": snapshot_date.isoformat(), "source_kind": SOURCE_KIND, "snapshot_count": 3, "transaction_count": 0, }, created_by="user", ) for target in targets: account_id = membership_account_id if target.asset_kind == "membership_asset" else str(target.account_id) if target.asset_kind == "bank_cash": snapshot_id = stable_id("cash-snapshot", confirmation_id, target.role) conn.execute( """INSERT INTO cash_account_snapshots( snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency, amount_chf,source,note,created_at,created_by,audit_id,semantic_identity ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( snapshot_id, account_id, "manual_balance", snapshot_date.isoformat(), _fmt(target.new_value), "CHF", _fmt(target.new_value), SOURCE_DB, "Datierter manueller Quellensnapshot; keine Transaktionsrekonstruktion.", now, "user", audit_id, stable_id("manual-source-snapshot", confirmation_id, target.role), ), ) else: snapshot_id = stable_id("account-value-snapshot", confirmation_id, target.role) conn.execute( """INSERT INTO account_value_snapshots( snapshot_id,account_id,valuation_date,total_value_chf,currency,source_type, quality_status,notes,created_at,updated_at,valuation_at,source_reference,is_active ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,1)""", ( snapshot_id, account_id, snapshot_date.isoformat(), _fmt(target.new_value), "CHF", SOURCE_DB, "confirmed", "Separates Mitgliedschaftsvermögen; kein frei verfügbares Bankguthaben.", now, None, snapshot_date.isoformat(), confirmation_id, ), ) created += 1 known_after = _decimal(preview["expected_known_wealth_after_chf"]) bank_after = _decimal(preview["bank_cash_after_chf"]) conn.execute( """INSERT INTO manual_snapshot_confirmations( confirmation_id,preview_id,input_fingerprint,payload_hash,snapshot_date,source_kind, known_wealth_after_chf,bank_cash_after_chf,separate_membership_asset_after_chf, created_snapshot_count,created_at,audit_id ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""", ( confirmation_id, preview_id, input_fingerprint, payload_hash, snapshot_date.isoformat(), SOURCE_KIND, _fmt(known_after), _fmt(bank_after), _fmt(membership_value_chf), created, now, audit_id, ), ) return { "status": "confirmed", "confirmation_id": confirmation_id, "snapshot_date": snapshot_date.isoformat(), "created_snapshot_count": created, "created_transaction_count": 0, "bank_cash_after_chf": _fmt(bank_after), "separate_membership_asset_after_chf": _fmt(membership_value_chf), "known_wealth_after_chf": _fmt(known_after), "audit_recorded": True, } def confirm_raiffeisen_manual_snapshot( conn: Connection, *, preview_id: str, confirmation_id: str, input_fingerprint: str, snapshot_date: date, private_account_value_chf: Decimal, savings_account_value_chf: Decimal, membership_value_chf: Decimal, ) -> dict[str, Any]: """Serialize stale-check, idempotency lookup, version allocation, and writes.""" if conn.in_transaction: raise ValueError("manual_snapshot_requires_clean_transaction") conn.execute("BEGIN IMMEDIATE") try: result = _confirm_raiffeisen_manual_snapshot_locked( conn, preview_id=preview_id, confirmation_id=confirmation_id, input_fingerprint=input_fingerprint, snapshot_date=snapshot_date, private_account_value_chf=private_account_value_chf, savings_account_value_chf=savings_account_value_chf, membership_value_chf=membership_value_chf, ) if conn.in_transaction: conn.commit() return result except Exception: if conn.in_transaction: conn.rollback() raise