from __future__ import annotations from datetime import UTC, datetime from decimal import Decimal, InvalidOperation import hashlib import json import re from sqlite3 import Connection, IntegrityError from typing import Any from fastapi import HTTPException from jarvis_finance.api.schemas.crypto_reconciliation import ( CryptoSnapshotConfirmRequest, CryptoSnapshotPreviewRequest, CryptoTransferPairConfirmRequest, CryptoTransferPairPreviewRequest, ) from jarvis_finance.crypto.current_balances import current_crypto_balance_basis ANCHOR_DATE = "2025-12-31" ACTIVITY_START = "2026-01-01T00:00:00Z" def _now() -> str: return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") def _parse_timestamp(value: str) -> datetime: text = str(value).strip().replace("Z", "+00:00") try: parsed = datetime.fromisoformat(text) except ValueError as exc: raise HTTPException(status_code=422, detail="observed_at must be an ISO timestamp") from exc if parsed.tzinfo is None: raise HTTPException(status_code=422, detail="observed_at must include a timezone") return parsed.astimezone(UTC) def _decimal(value: object, field: str) -> Decimal: try: parsed = Decimal(str(value)) except (InvalidOperation, ValueError) as exc: raise HTTPException(status_code=422, detail=f"{field} must be a decimal string") from exc if not parsed.is_finite() or parsed < 0: raise HTTPException(status_code=422, detail=f"{field} must be finite and non-negative") return parsed def _text(value: Decimal) -> str: return format(value, "f") def _id(prefix: str, *parts: object) -> str: raw = "|".join(str(part) for part in parts) return f"{prefix}_{hashlib.sha256(raw.encode()).hexdigest()[:24]}" def _redact_note(value: str) -> str: note = str(value or "").strip() note = re.sub(r"\b(?:0x)?[A-Fa-f0-9]{24,}\b", "[REDACTED]", note) note = re.sub(r"\b[a-zA-Z0-9]{40,}\b", "[REDACTED]", note) return note[:500] def _canonical_snapshot(request: CryptoSnapshotPreviewRequest) -> dict[str, Any]: observed_at = _parse_timestamp(request.observed_at).isoformat().replace("+00:00", "Z") wallets: list[dict[str, Any]] = [] seen_wallets: set[str] = set() seen_pairs: set[tuple[str, str]] = set() for wallet in sorted(request.wallets, key=lambda item: item.wallet_id): wallet_id = wallet.wallet_id.strip() if not wallet_id or wallet_id in seen_wallets: raise HTTPException(status_code=422, detail="wallets must contain unique canonical wallet IDs") evidence = wallet.evidence_source.strip() if not evidence: raise HTTPException(status_code=422, detail=f"evidence_source is required for {wallet_id}") seen_wallets.add(wallet_id) items: list[dict[str, str]] = [] for item in sorted(wallet.items, key=lambda row: row.asset_id): asset_id = item.asset_id.strip() pair = (wallet_id, asset_id) if not asset_id or pair in seen_pairs: raise HTTPException(status_code=422, detail="wallet/asset rows must be unique") seen_pairs.add(pair) items.append({"asset_id": asset_id, "quantity": _text(_decimal(item.quantity, "quantity"))}) wallets.append( { "wallet_id": wallet_id, "evidence_source": evidence[:120], "note": _redact_note(wallet.note), "items": items, } ) if not wallets: raise HTTPException(status_code=422, detail="at least one wallet observation is required") return {"observed_at": observed_at, "wallets": wallets} def _fingerprint(payload: dict[str, Any]) -> str: encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() return hashlib.sha256(encoded).hexdigest() def _known_ids(conn: Connection) -> tuple[set[str], set[str]]: wallets = {str(row[0]) for row in conn.execute("SELECT wallet_id FROM crypto_wallets WHERE is_active=1")} assets = {str(row[0]) for row in conn.execute("SELECT asset_id FROM crypto_assets WHERE is_active=1")} return wallets, assets def _baseline_quantities(conn: Connection) -> dict[tuple[str, str], Decimal]: return { (str(row["wallet_id"]), str(row["asset_id"])): Decimal(str(row["quantity"])) for row in conn.execute("SELECT wallet_id, asset_id, quantity FROM crypto_holdings") } def _activity_deltas(conn: Connection, observed_at: str) -> dict[tuple[str, str], Decimal]: deltas: dict[tuple[str, str], Decimal] = {} rows = conn.execute( """SELECT asset_id, quantity, fee_quantity, from_wallet_id, to_wallet_id FROM crypto_transactions WHERE confirmation_status='confirmed' AND transaction_datetime>=? AND transaction_datetime<=?""", (ACTIVITY_START, observed_at), ).fetchall() for row in rows: asset_id = str(row["asset_id"]) quantity = Decimal(str(row["quantity"] or "0")) fee = Decimal(str(row["fee_quantity"] or "0")) if row["from_wallet_id"]: key = (str(row["from_wallet_id"]), asset_id) deltas[key] = deltas.get(key, Decimal("0")) - quantity - fee if row["to_wallet_id"]: key = (str(row["to_wallet_id"]), asset_id) deltas[key] = deltas.get(key, Decimal("0")) + quantity return deltas def _snapshot_preview(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: known_wallets, known_assets = _known_ids(conn) submitted_wallets = {row["wallet_id"] for row in payload["wallets"]} unknown_wallets = sorted(submitted_wallets - known_wallets) if unknown_wallets: raise HTTPException(status_code=404, detail="unknown or inactive wallet ID") submitted: dict[tuple[str, str], Decimal] = {} for wallet in payload["wallets"]: for item in wallet["items"]: if item["asset_id"] not in known_assets: raise HTTPException(status_code=404, detail="unknown or inactive asset ID") submitted[(wallet["wallet_id"], item["asset_id"])] = Decimal(item["quantity"]) baseline = _baseline_quantities(conn) explained = _activity_deltas(conn, payload["observed_at"]) required_pairs = set(baseline) complete = submitted_wallets == known_wallets and required_pairs.issubset(submitted) differences = [] for key in sorted(set(baseline) | set(submitted) | set(explained)): before = baseline.get(key, Decimal("0")) observed = submitted.get(key) if observed is None: continue delta = observed - before explained_delta = explained.get(key, Decimal("0")) unexplained = delta - explained_delta asset = conn.execute("SELECT symbol, coin_name FROM crypto_assets WHERE asset_id=?", (key[1],)).fetchone() wallet = conn.execute("SELECT wallet_name FROM crypto_wallets WHERE wallet_id=?", (key[0],)).fetchone() differences.append( { "wallet_id": key[0], "wallet": wallet["wallet_name"], "asset_id": key[1], "asset": asset["symbol"], "asset_name": asset["coin_name"], "anchor_quantity": _text(before), "observed_quantity": _text(observed), "difference": _text(delta), "explained_difference": _text(explained_delta), "unexplained_difference": _text(unexplained), "status": "reconciled" if unexplained == 0 else "unexplained_balance_change", "possible_transfer_candidate": unexplained != 0, } ) fingerprint = _fingerprint(payload) missing_wallets = sorted(known_wallets - submitted_wallets) missing_pairs = sorted(required_pairs - set(submitted)) return { "preview_id": f"crypto_snapshot_preview_{fingerprint[:24]}", "input_fingerprint": fingerprint, "status": "complete" if complete else "partial", "observed_at": payload["observed_at"], "wallet_count": len(submitted_wallets), "item_count": len(submitted), "missing_wallet_ids": missing_wallets, "missing_wallet_asset_rows": [{"wallet_id": wallet, "asset_id": asset} for wallet, asset in missing_pairs], "differences": differences, "warnings": [ "Ungeklärte Differenzen bleiben unexplained_balance_change und erzeugen keine Aktivität oder Cashflow.", "Nur ein vollständiger Snapshot wird für die aktuelle Bewertung verwendet.", ], } def preview_crypto_snapshot(conn: Connection, request: CryptoSnapshotPreviewRequest) -> dict[str, Any]: return _snapshot_preview(conn, _canonical_snapshot(request)) def _insert_audit(conn: Connection, *, audit_id: str, action: str, entity_id: str, payload: dict[str, Any], note: str) -> None: timestamp = _now() conn.execute( """INSERT INTO audit_log( audit_id,timestamp,source,action,entity_type,entity_id,new_values_json, user_text_note,confirmed,confirmation_timestamp,created_by,quality_status,created_at ) VALUES (?,?, 'vue_dashboard',?,'crypto_balance_snapshot',?,?,?,1,?,'user','review_required',?)""", (audit_id, timestamp, action, entity_id, json.dumps(payload, sort_keys=True), note, timestamp, timestamp), ) def confirm_crypto_snapshot(conn: Connection, request: CryptoSnapshotConfirmRequest) -> dict[str, Any]: if not request.confirm: raise HTTPException(status_code=422, detail="confirm=true required") payload = _canonical_snapshot(request) preview = _snapshot_preview(conn, payload) if request.preview_id != preview["preview_id"]: raise HTTPException(status_code=409, detail="preview is stale; create a new preview") confirmation_key = request.confirmation_key.strip() if not confirmation_key: raise HTTPException(status_code=422, detail="confirmation_key is required") existing = conn.execute( "SELECT snapshot_id,input_fingerprint,audit_id,status,observed_at FROM crypto_balance_snapshots WHERE confirmation_key=?", (confirmation_key,), ).fetchone() if existing: if existing["input_fingerprint"] != preview["input_fingerprint"]: raise HTTPException(status_code=409, detail="confirmation_key already used for another payload") return { "status": "already_confirmed", "snapshot_id": existing["snapshot_id"], "audit_id": existing["audit_id"], "snapshot_status": existing["status"], "observed_at": existing["observed_at"], "idempotent_replay": True, } snapshot_id = _id("cryptosnap", confirmation_key, preview["input_fingerprint"]) audit_id = _id("audit", snapshot_id) now = _now() notes = "; ".join(row["note"] for row in payload["wallets"] if row["note"]) try: with conn: _insert_audit(conn, audit_id=audit_id, action="crypto_balance_snapshot_confirm", entity_id=snapshot_id, payload={"snapshot_id": snapshot_id, **preview}, note=notes) conn.execute( """INSERT INTO crypto_balance_snapshots( snapshot_id,observed_at,status,confirmation_key,input_fingerprint, wallet_count,item_count,audit_id,created_at) VALUES (?,?,?,?,?,?,?,?,?)""", (snapshot_id, payload["observed_at"], preview["status"], confirmation_key, preview["input_fingerprint"], preview["wallet_count"], preview["item_count"], audit_id, now), ) for wallet in payload["wallets"]: conn.execute( """INSERT INTO crypto_balance_snapshot_wallets( snapshot_wallet_id,snapshot_id,wallet_id,evidence_source,redacted_note,created_at) VALUES (?,?,?,?,?,?)""", (_id("cryptosnapwallet", snapshot_id, wallet["wallet_id"]), snapshot_id, wallet["wallet_id"], wallet["evidence_source"], wallet["note"], now), ) for item in wallet["items"]: conn.execute( """INSERT INTO crypto_balance_snapshot_items( snapshot_item_id,snapshot_id,wallet_id,asset_id,quantity,created_at) VALUES (?,?,?,?,?,?)""", (_id("cryptosnapitem", snapshot_id, wallet["wallet_id"], item["asset_id"]), snapshot_id, wallet["wallet_id"], item["asset_id"], item["quantity"], now), ) except IntegrityError as exc: raise HTTPException(status_code=409, detail="snapshot confirm conflict") from exc return { "status": "confirmed", "snapshot_id": snapshot_id, "audit_id": audit_id, "snapshot_status": preview["status"], "observed_at": payload["observed_at"], "idempotent_replay": False, } def _activity_summary(conn: Connection, wallet_id: str) -> dict[str, Any]: row = conn.execute( """SELECT COUNT(*) AS count, MIN(transaction_datetime) AS first_at, MAX(transaction_datetime) AS last_at FROM crypto_transactions WHERE transaction_datetime>=? AND (from_wallet_id=? OR to_wallet_id=?)""", (ACTIVITY_START, wallet_id, wallet_id), ).fetchone() count = int(row["count"] or 0) return { "count": count, "first_at": row["first_at"], "last_at": row["last_at"], "coverage": "missing" if count == 0 else "partial", } _EXPORT_ACTIONS = { "binance": "Spot/Convert/Earn sowie Deposit-/Withdrawal-Historie ab 01.01.2026 und aktuellen Balance-Export bereitstellen.", "coinbase": "Transaction History ab 01.01.2026 und aktuellen Balance-Export bereitstellen.", "kucoin": "Trade/Convert/Earn sowie Deposit-/Withdrawal-Historie ab 01.01.2026 und aktuellen Balance-Export bereitstellen.", "swissborg": "Account Statement/Transactions ab 01.01.2026 und aktuellen Portfolio-Snapshot bereitstellen.", "metamask": "Transaktions-, Token- und Internal-Tx-Nachweise je verwendeter EVM-Chain ab 01.01.2026 sowie aktuelle Bestände bestätigen.", "nami": "Cardano-Aktivitäten inklusive Rewards ab 01.01.2026 exportieren und aktuellen Bestand bestätigen.", "yoroi": "Cardano-Aktivitäten inklusive Rewards ab 01.01.2026 exportieren und aktuellen Bestand bestätigen.", "yoloi": "Cardano-Aktivitäten inklusive Rewards ab 01.01.2026 exportieren und aktuellen Bestand bestätigen.", "firefly": "Firefly-Aktivitätshistorie ab 01.01.2026 exportieren und aktuellen IOTA-Bestand bestätigen.", } def _next_action(name: str) -> str: lowered = name.casefold() for token, action in _EXPORT_ACTIONS.items(): if token in lowered: return action return "Aktivitäten ab 01.01.2026 exportieren und aktuellen Bestand je Asset bestätigen." def _current_valuation(conn: Connection) -> dict[str, Any]: basis = current_crypto_balance_basis(conn) totals: dict[str, Decimal] = {} for (_wallet_id, asset_id), quantity in basis.quantities.items(): totals[asset_id] = totals.get(asset_id, Decimal("0")) + quantity total = Decimal("0") missing: list[str] = [] timestamps: list[str] = [] for asset_id, quantity in totals.items(): if quantity == 0: continue price = conn.execute( """SELECT price,COALESCE(provider_timestamp,fetched_at) AS price_at FROM crypto_prices WHERE asset_id=? AND price_currency='CHF' ORDER BY COALESCE(provider_timestamp,fetched_at) DESC LIMIT 1""", (asset_id,), ).fetchone() if not price: missing.append(asset_id) continue total += quantity * Decimal(str(price["price"])) if price["price_at"]: timestamps.append(str(price["price_at"])) return { "value_chf": _text(total) if not missing else None, "valued_partial_chf": _text(total), "price_as_of": min(timestamps) if timestamps else None, "balance_as_of": basis.balance_as_of, "snapshot_id": basis.snapshot_id, "balance_confirmed_current": basis.confirmed_current, "missing_price_asset_ids": sorted(missing), "status_message": ( f"Preis aktuell, Bestandsmengen per {str(basis.balance_as_of)[:10]} – aktueller Portfoliowert nicht bestätigt." if not basis.confirmed_current else "Aktueller Schätzwert auf Basis des neuesten vollständig bestätigten Bestandssnapshots." ), } def get_crypto_reconciliation(conn: Connection) -> dict[str, Any]: latest = conn.execute( "SELECT * FROM crypto_balance_snapshots ORDER BY observed_at DESC, created_at DESC LIMIT 1" ).fetchone() latest_wallets: set[str] = set() latest_items: dict[tuple[str, str], Decimal] = {} if latest: latest_wallets = {str(row[0]) for row in conn.execute("SELECT wallet_id FROM crypto_balance_snapshot_wallets WHERE snapshot_id=?", (latest["snapshot_id"],))} latest_items = { (str(row["wallet_id"]), str(row["asset_id"])): Decimal(str(row["quantity"])) for row in conn.execute("SELECT wallet_id,asset_id,quantity FROM crypto_balance_snapshot_items WHERE snapshot_id=?", (latest["snapshot_id"],)) } baseline = _baseline_quantities(conn) explained = _activity_deltas(conn, latest["observed_at"] if latest else _now()) unpaired_by_wallet: dict[str, int] = {} unpaired_rows = conn.execute( """SELECT t.crypto_transaction_id,t.transaction_type,t.asset_id,t.quantity,t.fee_quantity, t.from_wallet_id,t.to_wallet_id,t.transaction_datetime,t.tx_hash FROM crypto_transactions t LEFT JOIN crypto_internal_transfer_pairs p ON p.withdrawal_transaction_id=t.crypto_transaction_id OR p.deposit_transaction_id=t.crypto_transaction_id WHERE p.transfer_pair_id IS NULL AND lower(t.transaction_type) IN ('deposit','withdrawal','einzahlung','auszahlung')""" ).fetchall() for row in unpaired_rows: wallet = row["from_wallet_id"] or row["to_wallet_id"] if wallet: unpaired_by_wallet[str(wallet)] = unpaired_by_wallet.get(str(wallet), 0) + 1 matrix = [] for wallet in conn.execute("SELECT wallet_id,wallet_name,wallet_type,platform_provider FROM crypto_wallets WHERE is_active=1 ORDER BY wallet_name"): wallet_id = str(wallet["wallet_id"]) activity = _activity_summary(conn, wallet_id) asset_count = len({asset for candidate_wallet, asset in baseline if candidate_wallet == wallet_id}) observed = wallet_id in latest_wallets unexplained_count = 0 if latest: keys = {key for key in set(baseline) | set(latest_items) if key[0] == wallet_id} unexplained_count = sum( 1 for key in keys if latest_items.get(key, Decimal("0")) - baseline.get(key, Decimal("0")) - explained.get(key, Decimal("0")) != 0 ) status = "missing" if observed or activity["count"]: status = "review_required" if unexplained_count or unpaired_by_wallet.get(wallet_id, 0) else "partial" matrix.append( { "wallet_id": wallet_id, "name": wallet["wallet_name"], "type": "Exchange" if "exchange" in str(wallet["wallet_type"] or "").casefold() else "Self-Custody", "last_confirmed_balance_at": latest["observed_at"] if observed else ANCHOR_DATE if asset_count else None, "asset_count": asset_count, "activity_coverage": activity["coverage"], "activity_first_at": activity["first_at"], "activity_last_at": activity["last_at"], "current_balance_evidence": observed, "unexplained_activities": unexplained_count, "unpaired_transfers": unpaired_by_wallet.get(wallet_id, 0), "status": status, "next_action": _next_action(str(wallet["wallet_name"])), } ) difference_rows = [] if latest: for key in sorted(set(baseline) | set(latest_items)): before = baseline.get(key, Decimal("0")) observed = latest_items.get(key, Decimal("0")) explained_delta = explained.get(key, Decimal("0")) unexplained = observed - before - explained_delta asset = conn.execute("SELECT symbol FROM crypto_assets WHERE asset_id=?", (key[1],)).fetchone() wallet = conn.execute("SELECT wallet_name FROM crypto_wallets WHERE wallet_id=?", (key[0],)).fetchone() difference_rows.append({"wallet_id": key[0], "wallet": wallet["wallet_name"], "asset_id": key[1], "asset": asset["symbol"], "anchor_quantity": _text(before), "observed_quantity": _text(observed), "difference": _text(observed-before), "explained_difference": _text(explained_delta), "unexplained_difference": _text(unexplained), "status": "reconciled" if unexplained == 0 else "unexplained_balance_change"}) transfer_candidates = [] for row in unpaired_rows: wallet_id = str(row["from_wallet_id"] or row["to_wallet_id"] or "") wallet = conn.execute("SELECT wallet_name FROM crypto_wallets WHERE wallet_id=?", (wallet_id,)).fetchone() asset = conn.execute("SELECT symbol FROM crypto_assets WHERE asset_id=?", (row["asset_id"],)).fetchone() transfer_candidates.append({"transaction_id": row["crypto_transaction_id"], "direction": "withdrawal" if row["from_wallet_id"] else "deposit", "wallet": wallet["wallet_name"] if wallet else "Unbekannt", "asset_id": row["asset_id"], "asset": asset["symbol"] if asset else row["asset_id"], "quantity": str(row["quantity"]), "fee_quantity": str(row["fee_quantity"] or "0"), "timestamp": row["transaction_datetime"], "has_event_reference": bool(row["tx_hash"])}) paired = int(conn.execute("SELECT COUNT(*) FROM crypto_internal_transfer_pairs").fetchone()[0]) entry_template = [] for wallet_row in conn.execute("SELECT wallet_id,wallet_name FROM crypto_wallets WHERE is_active=1 ORDER BY wallet_name"): wallet_id = str(wallet_row["wallet_id"]) items = [] for (candidate_wallet, asset_id), quantity in sorted(baseline.items()): if candidate_wallet != wallet_id: continue asset_row = conn.execute("SELECT symbol,coin_name FROM crypto_assets WHERE asset_id=?", (asset_id,)).fetchone() items.append({"asset_id": asset_id, "asset": asset_row["symbol"], "asset_name": asset_row["coin_name"], "quantity": _text(quantity)}) entry_template.append({"wallet_id": wallet_id, "name": wallet_row["wallet_name"], "evidence_source": "", "note": "", "items": items}) all_activity_complete = bool(matrix) and all(row["activity_coverage"] == "complete" for row in matrix) no_unexplained = bool(latest and latest["status"] == "complete") and all(row["status"] == "reconciled" for row in difference_rows) performance_open = all_activity_complete and no_unexplained and not unpaired_rows return { "anchor_date": ANCHOR_DATE, "latest_snapshot": dict(latest) if latest else None, "coverage": matrix, "entry_template": entry_template, "differences": difference_rows, "transfers": {"paired": paired, "unpaired": len(unpaired_rows), "ambiguous": 0, "candidates": transfer_candidates}, "valuation": _current_valuation(conn), "performance_gate": { "status": "open" if performance_open else "closed", "reason": None if performance_open else "Aktivitäten und Cashflows sind nicht vollständig reconciliiert; XIRR, TTWROR und Anlageergebnis bleiben nicht beurteilbar.", }, "render_provider_calls": False, } def _transfer_pair_preview(conn: Connection, request: CryptoTransferPairPreviewRequest) -> dict[str, Any]: if request.withdrawal_transaction_id == request.deposit_transaction_id: raise HTTPException(status_code=422, detail="two distinct transfer legs are required") rows = conn.execute( "SELECT * FROM crypto_transactions WHERE crypto_transaction_id IN (?,?)", (request.withdrawal_transaction_id, request.deposit_transaction_id), ).fetchall() by_id = {str(row["crypto_transaction_id"]): row for row in rows} if len(by_id) != 2: raise HTTPException(status_code=404, detail="transfer leg not found") withdrawal = by_id[request.withdrawal_transaction_id] deposit = by_id[request.deposit_transaction_id] if not withdrawal["from_wallet_id"] or withdrawal["to_wallet_id"] or not deposit["to_wallet_id"] or deposit["from_wallet_id"]: raise HTTPException(status_code=422, detail="legs must be one withdrawal and one deposit") if withdrawal["asset_id"] != deposit["asset_id"]: raise HTTPException(status_code=422, detail="transfer legs must use the same canonical asset") if withdrawal["from_wallet_id"] == deposit["to_wallet_id"]: raise HTTPException(status_code=422, detail="source and target wallets must be different") out_quantity = Decimal(str(withdrawal["quantity"])) in_quantity = Decimal(str(deposit["quantity"])) fee = Decimal(str(withdrawal["fee_quantity"] or "0")) if out_quantity - in_quantity != fee: raise HTTPException(status_code=422, detail="gross/net quantities do not reconcile with the documented fee") out_time = _parse_timestamp(str(withdrawal["transaction_datetime"])) in_time = _parse_timestamp(str(deposit["transaction_datetime"])) if abs((out_time - in_time).total_seconds()) > 7 * 86400: raise HTTPException(status_code=422, detail="transfer leg timestamps are not compatible") tx_hash_match = bool(withdrawal["tx_hash"] and withdrawal["tx_hash"] == deposit["tx_hash"]) evidence = request.evidence_reference.strip() if not tx_hash_match and len(evidence) < 6: raise HTTPException(status_code=422, detail="matching transaction hash or explicit evidence reference required") payload = { "withdrawal_transaction_id": request.withdrawal_transaction_id, "deposit_transaction_id": request.deposit_transaction_id, "evidence_reference": evidence, "note": _redact_note(request.note), } fingerprint = _fingerprint(payload) return {"preview_id": f"crypto_transfer_preview_{fingerprint[:24]}", "input_fingerprint": fingerprint, "asset_id": withdrawal["asset_id"], "quantity": _text(in_quantity), "fee_quantity": _text(fee), "tx_hash_match": tx_hash_match, "external_cashflow_effect": "0", "payload": payload} def preview_crypto_transfer_pair(conn: Connection, request: CryptoTransferPairPreviewRequest) -> dict[str, Any]: return _transfer_pair_preview(conn, request) def confirm_crypto_transfer_pair(conn: Connection, request: CryptoTransferPairConfirmRequest) -> dict[str, Any]: if not request.confirm: raise HTTPException(status_code=422, detail="confirm=true required") preview = _transfer_pair_preview(conn, request) if request.preview_id != preview["preview_id"]: raise HTTPException(status_code=409, detail="preview is stale; create a new preview") confirmation_key = request.confirmation_key.strip() if not confirmation_key: raise HTTPException(status_code=422, detail="confirmation_key is required") pair_id = _id("cryptopair", request.withdrawal_transaction_id, request.deposit_transaction_id) existing = conn.execute("SELECT audit_id FROM crypto_internal_transfer_pairs WHERE transfer_pair_id=?", (pair_id,)).fetchone() if existing: return {"status": "already_confirmed", "transfer_pair_id": pair_id, "audit_id": existing["audit_id"], "idempotent_replay": True} audit_id = _id("audit", pair_id) now = _now() with conn: timestamp = _now() conn.execute( """INSERT INTO audit_log(audit_id,timestamp,source,action,entity_type,entity_id,new_values_json,user_text_note,confirmed,confirmation_timestamp,created_by,quality_status,created_at) VALUES (?,?,'vue_dashboard','crypto_transfer_pair_confirm','crypto_internal_transfer_pair',?,?,?,1,?,'user','ok',?)""", (audit_id, timestamp, pair_id, json.dumps(preview, sort_keys=True), preview["payload"]["note"], timestamp, timestamp), ) conn.execute( """INSERT INTO crypto_internal_transfer_pairs(transfer_pair_id,withdrawal_transaction_id,deposit_transaction_id,asset_id,evidence_reference,audit_id,created_at) VALUES (?,?,?,?,?,?,?)""", (pair_id, request.withdrawal_transaction_id, request.deposit_transaction_id, preview["asset_id"], preview["payload"]["evidence_reference"], audit_id, now), ) return {"status": "confirmed", "transfer_pair_id": pair_id, "audit_id": audit_id, "idempotent_replay": False}