INDEX TREE b5c8843f388c2d46d20ea46eb19b6d272dabc37f CORE DIFF diff --git a/src/jarvis_finance/market_data/prices.py b/src/jarvis_finance/market_data/prices.py index 5506e61..c593caf 100644 --- a/src/jarvis_finance/market_data/prices.py +++ b/src/jarvis_finance/market_data/prices.py @@ -6,6 +6,7 @@ from decimal import Decimal, InvalidOperation from sqlite3 import Connection from typing import Protocol from urllib import error, parse, request +import hashlib import json from jarvis_finance.imports.common import stable_id, utc_now @@ -450,6 +451,137 @@ def _corporate_action_status(conn: Connection, *, instrument_id: str, price_date return "none_known" +def _economic_price_payload( + *, + instrument_id: str, + provider: str, + provider_symbol: str | None, + provider_market: str | None, + price_type: str, + close: Decimal | None, + adjusted_close: Decimal | None, + currency: str, + provider_timestamp: str, + quality_status: str, + source_reference: str, +) -> dict[str, str | None]: + return { + "provider": provider.lower(), + "instrument_id": instrument_id, + "provider_symbol": provider_symbol, + "provider_market": provider_market, + "price_type": price_type, + "close": format(close, "f") if close is not None else "", + "adjusted_close": format(adjusted_close, "f") if adjusted_close is not None else None, + "currency": currency.upper(), + "provider_timestamp": provider_timestamp, + "source_reference": source_reference, + "quality_status": quality_status, + } + + +def _normalise_provider_timestamp(value: str) -> str: + """Canonicalise equivalent timestamp spellings without inventing precision.""" + text = value.strip() + if len(text) == 10: + return date.fromisoformat(text).isoformat() + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc).isoformat() + + +def _store_economic_price_observation( + conn: Connection, + *, + instrument_id: str, + provider: str, + provider_symbol: str | None, + provider_market: str | None, + price_type: str, + close: Decimal | None, + adjusted_close: Decimal | None, + currency: str, + provider_timestamp: str, + quality_status: str, + source_reference: str | None, + job_reference: str | None, + created_at: str, +) -> str: + # Serialize identity/version allocation across parallel provider workers. + # The caller commits the complete observation + current-price projection atomically. + if not conn.in_transaction: + conn.execute("BEGIN IMMEDIATE") + canonical_provider_timestamp = _normalise_provider_timestamp(provider_timestamp) + origin_reference = source_reference or ":".join( + part for part in (provider.lower(), provider_symbol or "", provider_market or "") if part + ) + payload = _economic_price_payload( + instrument_id=instrument_id, + provider=provider, + provider_symbol=provider_symbol, + provider_market=provider_market, + price_type=price_type, + close=close, + adjusted_close=adjusted_close, + currency=currency, + provider_timestamp=canonical_provider_timestamp, + quality_status=quality_status, + source_reference=origin_reference, + ) + payload_json = json.dumps(payload, sort_keys=True, separators=(",", ":")) + payload_hash = hashlib.sha256(payload_json.encode()).hexdigest() + source_observation_id = stable_id( + "market-source-observation", + provider.lower(), + instrument_id, + canonical_provider_timestamp, + ) + same = conn.execute( + """SELECT observation_id FROM market_price_observations + WHERE source_observation_id=? AND economic_payload_hash=?""", + (source_observation_id, payload_hash), + ).fetchone() + if same: + return str(same["observation_id"]) + predecessor = conn.execute( + """SELECT observation_id,payload_version,economic_payload_json + FROM market_price_observations WHERE source_observation_id=? + ORDER BY payload_version DESC LIMIT 1""", + (source_observation_id,), + ).fetchone() + version = int(predecessor["payload_version"]) + 1 if predecessor else 1 + observation_id = stable_id("market-observation", source_observation_id, payload_hash) + conn.execute( + """INSERT INTO market_price_observations( + observation_id,source_observation_id,payload_version,supersedes_observation_id, + instrument_id,provider,provider_symbol,provider_market,price_type,close,adjusted_close, + currency,provider_timestamp,source_reference,quality_status,economic_payload_json, + economic_payload_hash,created_at,job_reference + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + observation_id, source_observation_id, version, + str(predecessor["observation_id"]) if predecessor else None, + instrument_id, provider, provider_symbol, provider_market, price_type, + payload["close"], payload["adjusted_close"], payload["currency"], canonical_provider_timestamp, + origin_reference, quality_status, payload_json, payload_hash, created_at, job_reference, + ), + ) + if predecessor: + record_audit_event( + conn, + source="market_price_observation_v2", + action="market_price_observation_corrected", + entity_type="market_price_observation", + entity_id=observation_id, + old_values={"supersedes_observation_id": predecessor["observation_id"]}, + new_values={"source_observation_id": source_observation_id, "payload_version": version}, + confirmed=True, + created_by="system", + ) + return observation_id + + def store_market_price( conn: Connection, *, @@ -467,6 +599,58 @@ def store_market_price( fetched_at: str | None = None, price_type: str = "unadjusted_close", run_id: str | None = None, +) -> str: + started_transaction = not conn.in_transaction + if started_transaction: + conn.execute("BEGIN IMMEDIATE") + conn.execute("UPDATE market_price_observation_mutex SET touched=touched WHERE mutex_id=1") + conn.execute("SAVEPOINT market_price_write") + try: + market_price_id = _store_market_price_locked( + conn, + instrument_id=instrument_id, + price_date=price_date, + close=close, + currency=currency, + provider=provider, + provider_symbol=provider_symbol, + provider_market=provider_market, + price_timestamp=price_timestamp, + adjusted_close=adjusted_close, + quality_status=quality_status, + error_message=error_message, + fetched_at=fetched_at, + price_type=price_type, + run_id=run_id, + ) + conn.execute("RELEASE SAVEPOINT market_price_write") + conn.commit() + return market_price_id + except Exception: + conn.execute("ROLLBACK TO SAVEPOINT market_price_write") + conn.execute("RELEASE SAVEPOINT market_price_write") + if started_transaction: + conn.rollback() + raise + + +def _store_market_price_locked( + conn: Connection, + *, + instrument_id: str, + price_date: str, + close: Decimal | None, + currency: str, + provider: str, + provider_symbol: str | None, + provider_market: str | None = None, + price_timestamp: str | None = None, + adjusted_close: Decimal | None = None, + quality_status: str = "fresh", + error_message: str | None = None, + fetched_at: str | None = None, + price_type: str = "unadjusted_close", + run_id: str | None = None, ) -> str: now = utc_now() fetched = fetched_at or now @@ -475,7 +659,26 @@ def store_market_price( (instrument_id, price_date, provider), ).fetchone() corp_status = _corporate_action_status(conn, instrument_id=instrument_id, price_date=price_date, provider=provider, close=close) if quality_status == "fresh" else "not_checked" - market_price_id = stable_id("marketprice", instrument_id, price_date, provider, provider_symbol or "", now) + provider_observed_at = price_timestamp or price_date + _store_economic_price_observation( + conn, + instrument_id=instrument_id, + provider=provider, + provider_symbol=provider_symbol, + provider_market=provider_market, + price_type=price_type, + close=close, + adjusted_close=adjusted_close, + currency=currency, + provider_timestamp=provider_observed_at, + quality_status=quality_status, + source_reference=None, + job_reference=run_id, + created_at=now, + ) + market_price_id = str(existing["market_price_id"]) if existing else stable_id( + "marketprice", instrument_id, price_date, provider, provider_symbol or "", now + ) conn.execute( """ INSERT INTO market_prices( @@ -520,7 +723,6 @@ def store_market_price( create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id=rule, message="Instrument local market price is not fresh.", evidence={"provider": provider, "provider_symbol": provider_symbol, "price_date": price_date, "quality_status": quality_status}, fingerprint=f"{rule}:{provider}:{provider_symbol}") elif quality_status == "fresh": _resolve_market_alerts(conn, instrument_id=instrument_id) - conn.commit() return market_price_id diff --git a/src/jarvis_finance/services/asset_price_refresh.py b/src/jarvis_finance/services/asset_price_refresh.py index a822fed..fe06cea 100644 --- a/src/jarvis_finance/services/asset_price_refresh.py +++ b/src/jarvis_finance/services/asset_price_refresh.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib import json import uuid +from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from pathlib import Path from sqlite3 import Connection, SQLITE_DELETE, SQLITE_DENY, SQLITE_INSERT, SQLITE_OK, SQLITE_UPDATE @@ -26,6 +27,26 @@ PROTECTED_TABLES = ( ) +@dataclass(frozen=True) +class SourceRunResult: + stale_candidates: int = 0 + updated_count: int = 0 + fresh_unchanged_count: int = 0 + stale_remaining_count: int = 0 + failed_count: int = 0 + diagnostics: tuple[str, ...] = field(default_factory=tuple) + + def __iter__(self): + yield self.stale_candidates + yield self.updated_count + + +def _source_result(value: SourceRunResult | tuple[int, int]) -> SourceRunResult: + if isinstance(value, SourceRunResult): + return value + return SourceRunResult(stale_candidates=int(value[0]), updated_count=int(value[1])) + + def _deny_protected_dml( action: int, table: str | None, @@ -90,6 +111,10 @@ def _status_payload(conn: Connection, job_id: str) -> dict[str, Any]: "stale_candidates": int(row["stale_candidates"]), "updated_count": int(row["updated_count"]), "error_code": str(row["error_code"]) if row["error_code"] else None, + "fresh_unchanged_count": int(row["fresh_unchanged_count"]), + "stale_remaining_count": int(row["stale_remaining_count"]), + "failed_count": int(row["failed_count"]), + "diagnostics": json.loads(str(row["diagnostics_json"] or "[]")), "started_at": str(row["started_at"]) if row["started_at"] else None, "completed_at": str(row["completed_at"]) if row["completed_at"] else None, } @@ -97,6 +122,15 @@ def _status_payload(conn: Connection, job_id: str) -> dict[str, Any]: ], "wealth_snapshot_created": bool(job["wealth_snapshot_id"]), "audit_recorded": bool(job["audit_id"]), + "successful_assets": sum(int(row["updated_count"]) for row in sources), + "fresh_unchanged_assets": sum(int(row["fresh_unchanged_count"]) for row in sources), + "stale_assets": sum(int(row["stale_remaining_count"]) for row in sources), + "failed_assets": sum(int(row["failed_count"]) for row in sources), + "next_action": ( + "Diagnose prüfen und nur betroffene Quelle erneut versuchen." + if any(int(row["failed_count"]) or int(row["stale_remaining_count"]) for row in sources) + else "Keine Aktion nötig; alle verfügbaren Kurse sind aktuell." + ), "provider_calls_on_read": False, } @@ -134,7 +168,7 @@ def create_asset_price_refresh_job(conn: Connection, *, stale_hours: int = 24) - return _status_payload(conn, job_id), _database_path(conn) -def _equity_source(conn: Connection, stale_before: str) -> tuple[int, int]: +def _equity_source(conn: Connection, stale_before: str) -> SourceRunResult: response = refresh_equity_quotes_batch( conn, QuoteRefreshRequest( @@ -147,12 +181,33 @@ def _equity_source(conn: Connection, stale_before: str) -> tuple[int, int]: ), ) 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) + failed = sum(1 for row in response.results if str(row.get("status")) in {"provider_error", "error"}) + stale = sum(1 for row in response.results if str(row.get("status")) in {"stale", "missing"}) + valuation_issues = [ + warning + for warning in response.warnings + if warning in {"portfolio_valuation_partial", "portfolio_valuation_failed"} + ] + return SourceRunResult( + stale_candidates=candidates, + updated_count=int(response.economic_updated), + fresh_unchanged_count=int(response.cached + response.updated - response.economic_updated), + stale_remaining_count=stale, + failed_count=max(failed, len(response.errors)) + len(valuation_issues), + diagnostics=tuple(sorted(set([*response.errors, *valuation_issues])))[:10], + ) -def _crypto_source(conn: Connection, stale_before: str) -> tuple[int, int]: +def _crypto_source(conn: Connection, stale_before: str) -> SourceRunResult: + held_asset_ids = [ + str(row["asset_id"]) + for row in conn.execute( + """SELECT a.asset_id FROM crypto_assets a + WHERE a.is_active=1 AND EXISTS( + SELECT 1 FROM crypto_holdings h WHERE h.asset_id=a.asset_id AND CAST(h.quantity AS REAL)<>0 + ) ORDER BY a.asset_id""" + ).fetchall() + ] stale_asset_ids = [ str(row["asset_id"]) for row in conn.execute( @@ -168,7 +223,7 @@ def _crypto_source(conn: Connection, stale_before: str) -> tuple[int, int]: ).fetchall() ] if not stale_asset_ids: - return 0, 0 + return SourceRunResult(fresh_unchanged_count=len(held_asset_ids)) cutoff = datetime.fromisoformat(stale_before.replace("Z", "+00:00")) if cutoff.tzinfo is None: cutoff = cutoff.replace(tzinfo=UTC) @@ -181,16 +236,30 @@ def _crypto_source(conn: Connection, stale_before: str) -> tuple[int, int]: asset_ids=stale_asset_ids, batch_size=100, ) - if result.error_count and result.updated_count == 0: - raise RuntimeError("crypto_provider_failed") - return len(stale_asset_ids), int(result.updated_count) + return SourceRunResult( + stale_candidates=len(stale_asset_ids), + updated_count=int(result.updated_count), + fresh_unchanged_count=max(0, len(held_asset_ids) - len(stale_asset_ids)) + int(result.cached_count), + stale_remaining_count=int(result.stale_count + result.missing_local_price_count), + failed_count=int(result.error_count), + diagnostics=tuple(result.errors[:10]), + ) -def _fx_source(conn: Connection, stale_before: str) -> tuple[int, int]: +def _fx_source(conn: Connection, stale_before: str) -> SourceRunResult: from jarvis_finance.fx.providers import FrankfurterFxProvider, TwelveDataFxProvider from jarvis_finance.fx.rates import resolve_fx_rate_to_chf cutoff_date = stale_before[:10] + all_currencies = [ + str(row["currency"]).upper() + for row in conn.execute( + """SELECT DISTINCT upper(i.currency) currency + FROM instruments i + WHERE i.is_active=1 AND upper(COALESCE(i.currency,'CHF'))!='CHF' + ORDER BY currency""" + ).fetchall() + ] currencies = [ str(row["currency"]).upper() for row in conn.execute( @@ -207,9 +276,18 @@ def _fx_source(conn: Connection, stale_before: str) -> tuple[int, int]: ).fetchall() ] updated = 0 + provider_unchanged = 0 failures = 0 for currency in currencies: try: + before = conn.execute( + """SELECT rate_date,rate,provider,rate_type,quality_status + FROM fx_rates + WHERE base_currency=? AND quote_currency='CHF' + ORDER BY rate_date DESC,COALESCE(fetched_at,created_at) DESC LIMIT 1""", + (currency,), + ).fetchone() + before_economic = tuple(before) if before is not None else None result = resolve_fx_rate_to_chf( conn, base_currency=currency, @@ -218,16 +296,31 @@ def _fx_source(conn: Connection, stale_before: str) -> tuple[int, int]: persist=True, resolve_fixed=True, ) - updated += int(result.status == "ok") + after = conn.execute( + """SELECT rate_date,rate,provider,rate_type,quality_status + FROM fx_rates + WHERE base_currency=? AND quote_currency='CHF' + ORDER BY rate_date DESC,COALESCE(fetched_at,created_at) DESC LIMIT 1""", + (currency,), + ).fetchone() + after_economic = tuple(after) if after is not None else None + if result.status == "ok" and after_economic != before_economic: + updated += 1 + elif result.status == "ok": + provider_unchanged += 1 except Exception: failures += 1 conn.commit() - if failures and updated == 0: - raise RuntimeError("fx_provider_failed") - return len(currencies), updated + return SourceRunResult( + stale_candidates=len(currencies), + updated_count=updated, + fresh_unchanged_count=max(0, len(all_currencies) - len(currencies)) + provider_unchanged, + stale_remaining_count=failures, + failed_count=failures, + ) -DEFAULT_RUNNERS: dict[str, Callable[[Connection, str], tuple[int, int]]] = { +DEFAULT_RUNNERS: dict[str, Callable[[Connection, str], SourceRunResult | tuple[int, int]]] = { "equity": _equity_source, "crypto": _crypto_source, "fx": _fx_source, @@ -238,7 +331,7 @@ def run_asset_price_refresh( db_path: str, job_id: str, *, - runners: dict[str, Callable[[Connection, str], tuple[int, int]]] | None = None, + runners: dict[str, Callable[[Connection, str], SourceRunResult | tuple[int, int]]] | None = None, ) -> None: """Background worker with source isolation, stored progress and mutation guard.""" conn = connect(db_path) @@ -264,6 +357,9 @@ def run_asset_price_refresh( conn.set_authorizer(_deny_protected_dml) completed = 0 failures = 0 + total_updated = 0 + total_fresh_unchanged = 0 + total_candidates = 0 for source in SOURCES: started = _now() conn.execute( @@ -271,25 +367,42 @@ def run_asset_price_refresh( (started, job_id, source), ) conn.commit() - candidates = updated = 0 + result = SourceRunResult() status = "complete" error_code = None try: - candidates, updated = selected[source](conn, stale_before) - if candidates == 0: + result = _source_result(selected[source](conn, stale_before)) + if result.stale_candidates == 0 and result.fresh_unchanged_count == 0: status = "skipped" + if result.failed_count: + failures += 1 + status = "failed" + error_code = f"{source}_instrument_failures" + elif result.stale_remaining_count: + failures += 1 + status = "failed" + error_code = f"{source}_stale_remaining" except Exception as exc: if conn.in_transaction: conn.rollback() status = "failed" failures += 1 - error_code = str(exc)[:120] or type(exc).__name__ + result = SourceRunResult(failed_count=1, diagnostics=(type(exc).__name__,)) + error_code = f"{source}_refresh_failed" + total_updated += result.updated_count + total_fresh_unchanged += result.fresh_unchanged_count + total_candidates += result.stale_candidates completed += 1 conn.execute( """UPDATE asset_price_refresh_sources - SET status=?,stale_candidates=?,updated_count=?,error_code=?,completed_at=? + SET status=?,stale_candidates=?,updated_count=?,error_code=?,completed_at=?, + fresh_unchanged_count=?,stale_remaining_count=?,failed_count=?,diagnostics_json=? WHERE job_id=? AND source=?""", - (status, candidates, updated, error_code, _now(), job_id, source), + ( + status, result.stale_candidates, result.updated_count, error_code, _now(), + result.fresh_unchanged_count, result.stale_remaining_count, result.failed_count, + json.dumps(list(result.diagnostics)), job_id, source, + ), ) conn.execute( "UPDATE asset_price_refresh_jobs SET progress_completed=? WHERE job_id=?", @@ -299,16 +412,18 @@ def run_asset_price_refresh( if _protected_fingerprint(conn) != protected_before: raise RuntimeError("protected_holdings_or_transactions_mutated") - successful_sources = failures < len(SOURCES) + usable_result = total_updated > 0 or total_fresh_unchanged > 0 or (total_candidates == 0 and failures == 0) wealth_snapshot_id = None - if successful_sources: + if total_updated > 0: 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", + """SELECT source,status,stale_candidates,updated_count,error_code, + fresh_unchanged_count,stale_remaining_count,failed_count + FROM asset_price_refresh_sources WHERE job_id=? ORDER BY source""", (job_id,), ).fetchall() ] @@ -325,7 +440,7 @@ def run_asset_price_refresh( json.dumps(source_rows, sort_keys=True), ), ) - final_status = "complete" if failures == 0 else "failed" if failures == len(SOURCES) else "partial" + final_status = "complete" if failures == 0 else "partial" if usable_result else "failed" audit_id = record_audit_event( conn, source="asset_price_refresh_job_v1", diff --git a/src/jarvis_finance/storage/migrations.py b/src/jarvis_finance/storage/migrations.py index 9f8994b..38c49aa 100644 --- a/src/jarvis_finance/storage/migrations.py +++ b/src/jarvis_finance/storage/migrations.py @@ -8,8 +8,8 @@ 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", @@ -2963,6 +2963,75 @@ def _create_professional_portfolio_cockpit_v1(conn: Connection) -> None: ) +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_observation_mutex ( + mutex_id INTEGER PRIMARY KEY CHECK(mutex_id=1), + touched INTEGER NOT NULL DEFAULT 0 + ); + INSERT OR IGNORE INTO market_price_observation_mutex(mutex_id,touched) VALUES(1,0); + + 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(): @@ -3005,6 +3074,7 @@ def _apply_compat_migrations(conn: Connection) -> None: _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: diff --git a/tests/unit/test_asset_price_refresh.py b/tests/unit/test_asset_price_refresh.py index c70b396..79d1f61 100644 --- a/tests/unit/test_asset_price_refresh.py +++ b/tests/unit/test_asset_price_refresh.py @@ -5,9 +5,11 @@ from decimal import Decimal from pathlib import Path from sqlite3 import Connection import threading +from types import SimpleNamespace from typing import Callable from jarvis_finance.market.providers import PriceQuote +from jarvis_finance.fx.rates import upsert_fx_rate from jarvis_finance.services import asset_price_refresh as asset_refresh from jarvis_finance.services.asset_price_refresh import ( asset_price_refresh_status, @@ -149,7 +151,118 @@ def test_concurrent_workers_claim_a_queued_job_exactly_once(tmp_path): assert conn.execute( "SELECT COUNT(*) FROM aggregated_wealth_refresh_snapshots WHERE job_id=?", (queued["job_id"],), - ).fetchone()[0] == 1 + ).fetchone()[0] == 0 + conn.close() + + +def test_all_failed_sources_create_no_false_wealth_snapshot(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, db_path = create_asset_price_refresh_job(conn) + conn.close() + + def failed(_conn, _stale_before): + raise RuntimeError("internal provider detail") + + run_asset_price_refresh(db_path, queued["job_id"], runners={source: failed for source in ("equity", "crypto", "fx")}) + conn = connect(path) + status = asset_price_refresh_status(conn, queued["job_id"]) + assert status["status"] == "failed" + assert status["wealth_snapshot_created"] is False + assert status["failed_assets"] == 3 + assert conn.execute("SELECT COUNT(*) FROM aggregated_wealth_refresh_snapshots").fetchone()[0] == 0 + assert "internal provider detail" not in str(status) + conn.close() + + +def test_partial_instrument_failure_preserves_success_and_quality_counts(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, db_path = create_asset_price_refresh_job(conn) + conn.close() + + def equity(_conn, _stale_before): + return asset_refresh.SourceRunResult( + stale_candidates=3, + updated_count=2, + stale_remaining_count=1, + failed_count=1, + diagnostics=("provider_error",), + ) + + def current(_conn, _stale_before): + return asset_refresh.SourceRunResult(fresh_unchanged_count=2) + + run_asset_price_refresh( + db_path, + queued["job_id"], + runners={"equity": equity, "crypto": current, "fx": current}, + ) + conn = connect(path) + status = asset_price_refresh_status(conn, queued["job_id"]) + assert status["status"] == "partial" + assert status["successful_assets"] == 2 + assert status["fresh_unchanged_assets"] == 4 + assert status["stale_assets"] == 1 + assert status["failed_assets"] == 1 + assert status["wealth_snapshot_created"] is True + conn.close() + + +def test_unresolved_stale_asset_makes_job_partial(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, db_path = create_asset_price_refresh_job(conn) + conn.close() + + def stale(_conn, _stale_before): + return asset_refresh.SourceRunResult( + stale_candidates=1, + fresh_unchanged_count=1, + stale_remaining_count=1, + ) + + def current(_conn, _stale_before): + return asset_refresh.SourceRunResult(fresh_unchanged_count=1) + + run_asset_price_refresh( + db_path, + queued["job_id"], + runners={"equity": stale, "crypto": current, "fx": current}, + ) + conn = connect(path) + status = asset_price_refresh_status(conn, queued["job_id"]) + assert status["status"] == "partial" + assert status["stale_assets"] == 1 + assert status["wealth_snapshot_created"] is False + source = next(item for item in status["sources"] if item["source"] == "equity") + assert source["status"] == "failed" + assert source["error_code"] == "equity_stale_remaining" + conn.close() + + +def test_equity_valuation_warning_marks_source_partial(tmp_path, monkeypatch): + conn = database(tmp_path / "finance.sqlite3") + response = type( + "Response", + (), + { + "total": 2, + "cached": 0, + "updated": 2, + "economic_updated": 2, + "results": [{"status": "fresh"}, {"status": "fresh"}], + "errors": [], + "warnings": ["portfolio_valuation_partial"], + }, + )() + monkeypatch.setattr(asset_refresh, "refresh_equity_quotes_batch", lambda *_args, **_kwargs: response) + + result = asset_refresh._equity_source(conn, "2026-08-26T00:00:00+00:00") + + assert result.updated_count == 2 + assert result.failed_count == 1 + assert result.diagnostics == ("portfolio_valuation_partial",) conn.close() @@ -190,7 +303,7 @@ def test_crypto_refresh_fetches_only_stale_held_asset_ids(tmp_path, monkeypatch) provider_id: PriceQuote( provider_id, currency, - Decimal("123.45"), + Decimal("100") if provider_id == "ethereum" else Decimal("123.45"), provider_timestamp=now.isoformat(), ) for provider_id in ids @@ -210,4 +323,49 @@ def test_crypto_refresh_fetches_only_stale_held_asset_ids(tmp_path, monkeypatch) assert provider.calls == [("bitcoin",)] assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='eth'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='btc'").fetchone()[0] == 1 + + replay = asset_refresh._crypto_source(conn, "2099-01-01T00:00:00+00:00") + assert replay.updated_count == 0 + assert replay.fresh_unchanged_count == 2 + assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='btc'").fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='eth'").fetchone()[0] == 1 + conn.close() + + +def test_fx_metadata_replay_is_not_an_economic_update(tmp_path, monkeypatch): + conn = database(tmp_path / "finance.sqlite3") + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) " + "VALUES('eur-asset','etf','Synthetic EUR','EUR',1,'2026-01-01')" + ) + upsert_fx_rate( + conn, + base_currency="EUR", + quote_currency="CHF", + rate_date="2026-08-27", + rate=Decimal("0.80448"), + provider="mockfx", + rate_type="close", + fetched_at="2026-08-27T10:00:00+00:00", + ) + conn.commit() + + def resolve_same(conn_arg, **_kwargs): + upsert_fx_rate( + conn_arg, + base_currency="EUR", + quote_currency="CHF", + rate_date="2026-08-27", + rate=Decimal("0.80448"), + provider="mockfx", + rate_type="close", + ) + return SimpleNamespace(status="ok") + + monkeypatch.setattr("jarvis_finance.fx.rates.resolve_fx_rate_to_chf", resolve_same) + result = asset_refresh._fx_source(conn, "2099-01-01T00:00:00+00:00") + + assert result.updated_count == 0 + assert result.fresh_unchanged_count == 1 + assert conn.execute("SELECT COUNT(*) FROM fx_rates WHERE base_currency='EUR'").fetchone()[0] == 1 conn.close() diff --git a/tests/unit/test_market_observation_idempotency.py b/tests/unit/test_market_observation_idempotency.py new file mode 100644 index 0000000..06cd3d0 --- /dev/null +++ b/tests/unit/test_market_observation_idempotency.py @@ -0,0 +1,251 @@ +from __future__ import annotations + +from decimal import Decimal +from pathlib import Path +import threading + +from jarvis_finance.market_data.prices import store_market_price +from jarvis_finance.services.portfolio_analytics import run_daily_market_valuation +from jarvis_finance.storage.database import connect +from jarvis_finance.storage.migrations import apply_migrations +from test_portfolio_market_analytics_v1 import Fx, database, quotes_for + + +def test_identical_economic_quote_ignores_request_metadata() -> None: + conn = connect(":memory:") + apply_migrations(conn) + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) VALUES('i','stock','Synthetic','CHF',1,'2026-01-01')" + ) + first = store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10.50"), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00+00:00", + fetched_at="2026-08-27T10:00:01+00:00", + run_id="job-a", + ) + second = store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10.50"), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00Z", + fetched_at="2026-08-27T10:05:00+00:00", + run_id="job-b", + ) + assert second == first + assert conn.execute("SELECT COUNT(*) FROM market_price_observations").fetchone()[0] == 1 + stored = conn.execute( + "SELECT economic_payload_json,job_reference FROM market_price_observations" + ).fetchone() + assert stored["job_reference"] == "job-a" + assert "job-a" not in stored["economic_payload_json"] + assert "job-b" not in stored["economic_payload_json"] + + store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10.50"), + currency="CHF", + provider="mock", + provider_symbol="SYN-CORRECTED", + price_timestamp="2026-08-27T10:00:00+00:00", + run_id="job-c", + ) + versions = conn.execute( + """SELECT payload_version,supersedes_observation_id FROM market_price_observations + ORDER BY payload_version""" + ).fetchall() + assert [row["payload_version"] for row in versions] == [1, 2] + assert versions[1]["supersedes_observation_id"] is not None + + +def test_two_same_day_provider_times_and_correction_are_append_only() -> None: + conn = connect(":memory:") + apply_migrations(conn) + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) VALUES('i','stock','Synthetic','CHF',1,'2026-01-01')" + ) + for timestamp, close in ( + ("2026-08-27T10:00:00+00:00", "10"), + ("2026-08-27T11:00:00+00:00", "11"), + ("2026-08-27T11:00:00+00:00", "11.1"), + ): + store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal(close), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp=timestamp, + ) + rows = conn.execute( + "SELECT observation_id,source_observation_id,payload_version,supersedes_observation_id,close FROM market_price_observations ORDER BY created_at,observation_id" + ).fetchall() + assert len(rows) == 3 + assert rows[0]["source_observation_id"] != rows[1]["source_observation_id"] + assert rows[1]["source_observation_id"] == rows[2]["source_observation_id"] + assert (rows[1]["payload_version"], rows[2]["payload_version"]) == (1, 2) + assert rows[2]["supersedes_observation_id"] == rows[1]["observation_id"] + audit = conn.execute( + "SELECT action,old_values_json,new_values_json FROM audit_log WHERE action='market_price_observation_corrected'" + ).fetchone() + assert audit is not None + assert rows[1]["observation_id"] in str(audit["old_values_json"]) + assert [row["close"] for row in rows] == ["10", "11", "11.1"] + + +def test_partial_daily_run_accepts_new_economic_payload_as_new_version(tmp_path: Path) -> None: + conn = database() + first_provider = quotes_for("2026-07-01", missing={"BENCH.S"}) + first = run_daily_market_valuation( + conn, + as_of="2026-07-01", + price_providers={"mock": first_provider}, + fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}), + lock_path=tmp_path / "job.lock", + ) + assert first.status == "partial" + store_market_price( + conn, + instrument_id="eur", + price_date="2026-07-01", + close=Decimal("51"), + currency="EUR", + provider="mock", + provider_symbol="EUR.S", + provider_market="SIX", + price_timestamp="2026-07-01T21:00:00+00:00", + run_id="provider-correction", + ) + corrected = quotes_for("2026-07-01") + second = run_daily_market_valuation( + conn, + as_of="2026-07-01", + price_providers={"mock": corrected}, + fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}), + lock_path=tmp_path / "job.lock", + ) + assert second.run_id == first.run_id + assert second.status == "complete" + versions = conn.execute( + "SELECT snapshot_version,supersedes_snapshot_id,value_original FROM portfolio_valuation_snapshots WHERE scope_kind='instrument' AND scope_id='eur' ORDER BY snapshot_version" + ).fetchall() + assert len(versions) == 2 + assert versions[1]["supersedes_snapshot_id"] is not None + assert [row["value_original"] for row in versions] == ["500", "510"] + + +def test_parallel_corrections_allocate_distinct_append_only_versions(tmp_path: Path) -> None: + path = tmp_path / "finance.sqlite3" + conn = connect(path) + apply_migrations(conn) + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) " + "VALUES('i','stock','Synthetic','CHF',1,'2026-01-01')" + ) + store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10"), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00+00:00", + ) + conn.close() + barrier = threading.Barrier(2) + errors: list[Exception] = [] + returned_ids: list[str] = [] + + def correct(value: str) -> None: + worker = connect(path) + try: + worker.execute("BEGIN") + barrier.wait(timeout=5) + returned_ids.append(store_market_price( + worker, + instrument_id="i", + price_date="2026-08-27", + close=Decimal(value), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00Z", + )) + except Exception as exc: + errors.append(exc) + finally: + worker.close() + + threads = [threading.Thread(target=correct, args=(value,)) for value in ("11", "12")] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + + assert errors == [] + assert len(set(returned_ids)) == 1 + conn = connect(path) + rows = conn.execute( + """SELECT payload_version,supersedes_observation_id + FROM market_price_observations ORDER BY payload_version""" + ).fetchall() + assert [row["payload_version"] for row in rows] == [1, 2, 3] + assert all(row["supersedes_observation_id"] for row in rows[1:]) + conn.close() + + +def test_projection_failure_rolls_back_pending_observation_inside_outer_transaction(tmp_path: Path) -> None: + conn = connect(tmp_path / "finance.sqlite3") + apply_migrations(conn) + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) " + "VALUES('i','stock','Synthetic','CHF',1,'2026-01-01')" + ) + conn.execute( + """CREATE TRIGGER reject_synthetic_market_projection + BEFORE INSERT ON market_prices WHEN NEW.instrument_id='i' + BEGIN SELECT RAISE(ABORT,'synthetic projection failure'); END""" + ) + conn.commit() + conn.execute("BEGIN") + + try: + store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10"), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00Z", + ) + except Exception as exc: + assert "synthetic projection failure" in str(exc) + else: + raise AssertionError("projection failure expected") + + assert conn.in_transaction is True + assert conn.execute("SELECT COUNT(*) FROM market_price_observations").fetchone()[0] == 0 + conn.execute( + "INSERT INTO platforms(platform_id,name,platform_type,created_at) " + "VALUES('unrelated','Unrelated','manual','2026-01-01')" + ) + conn.commit() + assert conn.execute("SELECT COUNT(*) FROM market_price_observations").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM market_prices").fetchone()[0] == 0 + conn.close() __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23.1__HERMES_CWD_8d46a20096ed__