from __future__ import annotations import hashlib import json import uuid from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from decimal import Decimal 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, PriceQuote, store_crypto_price from jarvis_finance.market_data.cache import upsert_crypto_price_point from jarvis_finance.market_data.prices import EquityPriceProvider, EquityPriceQuote from jarvis_finance.services.market_service import refresh_equity_quotes_batch from jarvis_finance.services.daily_valuations import run_daily_crypto_valuation from jarvis_finance.services.modelled_wealth import build_modelled_wealth_development from jarvis_finance.services.portfolio_analytics import run_daily_market_valuation from jarvis_finance.storage.database import connect SOURCES = ("equity", "crypto", "fx") PROTECTED_TABLES = ( "accounts", "transactions", "crypto_transactions", "crypto_holdings", "positions_snapshot", "postfinance_snapshot_positions", "truewealth_snapshot_positions", ) @dataclass(frozen=True) class SourceRunResult: stale_candidates: int = 0 updated_count: int = 0 fresh_unchanged_count: int = 0 stale_remaining_count: int = 0 failed_count: int = 0 diagnostics: tuple[str, ...] = field(default_factory=tuple) def __iter__(self): yield self.stale_candidates yield self.updated_count def _source_result(value: SourceRunResult | tuple[int, int]) -> SourceRunResult: if isinstance(value, SourceRunResult): return value return SourceRunResult(stale_candidates=int(value[0]), updated_count=int(value[1])) 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, "fresh_unchanged_count": int(row["fresh_unchanged_count"]), "stale_remaining_count": int(row["stale_remaining_count"]), "failed_count": int(row["failed_count"]), "diagnostics": json.loads(str(row["diagnostics_json"] or "[]")), "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"]), "successful_assets": sum(int(row["updated_count"]) for row in sources), "fresh_unchanged_assets": sum(int(row["fresh_unchanged_count"]) for row in sources), "stale_assets": sum(int(row["stale_remaining_count"]) for row in sources), "failed_assets": sum(int(row["failed_count"]) for row in sources), "next_action": ( "Diagnose prüfen und nur betroffene Quelle erneut versuchen." if any(int(row["failed_count"]) or int(row["stale_remaining_count"]) for row in sources) else "Keine Aktion nötig; alle verfügbaren Kurse sind aktuell." ), "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) -> SourceRunResult: 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, ), run_valuation=False, ) candidates = max(0, int(response.total) - int(response.cached)) failed = sum(1 for row in response.results if str(row.get("status")) in {"provider_error", "error"}) stale = sum(1 for row in response.results if str(row.get("status")) in {"stale", "missing"}) valuation_issues = [ warning for warning in response.warnings if warning in {"portfolio_valuation_partial", "portfolio_valuation_failed"} ] return SourceRunResult( stale_candidates=candidates, updated_count=int(response.economic_updated), fresh_unchanged_count=int(response.cached + response.updated - response.economic_updated), stale_remaining_count=stale, failed_count=max(failed, len(response.errors)) + len(valuation_issues), diagnostics=tuple(sorted(set([*response.errors, *valuation_issues])))[:10], ) def _crypto_source(conn: Connection, stale_before: str) -> SourceRunResult: held = conn.execute( """SELECT a.asset_id,a.coin_name,a.symbol,a.coingecko_id,a.notes 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 ) ORDER BY a.asset_id""" ).fetchall() def has_fresh(asset_id: str, currency: str) -> bool: return bool( conn.execute( """SELECT 1 FROM crypto_prices WHERE asset_id=? AND price_currency=? AND provider='CoinGecko' AND COALESCE(provider_timestamp,fetched_at)>=? AND quality_status='fresh' AND price IS NOT NULL AND price!='' LIMIT 1""", (asset_id, currency, stale_before), ).fetchone() ) stale = [ row for row in held if not row["coingecko_id"] or not has_fresh(str(row["asset_id"]), "CHF") or not has_fresh(str(row["asset_id"]), "USD") ] if not stale: return SourceRunResult(fresh_unchanged_count=len(held)) ids = [str(row["coingecko_id"]) for row in stale if row["coingecko_id"]] requested_ids = [*ids, *([] if "tether" in ids else ["tether"])] provider = CoinGeckoClient() if hasattr(provider, "get_crypto_market_bundle"): bundle = provider.get_crypto_market_bundle(requested_ids) else: # small compatibility seam for existing test providers fallback = { currency: provider.get_crypto_prices(requested_ids, currency) for currency in ("CHF", "USD") } bundle = { provider_id: { currency: fallback[currency].get( provider_id, PriceQuote(provider_id, currency, None, quality_status="missing"), ) for currency in ("CHF", "USD") } for provider_id in requested_ids } tether = (bundle.get("tether") or {}).get("USD") updated = unchanged = stale_remaining = failed = 0 diagnostics: list[str] = [] now = datetime.now(UTC) cutoff = datetime.fromisoformat(stale_before.replace("Z", "+00:00")) if cutoff.tzinfo is None: cutoff = cutoff.replace(tzinfo=UTC) def parsed(value: str | None) -> datetime | None: if not value: return None try: result = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None return result if result.tzinfo else result.replace(tzinfo=UTC) def quote_is_fresh(quote: PriceQuote) -> bool: observed = parsed(quote.provider_timestamp) return bool( quote.price is not None and quote.quality_status == "fresh" and observed is not None and cutoff <= observed <= now + timedelta(minutes=5) ) for asset in stale: asset_id = str(asset["asset_id"]) provider_id = str(asset["coingecko_id"] or "") if not provider_id: stale_remaining += 1 failed += 1 diagnostics.append(f"{asset['symbol']}: missing_coingecko_id") continue quotes = bundle.get(provider_id) or {} chf = quotes.get("CHF") or PriceQuote(provider_id, "CHF", None, quality_status="missing") usd = quotes.get("USD") or PriceQuote(provider_id, "USD", None, quality_status="missing") changed = False for quote in (chf, usd): if not quote_is_fresh(quote): continue before = conn.execute( "SELECT COUNT(*) FROM crypto_prices WHERE asset_id=? AND price_currency=?", (asset_id, quote.currency.upper()), ).fetchone()[0] store_crypto_price(conn, asset_id=asset_id, quote=quote) after = conn.execute( "SELECT COUNT(*) FROM crypto_prices WHERE asset_id=? AND price_currency=?", (asset_id, quote.currency.upper()), ).fetchone()[0] changed = changed or after > before upsert_crypto_price_point( conn, asset_id=asset_id, timestamp=quote.provider_timestamp or _now(), price=quote.price, currency=quote.currency, provider=quote.provider, provider_symbol=provider_id, ) if not quote_is_fresh(chf): stale_remaining += 1 failed += 1 diagnostics.append(f"{asset['symbol']}: CHF price unavailable") coin_ts = parsed(usd.provider_timestamp) tether_ts = parsed(tether.provider_timestamp if tether else None) compatible = bool( quote_is_fresh(usd) and tether and quote_is_fresh(tether) and tether.price is not None and tether.price > Decimal("0") and coin_ts and tether_ts and abs((coin_ts - tether_ts).total_seconds()) <= 300 and (now - coin_ts).total_seconds() <= 86_400 and (now - tether_ts).total_seconds() <= 86_400 and coin_ts <= now + timedelta(minutes=5) and tether_ts <= now + timedelta(minutes=5) ) try: notes = json.loads(str(asset["notes"] or "{}")) except json.JSONDecodeError: notes = {} if not isinstance(notes, dict): notes = {} info: dict[str, Any] = ( dict(notes["coingecko"]) if isinstance(notes.get("coingecko"), dict) else {} ) if compatible: assert usd.price is not None and tether and tether.price is not None usdt = usd.price / tether.price info.update( { "price_usd": format(usd.price, "f"), "price_usdt": format(usdt, "f"), "change_24h_pct": format(usd.change_24h_pct, "f") if usd.change_24h_pct is not None else None, "high_24h_usdt": format(usd.high_24h / tether.price, "f") if usd.high_24h is not None else None, "low_24h_usdt": format(usd.low_24h / tether.price, "f") if usd.low_24h is not None else None, "price_provider_timestamp": usd.provider_timestamp, "usdt_reference_usd": format(tether.price, "f"), "usdt_reference_timestamp": tether.provider_timestamp, "usdt_quality_status": "fresh", "usdt_source": "CoinGecko · USDT aus USD/Tether abgeleitet", } ) else: info.update( { "price_usd": format(usd.price, "f") if usd.price is not None else None, "price_usdt": None, "change_24h_pct": format(usd.change_24h_pct, "f") if usd.change_24h_pct is not None else None, "high_24h_usdt": None, "low_24h_usdt": None, "price_provider_timestamp": usd.provider_timestamp, "usdt_reference_usd": None, "usdt_reference_timestamp": None, "usdt_quality_status": "stale_reference", "usdt_source": "CoinGecko · USDT aus USD/Tether abgeleitet", } ) diagnostics.append(f"{asset['symbol']}: optional USDT reference unavailable or incompatible") notes["coingecko"] = info encoded = json.dumps(notes, sort_keys=True, separators=(",", ":")) if encoded != str(asset["notes"] or ""): conn.execute("UPDATE crypto_assets SET notes=? WHERE asset_id=?", (encoded, asset_id)) changed = True if changed: updated += 1 elif quote_is_fresh(chf): unchanged += 1 conn.commit() return SourceRunResult( stale_candidates=len(stale), updated_count=updated, fresh_unchanged_count=max(0, len(held) - len(stale)) + unchanged, stale_remaining_count=stale_remaining, failed_count=failed, diagnostics=tuple(diagnostics[:10]), ) def _fx_source(conn: Connection, stale_before: str) -> SourceRunResult: from jarvis_finance.fx.providers import FrankfurterFxProvider, TwelveDataFxProvider from jarvis_finance.fx.rates import resolve_fx_rate_to_chf cutoff_date = stale_before[:10] all_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' ORDER BY currency""" ).fetchall() ] 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 provider_unchanged = 0 failures = 0 for currency in currencies: try: before = conn.execute( """SELECT rate_date,rate,provider,rate_type,quality_status FROM fx_rates WHERE base_currency=? AND quote_currency='CHF' ORDER BY rate_date DESC,COALESCE(fetched_at,created_at) DESC LIMIT 1""", (currency,), ).fetchone() before_economic = tuple(before) if before is not None else None result = resolve_fx_rate_to_chf( conn, base_currency=currency, rate_date=None, providers=[FrankfurterFxProvider(), TwelveDataFxProvider()], persist=True, resolve_fixed=True, ) after = conn.execute( """SELECT rate_date,rate,provider,rate_type,quality_status FROM fx_rates WHERE base_currency=? AND quote_currency='CHF' ORDER BY rate_date DESC,COALESCE(fetched_at,created_at) DESC LIMIT 1""", (currency,), ).fetchone() after_economic = tuple(after) if after is not None else None if result.status == "ok" and after_economic != before_economic: updated += 1 elif result.status == "ok": provider_unchanged += 1 except Exception: failures += 1 conn.commit() return SourceRunResult( stale_candidates=len(currencies), updated_count=updated, fresh_unchanged_count=max(0, len(all_currencies) - len(currencies)) + provider_unchanged, stale_remaining_count=failures, failed_count=failures, ) DEFAULT_RUNNERS: dict[str, Callable[[Connection, str], SourceRunResult | tuple[int, int]]] = { "equity": _equity_source, "crypto": _crypto_source, "fx": _fx_source, } class _CachedFxProvider: """Read persisted FX only; final recalculation must not make a second provider pass.""" name = "asset_refresh_cache" def __init__(self, conn: Connection) -> None: self.conn = conn def get_rate(self, base_currency: str, quote_currency: str, rate_date: str | None = None) -> Decimal | None: row = self.conn.execute( """SELECT rate FROM fx_rates WHERE base_currency=? AND quote_currency=? AND (? IS NULL OR rate_date=?) ORDER BY rate_date DESC,COALESCE(fetched_at,created_at) DESC LIMIT 1""", (base_currency.upper(), quote_currency.upper(), rate_date, rate_date), ).fetchone() return Decimal(str(row["rate"])) if row else None class _CachedEquityProvider: """Expose persisted market observations through the valuation provider contract.""" def __init__(self, conn: Connection, name: str) -> None: self.conn = conn self.name = name def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote: row = self.conn.execute( """SELECT provider_symbol,currency,close,provider,provider_market,price_timestamp, adjusted_close,quality_status,error_message,price_date FROM market_prices WHERE provider_symbol=? AND (? IS NULL OR price_date=?) ORDER BY price_date DESC,COALESCE(fetched_at,created_at) DESC LIMIT 1""", (provider_symbol, price_date, price_date), ).fetchone() if not row: return EquityPriceQuote( provider_symbol=provider_symbol, currency="", close=None, provider=self.name, quality_status="missing", error_message="cached_price_missing", ) return EquityPriceQuote( provider_symbol=str(row["provider_symbol"]), currency=str(row["currency"]), close=Decimal(str(row["close"])), provider=str(row["provider"]), provider_market=str(row["provider_market"] or "") or None, price_timestamp=str(row["price_timestamp"] or row["price_date"]), adjusted_close=Decimal(str(row["adjusted_close"])) if row["adjusted_close"] is not None else None, quality_status=str(row["quality_status"] or "fresh"), error_message=str(row["error_message"]) if row["error_message"] else None, ) def _recalculate_cached_valuations( conn: Connection, source_results: dict[str, SourceRunResult], ) -> dict[str, tuple[str, ...]]: """Recalculate from just-persisted observations without further provider calls.""" issues: dict[str, tuple[str, ...]] = {} equity = source_results.get("equity", SourceRunResult()) fx = source_results.get("fx", SourceRunResult()) if equity.updated_count or fx.updated_count: try: provider_names = { str(row[0]) for row in conn.execute( """SELECT DISTINCT provider FROM instrument_price_mappings WHERE mapping_status='mapped' AND provider IS NOT NULL""" ).fetchall() } cached_providers: dict[str, EquityPriceProvider] = { name: _CachedEquityProvider(conn, name) for name in provider_names } market = run_daily_market_valuation( conn, price_providers=cached_providers, fx_provider=_CachedFxProvider(conn), ) if market.status != "complete": issues["equity"] = tuple(market.reason_codes[:10]) or ("portfolio_valuation_partial",) except Exception as exc: issues["equity"] = (type(exc).__name__,) if source_results.get("crypto", SourceRunResult()).updated_count: try: crypto = run_daily_crypto_valuation(conn) if crypto.status != "complete": issues["crypto"] = tuple(crypto.reason_codes[:10]) or ("crypto_valuation_partial",) except Exception as exc: issues["crypto"] = (type(exc).__name__,) return issues def _has_unmaterialized_price_updates(conn: Connection) -> bool: """Detect a prior interrupted refresh whose observations lack a wealth snapshot.""" last_snapshot = conn.execute( "SELECT MAX(captured_at) FROM aggregated_wealth_refresh_snapshots" ).fetchone()[0] row = conn.execute( """SELECT 1 FROM asset_price_refresh_jobs j JOIN asset_price_refresh_sources s ON s.job_id=j.job_id WHERE j.status='failed' AND j.wealth_snapshot_id IS NULL AND s.updated_count>0 AND (? IS NULL OR j.completed_at>?) LIMIT 1""", (last_snapshot, last_snapshot), ).fetchone() return row is not None def run_asset_price_refresh( db_path: str, job_id: str, *, runners: dict[str, Callable[[Connection, str], SourceRunResult | 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 total_updated = 0 total_fresh_unchanged = 0 total_candidates = 0 source_results: dict[str, SourceRunResult] = {} 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() result = SourceRunResult() status = "complete" error_code = None try: result = _source_result(selected[source](conn, stale_before)) if result.stale_candidates == 0 and result.fresh_unchanged_count == 0: status = "skipped" if result.failed_count: failures += 1 status = "failed" error_code = f"{source}_instrument_failures" elif result.stale_remaining_count: failures += 1 status = "failed" error_code = f"{source}_stale_remaining" except Exception as exc: if conn.in_transaction: conn.rollback() status = "failed" failures += 1 result = SourceRunResult(failed_count=1, diagnostics=(type(exc).__name__,)) error_code = f"{source}_refresh_failed" source_results[source] = result total_updated += result.updated_count total_fresh_unchanged += result.fresh_unchanged_count total_candidates += result.stale_candidates completed += 1 conn.execute( """UPDATE asset_price_refresh_sources SET status=?,stale_candidates=?,updated_count=?,error_code=?,completed_at=?, fresh_unchanged_count=?,stale_remaining_count=?,failed_count=?,diagnostics_json=? WHERE job_id=? AND source=?""", ( status, result.stale_candidates, result.updated_count, error_code, _now(), result.fresh_unchanged_count, result.stale_remaining_count, result.failed_count, json.dumps(list(result.diagnostics)), job_id, source, ), ) conn.execute( "UPDATE asset_price_refresh_jobs SET progress_completed=? WHERE job_id=?", (completed, job_id), ) conn.commit() if runners is None and total_updated > 0: for source, diagnostics in _recalculate_cached_valuations(conn, source_results).items(): current = conn.execute( "SELECT status,diagnostics_json FROM asset_price_refresh_sources WHERE job_id=? AND source=?", (job_id, source), ).fetchone() if current and str(current["status"]) != "failed": failures += 1 existing_diagnostics = json.loads(str(current["diagnostics_json"] or "[]")) conn.execute( """UPDATE asset_price_refresh_sources SET error_code=?,diagnostics_json=? WHERE job_id=? AND source=?""", ( f"{source}_valuation_partial", json.dumps([*existing_diagnostics, *diagnostics][:10]), job_id, source, ), ) conn.commit() if _protected_fingerprint(conn) != protected_before: raise RuntimeError("protected_holdings_or_transactions_mutated") usable_result = total_updated > 0 or total_fresh_unchanged > 0 or (total_candidates == 0 and failures == 0) wealth_snapshot_id = None snapshot_needed = total_updated > 0 or _has_unmaterialized_price_updates(conn) if snapshot_needed: 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, fresh_unchanged_count,stale_remaining_count,failed_count 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 "partial" if usable_result else "failed" 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) __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23.3-hotfix__HERMES_CWD_8d46a20096ed__