diff --git a/src/jarvis_finance/services/asset_price_refresh.py b/src/jarvis_finance/services/asset_price_refresh.py new file mode 100644 index 0000000..e5fd047 --- /dev/null +++ b/src/jarvis_finance/services/asset_price_refresh.py @@ -0,0 +1,367 @@ +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.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 + 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 + )""", + (stale_before,), + ).fetchone()[0] + ) + if candidates == 0: + 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) + + +def _fx_source(conn: Connection, stale_before: str) -> tuple[int, int]: + from jarvis_finance.fx.providers import FrankfurterFxProvider, TwelveDataFxProvider + from jarvis_finance.fx.rates import resolve_fx_rate_to_chf + + cutoff_date = stale_before[:10] + currencies = [ + str(row["currency"]).upper() + for row in conn.execute( + """SELECT DISTINCT upper(i.currency) currency + FROM instruments i + WHERE i.is_active=1 AND upper(COALESCE(i.currency,'CHF'))!='CHF' + AND NOT EXISTS( + SELECT 1 FROM fx_rates f + WHERE f.base_currency=upper(i.currency) AND f.quote_currency='CHF' + AND f.rate_date>=? AND f.quality_status IN ('fresh','ok') + ) + ORDER BY currency""", + (cutoff_date,), + ).fetchall() + ] + updated = 0 + failures = 0 + for currency in currencies: + try: + result = resolve_fx_rate_to_chf( + conn, + base_currency=currency, + rate_date=None, + providers=[FrankfurterFxProvider(), TwelveDataFxProvider()], + persist=True, + resolve_fixed=True, + ) + updated += int(result.status == "ok") + except Exception: + failures += 1 + conn.commit() + if failures and updated == 0: + raise RuntimeError("fx_provider_failed") + return len(currencies), updated + + +DEFAULT_RUNNERS: dict[str, Callable[[Connection, str], tuple[int, int]]] = { + "equity": _equity_source, + "crypto": _crypto_source, + "fx": _fx_source, +} + + +def run_asset_price_refresh( + db_path: str, + job_id: str, + *, + runners: dict[str, Callable[[Connection, str], tuple[int, int]]] | None = None, +) -> None: + """Background worker with source isolation, stored progress and mutation guard.""" + conn = connect(db_path) + selected = runners or DEFAULT_RUNNERS + try: + conn.execute("BEGIN IMMEDIATE") + claimed = conn.execute( + """UPDATE asset_price_refresh_jobs + SET status='running' + WHERE job_id=? AND status='queued'""", + (job_id,), + ).rowcount + conn.commit() + if claimed != 1: + return + stale_before = str( + conn.execute( + "SELECT stale_before FROM asset_price_refresh_jobs WHERE job_id=?", + (job_id,), + ).fetchone()[0] + ) + protected_before = _protected_fingerprint(conn) + conn.set_authorizer(_deny_protected_dml) + completed = 0 + failures = 0 + for source in SOURCES: + started = _now() + conn.execute( + "UPDATE asset_price_refresh_sources SET status='running',started_at=? WHERE job_id=? AND source=?", + (started, job_id, source), + ) + conn.commit() + candidates = updated = 0 + status = "complete" + error_code = None + try: + candidates, updated = selected[source](conn, stale_before) + if candidates == 0: + status = "skipped" + except Exception as exc: + if conn.in_transaction: + conn.rollback() + status = "failed" + failures += 1 + error_code = str(exc)[:120] or type(exc).__name__ + completed += 1 + conn.execute( + """UPDATE asset_price_refresh_sources + SET status=?,stale_candidates=?,updated_count=?,error_code=?,completed_at=? + WHERE job_id=? AND source=?""", + (status, candidates, updated, error_code, _now(), job_id, source), + ) + conn.execute( + "UPDATE asset_price_refresh_jobs SET progress_completed=? WHERE job_id=?", + (completed, job_id), + ) + conn.commit() + if _protected_fingerprint(conn) != protected_before: + raise RuntimeError("protected_holdings_or_transactions_mutated") + + successful_sources = failures < len(SOURCES) + wealth_snapshot_id = None + if successful_sources: + model = build_modelled_wealth_development(conn, period="1m") + current = model.get("current") or {} + wealth_snapshot_id = f"wealth-refresh-{uuid.uuid4().hex}" + source_rows = [ + dict(row) + for row in conn.execute( + "SELECT source,status,stale_candidates,updated_count,error_code FROM asset_price_refresh_sources WHERE job_id=? ORDER BY source", + (job_id,), + ).fetchall() + ] + conn.execute( + """INSERT INTO aggregated_wealth_refresh_snapshots( + wealth_snapshot_id,job_id,captured_at,known_wealth_chf,quality_status,source_status_json + ) VALUES(?,?,?,?,?,?)""", + ( + wealth_snapshot_id, + job_id, + _now(), + current.get("value_chf"), + "complete" if failures == 0 else "partial", + json.dumps(source_rows, sort_keys=True), + ), + ) + final_status = "complete" if failures == 0 else "failed" if failures == len(SOURCES) else "partial" + audit_id = record_audit_event( + conn, + source="asset_price_refresh_job_v1", + action="asset_prices_refresh_completed", + entity_type="asset_price_refresh_job", + entity_id=job_id, + old_values={}, + new_values={ + "status": final_status, + "source_count": len(SOURCES), + "failed_source_count": failures, + "wealth_snapshot_created": bool(wealth_snapshot_id), + "holdings_mutated": False, + "transactions_mutated": False, + "trades_created": 0, + }, + created_by="system", + ) + conn.execute( + """UPDATE asset_price_refresh_jobs + SET status=?,completed_at=?,wealth_snapshot_id=?,audit_id=? + WHERE job_id=?""", + (final_status, _now(), wealth_snapshot_id, audit_id, job_id), + ) + conn.commit() + except Exception as exc: + if conn.in_transaction: + conn.rollback() + audit_id = record_audit_event( + conn, + source="asset_price_refresh_job_v1", + action="asset_prices_refresh_failed", + entity_type="asset_price_refresh_job", + entity_id=job_id, + old_values={}, + new_values={"status": "failed", "error_code": str(exc)[:120]}, + created_by="system", + ) + conn.execute( + "UPDATE asset_price_refresh_jobs SET status='failed',completed_at=?,audit_id=? WHERE job_id=?", + (_now(), audit_id, job_id), + ) + conn.commit() + finally: + conn.close() + + +def asset_price_refresh_status(conn: Connection, job_id: str) -> dict[str, Any]: + """Stored status only: no provider call, write or lazy refresh.""" + return _status_payload(conn, job_id) diff --git a/src/jarvis_finance/services/modelled_wealth.py b/src/jarvis_finance/services/modelled_wealth.py index c3e84b4..acde6ca 100644 --- a/src/jarvis_finance/services/modelled_wealth.py +++ b/src/jarvis_finance/services/modelled_wealth.py @@ -1,48 +1,49 @@ from __future__ import annotations from calendar import monthrange from datetime import date, timedelta from decimal import Decimal import re from sqlite3 import Connection from typing import Any from jarvis_finance.services.cash_service import authoritative_cash_movements from jarvis_finance.services.daily_valuations import SOURCE_KEY as CRYPTO_VALUATION_SOURCE LEGACY_CRYPTO_VALUATION_SOURCE = "daily_crypto_current_valuation_v1" MONEY = Decimal("0.01") PERCENT = Decimal("0.0001") -MODEL_PERIODS = {"since_anchor", "1m", "3m", "1y", "all"} +MODEL_PERIODS = {"since_anchor", "1m", "3m", "ytd", "1y", "all"} SNAPSHOT_PRECEDENCE = { "reconciliation": 4, "manual_balance": 3, "csv_anchor_balance": 2, "calculated_balance": 1, } COMPONENT_LABELS = { "postfinance": "PostFinance", "truewealth": "True Wealth", "crypto": "Krypto", "bank_cash": "Bankguthaben", + "other_assets": "Weitere Anlagen", } def _money(value: Decimal | None) -> str | None: return None if value is None else format(value.quantize(MONEY), "f") def _subtract_months(day: date, months: int) -> date: absolute = day.year * 12 + day.month - 1 - months year, month_index = divmod(absolute, 12) month = month_index + 1 return date(year, month, min(day.day, monthrange(year, month)[1])) def _authoritative_cash_movements( conn: Connection, *, account_id: str, after: str | None, through: str ) -> dict[str, Any]: return authoritative_cash_movements( conn, account_id=account_id, after=after, through=through ) @@ -350,40 +351,114 @@ def _crypto_events(conn: Connection, *, through: str) -> dict[str, dict[str, Any through=through, source=source, ) ) latest_by_account_day: dict[tuple[str, str], dict[str, Any]] = {} for row in rows: key = (row["account_id"], row["date"]) if key not in latest_by_account_day or str(row["captured_at"]) > str( latest_by_account_day[key]["captured_at"] ): latest_by_account_day[key] = row values: dict[str, Decimal] = {} for row in latest_by_account_day.values(): values[row["date"]] = values.get(row["date"], Decimal("0")) + row["value"] return { day: {"value": value, "quality": "modelled", "source_date": day} for day, value in values.items() } +def _other_asset_events(conn: Connection, *, through: str) -> dict[str, dict[str, Any]]: + """Aggregate confirmed non-cash account values without inventing daily precision.""" + rows = conn.execute( + """WITH ranked AS ( + SELECT s.account_id,s.valuation_date,s.total_value_chf, + ROW_NUMBER() OVER( + PARTITION BY s.account_id,s.valuation_date + ORDER BY COALESCE(s.valuation_at,s.created_at) DESC,s.snapshot_id DESC + ) rn + FROM account_value_snapshots s + JOIN accounts a ON a.account_id=s.account_id + WHERE a.is_active=1 AND a.account_type IN ('other_asset','membership') + AND s.valuation_date<=? AND COALESCE(s.is_active,1)=1 + AND s.quality_status IN ('confirmed','ok','complete') + ) SELECT account_id,valuation_date,total_value_chf + FROM ranked WHERE rn=1 ORDER BY valuation_date,account_id""", + (through,), + ).fetchall() + latest: dict[str, Decimal] = {} + events: dict[str, dict[str, Any]] = {} + for row in rows: + try: + value = Decimal(str(row["total_value_chf"])) + except Exception: + continue + if not value.is_finite() or value < Decimal("0"): + continue + latest[str(row["account_id"])] = value + day = str(row["valuation_date"]) + events[day] = { + "value": sum(latest.values(), Decimal("0")), + "quality": "confirmed", + "source_date": day, + } + return events + + +def _manual_cash_correction_markers(conn: Connection, *, through: str) -> list[dict[str, str]]: + rows = conn.execute( + """SELECT snapshot_id,account_id,balance_date,amount_chf,created_at + FROM cash_account_snapshots + WHERE source='manual_screenshot_snapshot' AND balance_date<=? + ORDER BY balance_date,created_at,snapshot_id""", + (through,), + ).fetchall() + grouped: dict[str, dict[str, Decimal]] = {} + for row in rows: + previous = conn.execute( + """SELECT amount_chf FROM cash_account_snapshots + WHERE account_id=? AND ( + balance_date list[dict[str, str]]: markers = [] model_days = sorted(models) for row in official: predecessor_days = [ day for day in model_days if day < row["date"] or ( day == row["date"] and str(models[day]["captured_at"]) < str(row["captured_at"]) ) ] if not predecessor_days: continue @@ -462,127 +537,132 @@ def _earliest_evidence(conn: Connection, *, fallback: date) -> date: )""" ).fetchall() valid_days: list[date] = [] for row in rows: try: valid_days.append(date.fromisoformat(str(row[0]))) except (TypeError, ValueError): continue return min(valid_days, default=fallback) def _period_start( conn: Connection, *, period: str, as_of: date, latest_anchor: date | None ) -> date: if period == "since_anchor": return latest_anchor or as_of if period == "1m": return _subtract_months(as_of, 1) if period == "3m": return _subtract_months(as_of, 3) + if period == "ytd": + return date(as_of.year, 1, 1) if period == "1y": return _subtract_months(as_of, 12) if period == "all": return _earliest_evidence(conn, fallback=as_of) - raise ValueError("period must be since_anchor, 1m, 3m, 1y or all") + raise ValueError("period must be since_anchor, 1m, 3m, ytd, 1y or all") def build_modelled_wealth_development( conn: Connection, *, period: str = "1m", as_of: str | None = None ) -> dict[str, Any]: """Compose existing immutable valuation/snapshot sources into one read model.""" if period not in MODEL_PERIODS: - raise ValueError("period must be since_anchor, 1m, 3m, 1y or all") + raise ValueError("period must be since_anchor, 1m, 3m, ytd, 1y or all") reference = date.fromisoformat(as_of) if as_of else date.today() through = reference.isoformat() postfinance, pf_markers = _postfinance_events(conn, through=through) truewealth, tw_markers = _truewealth_events(conn, through=through) crypto = _crypto_events(conn, through=through) + other_assets = _other_asset_events(conn, through=through) investment_events = { "postfinance": postfinance, "truewealth": truewealth, "crypto": crypto, + "other_assets": other_assets, } expected_investment = { "postfinance": bool( _role_account_ids(conn, "postfinance_etrading_depot") or _role_account_ids(conn, "postfinance_etrading_cash") ), "truewealth": bool( _role_account_ids(conn, "canonical_truewealth_total_value") ), "crypto": bool(_role_account_ids(conn, "crypto_portfolio")), + "other_assets": bool(other_assets), } - confirmed_days = [ - date.fromisoformat(day) - for events in (postfinance, truewealth) - for day, event in events.items() - if event["quality"] == "confirmed" - ] - cash_snapshot_day = conn.execute( - "SELECT MAX(balance_date) FROM cash_account_snapshots WHERE balance_date<=?", - (through,), - ).fetchone()[0] - if cash_snapshot_day: - confirmed_days.append(date.fromisoformat(str(cash_snapshot_day))) + # Household anchors come from confirmed portfolio-import anchors. Component-only + # cash/membership corrections remain event markers and must not move the solid-line + # boundary or make mixed-date values look fully confirmed. + confirmed_days: list[date] = [] + for events in (postfinance, truewealth): + for day, event in events.items(): + if event["quality"] != "confirmed": + continue + try: + confirmed_days.append(date.fromisoformat(day)) + except ValueError: + continue latest_anchor = max(confirmed_days, default=None) start = _period_start( conn, period=period, as_of=reference, latest_anchor=latest_anchor ) if start > reference: start = reference if (reference - start).days > 5000: start = reference - timedelta(days=5000) bank_accounts = _bank_accounts(conn) points: list[dict[str, Any]] = [] unknown_identity_by_day: dict[str, frozenset[str]] = {} event_dates = { day for events in investment_events.values() for day in events if start.isoformat() <= day <= through } event_dates.update( str(row[0]) for row in conn.execute( "SELECT DISTINCT balance_date FROM cash_account_snapshots WHERE balance_date BETWEEN ? AND ?", (start.isoformat(), through), ).fetchall() ) for account in bank_accounts: movement_evidence = _authoritative_cash_movements( conn, account_id=account["account_id"], after=start.isoformat(), through=through, ) event_dates.update(movement_evidence["days"]) cursor = start while cursor <= reference: day = cursor.isoformat() components: list[dict[str, Any]] = [] qualities: list[str] = [] missing_investment: list[str] = [] known_total = Decimal("0") - for key in ("postfinance", "truewealth", "crypto"): + for key in ("postfinance", "truewealth", "crypto", "other_assets"): selected = _event_at_or_before(investment_events[key], day) value = selected["value"] if selected else None quality = selected["quality"] if selected else "unavailable" if value is not None: known_total += value qualities.append(quality) elif expected_investment[key]: missing_investment.append(key) components.append( { "key": key, "label": COMPONENT_LABELS[key], "value_chf": _money(value), "quality": quality, "source_date": selected["source_date"] if selected else None, } ) bank_total = Decimal("0") bank_qualities: list[str] = [] @@ -612,86 +692,86 @@ def build_modelled_wealth_development( bank_value = _money(bank_total) else: bank_quality = "unavailable" bank_value = None components.append( { "key": "bank_cash", "label": COMPONENT_LABELS["bank_cash"], "value_chf": bank_value, "quality": bank_quality, "source_date": min(bank_source_days, default=None), } ) if not qualities: cursor += timedelta(days=1) continue has_confirmed_anchor = any( component["quality"] == "confirmed" and component["source_date"] == day for component in components ) - has_modelled_value = any( + has_modelled_value = (latest_anchor is None or cursor > latest_anchor) and any( component["quality"] == "modelled" and component["source_date"] == day for component in components ) point_quality = ( "incomplete" if unknown_on_day or missing_investment else "modelled" - if "modelled" in qualities + if "modelled" in qualities and (latest_anchor is None or cursor > latest_anchor) else "carried" if "carried" in qualities else "confirmed" ) points.append( { "date": day, "value_chf": _money(known_total) or "0.00", "quality": point_quality, "has_confirmed_anchor": has_confirmed_anchor, "has_modelled_value": has_modelled_value, "components": components, "excluded_account_count": len(unknown_on_day) + len(missing_investment), } ) unknown_identity_by_day[day] = frozenset( [f"bank:{account_id}" for account_id in unknown_on_day] + [f"component:{key}" for key in missing_investment] ) cursor += timedelta(days=1) current_point = points[-1] if points else None current_unknown = [] for index, account in enumerate(bank_accounts, start=1): evidence = effective_cash_evidence( conn, account_id=account["account_id"], as_of=through ) if evidence["value_chf"] is None: current_unknown.append( { "key": f"unknown-bank-{index}", "label": _safe_bank_label(account["label"]), "reason_code": "confirmed_cash_evidence_missing", } ) - for key in ("postfinance", "truewealth", "crypto"): + for key in ("postfinance", "truewealth", "crypto", "other_assets"): if expected_investment[key] and not _event_at_or_before( investment_events[key], through ): current_unknown.append( { "key": f"unknown-component-{key}", "label": COMPONENT_LABELS[key], "reason_code": "stored_valuation_evidence_missing", } ) anchor_point = None if latest_anchor: anchor_point = next( (point for point in points if point["date"] == latest_anchor.isoformat()), None, ) anchor = ( { "date": anchor_point["date"], @@ -810,42 +890,43 @@ def build_modelled_wealth_development( "quality": quality, "as_of": item["source_date"], "unknown_account_count": sum( 1 for unknown in current_unknown if ( str(unknown["key"]).startswith("unknown-bank-") if key == "bank_cash" else unknown["key"] == f"unknown-component-{key}" ) ), } ) return { "status": "available" if points else "unavailable", "period": { "preset": period, "from": start.isoformat(), "to": through, }, + "last_confirmed_anchor_date": latest_anchor.isoformat() if latest_anchor else None, "anchor": anchor, "baseline": baseline, "current": current, "change_chf": _money(change), "change_pct": format(change_pct.quantize(PERCENT), "f") if change_pct is not None else None, "chart_visible": chart_visible, "points": points, "components": component_summaries, "correction_markers": sorted( [ marker - for marker in pf_markers + tw_markers + for marker in pf_markers + tw_markers + _manual_cash_correction_markers(conn, through=through) if start.isoformat() <= marker["date"] <= through ], key=lambda item: (item["date"], item["source_key"]), ), "unknown_accounts": current_unknown, "method": "modelled_wealth_daily_v1", "disclaimer": "Geschätzte Entwicklung aus bestätigten Ankern, gespeicherten Tagesbewertungen und fortgeschriebenen bekannten Salden; keine verifizierte TTWROR oder XIRR.", } diff --git a/src/jarvis_finance/services/portfolio_analysis_v1.py b/src/jarvis_finance/services/portfolio_analysis_v1.py new file mode 100644 index 0000000..3b30168 --- /dev/null +++ b/src/jarvis_finance/services/portfolio_analysis_v1.py @@ -0,0 +1,257 @@ +from __future__ import annotations + +import json +from collections import defaultdict +from decimal import Decimal, InvalidOperation +from sqlite3 import Connection +from typing import Any + +from jarvis_finance.services.modelled_wealth import ( + _bank_accounts, + build_modelled_wealth_development, + effective_cash_evidence, +) +from jarvis_finance.services.portfolio_policy import active_policy + +ZERO = Decimal("0") +HUNDRED = Decimal("100") +MONEY = Decimal("0.01") +PCT = Decimal("0.01") + + +def _decimal(value: object) -> Decimal: + try: + result = Decimal(str(value or "0")) + return result if result.is_finite() else ZERO + except (InvalidOperation, ValueError): + return ZERO + + +def _money(value: Decimal | None) -> str | None: + return None if value is None else format(value.quantize(MONEY), "f") + + +def _pct(value: Decimal | None) -> str | None: + return None if value is None else format(value.quantize(PCT), "f") + + +def _latest_positions(conn: Connection, as_of: str) -> tuple[list[dict[str, Any]], str]: + row = conn.execute( + """SELECT as_of,quality_status,summary_json FROM portfolio_analysis_snapshots + WHERE as_of<=? ORDER BY as_of DESC,created_at DESC,analysis_snapshot_id DESC LIMIT 1""", + (as_of,), + ).fetchone() + if not row: + return [], "unavailable" + try: + summary = json.loads(str(row["summary_json"] or "{}")) + positions = [item for item in summary.get("positions", []) if isinstance(item, dict)] + except (TypeError, ValueError, json.JSONDecodeError): + return [], "unavailable" + metadata = { + str(item["instrument_id"]): dict(item) + for item in conn.execute( + "SELECT instrument_id,country,sector,currency,asset_class FROM instruments WHERE is_active=1" + ).fetchall() + } + result: list[dict[str, Any]] = [] + for item in positions: + value = _decimal(item.get("value_chf")) + if value <= ZERO: + continue + instrument = metadata.get(str(item.get("instrument_id") or ""), {}) + result.append( + { + **item, + "value": value, + "asset_class": str(item.get("asset_class") or instrument.get("asset_class") or "").lower(), + "currency": str(item.get("currency") or instrument.get("currency") or "").upper(), + "country": str(instrument.get("country") or "").strip(), + "sector": str(instrument.get("sector") or "").strip(), + } + ) + quality = "complete" if str(row["quality_status"]) == "complete" else "partial" + return result, quality + + +def _current_components(modelled: dict[str, Any]) -> dict[str, Decimal]: + return { + str(item["key"]): _decimal(item.get("current_value_chf")) + for item in modelled.get("components", []) + if item.get("current_value_chf") is not None + } + + +def _policy_rows(conn: Connection) -> dict[str, dict[str, Any]]: + configured = active_policy(conn) + policy = configured.get("policy") if configured.get("configured") else None + if not policy: + return {} + return {str(item["asset_class"]): item for item in policy.get("allocations", [])} + + +def _allocation_row( + *, key: str, label: str, value: Decimal, total: Decimal, policy: dict[str, Any] | None +) -> dict[str, Any]: + current_pct = value / total * HUNDRED if total > ZERO else None + if not policy or current_pct is None: + return { + "key": key, "label": label, "current_value_chf": _money(value), + "current_pct": _pct(current_pct), "target_pct": None, "lower_pct": None, + "upper_pct": None, "deviation_pp": None, "deviation_chf": None, + "status": "unavailable", + } + target = _decimal(policy.get("target_pct")) + lower = _decimal(policy.get("lower_pct")) + upper = _decimal(policy.get("upper_pct")) + deviation_pp = current_pct - target + deviation_chf = value - total * target / HUNDRED + status = "below_corridor" if current_pct < lower else "above_corridor" if current_pct > upper else "within_corridor" + return { + "key": key, "label": label, "current_value_chf": _money(value), + "current_pct": _pct(current_pct), "target_pct": _pct(target), + "lower_pct": _pct(lower), "upper_pct": _pct(upper), + "deviation_pp": _pct(deviation_pp), "deviation_chf": _money(deviation_chf), + "status": status, + } + + +def _dimension( + positions: list[dict[str, Any]], field: str, total: Decimal, *, include_chf: Decimal = ZERO +) -> dict[str, Any]: + values: dict[str, Decimal] = defaultdict(lambda: ZERO) + assessed = ZERO + if include_chf > ZERO and field == "currency": + values["CHF"] += include_chf + assessed += include_chf + for item in positions: + label = str(item.get(field) or "").strip() + if not label: + continue + values[label] += item["value"] + assessed += item["value"] + rows = [ + {"label": label, "pct": _pct(value / total * HUNDRED) if total > ZERO else None} + for label, value in sorted(values.items(), key=lambda item: (-item[1], item[0])) + ] + if total <= ZERO or not rows: + status = "unavailable" + else: + status = "complete" if assessed >= total - Decimal("0.01") else "partial" + return {"status": status, "rows": rows} + + +def _concentrations(values: list[Decimal], total: Decimal) -> dict[str, str | None]: + ordered = sorted((value for value in values if value > ZERO), reverse=True) + def share(limit: int) -> str | None: + return _pct(sum(ordered[:limit], ZERO) / total * HUNDRED) if total > ZERO and ordered else None + return {"top1_pct": share(1), "top5_pct": share(5), "top10_pct": share(10)} + + +def _prioritized_hints(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + priority = {"above_corridor": 1, "below_corridor": 2, "unavailable": 3, "within_corridor": 4} + prefix = { + "above_corridor": "Reduktion prüfen", + "below_corridor": "Erhöhung prüfen", + "unavailable": "Daten ergänzen", + "within_corridor": "Im Zielkorridor", + } + candidates = sorted(rows, key=lambda row: (priority.get(str(row["status"]), 9), str(row["label"]))) + return [ + {"priority": index, "text": f"{prefix.get(str(row['status']), 'Daten ergänzen')}: {row['label']}."} + for index, row in enumerate(candidates[:5], start=1) + ] + + +def build_portfolio_analysis_v1( + conn: Connection, *, as_of: str, modelled: dict[str, Any] | None = None +) -> dict[str, Any]: + """Read-only v1 analysis over canonical stored valuations and active versioned policy.""" + model = modelled or build_modelled_wealth_development(conn, period="1m", as_of=as_of) + components = _current_components(model) + positions, position_quality = _latest_positions(conn, as_of) + truewealth_accounts = { + str(row["account_id"]) + for row in conn.execute( + "SELECT DISTINCT account_id FROM truewealth_portfolios WHERE is_active=1" + ).fetchall() + } + directly_classified = [ + item for item in positions + if str(item.get("account_id") or "") not in truewealth_accounts + ] + stock_value = sum( + (item["value"] for item in directly_classified if item["asset_class"] in {"equity", "stock"}), + ZERO, + ) + etf_value = sum( + (item["value"] for item in directly_classified if item["asset_class"] in {"etf", "fund"}), + ZERO, + ) + classified_pf = stock_value + etf_value + postfinance_total = components.get("postfinance", ZERO) + settlement_cash = max(ZERO, postfinance_total - classified_pf) + bank_cash = components.get("bank_cash", ZERO) + cash_value = bank_cash + settlement_cash + truewealth_value = components.get("truewealth", ZERO) + crypto_value = components.get("crypto", ZERO) + other_value = components.get("other_assets", ZERO) + total = cash_value + stock_value + etf_value + truewealth_value + crypto_value + other_value + policy = _policy_rows(conn) + + requested = [ + ("cash", "Cash", cash_value, policy.get("cash")), + ("stocks", "Aktien", stock_value, None), + ("etf", "ETF", etf_value, None), + ("truewealth", "True Wealth", truewealth_value, None), + ("crypto", "Krypto", crypto_value, policy.get("crypto")), + ("other", "Weitere Anlagen", other_value, policy.get("other")), + ] + allocation = [ + _allocation_row(key=key, label=label, value=value, total=total, policy=target) + for key, label, value, target in requested + ] + if policy.get("equity"): + allocation.append( + _allocation_row( + key="equity_policy_group", + label="Policy-Gruppe Aktien / ETF / True Wealth", + value=stock_value + etf_value + truewealth_value, + total=total, + policy=policy["equity"], + ) + ) + + concentration_values = [item["value"] for item in directly_classified] + concentration_values.extend(value for value in (truewealth_value, crypto_value, other_value) if value > ZERO) + for account in _bank_accounts(conn): + evidence = effective_cash_evidence(conn, account_id=account["account_id"], as_of=as_of) + if evidence["value_chf"] is not None: + concentration_values.append(_decimal(evidence["value_chf"])) + if settlement_cash > ZERO: + concentration_values.append(settlement_cash) + + contributions = [] + for item in model.get("components", []): + value = item.get("change_chf") + if value is None or _decimal(value) == ZERO: + continue + contributions.append( + {"key": str(item["key"]), "label": str(item["label"]), "value_chf": _money(_decimal(value)), "status": "modelled"} + ) + positives = sorted((row for row in contributions if _decimal(row["value_chf"]) > ZERO), key=lambda row: _decimal(row["value_chf"]), reverse=True)[:2] + negatives = sorted((row for row in contributions if _decimal(row["value_chf"]) < ZERO), key=lambda row: _decimal(row["value_chf"]))[:2] + + status = "unavailable" if total <= ZERO else "partial" if position_quality != "complete" or any(row["status"] == "unavailable" for row in allocation) else "complete" + return { + "status": status, + "allocation": allocation, + "concentrations": _concentrations(concentration_values, total), + "dimensions": { + "currency": _dimension(positions, "currency", total, include_chf=cash_value + other_value), + "region": _dimension(positions, "country", total), + "sector": _dimension(positions, "sector", total), + }, + "contributions": positives + negatives, + "hints": _prioritized_hints(allocation), + } diff --git a/src/jarvis_finance/services/system_ops.py b/src/jarvis_finance/services/system_ops.py index 2b379d8..5225e6a 100644 --- a/src/jarvis_finance/services/system_ops.py +++ b/src/jarvis_finance/services/system_ops.py @@ -1,112 +1,141 @@ from __future__ import annotations import json import os import socket import subprocess +import urllib.request +from urllib.parse import urlsplit from datetime import datetime, timezone from pathlib import Path from typing import Any from jarvis_finance.config.settings import find_repo_root, load_settings ALLOWED_ACTIONS: dict[str, str] = { "backend": "scripts/restart_backend.sh", "frontend": "scripts/restart_frontend.sh", "dashboard": "scripts/restart_dashboard.sh", } _ALLOWED_SCRIPT_NAMES = {"restart_backend.sh", "restart_frontend.sh", "restart_dashboard.sh", "restart_vue_dashboard.sh"} -_ALLOWED_ENV = {"PATH", "HOME", "USER", "SHELL", "LANG", "LC_ALL", "JARVIS_FINANCE_RUNTIME_DIR"} +_ALLOWED_ENV = {"PATH", "HOME", "USER", "SHELL", "LANG", "LC_ALL", "JARVIS_FINANCE_RUNTIME_DIR", + "VITE_API_BASE_URL", "BACKEND_HOST", "BACKEND_PORT", "FRONTEND_PORT", "JARVIS_FINANCE_API_URL"} _SAFE_OPS_KEYS = { "action", "component", "status", "started_at", "finished_at", "message", "error", "return_code", "worker_started", "log_available", } def _now() -> str: return datetime.now(timezone.utc).isoformat() def _port_open(port: int) -> bool: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.settimeout(0.3) return sock.connect_ex(("127.0.0.1", int(port))) == 0 def _runtime_dir(repo_root: Path | None = None) -> Path: return load_settings(repo_root=repo_root or find_repo_root()).runtime_paths.base_dir def _safe_env(runtime_dir: Path) -> dict[str, str]: env = {k: v for k, v in os.environ.items() if k in _ALLOWED_ENV} env["JARVIS_FINANCE_RUNTIME_DIR"] = str(runtime_dir) - env["VITE_API_BASE_URL"] = "http://100.85.29.67:8000" - env["BACKEND_HOST"] = "0.0.0.0" - env["BACKEND_PORT"] = "8000" - env["FRONTEND_PORT"] = "5173" return env +def _healthcheck(url: str) -> bool: + try: + with urllib.request.urlopen(url.rstrip("/") + "/health", timeout=0.5) as response: + return 200 <= int(response.status) < 300 + except Exception: + return False + + def _write_ops_log(runtime_dir: Path, entry: dict[str, Any]) -> None: log_dir = runtime_dir / "logs" log_dir.mkdir(parents=True, exist_ok=True) safe_entry = _sanitize_ops_entry(entry) with (log_dir / "ops_actions.jsonl").open("a", encoding="utf-8") as fh: fh.write(json.dumps(safe_entry, ensure_ascii=False, sort_keys=True) + "\n") def _sanitize_ops_entry(entry: dict[str, Any]) -> dict[str, Any]: return {key: value for key, value in entry.items() if key in _SAFE_OPS_KEYS} def _last_ops_action(runtime_dir: Path) -> dict[str, Any] | None: path = runtime_dir / "logs" / "ops_actions.jsonl" if not path.exists(): return None try: lines = [line for line in path.read_text(encoding="utf-8", errors="ignore").splitlines() if line.strip()] return _sanitize_ops_entry(json.loads(lines[-1])) if lines else None except Exception: return None -def system_status(*, runtime_dir: Path | None = None, repo_root: Path | None = None, api_url: str = "http://100.85.29.67:8000") -> dict[str, Any]: +def system_status( + *, + runtime_dir: Path | None = None, + repo_root: Path | None = None, + api_url: str | None = None, + frontend_url: str | None = None, + backend_reachable: bool | None = None, + frontend_reachable: bool | None = None, +) -> dict[str, Any]: repo = (repo_root or find_repo_root()).resolve() runtime = (runtime_dir or _runtime_dir(repo)).resolve() db_path = runtime / "data" / "finance.sqlite3" + configured_api = api_url or os.environ.get("JARVIS_FINANCE_API_URL") or os.environ.get("VITE_API_BASE_URL") + parsed = urlsplit(configured_api) if configured_api else None + backend_port = parsed.port if parsed and parsed.hostname else None + configured_frontend = frontend_url or os.environ.get("JARVIS_FINANCE_FRONTEND_URL") + frontend_parsed = urlsplit(configured_frontend) if configured_frontend else None + frontend_port = frontend_parsed.port if frontend_parsed and frontend_parsed.hostname else None + backend_running = backend_reachable if backend_reachable is not None else bool(configured_api and _healthcheck(configured_api)) + frontend_running = ( + frontend_reachable + if frontend_reachable is not None + else bool(configured_frontend and _healthcheck(configured_frontend.rstrip("/").removesuffix("/api"))) + ) + if frontend_reachable is None and not frontend_running and frontend_port: + frontend_running = _port_open(frontend_port) return { "purpose": "system_ops_status_v1", "status": "ok", - "api_url": api_url, + "api_url": configured_api, "runtime_db_available": db_path.exists(), "runtime_outside_repo": repo not in runtime.parents and runtime != repo, - "backend": {"status": "running" if _port_open(8000) else "offline", "port": 8000}, - "frontend": {"status": "running" if _port_open(5173) else "offline", "port": 5173}, + "backend": {"status": "running" if backend_running else "offline", "port": backend_port}, + "frontend": {"status": "running" if frontend_running else "offline", "port": frontend_port}, "last_restart": _last_ops_action(runtime), } def restart_system_component(action: str, *, runtime_dir: Path | None = None, repo_root: Path | None = None, timeout: int = 30) -> dict[str, Any]: if action not in ALLOWED_ACTIONS: raise ValueError("unsupported_system_action") repo = (repo_root or find_repo_root()).resolve() runtime = (runtime_dir or _runtime_dir(repo)).resolve() script_rel = ALLOWED_ACTIONS[action] script_path = (repo / script_rel).resolve() started_at = _now() base = { "action": f"restart_{action}", "component": action, "started_at": started_at, } if repo not in script_path.parents or script_path.name not in _ALLOWED_SCRIPT_NAMES: raise ValueError("script_not_allowed") if not script_path.exists(): diff --git a/src/jarvis_finance/storage/migrations.py b/src/jarvis_finance/storage/migrations.py index 90c34e3..9f8994b 100644 --- a/src/jarvis_finance/storage/migrations.py +++ b/src/jarvis_finance/storage/migrations.py @@ -1,32 +1,32 @@ from __future__ import annotations import hashlib import json from datetime import datetime, timezone from sqlite3 import Connection from .schema import INITIAL_SCHEMA_SQL from .postfinance_schema import create_postfinance_ledger_import_v1 -MIGRATION_VERSION = 51 -MIGRATION_NAME = "051_current_source_coverage_and_truewealth_activity_v1" +MIGRATION_VERSION = 52 +MIGRATION_NAME = "052_professional_portfolio_cockpit_v1" INSTRUMENT_OPTIONAL_COLUMNS = { "position_category": "TEXT", "ter": "TEXT", "distribution_policy": "TEXT", "index_name": "TEXT", "fund_domicile": "TEXT", "benchmark": "TEXT", "is_currency_hedged": "INTEGER NOT NULL DEFAULT 0", "hedged_to_currency": "TEXT", "hedge_status": "TEXT NOT NULL DEFAULT 'unknown'", "base_exposure_currency": "TEXT", "trading_currency": "TEXT", "instrument_status": "TEXT NOT NULL DEFAULT 'unknown'", "valuation_policy": "TEXT NOT NULL DEFAULT 'live_price'", "corporate_action_status": "TEXT NOT NULL DEFAULT 'not_checked'", "split_or_corporate_action_review_required": "INTEGER NOT NULL DEFAULT 0", } CATALOG_OPTIONAL_COLUMNS = { @@ -2820,40 +2820,166 @@ def _create_crypto_reconciliation_cockpit_v1(conn: Connection) -> None: BEGIN SELECT RAISE(ABORT, 'crypto snapshot wallets are immutable'); END; CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshot_wallets_no_delete BEFORE DELETE ON crypto_balance_snapshot_wallets BEGIN SELECT RAISE(ABORT, 'crypto snapshot wallets cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshot_items_no_update BEFORE UPDATE ON crypto_balance_snapshot_items BEGIN SELECT RAISE(ABORT, 'crypto snapshot items are immutable'); END; CREATE TRIGGER IF NOT EXISTS crypto_balance_snapshot_items_no_delete BEFORE DELETE ON crypto_balance_snapshot_items BEGIN SELECT RAISE(ABORT, 'crypto snapshot items cannot be deleted'); END; CREATE TRIGGER IF NOT EXISTS crypto_internal_transfer_pairs_no_update BEFORE UPDATE ON crypto_internal_transfer_pairs BEGIN SELECT RAISE(ABORT, 'crypto transfer pairs are immutable'); END; CREATE TRIGGER IF NOT EXISTS crypto_internal_transfer_pairs_no_delete BEFORE DELETE ON crypto_internal_transfer_pairs BEGIN SELECT RAISE(ABORT, 'crypto transfer pairs cannot be deleted'); END; """ ) +def _create_professional_portfolio_cockpit_v1(conn: Connection) -> None: + """Add bounded manual-snapshot and controlled refresh job lineage.""" + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS manual_snapshot_confirmations ( + confirmation_id TEXT PRIMARY KEY, + preview_id TEXT NOT NULL UNIQUE, + input_fingerprint TEXT NOT NULL, + payload_hash TEXT NOT NULL, + snapshot_date TEXT NOT NULL, + source_kind TEXT NOT NULL, + known_wealth_after_chf TEXT NOT NULL, + bank_cash_after_chf TEXT NOT NULL, + separate_membership_asset_after_chf TEXT NOT NULL, + created_snapshot_count INTEGER NOT NULL CHECK(created_snapshot_count=3), + created_at TEXT NOT NULL, + audit_id TEXT NOT NULL UNIQUE REFERENCES audit_log(audit_id), + CHECK(source_kind='dated_manual_screenshot') + ); + CREATE TRIGGER IF NOT EXISTS manual_snapshot_confirmations_no_update + BEFORE UPDATE ON manual_snapshot_confirmations + BEGIN SELECT RAISE(ABORT, 'manual snapshot confirmations are immutable'); END; + CREATE TRIGGER IF NOT EXISTS manual_snapshot_confirmations_no_delete + BEFORE DELETE ON manual_snapshot_confirmations + BEGIN SELECT RAISE(ABORT, 'manual snapshot confirmations cannot be deleted'); END; + CREATE TRIGGER IF NOT EXISTS manual_snapshot_confirmations_no_replace + BEFORE INSERT ON manual_snapshot_confirmations + WHEN EXISTS( + SELECT 1 FROM manual_snapshot_confirmations old + WHERE old.confirmation_id=NEW.confirmation_id + OR old.preview_id=NEW.preview_id + OR old.audit_id=NEW.audit_id + ) + BEGIN SELECT RAISE(ABORT, 'manual snapshot confirmations cannot be replaced'); END; + CREATE TRIGGER IF NOT EXISTS manual_cash_snapshots_no_update + BEFORE UPDATE ON cash_account_snapshots + WHEN OLD.source='manual_screenshot_snapshot' + BEGIN SELECT RAISE(ABORT, 'manual cash snapshots are immutable'); END; + CREATE TRIGGER IF NOT EXISTS manual_cash_snapshots_no_delete + BEFORE DELETE ON cash_account_snapshots + WHEN OLD.source='manual_screenshot_snapshot' + BEGIN SELECT RAISE(ABORT, 'manual cash snapshots cannot be deleted'); END; + CREATE TRIGGER IF NOT EXISTS manual_cash_snapshots_no_replace + BEFORE INSERT ON cash_account_snapshots + WHEN EXISTS( + SELECT 1 FROM cash_account_snapshots old + WHERE old.source='manual_screenshot_snapshot' + AND old.snapshot_id=NEW.snapshot_id + ) + BEGIN SELECT RAISE(ABORT, 'manual cash snapshots cannot be replaced'); END; + CREATE TRIGGER IF NOT EXISTS manual_asset_snapshots_no_update + BEFORE UPDATE ON account_value_snapshots + WHEN OLD.source_type='manual_screenshot_snapshot' + BEGIN SELECT RAISE(ABORT, 'manual asset snapshots are immutable'); END; + CREATE TRIGGER IF NOT EXISTS manual_asset_snapshots_no_delete + BEFORE DELETE ON account_value_snapshots + WHEN OLD.source_type='manual_screenshot_snapshot' + BEGIN SELECT RAISE(ABORT, 'manual asset snapshots cannot be deleted'); END; + CREATE TRIGGER IF NOT EXISTS manual_asset_snapshots_no_replace + BEFORE INSERT ON account_value_snapshots + WHEN EXISTS( + SELECT 1 FROM account_value_snapshots old + WHERE old.source_type='manual_screenshot_snapshot' + AND old.snapshot_id=NEW.snapshot_id + ) + BEGIN SELECT RAISE(ABORT, 'manual asset snapshots cannot be replaced'); END; + CREATE TABLE IF NOT EXISTS asset_price_refresh_jobs ( + job_id TEXT PRIMARY KEY, + status TEXT NOT NULL CHECK(status IN ('queued','running','complete','partial','failed')), + requested_at TEXT NOT NULL, + completed_at TEXT, + stale_before TEXT NOT NULL, + progress_total INTEGER NOT NULL DEFAULT 3, + progress_completed INTEGER NOT NULL DEFAULT 0, + wealth_snapshot_id TEXT, + audit_id TEXT REFERENCES audit_log(audit_id) + ); + CREATE UNIQUE INDEX IF NOT EXISTS uq_asset_price_refresh_single_active + ON asset_price_refresh_jobs((1)) + WHERE status IN ('queued','running'); + CREATE TABLE IF NOT EXISTS asset_price_refresh_sources ( + job_id TEXT NOT NULL REFERENCES asset_price_refresh_jobs(job_id), + source TEXT NOT NULL CHECK(source IN ('equity','crypto','fx')), + status TEXT NOT NULL CHECK(status IN ('pending','running','complete','failed','skipped')), + stale_candidates INTEGER NOT NULL DEFAULT 0, + updated_count INTEGER NOT NULL DEFAULT 0, + error_code TEXT, + started_at TEXT, + completed_at TEXT, + PRIMARY KEY(job_id,source) + ); + CREATE TABLE IF NOT EXISTS aggregated_wealth_refresh_snapshots ( + wealth_snapshot_id TEXT PRIMARY KEY, + job_id TEXT NOT NULL UNIQUE REFERENCES asset_price_refresh_jobs(job_id), + captured_at TEXT NOT NULL, + known_wealth_chf TEXT, + quality_status TEXT NOT NULL CHECK(quality_status IN ('complete','partial')), + source_status_json TEXT NOT NULL CHECK(json_valid(source_status_json)) + ); + CREATE INDEX IF NOT EXISTS idx_asset_price_refresh_jobs_requested + ON asset_price_refresh_jobs(requested_at DESC); + CREATE TRIGGER IF NOT EXISTS aggregated_wealth_refresh_snapshots_no_update + BEFORE UPDATE ON aggregated_wealth_refresh_snapshots + BEGIN SELECT RAISE(ABORT, 'wealth refresh snapshots are immutable'); END; + CREATE TRIGGER IF NOT EXISTS aggregated_wealth_refresh_snapshots_no_delete + BEFORE DELETE ON aggregated_wealth_refresh_snapshots + BEGIN SELECT RAISE(ABORT, 'wealth refresh snapshots cannot be deleted'); END; + CREATE TRIGGER IF NOT EXISTS sprint23_audit_no_update + BEFORE UPDATE ON audit_log + WHEN OLD.entity_type IN ('manual_source_snapshot','asset_price_refresh_job') + BEGIN SELECT RAISE(ABORT, 'sprint23 audit is immutable'); END; + CREATE TRIGGER IF NOT EXISTS sprint23_audit_no_delete + BEFORE DELETE ON audit_log + WHEN OLD.entity_type IN ('manual_source_snapshot','asset_price_refresh_job') + BEGIN SELECT RAISE(ABORT, 'sprint23 audit cannot be deleted'); END; + """ + ) + # Compatibility repair for an interrupted/pre-release schema-52 build where + # the table may already exist without the later payload-binding column. + _add_missing_columns( + conn, + "manual_snapshot_confirmations", + {"payload_hash": "TEXT NOT NULL DEFAULT ''"}, + ) + + def _apply_compat_migrations(conn: Connection) -> None: _add_missing_instrument_columns(conn) for table, text_columns in TEXT_AFFINITY_COLUMNS.items(): _rebuild_table_with_text_columns(conn, table, text_columns) _create_broker_bank_mapping_tables(conn) _create_broker_import_review_items(conn) _create_broker_import_execution_plans(conn) _add_transaction_void_columns(conn) _create_fx_market_data_tables(conn) _create_market_quote_chart_tables(conn) _create_account_value_snapshot_tables(conn) _create_cash_account_snapshot_tables(conn) _create_instrument_import_candidates(conn) _create_budget_phase1_tables(conn) _create_budget_phase11_tables(conn) _create_budget_phase14_tables(conn) _create_budget_phase15_tables(conn) _create_budget_phase18_tables(conn) _create_budget_phase19_tables(conn) _create_budget_import_production_v1_tables(conn) @@ -2861,38 +2987,39 @@ def _apply_compat_migrations(conn: Connection) -> None: _create_budget_planning_forecast_v1_tables(conn) _create_budget_fixed_costs_subscriptions_v1_tables(conn) _create_budget_monthly_import_rule_learning_v1_tables(conn) _create_transfer_pairing_v2_tables(conn) _create_portfolio_policy_tables(conn) _create_portfolio_performance_tables(conn) _create_portfolio_ingestion_reconciliation_tables(conn) _create_daily_market_analytics_tables(conn) _create_postfinance_baseline_mapping_audit_v1(conn) _create_truewealth_verified_snapshot_v1(conn) create_postfinance_ledger_import_v1(conn) _create_investment_performance_scope_v1(conn) _create_grocery_optimizer_v1_tables(conn) _add_grocery_price_provider_v1_columns(conn) _create_grocery_matching_learning_v2_tables(conn) _create_household_import_v1_tables(conn) _create_household_review_corrections_v1(conn) _create_annual_budget_recurring_semantics_v1(conn) _create_current_source_coverage_and_truewealth_activity_v1(conn) _create_crypto_reconciliation_cockpit_v1(conn) + _create_professional_portfolio_cockpit_v1(conn) def apply_migrations(conn: Connection) -> None: conn.executescript(INITIAL_SCHEMA_SQL) existing_initial = conn.execute("SELECT 1 FROM schema_migrations WHERE version = 1").fetchone() if not existing_initial: conn.execute( "INSERT INTO schema_migrations(version, name, applied_at, checksum) VALUES (?, ?, ?, ?)", (1, "001_initial_schema", utc_now(), checksum_sql(INITIAL_SCHEMA_SQL)), ) _apply_compat_migrations(conn) existing = conn.execute("SELECT 1 FROM schema_migrations WHERE version = ?", (MIGRATION_VERSION,)).fetchone() if not existing: conn.execute( "INSERT INTO schema_migrations(version, name, applied_at, checksum) VALUES (?, ?, ?, ?)", (MIGRATION_VERSION, MIGRATION_NAME, utc_now(), checksum_sql(MIGRATION_NAME)), ) conn.commit() __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23__HERMES_CWD_8d46a20096ed__