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()