diff --git a/src/jarvis_finance/api/schemas/market.py b/src/jarvis_finance/api/schemas/market.py index be9b5d6..08cbb5c 100644 --- a/src/jarvis_finance/api/schemas/market.py +++ b/src/jarvis_finance/api/schemas/market.py @@ -1,113 +1,122 @@ from __future__ import annotations from pydantic import BaseModel, ConfigDict, Field from typing import Literal class AssetPriceRefreshRequest(BaseModel): model_config = ConfigDict(extra="forbid") stale_hours: int = Field(default=24, ge=1, le=720) class AssetPriceRefreshSourceStatus(BaseModel): model_config = ConfigDict(extra="forbid") source: Literal["equity", "crypto", "fx"] status: Literal["pending", "running", "complete", "failed", "skipped"] stale_candidates: int updated_count: int error_code: str | None + fresh_unchanged_count: int + stale_remaining_count: int + failed_count: int + diagnostics: list[str] started_at: str | None completed_at: str | None class AssetPriceRefreshJobResponse(BaseModel): model_config = ConfigDict(extra="forbid") job_id: str status: Literal["queued", "running", "complete", "partial", "failed"] requested_at: str completed_at: str | None stale_before: str progress: dict[str, int] sources: list[AssetPriceRefreshSourceStatus] wealth_snapshot_created: bool audit_recorded: bool + successful_assets: int + fresh_unchanged_assets: int + stale_assets: int + failed_assets: int + next_action: str provider_calls_on_read: Literal[False] class MarketStatusResponse(BaseModel): equity_latest_update: str | None = None crypto_latest_update: str | None = None equity_cached_points: int = 0 crypto_cached_points: int = 0 mapped_equity_instruments: int = 0 mapped_crypto_assets: int = 0 render_provider_calls: bool = False warnings: list[str] = Field(default_factory=list) class QuoteRefreshRequest(BaseModel): provider: str = "auto" currency: str = "CHF" range: str = "1d" interval: str = "5m" limit: int = Field(default=100, ge=1, le=500) price_date: str | None = None only_missing: bool = True stale_before: str | None = None max_retries: int = Field(default=2, ge=0, le=3) pacing_seconds: float = Field(default=0.6, ge=0, le=5) max_parallelism: int = Field(default=3, ge=1, le=4) dry_run: bool = False class MarketBatchUpdateResponse(BaseModel): action: str provider: str mode: str = "apply" requested_at: str | None = None completed_at: str | None = None total: int = 0 updated: int = 0 skipped: int = 0 warnings: list[str] = Field(default_factory=list) errors: list[str] = Field(default_factory=list) target_date: str | None = None result_price_date_from: str | None = None result_price_date_to: str | None = None eligible_total: int = 0 limit_applied: bool = False provider_calls: int = 0 would_update: int = 0 persistence_performed: bool = False cached: int = 0 processed: int = 0 valued: int = 0 coverage_total: int = 0 complete: bool = False results: list[dict[str, str | int | bool | None]] = Field(default_factory=list) render_provider_calls: bool = False class MarketQuoteResponse(BaseModel): latest_price: str | None = None currency: str | None = None change_abs: str | None = None change_pct: str | None = None open: str | None = None high: str | None = None low: str | None = None close: str | None = None volume: str | None = None provider: str | None = None provider_symbol: str | None = None fetched_at: str | None = None quality_status: str = "missing" warnings: list[str] = Field(default_factory=list) chart_points: list[dict[str, str | None]] = Field(default_factory=list) class ChartPoint(BaseModel): timestamp: str price: str currency: str provider: str | None = None diff --git a/src/jarvis_finance/services/portfolio_analytics.py b/src/jarvis_finance/services/portfolio_analytics.py index a5b457b..1005cbc 100644 --- a/src/jarvis_finance/services/portfolio_analytics.py +++ b/src/jarvis_finance/services/portfolio_analytics.py @@ -320,215 +320,230 @@ def _active_policy(conn: Connection) -> dict[str, Any] | None: try: result["restrictions"] = json.loads(result.get("restrictions_json") or "[]") except json.JSONDecodeError: result["restrictions"] = [] return result def _benchmark_mapping(conn: Connection, policy: dict[str, Any] | None) -> tuple[dict[str, Any] | None, str | None]: if not policy or not policy.get("benchmarks"): return None, "benchmark_not_configured" benchmarks = policy["benchmarks"] if len(benchmarks) != 1 or not isinstance(benchmarks[0], dict): return None, "benchmark_mapping_required" reference = str(benchmarks[0].get("reference") or "").strip() if not reference: return None, "benchmark_mapping_required" rows = conn.execute( """ SELECT i.instrument_id, i.isin, lower(i.asset_class) AS asset_class, m.provider, m.provider_symbol, m.provider_market, upper(COALESCE(m.trading_currency, m.currency, i.trading_currency, i.currency, '')) AS currency FROM instruments i JOIN instrument_price_mappings m ON m.instrument_id=i.instrument_id WHERE i.is_active=1 AND m.mapping_status='mapped' AND COALESCE(i.instrument_status,'active') NOT IN ('inactive','delisted','suspended','merged') AND COALESCE(i.valuation_policy,'')!='exclude_from_auto_price_update' AND (i.isin=? OR i.instrument_id=? OR lower(i.name)=lower(?)) AND m.provider_symbol IS NOT NULL AND trim(m.provider_symbol)!='' ORDER BY m.updated_at DESC, m.mapping_id """, (reference, reference, reference), ).fetchall() if len(rows) != 1: return None, "benchmark_mapping_required" result = dict(rows[0]) result["reference"] = reference return result, None def _fingerprint( positions: list[ConfirmedPosition], policy: dict[str, Any] | None, cash_entries: list[dict[str, Any]], ) -> str: payload = { "positions": [ {"account_id": p.account_id, "instrument_id": p.instrument_id, "isin": p.isin, "quantity": _fmt(p.quantity)} for p in positions ], "policy_id": policy.get("policy_id") if policy else None, "benchmarks": policy.get("benchmarks") if policy else [], "cash": [ { "account_id": item["account_id"], "currency": item["currency"], "amount_chf": _fmt(item["amount_chf"]), "balance_date": item["balance_date"], "snapshot_id": item["snapshot_id"], } for item in cash_entries ], } return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest() def store_valuation_snapshot( conn: Connection, *, id_prefix: str, run_id: str, scope_kind: str, scope_id: str, account_id: str, value_original: str, currency: str, fx_rate_to_base: str, valuation_at: str, source: str, captured_at: str, quality_status: str, reason_codes: list[str], + source_observation_id: str | None = None, ) -> bool: reasons_json = json.dumps(sorted(set(reason_codes))) - observation = conn.execute( - """SELECT * FROM portfolio_valuation_snapshots - WHERE scope_kind=? AND scope_id=? AND account_id=? AND substr(valuation_at,1,10)=? - AND source=? AND source_reference=? - ORDER BY snapshot_version DESC,captured_at DESC,snapshot_id DESC LIMIT 1""", - (scope_kind, scope_id, account_id, valuation_at[:10], source, run_id), + economic_payload = { + "scope_kind": scope_kind, + "scope_id": scope_id, + "account_id": account_id, + "value_original": value_original, + "currency": currency, + "fx_rate_to_base": fx_rate_to_base, + "valuation_at": valuation_at, + "source": source, + "quality_status": quality_status, + "reason_codes": json.loads(reasons_json), + } + payload_hash = hashlib.sha256( + json.dumps(economic_payload, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + observation_identity = source_observation_id or stable_id( + "valuation-source-observation", + source, + scope_kind, + scope_id, + account_id, + valuation_at, + run_id, + ) + exact = conn.execute( + """SELECT 1 FROM portfolio_valuation_snapshots + WHERE source_observation_id=? AND economic_payload_hash=? LIMIT 1""", + (observation_identity, payload_hash), ).fetchone() - if observation: - same_payload = ( - str(observation["value_original"]) == value_original - and str(observation["currency"]) == currency - and str(observation["fx_rate_to_base"]) == fx_rate_to_base - and str(observation["quality_status"]) == quality_status - and str(observation["reason_codes_json"]) == reasons_json - ) - if same_payload: - return False - raise ValueError("valuation source observation conflicts with its existing payload") + if exact: + return False latest = conn.execute( """SELECT * FROM portfolio_valuation_snapshots WHERE scope_kind=? AND scope_id=? AND account_id=? AND substr(valuation_at,1,10)=? ORDER BY snapshot_version DESC,captured_at DESC,snapshot_id DESC LIMIT 1""", (scope_kind, scope_id, account_id, valuation_at[:10]), ).fetchone() if latest and ( str(latest["value_original"]) == value_original and str(latest["currency"]) == currency and str(latest["fx_rate_to_base"]) == fx_rate_to_base and str(latest["quality_status"]) == quality_status and str(latest["reason_codes_json"]) == reasons_json and str(latest["source"]) == source - and str(latest["source_reference"] or "") == run_id ): return False version = int( conn.execute( """SELECT COALESCE(MAX(snapshot_version),0)+1 FROM portfolio_valuation_snapshots WHERE scope_kind=? AND scope_id=? AND substr(valuation_at,1,10)=?""", (scope_kind, scope_id, valuation_at[:10]), ).fetchone()[0] ) snapshot_id = stable_id(id_prefix, run_id, account_id, scope_id, str(version)) conn.execute( """INSERT INTO portfolio_valuation_snapshots( snapshot_id,scope_kind,scope_id,account_id,value_original,currency,base_currency, fx_rate_to_base,fx_direction,valuation_at,source,captured_at,snapshot_version, - supersedes_snapshot_id,source_reference,quality_status,reason_codes_json) - VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + supersedes_snapshot_id,source_reference,quality_status,reason_codes_json, + source_observation_id,economic_payload_hash) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( snapshot_id, scope_kind, scope_id, account_id, value_original, currency, "CHF", fx_rate_to_base, "original_to_base", valuation_at, source, captured_at, version, str(latest["snapshot_id"]) if latest else None, run_id, quality_status, reasons_json, + observation_identity, payload_hash, ), ) return True @contextmanager def _exclusive_lock(lock_path: Path): lock_path.parent.mkdir(parents=True, exist_ok=True) with lock_path.open("a+") as handle: try: fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as exc: raise RuntimeError("market_job_already_running") from exc try: yield finally: fcntl.flock(handle.fileno(), fcntl.LOCK_UN) def _result_from_row(row: Any, *, idempotent: bool) -> MarketRunResult: return MarketRunResult( run_id=row["run_id"], status=row["status"], as_of=row["as_of"], price_total=int(row["price_total"]), price_stored=int(row["price_stored"]), fx_total=int(row["fx_total"]), fx_stored=int(row["fx_stored"]), benchmark_total=int(row["benchmark_total"]), benchmark_stored=int(row["benchmark_stored"]), valuation_stored=int(row["valuation_stored"]), missing_instruments=tuple(json.loads(row["missing_instruments_json"] or "[]")), reason_codes=tuple(json.loads(row["reason_codes_json"] or "[]")), idempotent=idempotent, ) def run_daily_market_valuation( conn: Connection, *, as_of: str | None = None, price_providers: dict[str, EquityPriceProvider] | None = None, fx_provider: Any | None = None, lock_path: Path | None = None, ) -> MarketRunResult: """Refresh market/FX inputs and immutable valuation analysis for confirmed positions only.""" requested = date.fromisoformat(as_of) if as_of else datetime.now(timezone.utc).date() effective = _effective_business_date(requested) lock = lock_path or Path("/tmp/jarvis-finance-market-job.lock") with _exclusive_lock(lock): positions = confirmed_canonical_positions(conn, as_of=effective.isoformat()) policy = _active_policy(conn) cash_entries = _latest_cash_entries(conn, as_of=effective) fingerprint = _fingerprint(positions, policy, cash_entries) existing = conn.execute( "SELECT * FROM market_data_runs WHERE source_key=? AND as_of=? AND input_fingerprint=?", (SOURCE_KEY, effective.isoformat(), fingerprint), ).fetchone() if existing and existing["status"] == "complete": return _result_from_row(existing, idempotent=True) now = utc_now() run_id = stable_id("market-run", SOURCE_KEY, effective.isoformat(), fingerprint) if not existing: conn.execute( "INSERT INTO market_data_runs(run_id,source_key,as_of,input_fingerprint,status,started_at) VALUES(?,?,?,?,?,?)", (run_id, SOURCE_KEY, effective.isoformat(), fingerprint, "running", now), ) conn.commit() else: run_id = existing["run_id"] missing: list[dict[str, str]] = [] position_details = {position.instrument_id: position for position in positions} reasons: set[str] = set() quotes: dict[tuple[str, str], tuple[ConfirmedPosition, EquityPriceQuote, date, dict[str, str]]] = {} cached_price_inputs: dict[str, dict[str, str | None]] = {} price_writes = 0 providers = price_providers or {} for position in positions: mapping, issue = _mapping(conn, position) if issue or not mapping: reason = issue or "mapping_required" missing.append({"instrument_id": position.instrument_id, "reason_code": reason}) reasons.add(reason) @@ -593,213 +608,241 @@ def run_daily_market_valuation( quote_age = _business_day_age(quote_as_of, effective) quality = "fresh" if 0 <= quote_age <= 2 else "stale" if quality == "stale": reasons.add("stale_price") existing_price = conn.execute( """SELECT close,currency,provider_symbol FROM market_prices WHERE instrument_id=? AND price_date=? AND provider=?""", (position.instrument_id, quote_as_of.isoformat(), quote.provider or mapping["provider"]), ).fetchone() same_persisted_price = bool( existing_price and _decimal(existing_price["close"]) == quote.close and str(existing_price["currency"] or "").upper() == quote_currency and str(existing_price["provider_symbol"] or "") == str(mapping["provider_symbol"] or "") ) if not from_cache and not same_persisted_price: store_market_price( conn, instrument_id=position.instrument_id, price_date=quote_as_of.isoformat(), close=quote.close, adjusted_close=quote.adjusted_close, currency=quote_currency, provider=quote.provider or mapping["provider"], provider_symbol=quote.provider_symbol or mapping["provider_symbol"], provider_market=quote.provider_market or mapping["provider_market"], price_timestamp=quote.price_timestamp, quality_status=quality, error_message=quote.error_message, fetched_at=now, price_type="unadjusted_close", run_id=run_id, ) price_writes += 1 quotes[(position.account_id, position.instrument_id)] = (position, quote, quote_as_of, mapping) currencies = sorted({quote.currency.upper() for _, quote, _, _ in quotes.values()}) rates: dict[str, Decimal] = {} rate_dates: dict[str, date] = {} fx = fx_provider or FrankfurterFxProvider() fx_writes = 0 for currency in currencies: provider_name = "identity" if currency == "CHF" else getattr(fx, "name", "fx_provider") rate, rate_as_of = _fetch_fx_rate(fx, currency, effective) if rate is None or rate_as_of is None: reasons.add("fx_rate_missing") continue rates[currency] = rate rate_dates[currency] = rate_as_of rate_age = _business_day_age(rate_as_of, effective) rate_quality = "fresh" if 0 <= rate_age <= 2 else "stale" if rate_quality == "stale": reasons.add("stale_fx") existing_rate = conn.execute( """SELECT rate FROM fx_rates WHERE base_currency=? AND quote_currency='CHF' AND rate_date=? AND provider=? AND rate_type='close'""", (currency, rate_as_of.isoformat(), provider_name), ).fetchone() if not existing_rate or _decimal(existing_rate["rate"]) != rate: upsert_fx_rate( conn, base_currency=currency, quote_currency="CHF", rate_date=rate_as_of.isoformat(), rate=rate, provider=provider_name, rate_type="close", quality_status=rate_quality, fetched_at=now, run_id=run_id, ) fx_writes += 1 conn.commit() values: list[dict[str, Any]] = [] account_values: dict[str, Decimal] = defaultdict(lambda: ZERO) account_missing: set[str] = set() for (_, instrument_id), (position, quote, quote_as_of, _) in quotes.items(): rate = rates.get(quote.currency.upper()) close = quote.close if rate is None or close is None: missing.append({"instrument_id": instrument_id, "reason_code": "fx_rate_missing"}) account_missing.add(position.account_id) continue value_original = position.quantity * close value_chf = value_original * rate account_values[position.account_id] += value_chf values.append({ "account_id": position.account_id, "instrument_id": instrument_id, "isin": position.isin, "name": position.name, "asset_class": position.asset_class, "currency": quote.currency.upper(), "quantity": _fmt(position.quantity), "close": _fmt(close), "fx_rate_to_chf": _fmt(rate), "value_chf": _fmt(value_chf), "as_of": quote_as_of.isoformat(), "quality_status": "fresh" if 0 <= _business_day_age(quote_as_of, effective) <= 2 else "stale", "provider": quote.provider, "provider_symbol": quote.provider_symbol, "provider_market": quote.provider_market, + "source_observation_id": stable_id( + "valuation-source-observation", + SOURCE_KEY, + position.account_id, + instrument_id, + str(quote.provider or ""), + str(quote.provider_symbol or ""), + str(quote.price_timestamp or quote_as_of.isoformat()), + quote.currency.upper(), + rate_dates[quote.currency.upper()].isoformat(), + ), **({"price_input_provenance": cached_price_inputs[instrument_id]} if instrument_id in cached_price_inputs else {}), }) missing_ids = {item["instrument_id"] for item in missing} for position in positions: if position.instrument_id in missing_ids: account_missing.add(position.account_id) for account_id, amount in _latest_cash_by_account(conn, as_of=effective).items(): account_values[account_id] += amount for item in missing: position = position_details.get(item["instrument_id"]) if position: item["label"] = position.name item["isin"] = position.isin or "" valuation_stored = 0 for item in values: value_original = _decimal(item["quantity"]) * _decimal(item["close"]) is_stale = item["quality_status"] == "stale" valuation_stored += int(store_valuation_snapshot( conn, id_prefix="instrument-valuation", run_id=run_id, scope_kind="instrument", scope_id=item["instrument_id"], account_id=item["account_id"], value_original=_fmt(value_original) or "0", currency=item["currency"], fx_rate_to_base=item["fx_rate_to_chf"], valuation_at=effective.isoformat(), source=SOURCE_KEY, captured_at=now, quality_status="partial" if is_stale else "complete", reason_codes=["stale_price"] if is_stale else [], + source_observation_id=item["source_observation_id"], )) for account_id, total in sorted(account_values.items()): if account_id in account_missing: reasons.add("account_valuation_not_materialized_incomplete_inputs") continue valuation_stored += int(store_valuation_snapshot( conn, id_prefix="portfolio-valuation", run_id=run_id, scope_kind="account", scope_id=account_id, account_id=account_id, value_original=_fmt(total) or "0", currency="CHF", fx_rate_to_base="1", valuation_at=effective.isoformat(), source=SOURCE_KEY, captured_at=now, quality_status="complete", reason_codes=[], + source_observation_id=stable_id( + "valuation-account-observation", + SOURCE_KEY, + account_id, + effective.isoformat(), + *sorted( + str(item["source_observation_id"]) + for item in values + if item["account_id"] == account_id + ), + *sorted( + str(entry["snapshot_id"]) + for entry in _latest_cash_entries(conn, as_of=effective) + if entry["account_id"] == account_id + ), + ), )) benchmark_mapping, benchmark_issue = _benchmark_mapping(conn, policy) benchmark_total = 1 if policy and policy.get("benchmarks") else 0 benchmark_stored = 0 if benchmark_issue: reasons.add(benchmark_issue) elif benchmark_mapping and policy: try: provider = providers.get(benchmark_mapping["provider"]) or equity_price_provider_by_name(benchmark_mapping["provider"]) quote = provider.get_price(benchmark_mapping["provider_symbol"], price_date=effective.isoformat()) except Exception: quote = None benchmark_quote_date = _quote_date(quote, effective) if quote else effective expected_benchmark_currency = str(benchmark_mapping["currency"] or "").upper() if quote and benchmark_quote_date > effective: reasons.add("benchmark_future_price") quote = None elif quote and _business_day_age(benchmark_quote_date, effective) > 2: reasons.add("benchmark_stale_price") quote = None elif quote and expected_benchmark_currency and str(quote.currency or "").upper() != expected_benchmark_currency: reasons.add("benchmark_currency_mismatch") quote = None elif quote and not exchange_matches(str(benchmark_mapping["provider_market"] or ""), quote.provider_market): reasons.add("benchmark_exchange_mismatch") quote = None if not quote or quote.close is None or quote.close <= ZERO: reasons.add("benchmark_quote_missing") elif quote.currency.upper() not in rates: benchmark_currency = quote.currency.upper() try: benchmark_rate, benchmark_rate_as_of = _fetch_fx_rate(fx, benchmark_currency, effective) except Exception: benchmark_rate = None benchmark_rate_as_of = None if benchmark_rate is None or benchmark_rate_as_of is None: reasons.add("benchmark_fx_missing") else: rates[benchmark_currency] = benchmark_rate rate_dates[benchmark_currency] = benchmark_rate_as_of if benchmark_currency not in currencies: currencies.append(benchmark_currency) existing_benchmark_rate = conn.execute( """SELECT rate FROM fx_rates WHERE base_currency=? AND quote_currency='CHF' AND rate_date=? AND provider=? AND rate_type='close'""", ( benchmark_currency, benchmark_rate_as_of.isoformat(), "identity" if benchmark_currency == "CHF" else getattr(fx, "name", "fx_provider"), ), ).fetchone() if not existing_benchmark_rate or _decimal(existing_benchmark_rate["rate"]) != benchmark_rate: upsert_fx_rate( conn, base_currency=benchmark_currency, quote_currency="CHF", rate_date=benchmark_rate_as_of.isoformat(), rate=benchmark_rate, provider="identity" if benchmark_currency == "CHF" else getattr(fx, "name", "fx_provider"), rate_type="close", quality_status="fresh", fetched_at=now, run_id=run_id, ) fx_writes += 1 if quote and quote.close is not None and quote.close > ZERO and quote.currency.upper() in rates: return_type = "etf_proxy" if benchmark_mapping["asset_class"] == "etf" else ("total_return" if quote.adjusted_close is not None else "price_return") if return_type == "total_return": assert quote.adjusted_close is not None benchmark_level = quote.adjusted_close else: benchmark_level = quote.close benchmark_value = benchmark_level * rates[quote.currency.upper()] conn.execute( """INSERT OR REPLACE INTO benchmark_snapshots( benchmark_snapshot_id,run_id,policy_id,benchmark_reference,provider,provider_symbol, price_currency,close,adjusted_close,fx_rate_to_chf,value_chf,return_type,as_of,source_as_of,fetched_at, quality_status,reason_codes_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (stable_id("benchmark", run_id, policy["policy_id"], benchmark_mapping["reference"]), run_id, policy["policy_id"], benchmark_mapping["reference"], quote.provider, benchmark_mapping["provider_symbol"], quote.currency.upper(), _fmt(quote.close), _fmt(quote.adjusted_close), _fmt(rates[quote.currency.upper()]), _fmt(benchmark_value), return_type, effective.isoformat(), benchmark_quote_date.isoformat(), now, "fresh", "[]"), ) benchmark_stored = 1 diff --git a/src/jarvis_finance/storage/migrations.py b/src/jarvis_finance/storage/migrations.py index 9f8994b..806bd30 100644 --- a/src/jarvis_finance/storage/migrations.py +++ b/src/jarvis_finance/storage/migrations.py @@ -1,92 +1,92 @@ 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 = 52 -MIGRATION_NAME = "052_professional_portfolio_cockpit_v1" +MIGRATION_VERSION = 53 +MIGRATION_NAME = "053_asset_refresh_observation_idempotency" 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 = { "last_price": "TEXT", "price_currency": "TEXT", "price_date": "TEXT", "price_source": "TEXT", "exchange_name": "TEXT", "mic": "TEXT", "security_type": "TEXT", } TEXT_AFFINITY_COLUMNS = { "transactions": {"quantity", "price_original", "gross_amount_original", "fee_original", "tax_original", "net_amount_original", "fx_rate_to_chf", "gross_amount_chf", "fee_chf", "tax_chf", "net_amount_chf"}, "crypto_holdings": {"quantity", "legacy_snapshot_value_original", "legacy_snapshot_value_chf"}, "crypto_transactions": {"quantity", "price_original", "gross_amount_original", "fee_quantity", "fee_original", "fx_rate_to_chf", "amount_chf"}, "crypto_prices": {"price", "market_cap", "volume_24h", "change_24h_pct"}, "fx_rates": {"rate"}, "market_prices": {"open", "high", "low", "close", "adjusted_close"}, "equity_price_points": {"price"}, "equity_intraday_candles": {"open", "close", "low", "high", "volume"}, "crypto_price_points": {"price"}, } def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def checksum_sql(sql: str) -> str: return hashlib.sha256(sql.encode("utf-8")).hexdigest() def get_schema_version(conn: Connection) -> int: row = conn.execute("SELECT MAX(version) AS version FROM schema_migrations").fetchone() return int(row["version"] or 0) if row else 0 def _table_columns(conn: Connection, table: str) -> dict[str, str]: return {row["name"]: (row["type"] or "") for row in conn.execute(f"PRAGMA table_info({table})").fetchall()} def _add_missing_instrument_columns(conn: Connection) -> None: existing = _table_columns(conn, "instruments") for name, col_type in INSTRUMENT_OPTIONAL_COLUMNS.items(): if name not in existing: conn.execute(f"ALTER TABLE instruments ADD COLUMN {name} {col_type}") def _rebuild_table_with_text_columns(conn: Connection, table: str, text_columns: set[str]) -> None: cols = conn.execute(f"PRAGMA table_info({table})").fetchall() if not cols: return needs_rebuild = any(row["name"] in text_columns and (row["type"] or "").upper() != "TEXT" for row in cols) if not needs_rebuild: return tmp = f"{table}__text_migration" col_defs: list[str] = [] pk_cols = [row["name"] for row in cols if row["pk"]] for row in cols: name = row["name"] col_type = "TEXT" if name in text_columns else (row["type"] or "TEXT") parts = [name, col_type] @@ -2886,140 +2886,204 @@ def _create_professional_portfolio_cockpit_v1(conn: Connection) -> None: 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 _create_asset_refresh_observation_v2(conn: Connection) -> None: + """Append-only economic quotes plus richer job quality counters.""" + + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS market_price_observations ( + observation_id TEXT PRIMARY KEY, + source_observation_id TEXT NOT NULL, + payload_version INTEGER NOT NULL CHECK(payload_version > 0), + supersedes_observation_id TEXT REFERENCES market_price_observations(observation_id), + instrument_id TEXT NOT NULL REFERENCES instruments(instrument_id), + provider TEXT NOT NULL, + provider_symbol TEXT, + provider_market TEXT, + price_type TEXT NOT NULL, + close TEXT NOT NULL, + adjusted_close TEXT, + currency TEXT NOT NULL CHECK(length(currency)=3 AND currency=upper(currency)), + provider_timestamp TEXT NOT NULL, + source_reference TEXT, + job_reference TEXT, + quality_status TEXT NOT NULL, + economic_payload_json TEXT NOT NULL CHECK(json_valid(economic_payload_json)), + economic_payload_hash TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(source_observation_id, payload_version), + UNIQUE(source_observation_id, economic_payload_hash) + ); + CREATE INDEX IF NOT EXISTS idx_market_price_observations_instrument_time + ON market_price_observations(instrument_id,provider_timestamp,created_at); + CREATE TRIGGER IF NOT EXISTS market_price_observations_no_update + BEFORE UPDATE ON market_price_observations + BEGIN SELECT RAISE(ABORT, 'market price observations are immutable'); END; + CREATE TRIGGER IF NOT EXISTS market_price_observations_no_delete + BEFORE DELETE ON market_price_observations + BEGIN SELECT RAISE(ABORT, 'market price observations cannot be deleted'); END; + """ + ) + _add_missing_columns( + conn, + "market_price_observations", + {"job_reference": "TEXT"}, + ) + _add_missing_columns( + conn, + "portfolio_valuation_snapshots", + { + "source_observation_id": "TEXT", + "economic_payload_hash": "TEXT", + }, + ) + _add_missing_columns( + conn, + "asset_price_refresh_sources", + { + "fresh_unchanged_count": "INTEGER NOT NULL DEFAULT 0", + "stale_remaining_count": "INTEGER NOT NULL DEFAULT 0", + "failed_count": "INTEGER NOT NULL DEFAULT 0", + "diagnostics_json": "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) _create_budget_categories_ux_fix_tables(conn) _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) + _create_asset_refresh_observation_v2(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() diff --git a/src/jarvis_finance/storage/schema.py b/src/jarvis_finance/storage/schema.py index d185981..a95c339 100644 --- a/src/jarvis_finance/storage/schema.py +++ b/src/jarvis_finance/storage/schema.py @@ -1,10 +1,10 @@ INITIAL_SCHEMA_SQL = "\nCREATE TABLE IF NOT EXISTS schema_migrations (\n version INTEGER PRIMARY KEY,\n name TEXT NOT NULL,\n applied_at TEXT NOT NULL,\n checksum TEXT NOT NULL\n);\n\nCREATE TABLE IF NOT EXISTS platforms (\n platform_id TEXT PRIMARY KEY,\n name TEXT NOT NULL UNIQUE,\n platform_type TEXT NOT NULL,\n country TEXT,\n default_currency TEXT NOT NULL DEFAULT 'CHF',\n is_active INTEGER NOT NULL DEFAULT 1,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS accounts (\n account_id TEXT PRIMARY KEY,\n platform_id TEXT NOT NULL REFERENCES platforms(platform_id),\n account_name TEXT NOT NULL,\n account_type TEXT NOT NULL,\n currency TEXT NOT NULL DEFAULT 'CHF',\n performance_included INTEGER NOT NULL DEFAULT 0,\n is_health_reserve INTEGER NOT NULL DEFAULT 0,\n target_cash_min_chf NUMERIC,\n target_cash_max_chf NUMERIC,\n is_active INTEGER NOT NULL DEFAULT 1,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT,\n UNIQUE(platform_id, account_name)\n);\n\nCREATE TABLE IF NOT EXISTS instruments (\n instrument_id TEXT PRIMARY KEY,\n asset_class TEXT NOT NULL,\n name TEXT NOT NULL,\n ticker TEXT,\n isin TEXT,\n exchange TEXT,\n currency TEXT NOT NULL,\n country TEXT,\n sector TEXT,\n industry TEXT,\n provider_symbol TEXT,\n data_provider_primary TEXT,\n position_category TEXT,\n ter TEXT,\n distribution_policy TEXT,\n index_name TEXT,\n fund_domicile TEXT,\n benchmark TEXT,\n data_provider_fallback TEXT,\n is_active INTEGER NOT NULL DEFAULT 1,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS transactions (\n transaction_id TEXT PRIMARY KEY,\n transaction_type TEXT NOT NULL,\n account_id TEXT NOT NULL REFERENCES accounts(account_id),\n instrument_id TEXT REFERENCES instruments(instrument_id),\n trade_date TEXT NOT NULL,\n settlement_date TEXT,\n quantity TEXT,\n price_original TEXT,\n gross_amount_original TEXT,\n fee_original TEXT DEFAULT '0',\n tax_original TEXT DEFAULT '0',\n net_amount_original TEXT,\n currency_original TEXT NOT NULL,\n fx_rate_to_chf TEXT,\n fx_source TEXT,\n fx_status TEXT NOT NULL DEFAULT 'ok',\n gross_amount_chf TEXT,\n fee_chf TEXT,\n tax_chf TEXT,\n net_amount_chf TEXT,\n source_type TEXT NOT NULL,\n source_id TEXT,\n external_transaction_id TEXT,\n row_hash TEXT,\n is_confirmed INTEGER NOT NULL DEFAULT 1,\n quality_status TEXT NOT NULL DEFAULT 'ok',\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS positions_snapshot (\n position_snapshot_id TEXT PRIMARY KEY,\n snapshot_date TEXT NOT NULL,\n account_id TEXT NOT NULL REFERENCES accounts(account_id),\n platform_id TEXT NOT NULL REFERENCES platforms(platform_id),\n instrument_id TEXT NOT NULL REFERENCES instruments(instrument_id),\n quantity NUMERIC NOT NULL,\n average_cost_original NUMERIC,\n cost_basis_original NUMERIC,\n cost_basis_chf NUMERIC,\n market_price_original NUMERIC,\n market_value_original NUMERIC,\n market_fx_rate_to_chf NUMERIC,\n market_value_chf NUMERIC,\n unrealized_price_pnl_chf NUMERIC,\n unrealized_fx_pnl_chf NUMERIC,\n realized_pnl_chf NUMERIC,\n income_chf NUMERIC,\n fees_chf NUMERIC,\n taxes_chf NUMERIC,\n total_return_chf NUMERIC,\n portfolio_weight_pct NUMERIC,\n category TEXT,\n data_quality_status TEXT NOT NULL DEFAULT 'ok',\n created_at TEXT NOT NULL,\n UNIQUE(snapshot_date, account_id, instrument_id)\n);\n\nCREATE TABLE IF NOT EXISTS cash_balances (\n cash_balance_id TEXT PRIMARY KEY,\n account_id TEXT NOT NULL REFERENCES accounts(account_id),\n balance_date TEXT NOT NULL,\n currency TEXT NOT NULL,\n amount_original NUMERIC NOT NULL,\n fx_rate_to_chf NUMERIC,\n amount_chf NUMERIC,\n source_type TEXT NOT NULL,\n quality_status TEXT NOT NULL DEFAULT 'ok',\n notes TEXT,\n created_at TEXT NOT NULL,\n UNIQUE(account_id, balance_date, currency, source_type)\n);\n\nCREATE TABLE IF NOT EXISTS fx_rates (\n fx_rate_id TEXT PRIMARY KEY,\n base_currency TEXT NOT NULL,\n quote_currency TEXT NOT NULL DEFAULT 'CHF',\n rate_date TEXT NOT NULL,\n rate_timestamp TEXT,\n rate NUMERIC NOT NULL,\n provider TEXT NOT NULL,\n rate_type TEXT NOT NULL,\n quality_status TEXT NOT NULL DEFAULT 'ok',\n created_at TEXT NOT NULL,\n UNIQUE(base_currency, quote_currency, rate_date, provider, rate_type)\n);\n\nCREATE TABLE IF NOT EXISTS market_prices (\n market_price_id TEXT PRIMARY KEY,\n instrument_id TEXT NOT NULL REFERENCES instruments(instrument_id),\n price_date TEXT NOT NULL,\n price_timestamp TEXT,\n open NUMERIC,\n high NUMERIC,\n low NUMERIC,\n close NUMERIC NOT NULL,\n adjusted_close NUMERIC,\n currency TEXT NOT NULL,\n provider TEXT NOT NULL,\n provider_symbol TEXT,\n quality_status TEXT NOT NULL DEFAULT 'ok',\n created_at TEXT NOT NULL,\n UNIQUE(instrument_id, price_date, provider)\n);\n\nCREATE TABLE IF NOT EXISTS crypto_wallets (\n wallet_id TEXT PRIMARY KEY,\n wallet_name TEXT NOT NULL UNIQUE,\n wallet_type TEXT NOT NULL,\n platform_provider TEXT,\n network_chain TEXT,\n wallet_address TEXT,\n owner TEXT,\n is_active INTEGER NOT NULL DEFAULT 1,\n last_verified_at TEXT,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS crypto_assets (\n asset_id TEXT PRIMARY KEY,\n coin_name TEXT NOT NULL,\n symbol TEXT NOT NULL,\n coingecko_id TEXT UNIQUE,\n network_chain_default TEXT,\n is_stablecoin INTEGER NOT NULL DEFAULT 0,\n price_provider_primary TEXT NOT NULL DEFAULT 'CoinGecko',\n price_provider_fallback TEXT,\n is_active INTEGER NOT NULL DEFAULT 1,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS crypto_holdings (\n crypto_holding_id TEXT PRIMARY KEY,\n asset_id TEXT NOT NULL REFERENCES crypto_assets(asset_id),\n wallet_id TEXT NOT NULL REFERENCES crypto_wallets(wallet_id),\n quantity TEXT NOT NULL,\n acquisition_source TEXT,\n last_verified_at TEXT,\n verification_status TEXT NOT NULL DEFAULT 'unverified',\n legacy_snapshot_value_original TEXT,\n legacy_snapshot_value_chf TEXT,\n legacy_snapshot_currency TEXT,\n legacy_snapshot_date TEXT,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT,\n UNIQUE(asset_id, wallet_id)\n);\n\nCREATE TABLE IF NOT EXISTS crypto_transactions (\n crypto_transaction_id TEXT PRIMARY KEY,\n transaction_id TEXT REFERENCES transactions(transaction_id),\n transaction_type TEXT NOT NULL,\n asset_id TEXT NOT NULL REFERENCES crypto_assets(asset_id),\n quantity TEXT NOT NULL,\n price_original TEXT,\n currency_original TEXT,\n gross_amount_original TEXT,\n fee_quantity TEXT,\n fee_original TEXT,\n fee_currency TEXT,\n fx_rate_to_chf TEXT,\n fx_source TEXT,\n amount_chf TEXT,\n from_wallet_id TEXT REFERENCES crypto_wallets(wallet_id),\n to_wallet_id TEXT REFERENCES crypto_wallets(wallet_id),\n transaction_datetime TEXT NOT NULL,\n tx_hash TEXT,\n source TEXT NOT NULL,\n confirmation_status TEXT NOT NULL DEFAULT 'confirmed',\n parse_confidence NUMERIC,\n original_input_text TEXT,\n notes TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS crypto_prices (\n crypto_price_id TEXT PRIMARY KEY,\n asset_id TEXT NOT NULL REFERENCES crypto_assets(asset_id),\n coingecko_id TEXT,\n price_currency TEXT NOT NULL,\n price TEXT NOT NULL,\n provider TEXT NOT NULL DEFAULT 'CoinGecko',\n provider_timestamp TEXT,\n fetched_at TEXT NOT NULL,\n quality_status TEXT NOT NULL DEFAULT 'fresh',\n error_message TEXT\n);\n\nCREATE TABLE IF NOT EXISTS watchlist (\n watchlist_id TEXT PRIMARY KEY,\n instrument_id TEXT REFERENCES instruments(instrument_id),\n crypto_asset_id TEXT REFERENCES crypto_assets(asset_id),\n name TEXT NOT NULL,\n asset_class TEXT NOT NULL,\n reason TEXT NOT NULL,\n target_entry_price NUMERIC,\n target_entry_currency TEXT,\n desired_position_size_chf NUMERIC,\n desired_weight_pct NUMERIC,\n trigger_rules TEXT,\n risk_notes TEXT,\n investment_case TEXT,\n bear_case TEXT,\n sources TEXT,\n status TEXT NOT NULL DEFAULT 'active',\n next_review_date TEXT,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE TABLE IF NOT EXISTS reports (\n report_id TEXT PRIMARY KEY,\n report_type TEXT NOT NULL,\n title TEXT NOT NULL,\n period_start TEXT,\n period_end TEXT,\n generated_at TEXT NOT NULL,\n file_path TEXT,\n format TEXT NOT NULL,\n data_quality_status TEXT NOT NULL DEFAULT 'ok',\n summary_json TEXT,\n created_at TEXT NOT NULL\n);\n\nCREATE TABLE IF NOT EXISTS alerts (\n alert_id TEXT PRIMARY KEY,\n priority TEXT NOT NULL,\n category TEXT NOT NULL,\n entity_type TEXT,\n entity_id TEXT,\n rule_id TEXT,\n message TEXT NOT NULL,\n evidence_json TEXT,\n status TEXT NOT NULL DEFAULT 'active',\n created_at TEXT NOT NULL,\n resolved_at TEXT,\n muted_until TEXT,\n last_seen_at TEXT,\n occurrence_count INTEGER NOT NULL DEFAULT 1,\n fingerprint TEXT,\n dedup_key TEXT\n);\n\nCREATE INDEX IF NOT EXISTS idx_alerts_dedup ON alerts(dedup_key, status);\n\nCREATE TABLE IF NOT EXISTS audit_log (\n audit_id TEXT PRIMARY KEY,\n timestamp TEXT NOT NULL,\n source TEXT NOT NULL,\n action TEXT NOT NULL,\n entity_type TEXT NOT NULL,\n entity_id TEXT NOT NULL,\n old_values_json TEXT,\n new_values_json TEXT,\n user_text_note TEXT,\n original_input_text TEXT,\n confirmed INTEGER NOT NULL DEFAULT 1,\n confirmation_timestamp TEXT,\n auto_parsed INTEGER NOT NULL DEFAULT 0,\n parse_confidence NUMERIC,\n created_by TEXT NOT NULL DEFAULT 'system',\n quality_status TEXT NOT NULL DEFAULT 'ok',\n created_at TEXT NOT NULL\n);\n\nCREATE TABLE IF NOT EXISTS import_sessions (\n import_session_id TEXT PRIMARY KEY,\n import_type TEXT NOT NULL,\n source_filename TEXT NOT NULL,\n source_hash TEXT,\n started_at TEXT NOT NULL,\n finished_at TEXT,\n status TEXT NOT NULL,\n rows_total INTEGER DEFAULT 0,\n rows_imported INTEGER DEFAULT 0,\n rows_failed INTEGER DEFAULT 0,\n errors_json TEXT,\n notes TEXT\n);\n\nCREATE TABLE IF NOT EXISTS decision_journal (\n decision_id TEXT PRIMARY KEY,\n decision_date TEXT NOT NULL,\n decision_type TEXT NOT NULL,\n entity_type TEXT,\n entity_id TEXT,\n system_recommendation TEXT,\n human_decision TEXT NOT NULL,\n rationale TEXT NOT NULL,\n investment_case TEXT,\n risks TEXT,\n exit_rule TEXT,\n alternatives_considered TEXT,\n sources TEXT,\n review_date TEXT,\n outcome_status TEXT,\n outcome_return_chf NUMERIC,\n created_at TEXT NOT NULL,\n updated_at TEXT\n);\n\nCREATE INDEX IF NOT EXISTS idx_accounts_platform_id ON accounts(platform_id);\nCREATE INDEX IF NOT EXISTS idx_transactions_account_date ON transactions(account_id, trade_date);\nCREATE INDEX IF NOT EXISTS idx_transactions_instrument_date ON transactions(instrument_id, trade_date);\nCREATE INDEX IF NOT EXISTS idx_transactions_type ON transactions(transaction_type);\nCREATE UNIQUE INDEX IF NOT EXISTS idx_transactions_external_id ON transactions(external_transaction_id) WHERE external_transaction_id IS NOT NULL AND external_transaction_id != '';\nCREATE UNIQUE INDEX IF NOT EXISTS idx_transactions_row_hash ON transactions(row_hash) WHERE row_hash IS NOT NULL AND row_hash != '';\nCREATE INDEX IF NOT EXISTS idx_crypto_transactions_asset_datetime ON crypto_transactions(asset_id, transaction_datetime);\nCREATE INDEX IF NOT EXISTS idx_alerts_priority_status ON alerts(priority, status);\nCREATE INDEX IF NOT EXISTS idx_audit_entity ON audit_log(entity_type, entity_id);\n" REQUIRED_TABLES = [ "schema_migrations", "platforms", "accounts", "instruments", "transactions", - "positions_snapshot", "cash_balances", "fx_rates", "market_prices", "equity_price_points", + "positions_snapshot", "cash_balances", "fx_rates", "market_prices", "market_price_observations", "equity_price_points", "crypto_wallets", "crypto_assets", "crypto_holdings", "crypto_transactions", "crypto_prices", "crypto_price_points", "watchlist", "reports", "alerts", "audit_log", "import_sessions", "decision_journal", "instrument_mappings", "platform_account_mappings", "broker_import_dry_runs", ] __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23.1__HERMES_CWD_8d46a20096ed__