from __future__ import annotations import hashlib import json import os from datetime import date, datetime, time, timedelta from pathlib import Path from sqlite3 import Connection from typing import Any from zoneinfo import ZoneInfo from jarvis_finance.audit.log import record_audit_event from jarvis_finance.imports.common import utc_now from jarvis_finance.market.providers import MarketDataProvider from jarvis_finance.services.crypto_market_recovery import ( is_crypto_market_source_activated as _crypto_market_source_enabled, ) from jarvis_finance.services.daily_valuations import run_daily_crypto_valuation from jarvis_finance.services.performance_hardening import ( build_postfinance_component_preview, preview_performance_reclassification, ) from jarvis_finance.services.performance_scope import ( set_performance_cashflow_coverage, set_performance_scope_classification, ) from jarvis_finance.services.portfolio_analytics import run_daily_market_valuation SOURCES = {"postfinance", "truewealth", "crypto"} ROLE_BY_SOURCE = { "postfinance": ("postfinance_etrading_depot", "postfinance_etrading_cash"), "truewealth": ("canonical_truewealth_total_value",), "crypto": ("crypto_portfolio",), } def build_daily_valuation_job_status(conn: Connection) -> dict[str, Any]: enabled = os.environ.get("JARVIS_FINANCE_DAILY_VALUATION_ENABLED") == "1" sources: list[dict[str, Any]] = [] source_specs = ( ("daily_market_fx_v4", "PostFinance / Marktpreise / FX", False), ("daily_crypto_valuation_v1", "Krypto-Performance (historischer Pfad)", False), ("daily_crypto_current_valuation_v1", "Krypto", _crypto_market_source_enabled(conn)), ) for source_key, label, source_enabled in source_specs: row = conn.execute( """SELECT as_of,status,started_at,completed_at,reason_codes_json, price_total,price_stored,valuation_stored,missing_instruments_json FROM market_data_runs WHERE source_key=? ORDER BY COALESCE(completed_at,started_at) DESC,run_id DESC LIMIT 1""", (source_key,), ).fetchone() last_complete = conn.execute( """SELECT as_of,started_at,completed_at,price_total,price_stored,valuation_stored FROM market_data_runs WHERE source_key=? AND status='complete' ORDER BY as_of DESC,COALESCE(completed_at,started_at) DESC,run_id DESC LIMIT 1""", (source_key,), ).fetchone() operational_status = "paused" next_action = "Quelle kontrolliert aktivieren" if enabled and source_enabled: if row and str(row["status"]) == "complete": operational_status = "active" next_action = "Keine Aktion erforderlich" elif row and str(row["status"]) == "partial": operational_status = "partial" next_action = "Fehlende Kurszuordnungen oder Preise prüfen" elif row and str(row["status"]) in {"failed", "blocked", "unavailable"}: operational_status = "failed" next_action = "Fehlerstatus prüfen und kontrollierten Lauf wiederholen" else: operational_status = "active" next_action = "Ersten abgesicherten Krypto-Lauf ausführen" elif source_enabled: next_action = "Globales Tagesjob-Gate und Timer kontrolliert aktivieren" price_as_of = None price_age_seconds = None total_value_chf = None price_provider = None price_currency = None fx_provider = None if source_key == "daily_crypto_current_valuation_v1" and last_complete: price_row = conn.execute( """SELECT MIN(provider_timestamp) AS oldest_provider_timestamp, COUNT(*) AS price_count, COUNT(provider_timestamp) AS timestamp_count, COUNT(provider) AS provider_count, COUNT(price_currency) AS currency_count, MIN(provider) AS min_provider,MAX(provider) AS max_provider, MIN(price_currency) AS min_currency,MAX(price_currency) AS max_currency FROM crypto_prices WHERE fetched_at=? AND quality_status='fresh' AND provider='CoinGecko'""", (last_complete["started_at"],), ).fetchone() expected_price_count = int(last_complete["price_stored"] or 0) price_rows_complete = bool( price_row and int(price_row["price_count"] or 0) == expected_price_count and int(price_row["timestamp_count"] or 0) == expected_price_count and int(price_row["provider_count"] or 0) == expected_price_count and int(price_row["currency_count"] or 0) == expected_price_count and expected_price_count > 0 ) price_as_of = ( str(price_row["oldest_provider_timestamp"]) if price_rows_complete and price_row["oldest_provider_timestamp"] else None ) if price_rows_complete and price_row["min_provider"] == price_row["max_provider"]: price_provider = str(price_row["min_provider"]) if price_rows_complete and price_row["min_currency"] == price_row["max_currency"]: price_currency = str(price_row["min_currency"]) if price_currency == "CHF": fx_provider = "Direkte CHF-Notierung (kein FX-Lauf)" if price_as_of: try: parsed = datetime.fromisoformat(price_as_of) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=ZoneInfo("UTC")) price_age_seconds = max(0, int((datetime.now(ZoneInfo("UTC")) - parsed).total_seconds())) except ValueError: price_age_seconds = None value_row = conn.execute( """SELECT value_original FROM portfolio_valuation_snapshots WHERE source=? AND substr(valuation_at,1,10)=? ORDER BY captured_at DESC,snapshot_id DESC LIMIT 1""", (source_key, str(last_complete["as_of"])), ).fetchone() total_value_chf = str(value_row["value_original"]) if value_row else None missing_items = json.loads(row["missing_instruments_json"] or "[]") if row else [] missing_assets = len(missing_items) valued_assets = max(0, int(row["price_total"] or 0) - missing_assets) if row else 0 sources.append( { "source_key": source_key, "label": label, "source_enabled": bool(source_enabled), "operational_status": operational_status, "last_confirmed_date": str(last_complete["as_of"]) if last_complete else None, "last_successful_at": str(last_complete["completed_at"] or last_complete["started_at"]) if last_complete else None, "last_run_at": str(row["completed_at"] or row["started_at"]) if row else None, "status": str(row["status"]) if row else "never_run", "reason_codes": json.loads(row["reason_codes_json"] or "[]") if row else [], "price_as_of": price_as_of, "price_age_seconds": price_age_seconds, "price_provider": price_provider, "price_currency": price_currency, "fx_provider": fx_provider, "valued_assets": valued_assets, "missing_assets": missing_assets, "total_value_chf": total_value_chf, "next_action": next_action, } ) next_run = None if enabled: now = datetime.now(ZoneInfo("Europe/Zurich")) candidate = datetime.combine(now.date(), time(23, 30), tzinfo=now.tzinfo) if candidate <= now: candidate += timedelta(days=1) next_run = candidate.isoformat() return { "job_key": "canonical_daily_investment_valuation_v1", "configured": True, "enabled": enabled, "activation_required": not enabled, "schedule": "daily 23:30 Europe/Zurich", "next_run": next_run, "sources": sources, } def _days(start: date, end: date, *, business_only: bool) -> list[str]: result: list[str] = [] current = start while current <= end: if not business_only or current.weekday() < 5: result.append(current.isoformat()) current += timedelta(days=1) return result def preview_performance_source_activation( conn: Connection, *, source: str, tracking_mode: str, period_from: str, period_to: str, evidence_reference: str, confirmed_start_date: str | None = None, attest_complete_external_flows: bool = False, ) -> dict[str, Any]: if source not in SOURCES: raise ValueError("Unsupported performance source") allowed = { "postfinance": {"transactions"}, "truewealth": {"managed_total_value"}, "crypto": {"transactions", "confirmed_start_snapshot"}, } if tracking_mode not in allowed[source]: raise ValueError("Tracking mode is not approved for this source") try: start, end = date.fromisoformat(period_from), date.fromisoformat(period_to) except ValueError as exc: raise ValueError("Activation requires ISO dates") from exc if start > end: raise ValueError("Activation start must not follow its end") blockers: list[str] = [] if not evidence_reference.strip(): blockers.append("evidence_reference_required") if not attest_complete_external_flows: blockers.append("external_flow_attestation_required") account_ids = _accounts(conn, source) create_crypto_scope = source == "crypto" and not account_ids if source != "crypto" and not account_ids: blockers.append(f"{source}_scope_missing") if source == "postfinance" and len(account_ids) != 2: blockers.append("postfinance_component_scope_incomplete") evidence: dict[str, Any] = {} if source == "postfinance" and account_ids: placeholders = ",".join("?" for _ in account_ids) row = conn.execute( f"""SELECT MIN(trade_date),MAX(trade_date),COUNT(*) FROM transactions WHERE account_id IN ({placeholders}) AND COALESCE(is_confirmed,0)=1 AND COALESCE(is_voided,0)=0""", account_ids, ).fetchone() evidence = {"first_activity": row[0], "last_activity": row[1], "activity_count": int(row[2])} if not row[0] or str(row[0]) > period_from: blockers.append("postfinance_history_starts_too_late") reclassification = preview_performance_reclassification( conn, source="postfinance", period_from=period_from, period_to=period_to, ) evidence["reclassification"] = reclassification if not reclassification["can_activate"]: blockers.append("postfinance_performance_classification_unclear") latest_official = conn.execute( """SELECT substr(snapshot_at,1,10) FROM postfinance_snapshots WHERE substr(snapshot_at,1,10)<=? ORDER BY snapshot_at DESC,created_at DESC,snapshot_id DESC LIMIT 1""", (period_to,), ).fetchone() if latest_official: evidence["component_provenance"] = build_postfinance_component_preview( conn, day=str(latest_official[0]) ) if evidence["component_provenance"]["missing_roles"]: blockers.append("postfinance_component_scope_incomplete") elif source == "truewealth" and account_ids: placeholders = ",".join("?" for _ in account_ids) row = conn.execute( f"""SELECT MIN(valuation_date),MAX(valuation_date),COUNT(*) FROM account_value_snapshots WHERE account_id IN ({placeholders}) AND updated_at IS NULL AND COALESCE(is_active,1)=1 AND source_type<>'truewealth_manual_provisional'""", account_ids, ).fetchone() evidence = {"first_value": row[0], "last_value": row[1], "value_count": int(row[2])} if not row[0] or str(row[0]) > period_from: blockers.append("truewealth_opening_value_missing") elif source == "crypto": if tracking_mode == "confirmed_start_snapshot": if confirmed_start_date != period_from: blockers.append("crypto_confirmed_start_must_equal_period_start") rows = conn.execute( """SELECT verification_status, COALESCE(substr(last_verified_at,1,10),legacy_snapshot_date) verified_date FROM crypto_holdings ORDER BY crypto_holding_id""" ).fetchall() evidence = {"holding_count": len(rows), "confirmed_start_date": confirmed_start_date} if not rows or any(row["verification_status"] != "verified" or row["verified_date"] != confirmed_start_date for row in rows): blockers.append("confirmed_crypto_start_balance_missing") else: row = conn.execute( """SELECT MIN(substr(transaction_datetime,1,10)),MAX(substr(transaction_datetime,1,10)),COUNT(*) FROM crypto_transactions WHERE confirmation_status='confirmed'""" ).fetchone() evidence = {"first_activity": row[0], "last_activity": row[1], "activity_count": int(row[2])} if not row[0] or str(row[0]) > period_from: blockers.append("crypto_history_starts_too_late") state = { "source": source, "tracking_mode": tracking_mode, "period_from": period_from, "period_to": period_to, "evidence_reference": evidence_reference.strip(), "confirmed_start_date": confirmed_start_date, "attest_complete_external_flows": attest_complete_external_flows, "account_ids": account_ids, "create_crypto_scope": create_crypto_scope, "evidence": evidence, "local_input_revision": _local_backfill_input_revision( conn, source=source, account_ids=account_ids, period_from=period_from, period_to=period_to, ), "blockers": sorted(set(blockers)), } fingerprint = hashlib.sha256(json.dumps(state, sort_keys=True, separators=(",", ":")).encode()).hexdigest() return { **state, "preview_id": f"performance-activation-{fingerprint[:24]}", "input_fingerprint": fingerprint, "preview_created_at": utc_now(), "can_confirm": not blockers, "planned_changes": { "create_crypto_scope": create_crypto_scope, "scope_classifications": 1 if create_crypto_scope else len(account_ids), "cashflow_coverage_records": 1 if create_crypto_scope else len(account_ids), "financial_snapshots": 0, }, } def confirm_performance_source_activation(conn: Connection, request: dict[str, Any]) -> dict[str, Any]: if request.get("confirm") is not True: raise ValueError("Explicit performance activation confirmation is required") confirmation_id = str(request.get("confirmation_id") or "").strip() if not confirmation_id: raise ValueError("confirmation_id is required") prior = conn.execute( """SELECT audit_id FROM audit_log WHERE action='performance_source_activation_confirmed' AND entity_id=? ORDER BY created_at DESC LIMIT 1""", (confirmation_id,), ).fetchone() if prior: return {"confirmation_id": confirmation_id, "idempotent": True, "audit_id": str(prior["audit_id"]), "changed": False} current = preview_performance_source_activation( conn, source=str(request.get("source") or ""), tracking_mode=str(request.get("tracking_mode") or ""), period_from=str(request.get("period_from") or ""), period_to=str(request.get("period_to") or ""), evidence_reference=str(request.get("evidence_reference") or ""), confirmed_start_date=(str(request["confirmed_start_date"]) if request.get("confirmed_start_date") else None), attest_complete_external_flows=request.get("attest_complete_external_flows") is True, ) if request.get("preview_id") != current["preview_id"] or request.get("input_fingerprint") != current["input_fingerprint"]: raise ValueError("Performance activation preview is stale; create a new preview") if not current["can_confirm"]: raise ValueError("Performance activation preview has blockers") source = str(request["source"]) account_ids = list(current["account_ids"]) changed = False if current["create_crypto_scope"]: platform_id = "performance-crypto-platform-v1" account_id = "performance-crypto-account-v1" now = utc_now() if not conn.execute("SELECT 1 FROM platforms WHERE platform_id=?", (platform_id,)).fetchone(): conn.execute( """INSERT INTO platforms(platform_id,name,platform_type,default_currency,created_at) VALUES(?,?,'crypto','CHF',?)""", (platform_id, "Crypto Performance", now), ) if not conn.execute("SELECT 1 FROM accounts WHERE account_id=?", (account_id,)).fetchone(): conn.execute( """INSERT INTO accounts( account_id,platform_id,account_name,account_type,currency,is_active,created_at ) VALUES(?,?,'Crypto Performance','brokerage','CHF',1,?)""", (account_id, platform_id, now), ) account_ids = [account_id] role = { "postfinance": "postfinance_etrading_depot", "truewealth": "canonical_truewealth_total_value", "crypto": "crypto_portfolio", }[source] coverage_source = { "transactions": "confirmed_transaction_history_v1", "managed_total_value": "confirmed_managed_total_value_v1", "confirmed_start_snapshot": "confirmed_start_snapshot_v1", }[str(request["tracking_mode"])] for account_id in account_ids: current_role = conn.execute( "SELECT classification_role FROM performance_scope_classifications WHERE account_id=?", (account_id,), ).fetchone() target_role = str(current_role[0]) if current_role else role changed = set_performance_scope_classification( conn, account_id=account_id, included=True, classification_role=target_role, source="performance_activation_v1", note=str(request["evidence_reference"]), ) or changed changed = set_performance_cashflow_coverage( conn, account_id=account_id, coverage_from=str(request["period_from"]), coverage_to=str(request["period_to"]), status="complete", source=coverage_source, note=str(request["evidence_reference"]), ) or changed audit_id = record_audit_event( conn, source="performance_activation_v1", action="performance_source_activation_confirmed", entity_type="performance_source_activation", entity_id=confirmation_id, old_values={}, new_values={ "source": source, "tracking_mode": request["tracking_mode"], "period_from": request["period_from"], "period_to": request["period_to"], "input_fingerprint": current["input_fingerprint"], "account_count": len(account_ids), "financial_snapshots_written": 0, }, user_text_note=str(request["evidence_reference"]), created_by="user", ) conn.commit() return {"confirmation_id": confirmation_id, "idempotent": False, "audit_id": audit_id, "changed": changed} def _accounts(conn: Connection, source: str) -> list[str]: roles = ROLE_BY_SOURCE[source] placeholders = ",".join("?" for _ in roles) return [ str(row[0]) for row in conn.execute( f"""SELECT account_id FROM performance_scope_classifications WHERE included=1 AND decision_version='investment_performance_scope_v1' AND classification_role IN ({placeholders}) ORDER BY account_id""", roles, ).fetchall() ] def _coverage(conn: Connection, account_ids: list[str], start: str, end: str) -> list[dict[str, Any]]: if not account_ids: return [] placeholders = ",".join("?" for _ in account_ids) return [ dict(row) for row in conn.execute( f"""SELECT account_id,coverage_from,coverage_to,status,source,audit_id,recorded_at FROM performance_cashflow_coverage WHERE account_id IN ({placeholders}) AND coverage_from<=? AND coverage_to>=? ORDER BY account_id""", (*account_ids, start, end), ).fetchall() ] def _complete_valuation_days(conn: Connection, account_ids: list[str], start: str, end: str) -> set[str]: if not account_ids: return set() placeholders = ",".join("?" for _ in account_ids) rows = conn.execute( f"""WITH ranked AS ( SELECT scope_id,substr(valuation_at,1,10) AS day,source,quality_status, ROW_NUMBER() OVER ( PARTITION BY scope_id,substr(valuation_at,1,10),source ORDER BY snapshot_version DESC,captured_at DESC,snapshot_id DESC ) rn FROM portfolio_valuation_snapshots WHERE scope_kind='account' AND scope_id IN ({placeholders}) AND substr(valuation_at,1,10) BETWEEN ? AND ? ) SELECT day,source,COUNT(DISTINCT scope_id) AS covered FROM ranked WHERE rn=1 AND quality_status='complete' GROUP BY day,source HAVING covered=?""", (*account_ids, start, end, len(account_ids)), ).fetchall() days = {str(row["day"]) for row in rows} role_rows = conn.execute( """SELECT classification_role,account_id FROM performance_scope_classifications WHERE included=1 AND decision_version='investment_performance_scope_v1' AND classification_role IN ('postfinance_etrading_depot','postfinance_etrading_cash')""" ).fetchall() role_accounts = {str(row["classification_role"]): str(row["account_id"]) for row in role_rows} postfinance_accounts = { role_accounts.get("postfinance_etrading_depot"), role_accounts.get("postfinance_etrading_cash"), } if None not in postfinance_accounts and postfinance_accounts.issubset(set(account_ids)): days.update( str(row[0]) for row in conn.execute( """SELECT DISTINCT substr(snapshot_at,1,10) FROM postfinance_snapshots WHERE substr(snapshot_at,1,10) BETWEEN ? AND ? AND reconciliation_status IN ('matched','complete','verified')""", (start, end), ).fetchall() ) return days def _local_backfill_input_revision( conn: Connection, *, source: str, account_ids: list[str], period_from: str, period_to: str, ) -> str: hasher = hashlib.sha256() def add(query: str, params: tuple[Any, ...] = ()) -> None: for row in conn.execute(query, params).fetchall(): hasher.update(json.dumps(list(row), default=str, separators=(",", ":")).encode()) if account_ids: placeholders = ",".join("?" for _ in account_ids) add( f"""SELECT * FROM accounts WHERE account_id IN ({placeholders}) ORDER BY account_id""", tuple(account_ids), ) add( f"""SELECT * FROM performance_scope_classifications WHERE account_id IN ({placeholders}) ORDER BY account_id""", tuple(account_ids), ) add( f"""SELECT * FROM performance_cashflow_coverage WHERE account_id IN ({placeholders}) ORDER BY account_id""", tuple(account_ids), ) add( f"""SELECT * FROM portfolio_valuation_snapshots WHERE scope_kind='account' AND scope_id IN ({placeholders}) AND substr(valuation_at,1,10) BETWEEN ? AND ? ORDER BY snapshot_id""", (*account_ids, period_from, period_to), ) if source == "postfinance" and account_ids: placeholders = ",".join("?" for _ in account_ids) add("SELECT * FROM instruments ORDER BY instrument_id") add("SELECT * FROM instrument_price_mappings ORDER BY mapping_id") add( f"""SELECT * FROM transactions WHERE account_id IN ({placeholders}) AND trade_date<=? ORDER BY transaction_id""", (*account_ids, period_to), ) add( f"""SELECT * FROM positions_snapshot WHERE account_id IN ({placeholders}) AND snapshot_date<=? ORDER BY position_snapshot_id""", (*account_ids, period_to), ) add( f"""SELECT * FROM cash_account_snapshots WHERE account_id IN ({placeholders}) AND balance_date<=? ORDER BY snapshot_id""", (*account_ids, period_to), ) add("SELECT * FROM market_prices WHERE price_date BETWEEN ? AND ? ORDER BY market_price_id", (period_from, period_to)) add("SELECT * FROM fx_rates WHERE rate_date BETWEEN ? AND ? ORDER BY fx_rate_id", (period_from, period_to)) elif source == "crypto": add("SELECT * FROM crypto_holdings ORDER BY crypto_holding_id") add( """SELECT * FROM crypto_transactions WHERE confirmation_status='confirmed' AND substr(transaction_datetime,1,10)<=? ORDER BY crypto_transaction_id""", (period_to,), ) add( """SELECT * FROM crypto_prices WHERE substr(COALESCE(provider_timestamp,fetched_at),1,10) BETWEEN ? AND ? ORDER BY crypto_price_id""", (period_from, period_to), ) elif source == "truewealth" and account_ids: placeholders = ",".join("?" for _ in account_ids) add( f"""SELECT * FROM account_value_snapshots WHERE account_id IN ({placeholders}) AND valuation_date BETWEEN ? AND ? ORDER BY snapshot_id""", (*account_ids, period_from, period_to), ) return hasher.hexdigest() def preview_performance_backfill( conn: Connection, *, source: str, period_from: str, period_to: str, ) -> dict[str, Any]: if source not in SOURCES: raise ValueError("Unsupported performance source") try: start, end = date.fromisoformat(period_from), date.fromisoformat(period_to) except ValueError as exc: raise ValueError("Backfill requires ISO dates") from exc if start > end: raise ValueError("Backfill start must not follow its end") if (end - start).days > 366: raise ValueError("Backfill preview is limited to 367 calendar days") account_ids = _accounts(conn, source) coverage = _coverage(conn, account_ids, period_from, period_to) candidate_days = _days(start, end, business_only=source == "postfinance") complete_days = _complete_valuation_days(conn, account_ids, period_from, period_to) missing_days = [day for day in candidate_days if day not in complete_days] blockers: list[str] = [] if not account_ids: blockers.append(f"{source}_activation_required") if len(coverage) != len(account_ids) or any(row["status"] != "complete" for row in coverage): blockers.append(f"{source}_cashflow_history_incomplete") if source == "truewealth" and missing_days: blockers.append("truewealth_requires_new_confirmed_values") state = { "source": source, "period_from": period_from, "period_to": period_to, "account_ids": account_ids, "coverage": coverage, "candidate_days": candidate_days, "complete_days": sorted(complete_days), "missing_days": missing_days, "local_input_revision": _local_backfill_input_revision( conn, source=source, account_ids=account_ids, period_from=period_from, period_to=period_to, ), "blockers": sorted(set(blockers)), } fingerprint = hashlib.sha256( json.dumps(state, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() return { "preview_id": f"performance-backfill-{fingerprint[:24]}", "input_fingerprint": fingerprint, "preview_created_at": utc_now(), "source": source, "period_from": period_from, "period_to": period_to, "candidate_days": len(candidate_days), "already_complete_days": len(complete_days), "days_to_materialize": len(missing_days), "missing_from": missing_days[0] if missing_days else None, "missing_to": missing_days[-1] if missing_days else None, "blockers": sorted(set(blockers)), "can_confirm": not blockers and bool(missing_days), "required_input": { "postfinance": "Vollständige bestätigte PostFinance-Aktivitäten für den Zeitraum", "truewealth": "Bestätigte True-Wealth-Gesamtwerte; Werte werden nie künstlich fortgeschrieben", "crypto": "Vollständige Transaktionen oder bestätigter Startbestand plus exakte historische Tageskurse", }[source], } def confirm_performance_backfill( conn: Connection, request: dict[str, Any], *, price_providers: dict[str, Any] | None = None, fx_provider: Any | None = None, crypto_price_provider: MarketDataProvider | None = None, lock_dir: Path | None = None, ) -> dict[str, Any]: if request.get("confirm") is not True: raise ValueError("Explicit backfill confirmation is required") confirmation_id = str(request.get("confirmation_id") or "").strip() if not confirmation_id: raise ValueError("confirmation_id is required") source = str(request.get("source") or "") prior = conn.execute( """SELECT audit_id FROM audit_log WHERE action='performance_backfill_confirmed' AND entity_id=? ORDER BY created_at DESC LIMIT 1""", (confirmation_id,), ).fetchone() if prior: return { "confirmation_id": confirmation_id, "source": source, "idempotent": True, "runs": [], "audit_id": str(prior["audit_id"]), "completed": True, "remaining_days": 0, } attempted = conn.execute( """SELECT audit_id,new_values_json FROM audit_log WHERE action='performance_backfill_attempted' AND entity_id=? ORDER BY created_at DESC LIMIT 1""", (confirmation_id,), ).fetchone() attempted_values: dict[str, Any] = {} if attempted: attempted_values = json.loads(str(attempted["new_values_json"] or "{}")) if attempted_values.get("preview_id") != request.get("preview_id") or attempted_values.get("input_fingerprint") != request.get("input_fingerprint"): raise ValueError("confirmation_id was already used for a different backfill preview") current = preview_performance_backfill( conn, source=source, period_from=str(request.get("period_from") or ""), period_to=str(request.get("period_to") or ""), ) current_revision = _local_backfill_input_revision( conn, source=source, account_ids=_accounts(conn, source), period_from=str(request.get("period_from") or ""), period_to=str(request.get("period_to") or ""), ) if attempted and attempted_values.get("resume_input_revision") != current_revision: raise ValueError("Backfill inputs changed after the partial attempt; create a new preview") if not attempted and (request.get("preview_id") != current["preview_id"] or request.get("input_fingerprint") != current["input_fingerprint"]): raise ValueError("Backfill preview is stale; create a new preview") if current["blockers"] or (not current["can_confirm"] and current["days_to_materialize"] > 0): raise ValueError("Backfill preview has blockers or no missing days") missing = [] if current["missing_from"] and current["missing_to"]: start, end = date.fromisoformat(current["missing_from"]), date.fromisoformat(current["missing_to"]) missing = _days(start, end, business_only=source == "postfinance") lock_root = lock_dir or Path("/tmp") runs: list[dict[str, Any]] = [] for day in missing: if day in _complete_valuation_days(conn, _accounts(conn, source), day, day): continue if source == "postfinance": result = run_daily_market_valuation( conn, as_of=day, price_providers=price_providers, fx_provider=fx_provider, lock_path=lock_root / "performance-backfill-market.lock", ) elif source == "crypto": provider = crypto_price_provider if day == datetime.now(ZoneInfo("Europe/Zurich")).date().isoformat() else None result = run_daily_crypto_valuation( conn, as_of=day, price_provider=provider, lock_path=lock_root / "performance-backfill-crypto.lock", ) else: raise ValueError("True Wealth can only be extended by a newly confirmed value") runs.append( { "run_id": result.run_id, "as_of": result.as_of, "status": result.status, "valuation_stored": result.valuation_stored, "idempotent": result.idempotent, } ) refreshed = preview_performance_backfill( conn, source=source, period_from=str(request.get("period_from") or ""), period_to=str(request.get("period_to") or ""), ) completed = refreshed["days_to_materialize"] == 0 audit_id = record_audit_event( conn, source="performance_backfill_v1", action="performance_backfill_confirmed" if completed else "performance_backfill_attempted", entity_type="performance_backfill", entity_id=confirmation_id, old_values={}, new_values={ "source": source, "period_from": current["period_from"], "period_to": current["period_to"], "preview_id": request["preview_id"], "input_fingerprint": request["input_fingerprint"], "run_ids": [item["run_id"] for item in runs], "completed": completed, "remaining_days": refreshed["days_to_materialize"], "resume_input_revision": _local_backfill_input_revision( conn, source=source, account_ids=_accounts(conn, source), period_from=str(request.get("period_from") or ""), period_to=str(request.get("period_to") or ""), ), }, user_text_note=str(request.get("note") or "Confirmed performance valuation backfill"), created_by="user", ) conn.commit() return { "confirmation_id": confirmation_id, "source": source, "idempotent": False, "runs": runs, "audit_id": audit_id, "completed": completed, "remaining_days": refreshed["days_to_materialize"], }