tree src/jarvis_finance/market/providers.py | 27 +++++++- src/jarvis_finance/services/asset_price_refresh.py | 37 +++++++---- .../services/raiffeisen_manual_snapshot.py | 75 ++++++++++++---------- tests/unit/test_asset_price_refresh.py | 64 ++++++++++++++++++ tests/unit/test_raiffeisen_manual_snapshot.py | 23 +++++++ 5 files changed, 176 insertions(+), 50 deletions(-) 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 @@ -119,189 +119,212 @@ class CoinGeckoClient: if value is None: return PriceQuote(coingecko_id, currency.upper(), None, quality_status="missing", error_message="price missing") try: price = Decimal(str(value)) except InvalidOperation: return PriceQuote(coingecko_id, currency.upper(), None, quality_status="error", error_message="invalid price") ts = data.get("last_updated_at") provider_ts = None if ts: try: provider_ts = datetime.fromtimestamp(int(ts), tz=timezone.utc).isoformat() except (TypeError, ValueError): provider_ts = str(ts) return PriceQuote(coingecko_id, currency.upper(), price, provider_timestamp=provider_ts) def _parse_dt(value: str | None) -> datetime | None: if not value: return None try: dt = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) def is_price_stale(timestamp: str | None, *, max_age_seconds: int = 86_400) -> bool: dt = _parse_dt(timestamp) if dt is None: return True return (datetime.now(timezone.utc) - dt).total_seconds() > max_age_seconds def latest_crypto_price_row(conn: Connection, asset_id: str, currency: str = "CHF"): return conn.execute( """ SELECT * FROM crypto_prices WHERE asset_id=? AND price_currency=? ORDER BY COALESCE(provider_timestamp, fetched_at, '') DESC, fetched_at DESC LIMIT 1 """, (asset_id, currency.upper()), ).fetchone() def store_crypto_price(conn: Connection, *, asset_id: str, quote: PriceQuote) -> str: now = utc_now() price_id = stable_id("cryptoprice", asset_id, quote.coingecko_id, quote.currency, quote.provider, now) conn.execute( """ INSERT INTO crypto_prices(crypto_price_id, asset_id, coingecko_id, price_currency, price, provider, provider_timestamp, fetched_at, quality_status, error_message) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (price_id, asset_id, quote.coingecko_id, quote.currency.upper(), format(quote.price, "f") if quote.price is not None else "", quote.provider, quote.provider_timestamp, now, quote.quality_status, quote.error_message), ) if quote.quality_status in {"missing", "stale", "error", "conflict"}: create_alert(conn, priority="warnung", category="market_data", entity_type="crypto_asset", entity_id=asset_id, rule_id=f"crypto_price_{quote.quality_status}", message=f"Crypto price quality is {quote.quality_status}.", evidence={"coingecko_id": quote.coingecko_id, "currency": quote.currency, "error": quote.error_message}) conn.commit() return price_id def _create_missing_local_price_alert(conn: Connection, *, asset_id: str, symbol: str | None, coingecko_id: str | None, currency: str) -> None: create_alert( conn, priority="warnung", category="market_data", entity_type="crypto_asset", entity_id=asset_id, rule_id="crypto_price_missing_local", message="Crypto asset has no fresh local price after refresh.", evidence={"symbol": symbol, "coingecko_id": coingecko_id, "currency": currency}, fingerprint=f"crypto_price_missing_local:{currency}", ) def _has_fresh_local_price(conn: Connection, asset_id: str, currency: str, max_age_seconds: int) -> bool: latest = latest_crypto_price_row(conn, asset_id, currency) 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: if wanted_symbol and (asset["symbol"] or "").upper() != wanted_symbol: continue latest = latest_crypto_price_row(conn, asset["asset_id"], currency) has_any = latest is not None and latest["price"] not in (None, "") and latest["quality_status"] == "fresh" is_stale = bool(latest and latest["quality_status"] == "fresh" and is_price_stale(latest["fetched_at"], max_age_seconds=max_age_seconds)) if only_missing and has_any: continue if only_stale and not is_stale: continue filtered.append(asset) if limit is not None and len(filtered) >= limit: break return assets, filtered def _provider_get_batch(provider: MarketDataProvider, coingecko_ids: Sequence[str], currency: str) -> dict[str, PriceQuote]: if hasattr(provider, "get_crypto_prices"): return provider.get_crypto_prices(coingecko_ids, currency) # type: ignore[attr-defined] return {cid: provider.get_crypto_price(cid, currency) for cid in coingecko_ids} def refresh_crypto_prices( conn: Connection, *, provider: MarketDataProvider, currency: str = "CHF", max_age_seconds: int = 3600, only_missing: bool = False, only_stale: bool = False, only_symbol: str | None = None, limit: int | None = None, 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)) request_assets = [] for asset in assets: asset_id = asset["asset_id"] if not asset["coingecko_id"]: if not dry_run: create_alert(conn, priority="warnung", category="crypto", entity_type="crypto_asset", entity_id=asset_id, rule_id="missing_coingecko_id", message="Crypto asset has no CoinGecko ID; price refresh skipped.", evidence={"symbol": asset["symbol"]}, fingerprint="missing_coingecko_id") _create_missing_local_price_alert(conn, asset_id=asset_id, symbol=asset["symbol"], coingecko_id=None, currency=currency) result.skipped_count += 1 result.warning_count += 1 result.warnings.append(f"{asset['symbol']}: missing_coingecko_id") continue latest = latest_crypto_price_row(conn, asset_id, currency) if latest and latest["quality_status"] == "fresh" and not is_price_stale(latest["fetched_at"], max_age_seconds=max_age_seconds): result.cached_count += 1 continue request_assets.append(asset) refreshed_fresh_asset_ids: set[str] = set() for start in range(0, len(request_assets), max(1, batch_size)): if start and sleep_seconds > 0: time.sleep(sleep_seconds) chunk = request_assets[start : start + max(1, batch_size)] ids = [asset["coingecko_id"] for asset in chunk] try: quotes = _provider_get_batch(provider, ids, currency) except Exception as exc: quotes = {cid: PriceQuote(cid, currency, None, quality_status="error", error_message=str(exc)) for cid in ids} for asset in chunk: quote = quotes.get(asset["coingecko_id"]) or PriceQuote(asset["coingecko_id"], currency, None, quality_status="missing", error_message="price missing") if not dry_run: price_id = store_crypto_price(conn, asset_id=asset["asset_id"], quote=quote) result.written_price_ids.append(price_id) if quote.price is not None and quote.quality_status == "fresh": result.updated_count += 1 refreshed_fresh_asset_ids.add(asset["asset_id"]) if not dry_run: resolve_fixed_crypto_price_alerts(conn, currency=currency, max_age_seconds=max_age_seconds) elif quote.quality_status in {"missing", "stale"}: result.warning_count += 1 result.warnings.append(f"{asset['symbol']}: {quote.quality_status}") else: result.error_count += 1 result.errors.append(f"{asset['symbol']}: {quote.error_message or quote.quality_status}") if quote.quality_status == "stale": result.stale_count += 1 if (quote.price is None or quote.quality_status != "fresh") and not dry_run: _create_missing_local_price_alert(conn, asset_id=asset["asset_id"], symbol=asset["symbol"], coingecko_id=asset["coingecko_id"], currency=currency) if not dry_run: for asset in assets: if asset["asset_id"] in refreshed_fresh_asset_ids: continue if not _has_fresh_local_price(conn, asset["asset_id"], currency, max_age_seconds): result.missing_local_price_count += 1 _create_missing_local_price_alert(conn, asset_id=asset["asset_id"], symbol=asset["symbol"], coingecko_id=asset["coingecko_id"], currency=currency) conn.commit() else: # Dry-run reports current known gaps plus simulated failed quotes without mutating alerts/prices. for asset in assets: if not _has_fresh_local_price(conn, asset["asset_id"], currency, max_age_seconds): result.missing_local_price_count += 1 return result diff --git a/src/jarvis_finance/services/asset_price_refresh.py b/src/jarvis_finance/services/asset_price_refresh.py index e5fd047..a822fed 100644 --- a/src/jarvis_finance/services/asset_price_refresh.py +++ b/src/jarvis_finance/services/asset_price_refresh.py @@ -1,255 +1,266 @@ 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 -from jarvis_finance.services.crypto_market_recovery import run_crypto_market_one_shot +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]: - candidates = int( - conn.execute( - """SELECT COUNT(*) FROM crypto_assets a + 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,), - ).fetchone()[0] - ) - if candidates == 0: + ).fetchall() + ] + if not stale_asset_ids: return 0, 0 - result = run_crypto_market_one_shot(conn, provider=CoinGeckoClient()) - if result.status != "complete": - raise RuntimeError("crypto_provider_" + result.status) - return candidates, int(result.price_stored) + 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 diff --git a/src/jarvis_finance/services/raiffeisen_manual_snapshot.py b/src/jarvis_finance/services/raiffeisen_manual_snapshot.py index 20f8b61..c197c3b 100644 --- a/src/jarvis_finance/services/raiffeisen_manual_snapshot.py +++ b/src/jarvis_finance/services/raiffeisen_manual_snapshot.py @@ -90,324 +90,329 @@ def _membership_account(conn: Connection, platform_id: str) -> Any | None: """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(conn: Connection, payload: dict[str, Any], targets: list[_Target]) -> str: +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: - latest = None - if target.account_id: - table = "cash_account_snapshots" if target.asset_kind == "bank_cash" else "account_value_snapshots" - date_column = "balance_date" if target.asset_kind == "bank_cash" else "valuation_date" - value_column = "amount_chf" if target.asset_kind == "bank_cash" else "total_value_chf" - latest = conn.execute( - f"SELECT snapshot_id,{date_column} AS at,{value_column} AS value FROM {table} " - f"WHERE account_id=? AND {date_column}<=? " - + ("AND is_active=1 " if table == "account_value_snapshots" else "") - + f"ORDER BY {date_column} DESC,created_at DESC,snapshot_id DESC LIMIT 1", - (target.account_id, snapshot_date), - ).fetchone() details.append( { "role": target.role, "account_id": target.account_id, - "latest": dict(latest) if latest else None, + "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}) + 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 _known_total(conn: Connection, as_of: str) -> Decimal: - model = _wealth_model(conn, as_of) - # A component correction must not promote a mixed-date/modelled point to a - # confirmed household anchor. Current is only a sparse-fixture fallback. - baseline = model.get("anchor") or model.get("current") - return _decimal(baseline.get("value_chf")) if isinstance(baseline, dict) else ZERO - - -def _bank_total(conn: Connection, as_of: str) -> Decimal: - model = _wealth_model(conn, as_of) - for component in model.get("components") or []: - if component.get("key") == "bank_cash": - return _decimal(component.get("current_value_chf")) - return ZERO - - 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) - input_fingerprint = _baseline(conn, payload, targets) - known_before = _known_total(conn, snapshot_date) + 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_total(conn, snapshot_date) + cash_delta + 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 _baseline(conn, payload, targets) != input_fingerprint: + 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)""", diff --git a/tests/unit/test_asset_price_refresh.py b/tests/unit/test_asset_price_refresh.py index e1a13fc..c70b396 100644 --- a/tests/unit/test_asset_price_refresh.py +++ b/tests/unit/test_asset_price_refresh.py @@ -1,149 +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 index 76d70f8..a537ed0 100644 --- a/tests/unit/test_raiffeisen_manual_snapshot.py +++ b/tests/unit/test_raiffeisen_manual_snapshot.py @@ -135,160 +135,183 @@ def test_confirm_is_append_only_audited_and_idempotent(): 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): __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23__HERMES_CWD_8d46a20096ed__