diff --git a/src/jarvis_finance/market/providers.py b/src/jarvis_finance/market/providers.py index ce60256..82fbc88 100644 --- a/src/jarvis_finance/market/providers.py +++ b/src/jarvis_finance/market/providers.py @@ -196,8 +196,21 @@ def _has_fresh_local_price(conn: Connection, asset_id: str, currency: str, max_a return bool(latest and latest["quality_status"] == "fresh" and not is_price_stale(latest["fetched_at"], max_age_seconds=max_age_seconds)) -def _filter_assets(conn: Connection, *, currency: str, max_age_seconds: int, only_missing: bool, only_stale: bool, only_symbol: str | None, limit: int | None): +def _filter_assets( + conn: Connection, + *, + currency: str, + max_age_seconds: int, + only_missing: bool, + only_stale: bool, + only_symbol: str | None, + limit: int | None, + asset_ids: Sequence[str] | None, +): assets = conn.execute("SELECT asset_id, coingecko_id, symbol FROM crypto_assets WHERE is_active=1 ORDER BY symbol").fetchall() + if asset_ids is not None: + wanted_ids = {str(asset_id) for asset_id in asset_ids} + assets = [asset for asset in assets if str(asset["asset_id"]) in wanted_ids] filtered = [] wanted_symbol = only_symbol.upper() if only_symbol else None for asset in assets: @@ -235,10 +248,20 @@ def refresh_crypto_prices( dry_run: bool = False, sleep_seconds: float = 0.0, batch_size: int = 100, + asset_ids: Sequence[str] | None = None, ) -> PriceRefreshResult: currency = currency.upper() result = PriceRefreshResult(currency=currency, dry_run=dry_run) - all_assets, assets = _filter_assets(conn, currency=currency, max_age_seconds=max_age_seconds, only_missing=only_missing, only_stale=only_stale, only_symbol=only_symbol, limit=limit) + all_assets, assets = _filter_assets( + conn, + currency=currency, + max_age_seconds=max_age_seconds, + only_missing=only_missing, + only_stale=only_stale, + only_symbol=only_symbol, + limit=limit, + asset_ids=asset_ids, + ) result.total_assets = len(all_assets) if only_missing or only_stale or only_symbol or limit is not None: result.skipped_count += max(0, len(all_assets) - len(assets)) diff --git a/src/jarvis_finance/services/asset_price_refresh.py b/src/jarvis_finance/services/asset_price_refresh.py new file mode 100644 index 0000000..a822fed --- /dev/null +++ b/src/jarvis_finance/services/asset_price_refresh.py @@ -0,0 +1,378 @@ +from __future__ import annotations + +import hashlib +import json +import uuid +from datetime import UTC, datetime, timedelta +from pathlib import Path +from sqlite3 import Connection, SQLITE_DELETE, SQLITE_DENY, SQLITE_INSERT, SQLITE_OK, SQLITE_UPDATE +from typing import Any, Callable + +from jarvis_finance.api.schemas.market import QuoteRefreshRequest +from jarvis_finance.audit.log import record_audit_event +from jarvis_finance.market.providers import CoinGeckoClient, refresh_crypto_prices +from jarvis_finance.services.market_service import refresh_equity_quotes_batch +from jarvis_finance.services.modelled_wealth import build_modelled_wealth_development +from jarvis_finance.storage.database import connect + +SOURCES = ("equity", "crypto", "fx") +PROTECTED_TABLES = ( + "accounts", + "transactions", + "crypto_holdings", + "positions_snapshot", + "postfinance_snapshot_positions", + "truewealth_snapshot_positions", +) + + +def _deny_protected_dml( + action: int, + table: str | None, + _column: str | None, + _database: str | None, + _trigger: str | None, +) -> int: + if action in {SQLITE_INSERT, SQLITE_UPDATE, SQLITE_DELETE} and table in PROTECTED_TABLES: + return SQLITE_DENY + return SQLITE_OK + + +def _now() -> str: + return datetime.now(UTC).isoformat() + + +def _database_path(conn: Connection) -> str: + row = next((row for row in conn.execute("PRAGMA database_list") if str(row[1]) == "main"), None) + if not row or not str(row[2] or ""): + raise ValueError("asset_refresh_requires_persistent_database") + return str(Path(str(row[2])).resolve()) + + +def _protected_fingerprint(conn: Connection) -> str: + payload: dict[str, list[dict[str, Any]]] = {} + available = { + str(row[0]) + for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() + } + for table in PROTECTED_TABLES: + if table not in available: + continue + rows = conn.execute(f'SELECT * FROM "{table}" ORDER BY rowid').fetchall() + payload[table] = [dict(row) for row in rows] + return hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") + ).hexdigest() + + +def _status_payload(conn: Connection, job_id: str) -> dict[str, Any]: + job = conn.execute("SELECT * FROM asset_price_refresh_jobs WHERE job_id=?", (job_id,)).fetchone() + if not job: + raise ValueError("asset_price_refresh_job_not_found") + sources = [ + dict(row) + for row in conn.execute( + "SELECT * FROM asset_price_refresh_sources WHERE job_id=? ORDER BY CASE source WHEN 'equity' THEN 1 WHEN 'crypto' THEN 2 ELSE 3 END", + (job_id,), + ).fetchall() + ] + return { + "job_id": str(job["job_id"]), + "status": str(job["status"]), + "requested_at": str(job["requested_at"]), + "completed_at": str(job["completed_at"]) if job["completed_at"] else None, + "stale_before": str(job["stale_before"]), + "progress": {"completed": int(job["progress_completed"]), "total": int(job["progress_total"])}, + "sources": [ + { + "source": str(row["source"]), + "status": str(row["status"]), + "stale_candidates": int(row["stale_candidates"]), + "updated_count": int(row["updated_count"]), + "error_code": str(row["error_code"]) if row["error_code"] else None, + "started_at": str(row["started_at"]) if row["started_at"] else None, + "completed_at": str(row["completed_at"]) if row["completed_at"] else None, + } + for row in sources + ], + "wealth_snapshot_created": bool(job["wealth_snapshot_id"]), + "audit_recorded": bool(job["audit_id"]), + "provider_calls_on_read": False, + } + + +def create_asset_price_refresh_job(conn: Connection, *, stale_hours: int = 24) -> tuple[dict[str, Any], str]: + """Persist a queued job only. No provider call occurs before the HTTP response.""" + if conn.in_transaction: + raise ValueError("asset_refresh_requires_clean_transaction") + conn.execute("BEGIN IMMEDIATE") + try: + if conn.execute( + "SELECT 1 FROM asset_price_refresh_jobs WHERE status IN ('queued','running') LIMIT 1" + ).fetchone(): + raise ValueError("asset_price_refresh_job_already_running") + now = datetime.now(UTC) + job_id = f"asset-refresh-{uuid.uuid4().hex}" + stale_before = (now - timedelta(hours=max(1, min(stale_hours, 720)))).isoformat() + conn.execute( + """INSERT INTO asset_price_refresh_jobs( + job_id,status,requested_at,stale_before,progress_total,progress_completed + ) VALUES(?,'queued',?,?,3,0)""", + (job_id, now.isoformat(), stale_before), + ) + conn.executemany( + """INSERT INTO asset_price_refresh_sources( + job_id,source,status,stale_candidates,updated_count + ) VALUES(?,?,'pending',0,0)""", + [(job_id, source) for source in SOURCES], + ) + conn.commit() + except Exception: + if conn.in_transaction: + conn.rollback() + raise + return _status_payload(conn, job_id), _database_path(conn) + + +def _equity_source(conn: Connection, stale_before: str) -> tuple[int, int]: + response = refresh_equity_quotes_batch( + conn, + QuoteRefreshRequest( + provider="auto", + only_missing=True, + stale_before=stale_before, + limit=500, + max_retries=1, + pacing_seconds=0.15, + ), + ) + candidates = max(0, int(response.total) - int(response.cached)) + if response.errors and response.updated == 0 and candidates > 0: + raise RuntimeError("equity_provider_failed") + return candidates, int(response.updated) + + +def _crypto_source(conn: Connection, stale_before: str) -> tuple[int, int]: + stale_asset_ids = [ + str(row["asset_id"]) + for row in conn.execute( + """SELECT a.asset_id FROM crypto_assets a + WHERE a.is_active=1 AND EXISTS( + SELECT 1 FROM crypto_holdings h WHERE h.asset_id=a.asset_id AND CAST(h.quantity AS REAL)<>0 + ) AND NOT EXISTS( + SELECT 1 FROM crypto_prices p + WHERE p.asset_id=a.asset_id AND p.fetched_at>=? + AND p.quality_status='fresh' AND p.price IS NOT NULL + ) ORDER BY a.asset_id""", + (stale_before,), + ).fetchall() + ] + if not stale_asset_ids: + return 0, 0 + cutoff = datetime.fromisoformat(stale_before.replace("Z", "+00:00")) + if cutoff.tzinfo is None: + cutoff = cutoff.replace(tzinfo=UTC) + max_age_seconds = max(1, int((datetime.now(UTC) - cutoff.astimezone(UTC)).total_seconds())) + result = refresh_crypto_prices( + conn, + provider=CoinGeckoClient(), + currency="CHF", + max_age_seconds=max_age_seconds, + asset_ids=stale_asset_ids, + batch_size=100, + ) + if result.error_count and result.updated_count == 0: + raise RuntimeError("crypto_provider_failed") + return len(stale_asset_ids), int(result.updated_count) + + +def _fx_source(conn: Connection, stale_before: str) -> tuple[int, int]: + from jarvis_finance.fx.providers import FrankfurterFxProvider, TwelveDataFxProvider + from jarvis_finance.fx.rates import resolve_fx_rate_to_chf + + cutoff_date = stale_before[:10] + currencies = [ + str(row["currency"]).upper() + for row in conn.execute( + """SELECT DISTINCT upper(i.currency) currency + FROM instruments i + WHERE i.is_active=1 AND upper(COALESCE(i.currency,'CHF'))!='CHF' + AND NOT EXISTS( + SELECT 1 FROM fx_rates f + WHERE f.base_currency=upper(i.currency) AND f.quote_currency='CHF' + AND f.rate_date>=? AND f.quality_status IN ('fresh','ok') + ) + ORDER BY currency""", + (cutoff_date,), + ).fetchall() + ] + updated = 0 + failures = 0 + for currency in currencies: + try: + result = resolve_fx_rate_to_chf( + conn, + base_currency=currency, + rate_date=None, + providers=[FrankfurterFxProvider(), TwelveDataFxProvider()], + persist=True, + resolve_fixed=True, + ) + updated += int(result.status == "ok") + except Exception: + failures += 1 + conn.commit() + if failures and updated == 0: + raise RuntimeError("fx_provider_failed") + return len(currencies), updated + + +DEFAULT_RUNNERS: dict[str, Callable[[Connection, str], tuple[int, int]]] = { + "equity": _equity_source, + "crypto": _crypto_source, + "fx": _fx_source, +} + + +def run_asset_price_refresh( + db_path: str, + job_id: str, + *, + runners: dict[str, Callable[[Connection, str], tuple[int, int]]] | None = None, +) -> None: + """Background worker with source isolation, stored progress and mutation guard.""" + conn = connect(db_path) + selected = runners or DEFAULT_RUNNERS + try: + conn.execute("BEGIN IMMEDIATE") + claimed = conn.execute( + """UPDATE asset_price_refresh_jobs + SET status='running' + WHERE job_id=? AND status='queued'""", + (job_id,), + ).rowcount + conn.commit() + if claimed != 1: + return + stale_before = str( + conn.execute( + "SELECT stale_before FROM asset_price_refresh_jobs WHERE job_id=?", + (job_id,), + ).fetchone()[0] + ) + protected_before = _protected_fingerprint(conn) + conn.set_authorizer(_deny_protected_dml) + completed = 0 + failures = 0 + for source in SOURCES: + started = _now() + conn.execute( + "UPDATE asset_price_refresh_sources SET status='running',started_at=? WHERE job_id=? AND source=?", + (started, job_id, source), + ) + conn.commit() + candidates = updated = 0 + status = "complete" + error_code = None + try: + candidates, updated = selected[source](conn, stale_before) + if candidates == 0: + status = "skipped" + except Exception as exc: + if conn.in_transaction: + conn.rollback() + status = "failed" + failures += 1 + error_code = str(exc)[:120] or type(exc).__name__ + completed += 1 + conn.execute( + """UPDATE asset_price_refresh_sources + SET status=?,stale_candidates=?,updated_count=?,error_code=?,completed_at=? + WHERE job_id=? AND source=?""", + (status, candidates, updated, error_code, _now(), job_id, source), + ) + conn.execute( + "UPDATE asset_price_refresh_jobs SET progress_completed=? WHERE job_id=?", + (completed, job_id), + ) + conn.commit() + if _protected_fingerprint(conn) != protected_before: + raise RuntimeError("protected_holdings_or_transactions_mutated") + + successful_sources = failures < len(SOURCES) + wealth_snapshot_id = None + if successful_sources: + model = build_modelled_wealth_development(conn, period="1m") + current = model.get("current") or {} + wealth_snapshot_id = f"wealth-refresh-{uuid.uuid4().hex}" + source_rows = [ + dict(row) + for row in conn.execute( + "SELECT source,status,stale_candidates,updated_count,error_code FROM asset_price_refresh_sources WHERE job_id=? ORDER BY source", + (job_id,), + ).fetchall() + ] + conn.execute( + """INSERT INTO aggregated_wealth_refresh_snapshots( + wealth_snapshot_id,job_id,captured_at,known_wealth_chf,quality_status,source_status_json + ) VALUES(?,?,?,?,?,?)""", + ( + wealth_snapshot_id, + job_id, + _now(), + current.get("value_chf"), + "complete" if failures == 0 else "partial", + json.dumps(source_rows, sort_keys=True), + ), + ) + final_status = "complete" if failures == 0 else "failed" if failures == len(SOURCES) else "partial" + audit_id = record_audit_event( + conn, + source="asset_price_refresh_job_v1", + action="asset_prices_refresh_completed", + entity_type="asset_price_refresh_job", + entity_id=job_id, + old_values={}, + new_values={ + "status": final_status, + "source_count": len(SOURCES), + "failed_source_count": failures, + "wealth_snapshot_created": bool(wealth_snapshot_id), + "holdings_mutated": False, + "transactions_mutated": False, + "trades_created": 0, + }, + created_by="system", + ) + conn.execute( + """UPDATE asset_price_refresh_jobs + SET status=?,completed_at=?,wealth_snapshot_id=?,audit_id=? + WHERE job_id=?""", + (final_status, _now(), wealth_snapshot_id, audit_id, job_id), + ) + conn.commit() + except Exception as exc: + if conn.in_transaction: + conn.rollback() + audit_id = record_audit_event( + conn, + source="asset_price_refresh_job_v1", + action="asset_prices_refresh_failed", + entity_type="asset_price_refresh_job", + entity_id=job_id, + old_values={}, + new_values={"status": "failed", "error_code": str(exc)[:120]}, + created_by="system", + ) + conn.execute( + "UPDATE asset_price_refresh_jobs SET status='failed',completed_at=?,audit_id=? WHERE job_id=?", + (_now(), audit_id, job_id), + ) + conn.commit() + finally: + conn.close() + + +def asset_price_refresh_status(conn: Connection, job_id: str) -> dict[str, Any]: + """Stored status only: no provider call, write or lazy refresh.""" + return _status_payload(conn, job_id) diff --git a/src/jarvis_finance/services/raiffeisen_manual_snapshot.py b/src/jarvis_finance/services/raiffeisen_manual_snapshot.py new file mode 100644 index 0000000..c197c3b --- /dev/null +++ b/src/jarvis_finance/services/raiffeisen_manual_snapshot.py @@ -0,0 +1,503 @@ +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 diff --git a/tests/unit/test_asset_price_refresh.py b/tests/unit/test_asset_price_refresh.py new file mode 100644 index 0000000..c70b396 --- /dev/null +++ b/tests/unit/test_asset_price_refresh.py @@ -0,0 +1,213 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from decimal import Decimal +from pathlib import Path +from sqlite3 import Connection +import threading +from typing import Callable + +from jarvis_finance.market.providers import PriceQuote +from jarvis_finance.services import asset_price_refresh as asset_refresh +from jarvis_finance.services.asset_price_refresh import ( + asset_price_refresh_status, + create_asset_price_refresh_job, + run_asset_price_refresh, +) +from jarvis_finance.storage.database import connect +from jarvis_finance.storage.migrations import apply_migrations + + +def database(path: Path): + conn = connect(path) + apply_migrations(conn) + return conn + + +def test_refresh_job_is_queued_then_isolates_source_failure_and_creates_one_snapshot(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, db_path = create_asset_price_refresh_job(conn, stale_hours=12) + assert queued["status"] == "queued" + assert all(row["status"] == "pending" for row in queued["sources"]) + assert queued["provider_calls_on_read"] is False + job_id = queued["job_id"] + conn.close() + + calls: list[str] = [] + cutoffs: list[str] = [] + + def success(source: str): + def runner(_conn, _stale_before): + calls.append(source) + cutoffs.append(_stale_before) + return 2, 1 + return runner + + def failure(_conn, _stale_before): + calls.append("crypto") + cutoffs.append(_stale_before) + _conn.execute("UPDATE transactions SET net_amount_chf='999.00'") + raise AssertionError("protected-table authorizer should reject before this line") + + run_asset_price_refresh( + db_path, + job_id, + runners={"equity": success("equity"), "crypto": failure, "fx": success("fx")}, + ) + + conn = connect(path) + status = asset_price_refresh_status(conn, job_id) + assert calls == ["equity", "crypto", "fx"] + assert cutoffs == [queued["stale_before"]] * 3 + assert status["status"] == "partial" + assert status["progress"] == {"completed": 3, "total": 3} + assert status["wealth_snapshot_created"] is True + assert status["audit_recorded"] is True + assert [row["status"] for row in status["sources"]] == ["complete", "failed", "complete"] + assert conn.execute("SELECT COUNT(*) FROM aggregated_wealth_refresh_snapshots WHERE job_id=?", (job_id,)).fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM transactions").fetchone()[0] == 0 + conn.close() + + +def test_status_read_does_not_write_or_call_runner(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, _ = create_asset_price_refresh_job(conn) + before = conn.total_changes + first = asset_price_refresh_status(conn, queued["job_id"]) + second = asset_price_refresh_status(conn, queued["job_id"]) + assert conn.total_changes == before + assert first == second + assert first["provider_calls_on_read"] is False + conn.close() + + +def test_concurrent_starts_create_exactly_one_active_job(tmp_path): + path = tmp_path / "finance.sqlite3" + database(path).close() + barrier = threading.Barrier(2) + outcomes: list[str] = [] + + def worker() -> None: + conn = connect(path) + conn.execute("PRAGMA busy_timeout=5000") + barrier.wait() + try: + create_asset_price_refresh_job(conn) + outcomes.append("created") + except ValueError as exc: + outcomes.append(str(exc)) + finally: + conn.close() + + first = threading.Thread(target=worker) + second = threading.Thread(target=worker) + first.start() + second.start() + first.join() + second.join() + + assert sorted(outcomes) == ["asset_price_refresh_job_already_running", "created"] + conn = connect(path) + assert conn.execute( + "SELECT COUNT(*) FROM asset_price_refresh_jobs WHERE status IN ('queued','running')" + ).fetchone()[0] == 1 + conn.close() + + +def test_concurrent_workers_claim_a_queued_job_exactly_once(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, db_path = create_asset_price_refresh_job(conn) + conn.close() + calls: list[str] = [] + barrier = threading.Barrier(2) + + def runner(source: str) -> Callable[[Connection, str], tuple[int, int]]: + def execute(_conn: Connection, _stale_before: str) -> tuple[int, int]: + calls.append(source) + return 0, 0 + + return execute + + runners = {source: runner(source) for source in ("equity", "crypto", "fx")} + + def worker() -> None: + barrier.wait() + run_asset_price_refresh(db_path, queued["job_id"], runners=runners) + + first = threading.Thread(target=worker) + second = threading.Thread(target=worker) + first.start() + second.start() + first.join() + second.join() + + assert sorted(calls) == ["crypto", "equity", "fx"] + conn = connect(path) + assert conn.execute( + "SELECT COUNT(*) FROM aggregated_wealth_refresh_snapshots WHERE job_id=?", + (queued["job_id"],), + ).fetchone()[0] == 1 + conn.close() + + +def test_crypto_refresh_fetches_only_stale_held_asset_ids(tmp_path, monkeypatch): + path = tmp_path / "finance.sqlite3" + conn = database(path) + conn.execute( + """INSERT INTO crypto_assets(asset_id,coin_name,symbol,coingecko_id,is_active,created_at) + VALUES('btc','Bitcoin','BTC','bitcoin',1,'2026-01-01'), + ('eth','Ethereum','ETH','ethereum',1,'2026-01-01')""" + ) + conn.execute( + "INSERT INTO crypto_wallets(wallet_id,wallet_name,wallet_type,created_at) VALUES('wallet','Wallet','exchange','2026-01-01')" + ) + conn.execute( + """INSERT INTO crypto_holdings( + crypto_holding_id,asset_id,wallet_id,quantity,last_verified_at,verification_status,created_at + ) VALUES('btc-held','btc','wallet','1','2026-08-27','verified','2026-01-01'), + ('eth-held','eth','wallet','1','2026-08-27','verified','2026-01-01')""" + ) + now = datetime.now(UTC) + conn.execute( + """INSERT INTO crypto_prices( + crypto_price_id,asset_id,coingecko_id,price_currency,price,provider, + provider_timestamp,fetched_at,quality_status + ) VALUES('eth-fresh','eth','ethereum','CHF','100','CoinGecko',?,?, 'fresh')""", + (now.isoformat(), now.isoformat()), + ) + conn.commit() + + class Provider: + calls: list[tuple[str, ...]] = [] + + def get_crypto_prices(self, coingecko_ids, currency="CHF"): + ids = tuple(coingecko_ids) + self.calls.append(ids) + return { + provider_id: PriceQuote( + provider_id, + currency, + Decimal("123.45"), + provider_timestamp=now.isoformat(), + ) + for provider_id in ids + } + + def get_crypto_price(self, coingecko_id, currency="CHF"): + return self.get_crypto_prices([coingecko_id], currency)[coingecko_id] + + provider = Provider() + monkeypatch.setattr(asset_refresh, "CoinGeckoClient", lambda: provider) + candidates, updated = asset_refresh._crypto_source( + conn, + (now - timedelta(hours=24)).isoformat(), + ) + + assert (candidates, updated) == (1, 1) + assert provider.calls == [("bitcoin",)] + assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='eth'").fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='btc'").fetchone()[0] == 1 + conn.close() diff --git a/tests/unit/test_raiffeisen_manual_snapshot.py b/tests/unit/test_raiffeisen_manual_snapshot.py new file mode 100644 index 0000000..a537ed0 --- /dev/null +++ b/tests/unit/test_raiffeisen_manual_snapshot.py @@ -0,0 +1,363 @@ +from __future__ import annotations + +from datetime import date +from decimal import Decimal +from pathlib import Path +import sqlite3 +import threading + +import pytest + +from jarvis_finance.services.modelled_wealth import build_modelled_wealth_development +from jarvis_finance.services.performance_scope import set_performance_scope_classification +from jarvis_finance.services.raiffeisen_manual_snapshot import ( + confirm_raiffeisen_manual_snapshot, + preview_raiffeisen_manual_snapshot, +) +from jarvis_finance.storage.database import connect, connect_memory +from jarvis_finance.storage.migrations import apply_migrations + +NOW = "2026-08-20T12:00:00+00:00" + + +def database(): + conn = connect_memory() + apply_migrations(conn) + conn.execute( + "INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES('bank','Raiffeisen','bank',?)", + (NOW,), + ) + for account_id, name in ( + ("private", "Privatkonto ••••5632"), + ("savings", "Sparkonto ••••5031"), + ): + conn.execute( + """INSERT INTO accounts( + account_id,platform_id,account_name,account_type,currency,performance_included, + is_active,created_at,balance_mode,portfolio_bucket + ) VALUES(?, 'bank', ?, 'cash','CHF',0,1,?,'snapshot','cash')""", + (account_id, name, NOW), + ) + for snapshot_id, account_id, value in ( + ("old-private", "private", "100.00"), + ("old-savings", "savings", "10.00"), + ): + conn.execute( + """INSERT INTO cash_account_snapshots( + snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency, + amount_chf,source,created_at,created_by,semantic_identity + ) VALUES(?,?, 'manual_balance','2026-08-20',?,'CHF',?,'test',?,'test',?)""", + (snapshot_id, account_id, value, value, NOW, snapshot_id), + ) + conn.commit() + return conn + + +def source_facts(): + return { + "snapshot_date": date(2026, 8, 21), + "private_account_value_chf": Decimal("90.00"), + "savings_account_value_chf": Decimal("20.00"), + "membership_value_chf": Decimal("5.00"), + } + + +def test_preview_is_read_only_and_keeps_targets_separate(): + conn = database() + before = conn.total_changes + + preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) + + assert conn.total_changes == before + assert preview["bank_cash_after_chf"] == "110.00" + assert preview["separate_membership_asset_after_chf"] == "5.00" + assert preview["known_wealth_before_chf"] == "110.00" + assert preview["expected_known_wealth_after_chf"] == "115.00" + assert preview["expected_total_wealth_change_chf"] == "5.00" + assert preview["creates_transactions"] is False + assert [row["account_label"] for row in preview["affected_accounts"]] == [ + "Bankkonto ••••5632", + "Bankkonto ••••5031", + "Raiffeisen Genossenschaftsanteil", + ] + assert preview["affected_accounts"][2]["previous_status"] == "not_created" + + +def test_preview_uses_confirmed_cash_movements_after_latest_snapshot(): + conn = database() + conn.execute( + """INSERT INTO transactions( + transaction_id,transaction_type,account_id,trade_date,net_amount_original, + currency_original,fx_rate_to_chf,net_amount_chf,source_type,is_confirmed, + quality_status,created_at,is_voided + ) VALUES('movement','cash_movement','private','2026-08-21','5','CHF','1','5', + 'household_csv',1,'ok',?,0)""", + (NOW,), + ) + conn.commit() + + preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) + + private = preview["affected_accounts"][0] + assert private["previous_value_chf"] == "105.00" + assert preview["bank_cash_after_chf"] == "110.00" + assert preview["known_wealth_before_chf"] == "115.00" + assert preview["expected_total_wealth_change_chf"] == "0.00" + assert preview["expected_known_wealth_after_chf"] == "115.00" + + +def test_confirm_is_append_only_audited_and_idempotent(): + conn = database() + preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) + with pytest.raises(ValueError, match="confirmation_token_mismatch"): + confirm_raiffeisen_manual_snapshot( + conn, + **source_facts(), + preview_id=f"raiffeisen-preview-{'0' * 32}", + confirmation_id=f"raiffeisen-confirm-{'0' * 32}", + input_fingerprint=preview["input_fingerprint"], + ) + request = { + **source_facts(), + "preview_id": preview["preview_id"], + "confirmation_id": preview["confirmation_id"], + "input_fingerprint": preview["input_fingerprint"], + } + + result = confirm_raiffeisen_manual_snapshot(conn, **request) + replay = confirm_raiffeisen_manual_snapshot(conn, **request) + + assert result["status"] == "confirmed" + assert result["created_snapshot_count"] == 3 + assert result["created_transaction_count"] == 0 + assert result["known_wealth_after_chf"] == "115.00" + assert replay["status"] == "already_applied" + assert conn.execute("SELECT COUNT(*) FROM transactions").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM cash_account_snapshots WHERE source='manual_screenshot_snapshot'").fetchone()[0] == 2 + assert conn.execute("SELECT COUNT(*) FROM account_value_snapshots WHERE source_type='manual_screenshot_snapshot'").fetchone()[0] == 1 + membership = conn.execute( + "SELECT account_type,portfolio_bucket FROM accounts WHERE lower(account_name) LIKE '%genossenschaft%'" + ).fetchone() + assert dict(membership) == {"account_type": "other_asset", "portfolio_bucket": "other"} + assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity_id=?", (preview["confirmation_id"],)).fetchone()[0] == 1 + with pytest.raises(ValueError, match="confirmation_id_reused_with_different_input"): + confirm_raiffeisen_manual_snapshot( + conn, + **{**request, "membership_value_chf": Decimal("6.00")}, + ) + with pytest.raises(sqlite3.IntegrityError, match="manual cash snapshots are immutable"): + conn.execute( + "UPDATE cash_account_snapshots SET amount_chf='999.00' WHERE source='manual_screenshot_snapshot'" + ) + conn.rollback() + with pytest.raises(sqlite3.IntegrityError, match="manual asset snapshots cannot be deleted"): + conn.execute( + "DELETE FROM account_value_snapshots WHERE source_type='manual_screenshot_snapshot'" + ) + conn.rollback() + with pytest.raises(sqlite3.IntegrityError, match="manual cash snapshots cannot be replaced"): + conn.execute( + """INSERT OR REPLACE 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 + ) SELECT snapshot_id,account_id,snapshot_type,balance_date,'999.00',currency, + '999.00',source,note,created_at,created_by,audit_id,semantic_identity + FROM cash_account_snapshots + WHERE source='manual_screenshot_snapshot' LIMIT 1""" + ) + conn.rollback() + with pytest.raises(sqlite3.IntegrityError, match="manual asset snapshots cannot be replaced"): + conn.execute( + """INSERT OR REPLACE 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 + ) SELECT snapshot_id,account_id,valuation_date,'999.00',currency,source_type, + quality_status,notes,created_at,updated_at,valuation_at,source_reference,is_active + FROM account_value_snapshots + WHERE source_type='manual_screenshot_snapshot' LIMIT 1""" + ) + conn.rollback() + with pytest.raises(sqlite3.IntegrityError, match="manual snapshot confirmations cannot be replaced"): + conn.execute( + """INSERT OR REPLACE 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 + ) SELECT 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 + FROM manual_snapshot_confirmations LIMIT 1""" + ) + conn.rollback() + + +def test_confirm_fails_closed_when_baseline_changed(): + conn = database() + preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) + conn.execute( + """INSERT INTO cash_account_snapshots( + snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency,amount_chf, + source,created_at,created_by,semantic_identity + ) VALUES('changed','private','manual_balance','2026-08-21','95','CHF','95','test',?,'test','changed')""", + (NOW,), + ) + conn.commit() + with pytest.raises(ValueError, match="baseline_changed"): + confirm_raiffeisen_manual_snapshot( + conn, + **source_facts(), + preview_id=preview["preview_id"], + confirmation_id=preview["confirmation_id"], + input_fingerprint=preview["input_fingerprint"], + ) + + +def test_confirm_fails_closed_when_cash_movement_changes_preview_projection(): + conn = database() + preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) + conn.execute( + """INSERT INTO transactions( + transaction_id,transaction_type,account_id,trade_date,currency_original, + net_amount_chf,fx_status,source_type,is_confirmed,quality_status,created_at,updated_at + ) VALUES('movement-after-preview','cash','private','2026-08-21','CHF', + '3.00','ok','test_manual_adjustment',1,'ok',?,?)""", + (NOW, NOW), + ) + conn.commit() + + with pytest.raises(ValueError, match="baseline_changed"): + confirm_raiffeisen_manual_snapshot( + conn, + **source_facts(), + preview_id=preview["preview_id"], + confirmation_id=preview["confirmation_id"], + input_fingerprint=preview["input_fingerprint"], + ) + + +def test_acceptance_sums_include_unchanged_bank_cash_and_keep_component_correction_after_anchor(): + conn = database() + conn.execute( + "UPDATE cash_account_snapshots SET amount_original='29042.53',amount_chf='29042.53' WHERE account_id='private'" + ) + conn.execute( + "UPDATE cash_account_snapshots SET amount_original='19.84',amount_chf='19.84' WHERE account_id='savings'" + ) + conn.execute( + """INSERT INTO accounts( + account_id,platform_id,account_name,account_type,currency,performance_included, + is_active,created_at,balance_mode,portfolio_bucket + ) VALUES('unchanged-bank','bank','Unchanged bank account','cash','CHF',0,1,?,'snapshot','cash')""", + (NOW,), + ) + conn.execute( + """INSERT INTO cash_account_snapshots( + snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency, + amount_chf,source,created_at,created_by,semantic_identity + ) VALUES('unchanged-bank-value','unchanged-bank','manual_balance','2026-08-20', + '74900.12','CHF','74900.12','test',?,'test','unchanged-bank-value')""", + (NOW,), + ) + 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,balance_mode,portfolio_bucket + ) VALUES('pf-depot','pf','PostFinance Depot','brokerage','CHF',0,1,?,'snapshot','equity')""", + (NOW,), + ) + set_performance_scope_classification( + conn, + account_id="pf-depot", + included=True, + classification_role="postfinance_etrading_depot", + source="test", + note="acceptance anchor", + classified_at=NOW, + ) + conn.execute( + """INSERT INTO account_value_snapshots( + snapshot_id,account_id,valuation_date,total_value_chf,currency,source_type, + quality_status,created_at,valuation_at,is_active + ) VALUES('pf-anchor','pf-depot','2026-08-26','523788.47','CHF', + 'postfinance_official_import','ok',?,'2026-08-26T12:00:00+00:00',1)""", + (NOW,), + ) + conn.commit() + facts = { + "snapshot_date": date(2026, 8, 27), + "private_account_value_chf": Decimal("29059.44"), + "savings_account_value_chf": Decimal("19.84"), + "membership_value_chf": Decimal("200.00"), + } + + preview = preview_raiffeisen_manual_snapshot(conn, **facts) + + assert preview["bank_cash_after_chf"] == "103979.40" + assert preview["separate_membership_asset_after_chf"] == "200.00" + assert preview["known_wealth_before_chf"] == "627750.96" + assert preview["expected_total_wealth_change_chf"] == "216.91" + assert preview["expected_known_wealth_after_chf"] == "627967.87" + confirmed = confirm_raiffeisen_manual_snapshot( + conn, + **facts, + preview_id=preview["preview_id"], + confirmation_id=preview["confirmation_id"], + input_fingerprint=preview["input_fingerprint"], + ) + assert confirmed["bank_cash_after_chf"] == "103979.40" + assert confirmed["known_wealth_after_chf"] == "627967.87" + model = build_modelled_wealth_development(conn, as_of="2026-08-27", period="all") + assert model["last_confirmed_anchor_date"] == "2026-08-26" + + +def test_concurrent_identical_confirm_is_one_write_and_one_truthful_replay(tmp_path: Path): + source = database() + db_path = tmp_path / "concurrent.sqlite3" + target = connect(db_path) + source.backup(target) + source.close() + preview = preview_raiffeisen_manual_snapshot(target, **source_facts()) + target.close() + request = { + **source_facts(), + "preview_id": preview["preview_id"], + "confirmation_id": preview["confirmation_id"], + "input_fingerprint": preview["input_fingerprint"], + } + barrier = threading.Barrier(2) + statuses: list[str] = [] + failures: list[Exception] = [] + + def worker() -> None: + conn = connect(db_path) + conn.execute("PRAGMA busy_timeout=5000") + barrier.wait() + try: + statuses.append(confirm_raiffeisen_manual_snapshot(conn, **request)["status"]) + except Exception as exc: # pragma: no cover - asserted empty below + failures.append(exc) + finally: + conn.close() + + first = threading.Thread(target=worker) + second = threading.Thread(target=worker) + first.start() + second.start() + first.join() + second.join() + + assert failures == [] + assert sorted(statuses) == ["already_applied", "confirmed"] + conn = connect(db_path) + assert conn.execute("SELECT COUNT(*) FROM manual_snapshot_confirmations").fetchone()[0] == 1 + assert conn.execute( + "SELECT COUNT(*) FROM cash_account_snapshots WHERE source='manual_screenshot_snapshot'" + ).fetchone()[0] == 2 + assert conn.execute( + "SELECT COUNT(*) FROM account_value_snapshots WHERE source_type='manual_screenshot_snapshot'" + ).fetchone()[0] == 1 + conn.close() __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23__HERMES_CWD_8d46a20096ed__