from __future__ import annotations

import fcntl
import json
from decimal import Decimal
from pathlib import Path
from sqlite3 import Connection
from typing import Any, cast

import pytest
from fastapi.testclient import TestClient

from jarvis_finance.api.dependencies import get_db
from jarvis_finance.api.main import create_app
from jarvis_finance.dashboard import data as dashboard_data
from jarvis_finance.fx.rates import upsert_fx_rate
from jarvis_finance.market_data.prices import EquityPriceQuote, store_market_price
from jarvis_finance.services.portfolio_analytics import (
    build_portfolio_analytics,
    confirmed_canonical_positions,
    run_daily_market_valuation,
)
from jarvis_finance.services.portfolio_performance import _load_valuations, build_portfolio_performance
from jarvis_finance.services.performance_activation import _local_backfill_input_revision
from jarvis_finance.services.performance_scope import (
    set_performance_cashflow_coverage,
    set_performance_scope_classification,
)
from jarvis_finance.storage.database import connect
from jarvis_finance.storage.migrations import apply_migrations

NOW = "2026-07-01T20:00:00+00:00"


class DatedProvider:
    name = "mock"

    def __init__(self, quotes: dict[tuple[str, str], tuple[str, str, str | None]]) -> None:
        self.quotes = quotes

    def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote:
        row = self.quotes.get((provider_symbol, price_date or ""))
        if row is None:
            return EquityPriceQuote(provider_symbol=provider_symbol, currency="", close=None, provider=self.name, quality_status="missing")
        close, currency, adjusted = row
        return EquityPriceQuote(
            provider_symbol=provider_symbol,
            currency=currency,
            close=Decimal(close),
            adjusted_close=Decimal(adjusted) if adjusted is not None else None,
            provider=self.name,
            provider_market="SIX",
            price_timestamp=f"{price_date}T20:00:00+00:00",
        )


class Fx:
    name = "mockfx"

    def __init__(self, rates: dict[tuple[str, str], str | None]) -> None:
        self.rates = rates

    def get_rate(self, base_currency: str, quote_currency: str, rate_date: str | None = None) -> Decimal | None:
        value = self.rates.get((base_currency, rate_date or ""))
        return Decimal(value) if value is not None else None


class StaleProvider(DatedProvider):
    def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote:
        quote = super().get_price(provider_symbol, price_date=price_date)
        return EquityPriceQuote(
            provider_symbol=quote.provider_symbol, currency=quote.currency, close=quote.close,
            adjusted_close=quote.adjusted_close, provider=quote.provider, provider_market=quote.provider_market,
            price_timestamp="2026-06-20T20:00:00+00:00", quality_status=quote.quality_status,
        )


class PriorBusinessDayProvider(DatedProvider):
    def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote:
        quote = super().get_price(provider_symbol, price_date=price_date)
        return EquityPriceQuote(
            provider_symbol=quote.provider_symbol, currency=quote.currency, close=quote.close,
            adjusted_close=quote.adjusted_close, provider=quote.provider, provider_market=quote.provider_market,
            price_timestamp="2026-07-03T20:00:00+00:00", quality_status=quote.quality_status,
        )


def database(*, policy: bool = True) -> Connection:
    conn = connect(":memory:")
    apply_migrations(conn)
    conn.execute("INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES('p','Synthetic Broker','broker',?)", (NOW,))
    conn.execute(
        """INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,performance_included,is_active,created_at)
           VALUES('a','p','Synthetic Account','brokerage','CHF',0,1,?)""",
        (NOW,),
    )
    set_performance_scope_classification(
        conn,
        account_id="a",
        included=True,
        classification_role="crypto_portfolio",
        source="synthetic_test_fixture",
        note="Synthetic investment account",
        classified_at=NOW,
    )
    set_performance_cashflow_coverage(
        conn,
        account_id="a",
        coverage_from="1900-01-01",
        coverage_to="2100-12-31",
        status="complete",
        source="synthetic_test_fixture",
        note="Complete synthetic cashflow history",
        recorded_at=NOW,
    )
    instruments = [
        ("chf", "stock", "Synthetic CHF", "CH0000000001", "CHF", "CHF.S"),
        ("eur", "etf", "Synthetic EUR", "DE0000000002", "EUR", "EUR.S"),
        ("usd", "stock", "Synthetic USD", "US0000000003", "USD", "USD.S"),
        ("bench", "etf", "Synthetic Benchmark Proxy", "CHBENCH00004", "USD", "BENCH.S"),
    ]
    for instrument_id, asset_class, name, isin, currency, symbol in instruments:
        conn.execute(
            """INSERT INTO instruments(instrument_id,asset_class,name,isin,currency,trading_currency,is_active,instrument_status,valuation_policy,created_at)
               VALUES(?,?,?,?,?,?,1,'active','live_price',?)""",
            (instrument_id, asset_class, name, isin, currency, currency, NOW),
        )
        conn.execute(
            """INSERT INTO instrument_price_mappings(mapping_id,instrument_id,isin,currency,trading_currency,provider,provider_symbol,provider_market,mapping_status,confidence,created_at,updated_at)
               VALUES(?,?,?,?,?,'mock',?,'SIX','mapped','1',?,?)""",
            (f"m-{instrument_id}", instrument_id, isin, currency, currency, symbol, NOW, NOW),
        )
    for index, instrument_id in enumerate(("chf", "eur", "usd"), start=1):
        conn.execute(
            """INSERT INTO transactions(transaction_id,transaction_type,account_id,instrument_id,trade_date,quantity,currency_original,
               source_type,source_id,row_hash,is_confirmed,quality_status,created_at)
               VALUES(?, 'initial_position_snapshot','a',?,'2026-06-30','10','CHF','synthetic_confirmed',?, ?,1,'ok',?)""",
            (f"t{index}", instrument_id, f"s{index}", f"h{index}", NOW),
        )
    # A mapped but unconfirmed record models preview/staging data and must never enter the job.
    conn.execute(
        """INSERT INTO transactions(transaction_id,transaction_type,account_id,instrument_id,trade_date,quantity,currency_original,
           source_type,source_id,row_hash,is_confirmed,quality_status,created_at)
           VALUES('preview','buy','a','bench','2026-07-01','999','USD','preview_only','preview','preview',0,'pending',?)""",
        (NOW,),
    )
    if policy:
        conn.execute(
            """INSERT INTO portfolio_policies(policy_id,version,is_active,effective_from,base_currency,max_single_position_pct,
               benchmarks_json,restrictions_json,request_fingerprint,audit_id,created_at)
               VALUES('policy',1,1,'2026-06-01','CHF','40',?,'[]','policy-fp','audit-policy',?)""",
            (json.dumps([{"reference": "CHBENCH00004"}]), NOW),
        )
        conn.execute(
            "INSERT INTO portfolio_policy_allocations(allocation_id,policy_id,asset_class,target_pct,lower_pct,upper_pct) VALUES('alloc','policy','stock','50','20','70')"
        )
    conn.commit()
    return conn


def quotes_for(day: str, *, missing: set[str] | None = None) -> DatedProvider:
    values = {
        ("CHF.S", day): ("100", "CHF", None),
        ("EUR.S", day): ("50", "EUR", None),
        ("USD.S", day): ("20", "USD", None),
        ("BENCH.S", day): ("200", "USD", "210"),
    }
    for symbol in missing or set():
        values.pop((symbol, day), None)
    return DatedProvider(values)


def test_daily_job_uses_confirmed_chf_eur_usd_decimal_fx_and_unadjusted_close_idempotently(tmp_path: Path) -> None:
    conn = database()
    conn.executemany(
        """INSERT INTO cash_account_snapshots(snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency,amount_chf,source,created_at)
           VALUES(?,'a','confirmed','2026-07-01',?,?,?,?,?)""",
        [
            ("cash-chf", "0.1", "CHF", "0.1", "synthetic", NOW),
            ("cash-eur", "0.2", "EUR", "0.2", "synthetic", NOW),
        ],
    )
    conn.commit()
    provider = quotes_for("2026-07-01")
    result = run_daily_market_valuation(
        conn, as_of="2026-07-01", price_providers={"mock": provider},
        fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}),
        lock_path=tmp_path / "job.lock",
    )
    replay = run_daily_market_valuation(
        conn, as_of="2026-07-01", price_providers={"mock": provider},
        fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}),
        lock_path=tmp_path / "job.lock",
    )
    assert result.status == "complete"
    assert (result.price_stored, result.fx_stored, result.benchmark_stored) == (3, 3, 1)
    assert replay.idempotent is True and replay.run_id == result.run_id
    assert conn.execute("SELECT COUNT(*) FROM market_data_runs").fetchone()[0] == 1
    assert conn.execute("SELECT COUNT(*) FROM market_prices").fetchone()[0] == 3
    assert conn.execute("SELECT COUNT(*) FROM fx_rates").fetchone()[0] == 3
    assert conn.execute("SELECT rate FROM fx_rates WHERE base_currency='CHF' AND quote_currency='CHF'").fetchone()[0] == "1"
    # EUR position: quantity 10 × unadjusted close 50 × EUR→CHF 0.95.
    payload = json.loads(conn.execute("SELECT summary_json FROM portfolio_analysis_snapshots").fetchone()[0])
    eur = next(item for item in payload["positions"] if item["instrument_id"] == "eur")
    assert eur["value_chf"] == "475.00"
    assert tuple(conn.execute("SELECT close,adjusted_close,price_type FROM market_prices WHERE instrument_id='eur'").fetchone()) == ("50", None, "unadjusted_close")
    assert conn.execute("SELECT total_value_chf FROM portfolio_analysis_snapshots").fetchone()[0] == "1635.30"
    assert {item["label"] for item in payload["risk"]["currencies"]} == {"CHF", "EUR", "USD"}
    assert {item.instrument_id for item in confirmed_canonical_positions(conn)} == {"chf", "eur", "usd"}
    store_market_price(
        conn, instrument_id="eur", price_date="2026-07-01", close=Decimal("50.1"), currency="EUR",
        provider="mock", provider_symbol="EUR.S", run_id="provider-correction",
    )
    upsert_fx_rate(
        conn, base_currency="EUR", quote_currency="CHF", rate_date="2026-07-01", rate=Decimal("0.951"),
        provider="mockfx", rate_type="close", run_id="provider-correction",
    )
    assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE action IN ('market_price_provider_correction','fx_rate_provider_correction')").fetchone()[0] == 2


def scenario_exact_date_cache_recovers_transient_provider_gap_without_rewriting_price(tmp_path: Path) -> None:
    stale_conn = database()
    store_market_price(
        stale_conn,
        instrument_id="eur",
        price_date="2026-07-01",
        close=Decimal("50"),
        currency="EUR",
        provider="mock",
        provider_symbol="EUR.S",
        quality_status="stale",
    )
    stale_result = run_daily_market_valuation(
        stale_conn,
        as_of="2026-07-01",
        price_providers={"mock": quotes_for("2026-07-01", missing={"EUR.S"})},
        fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}),
        lock_path=tmp_path / "stale-job.lock",
    )
    assert stale_result.status == "partial"
    assert any(item["instrument_id"] == "eur" for item in stale_result.missing_instruments)
    assert "cached_cutoff_safe_price" not in stale_result.reason_codes

    conn = database()
    store_market_price(
        conn,
        instrument_id="eur",
        price_date="2026-07-01",
        close=Decimal("50"),
        currency="EUR",
        provider="mock",
        provider_symbol="EUR.S",
        price_timestamp="2026-07-01T18:00:00+00:00",
    )
    cached_id = conn.execute("SELECT market_price_id FROM market_prices WHERE instrument_id='eur'").fetchone()[0]
    store_market_price(
        conn, instrument_id="chf", price_date="2026-07-01", close=Decimal("100"), currency="CHF",
        provider="mock", provider_symbol="CHF.S", run_id="legacy-v1",
    )
    store_market_price(
        conn, instrument_id="usd", price_date="2026-07-01", close=Decimal("20"), currency="USD",
        provider="mock", provider_symbol="USD.S", run_id="legacy-v1",
    )
    for currency, rate in (("CHF", "1"), ("EUR", "0.95"), ("USD", "0.80")):
        upsert_fx_rate(
            conn, base_currency=currency, quote_currency="CHF", rate_date="2026-07-01",
            rate=Decimal(rate), provider="identity" if currency == "CHF" else "mockfx",
            rate_type="close", run_id="legacy-v1",
        )
    conn.commit()
    result = run_daily_market_valuation(
        conn,
        as_of="2026-07-01",
        price_providers={"mock": quotes_for("2026-07-01", missing={"EUR.S"})},
        fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}),
        lock_path=tmp_path / "job.lock",
    )
    assert result.status == "complete"
    assert result.price_stored == 0
    assert result.fx_stored == 0
    assert "cached_cutoff_safe_price" in result.reason_codes
    cached = conn.execute(
        "SELECT market_price_id,run_id,close,currency FROM market_prices WHERE instrument_id='eur'"
    ).fetchone()
    assert tuple(cached) == (cached_id, None, "50", "EUR")
    assert {
        tuple(row) for row in conn.execute(
            "SELECT instrument_id,run_id FROM market_prices WHERE instrument_id IN ('chf','usd')"
        )
    } == {("chf", "legacy-v1"), ("usd", "legacy-v1")}
    assert {
        tuple(row) for row in conn.execute(
            "SELECT base_currency,run_id FROM fx_rates WHERE rate_date='2026-07-01'"
        )
    } == {("CHF", "legacy-v1"), ("EUR", "legacy-v1"), ("USD", "legacy-v1")}
    payload = json.loads(conn.execute("SELECT summary_json FROM portfolio_analysis_snapshots").fetchone()[0])
    eur = next(item for item in payload["positions"] if item["instrument_id"] == "eur")
    assert (eur["close"], eur["fx_rate_to_chf"], eur["value_chf"]) == ("50", "0.95", "475.00")
    assert eur["price_input_provenance"]["market_price_id"] == cached_id
    assert eur["price_input_provenance"]["original_run_id"] is None
    assert eur["price_input_provenance"]["fetched_at"]
    audit = json.loads(conn.execute(
        "SELECT new_values_json FROM audit_log WHERE action='daily_market_valuation_completed'"
    ).fetchone()[0])
    assert audit["price_inputs_accepted"] == 3 and audit["price_rows_written"] == 0
    assert audit["fx_inputs_accepted"] == 3 and audit["fx_rows_written"] == 0
    assert audit["cached_price_inputs"]["eur"]["market_price_id"] == cached_id


def scenario_partial_provider_failure_unknown_isin_missing_fx_and_unconfirmed_record_are_excluded(tmp_path: Path) -> None:
    conn = database(policy=False)
    conn.execute(
        """INSERT INTO instruments(instrument_id,asset_class,name,currency,trading_currency,is_active,instrument_status,valuation_policy,created_at)
           VALUES('unknown','stock','Unknown ISIN','USD','USD',1,'active','live_price',?)""", (NOW,)
    )
    conn.execute(
        """INSERT INTO transactions(transaction_id,transaction_type,account_id,instrument_id,trade_date,quantity,currency_original,
           source_type,source_id,row_hash,is_confirmed,quality_status,created_at)
           VALUES('t-unknown','buy','a','unknown','2026-06-30','1','USD','confirmed','x','x',1,'ok',?)""", (NOW,)
    )
    conn.commit()
    result = run_daily_market_valuation(
        conn, as_of="2026-07-01", price_providers={"mock": quotes_for("2026-07-01", missing={"EUR.S"})},
        fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): None}),
        lock_path=tmp_path / "job.lock",
    )
    reasons = {item["reason_code"] for item in result.missing_instruments}
    assert result.status == "partial"
    assert {"price_missing", "mapping_required", "fx_rate_missing"}.issubset(set(result.reason_codes) | reasons)
    assert all(item["instrument_id"] != "bench" for item in result.missing_instruments)
    analytics = build_portfolio_analytics(conn)
    assert analytics["notice"].startswith("PostFinance-Vorschau noch nicht bestätigt")
    assert Decimal(analytics["coverage"]["price_pct"]) < 100


def scenario_weekend_uses_friday_without_staleness_and_benchmark_is_explicit_etf_proxy(tmp_path: Path) -> None:
    conn = database()
    result = run_daily_market_valuation(
        conn, as_of="2026-07-05", price_providers={"mock": quotes_for("2026-07-03")},
        fx_provider=Fx({("EUR", "2026-07-03"): "0.95", ("USD", "2026-07-03"): "0.80"}),
        lock_path=tmp_path / "job.lock",
    )
    assert result.as_of == "2026-07-03"
    assert "stale_price" not in result.reason_codes
    benchmark = conn.execute("SELECT return_type,as_of,adjusted_close FROM benchmark_snapshots").fetchone()
    assert tuple(benchmark) == ("etf_proxy", "2026-07-03", "210")


def scenario_market_holiday_uses_prior_business_day_fx_without_false_staleness(tmp_path: Path) -> None:
    conn = database()
    provider = PriorBusinessDayProvider(quotes_for("2026-07-06").quotes)
    result = run_daily_market_valuation(
        conn, as_of="2026-07-06", price_providers={"mock": provider},
        fx_provider=Fx({("EUR", "2026-07-03"): "0.95", ("USD", "2026-07-03"): "0.80"}),
        lock_path=tmp_path / "job.lock",
    )
    assert "stale_price" not in result.reason_codes and "stale_fx" not in result.reason_codes
    assert {row[0] for row in conn.execute("SELECT DISTINCT rate_date FROM fx_rates").fetchall()} == {"2026-07-03", "2026-07-06"}


def scenario_same_instrument_in_two_accounts_is_valued_without_overwrite(tmp_path: Path) -> None:
    conn = database()
    conn.execute(
        """INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,performance_included,is_active,created_at)
           VALUES('b','p','Second Synthetic Account','brokerage','CHF',0,1,?)""", (NOW,),
    )
    set_performance_scope_classification(
        conn,
        account_id="b",
        included=True,
        classification_role="postfinance_etrading_depot",
        source="synthetic_test_fixture",
        note="Second synthetic investment account",
        classified_at=NOW,
    )
    set_performance_cashflow_coverage(
        conn,
        account_id="b",
        coverage_from="1900-01-01",
        coverage_to="2100-12-31",
        status="complete",
        source="synthetic_test_fixture",
        note="Complete synthetic cashflow history",
        recorded_at=NOW,
    )
    conn.execute(
        """INSERT INTO transactions(transaction_id,transaction_type,account_id,instrument_id,trade_date,quantity,currency_original,
           source_type,source_id,row_hash,is_confirmed,quality_status,created_at)
           VALUES('t-second','buy','b','chf','2026-06-30','5','CHF','confirmed','b','b',1,'ok',?)""", (NOW,),
    )
    conn.commit()
    result = run_daily_market_valuation(
        conn, as_of="2026-07-01", price_providers={"mock": quotes_for("2026-07-01")},
        fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}),
        lock_path=tmp_path / "job.lock",
    )
    assert result.price_stored == 3  # one immutable quote row per instrument/date/provider, not per account
    assert conn.execute(
        "SELECT COUNT(*) FROM portfolio_valuation_snapshots WHERE scope_kind='instrument' AND scope_id='chf'"
    ).fetchone()[0] == 2
    loaded = _load_valuations(
        conn,
        account_ids=["a", "b"],
        from_date="2026-07-01",
        to_date="2026-07-01",
        data_cutoff="9999-12-31T23:59:59+00:00",
        base_currency="CHF",
    )
    assert sorted(
        (item.account_id, item.value) for item in loaded
        if item.scope_kind == "instrument" and item.scope_id == "chf"
    ) == [("a", Decimal("1000")), ("b", Decimal("500"))]
    values = json.loads(conn.execute("SELECT summary_json FROM portfolio_analysis_snapshots").fetchone()[0])["positions"]
    assert sorted(Decimal(item["value_chf"]) for item in values if item["instrument_id"] == "chf") == [Decimal("500"), Decimal("1000")]


def test_backfill_input_revision_binds_consumed_price_mapping() -> None:
    conn = database()
    before = _local_backfill_input_revision(
        conn,
        source="postfinance",
        account_ids=["a"],
        period_from="2026-01-01",
        period_to="2026-07-01",
    )
    conn.execute(
        "UPDATE instrument_price_mappings SET provider_symbol=provider_symbol || '-changed'"
    )
    conn.commit()
    after = _local_backfill_input_revision(
        conn,
        source="postfinance",
        account_ids=["a"],
        period_from="2026-01-01",
        period_to="2026-07-01",
    )
    assert after != before


def scenario_stale_prices_are_visible_but_not_counted_as_current_coverage(tmp_path: Path) -> None:
    conn = database(policy=False)
    provider = StaleProvider(quotes_for("2026-07-01").quotes)
    result = run_daily_market_valuation(
        conn, as_of="2026-07-01", price_providers={"mock": provider},
        fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}),
        lock_path=tmp_path / "job.lock",
    )
    assert result.status == "partial" and "stale_price" in result.reason_codes
    analytics = build_portfolio_analytics(conn)
    assert analytics["coverage"]["price_pct"] == "0.00"


def scenario_non_etf_benchmark_return_type_is_explicit_total_return(tmp_path: Path) -> None:
    conn = database()
    conn.execute("UPDATE instruments SET asset_class='index' WHERE instrument_id='bench'")
    conn.commit()
    run_daily_market_valuation(
        conn, as_of="2026-07-01", price_providers={"mock": quotes_for("2026-07-01")},
        fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}),
        lock_path=tmp_path / "job.lock",
    )
    assert conn.execute("SELECT return_type FROM benchmark_snapshots").fetchone()[0] == "total_return"


def scenario_benchmark_rejects_currency_exchange_and_stale_quotes(tmp_path: Path) -> None:
    class BadBenchmarkProvider(DatedProvider):
        def __init__(self, mode: str) -> None:
            super().__init__(quotes_for("2026-07-01").quotes)
            self.mode = mode

        def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote:
            quote = super().get_price(provider_symbol, price_date=price_date)
            if provider_symbol != "BENCH.S":
                return quote
            return EquityPriceQuote(
                provider_symbol=provider_symbol,
                currency="EUR" if self.mode == "currency" else "USD",
                close=quote.close,
                adjusted_close=quote.adjusted_close,
                provider=self.name,
                provider_market="XNAS" if self.mode == "exchange" else "SIX",
                price_timestamp="2026-06-20T20:00:00+00:00" if self.mode == "stale" else f"{price_date}T20:00:00+00:00",
            )

    expected = {"currency": "benchmark_currency_mismatch", "exchange": "benchmark_exchange_mismatch", "stale": "benchmark_stale_price"}
    for mode, reason in expected.items():
        conn = database()
        conn.execute("UPDATE instrument_price_mappings SET provider_market='SIX' WHERE instrument_id='bench'")
        conn.commit()
        result = run_daily_market_valuation(
            conn,
            as_of="2026-07-01",
            price_providers={"mock": BadBenchmarkProvider(mode)},
            fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}),
            lock_path=tmp_path / f"{mode}.lock",
        )
        assert result.benchmark_stored == 0 and reason in result.reason_codes
        assert conn.execute("SELECT COUNT(*) FROM benchmark_snapshots").fetchone()[0] == 0


def scenario_benchmark_and_portfolio_align_on_same_chf_period_and_risk_short_history_is_suppressed(tmp_path: Path) -> None:
    conn = database()
    fx = Fx({
        ("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80",
        ("EUR", "2026-07-02"): "0.96", ("USD", "2026-07-02"): "0.81",
    })
    first = quotes_for("2026-07-01")
    second = DatedProvider({
        ("CHF.S", "2026-07-02"): ("102", "CHF", None),
        ("EUR.S", "2026-07-02"): ("51", "EUR", None),
        ("USD.S", "2026-07-02"): ("21", "USD", None),
        ("BENCH.S", "2026-07-02"): ("204", "USD", "214"),
    })
    run_daily_market_valuation(conn, as_of="2026-07-01", price_providers={"mock": first}, fx_provider=fx, lock_path=tmp_path / "job.lock")
    run_daily_market_valuation(conn, as_of="2026-07-02", price_providers={"mock": second}, fx_provider=fx, lock_path=tmp_path / "job.lock")
    analytics = build_portfolio_analytics(conn)
    canonical = build_portfolio_performance(
        conn,
        from_date="2026-07-01",
        to_date="2026-07-02",
        method="twr",
        data_cutoff="9999-12-31T23:59:59+00:00",
    )
    assert canonical["engine_version"] == "portfolio_performance_v2"
    assert [point["at"] for point in cast(list[dict[str, str]], canonical["time_series"])] == ["2026-07-01", "2026-07-02"]
    assert [point["at"] for point in cast(list[dict[str, str]], canonical["ttwror_series"])] == ["2026-07-01", "2026-07-02"]
    assert analytics["performance"]["return_type"] == "etf_proxy"
    assert analytics["performance"]["deprecated"] is True
    assert analytics["performance"]["source"] == "portfolio_performance_v2"
    assert [point["date"] for point in analytics["performance"]["valuation_points"]] == ["2026-07-01", "2026-07-02"]
    assert [point["date"] for point in analytics["performance"]["benchmark_points"]] == ["2026-07-01", "2026-07-02"]
    assert [point["date"] for point in analytics["performance"]["normalized"]] == ["2026-07-01", "2026-07-02"]
    assert analytics["risk"]["volatility_pct"] is None
    assert analytics["risk"]["max_drawdown_pct"] is None
    assert "insufficient_history_30_observations" in analytics["risk"]["reason_codes"]
    assert "Positionslimit überschritten" in analytics["risk"]["policy_breaches"]
    set_performance_scope_classification(
        conn,
        account_id="a",
        included=False,
        classification_role="not_in_investment_performance_scope",
        source="synthetic_test_fixture",
        note="Exercise analytics without included performance accounts",
        classified_at="2026-08-01T00:00:00+00:00",
    )
    excluded = build_portfolio_performance(
        conn,
        from_date="2026-07-01",
        to_date="2026-07-02",
        method="twr",
        data_cutoff="9999-12-31T23:59:59+00:00",
    )
    excluded_quality = cast(dict[str, Any], excluded["quality"])
    excluded_summary = cast(dict[str, Any], excluded["summary"])
    assert cast(dict[str, Any], excluded_quality["ttwror"])["status"] == "unavailable"
    assert excluded_summary["ttwror_cumulative"] is None


def scenario_analytics_get_projection_is_read_only_and_scheduler_lock_blocks_parallel_run(tmp_path: Path) -> None:
    conn = database(policy=False)
    before = conn.total_changes
    first = build_portfolio_analytics(conn)
    second = build_portfolio_analytics(conn)
    assert first == second and conn.total_changes == before
    app = create_app(write_mode="test")
    app.dependency_overrides[get_db] = lambda: conn
    response = TestClient(app).get("/api/portfolio/analytics?period=1y")
    assert response.status_code == 200
    assert response.json()["status"] == "unavailable"
    assert conn.total_changes == before
    for path in ("/api/overview",):
        response = TestClient(app).get(path)
        assert response.status_code == 200
        assert conn.total_changes == before
    dashboard_data.get_positions(conn)
    assert conn.total_changes == before

    lock_path = tmp_path / "job.lock"
    with lock_path.open("a+") as handle:
        fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        with pytest.raises(RuntimeError, match="market_job_already_running"):
            run_daily_market_valuation(
                conn, as_of="2026-07-01", price_providers={"mock": quotes_for("2026-07-01")},
                fx_provider=Fx({}), lock_path=lock_path,
            )
        fcntl.flock(handle.fileno(), fcntl.LOCK_UN)


def test_market_job_cash_cutoff_rejects_future_balance_and_cash_changes_break_idempotency(tmp_path: Path) -> None:
    conn = database(policy=False)
    conn.execute(
        """INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,is_active,created_at)
           VALUES('bank','p','Synthetic Bank Account','cash','CHF',1,?)""",
        (NOW,),
    )
    conn.execute(
        """INSERT INTO cash_account_snapshots(
               snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency,
               amount_chf,source,created_at
           ) VALUES('bank-cash','bank','manual_balance','2026-07-01','5000','CHF','5000','synthetic',?)""",
        (NOW,),
    )
    conn.execute(
        """INSERT INTO cash_account_snapshots(
               snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency,
               amount_chf,source,created_at
           ) VALUES('future-cash','a','confirmed','2026-07-10','999','CHF','999','synthetic',?)""",
        (NOW,),
    )
    conn.commit()
    first = run_daily_market_valuation(
        conn,
        as_of="2026-07-01",
        price_providers={"mock": quotes_for("2026-07-01")},
        fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}),
        lock_path=tmp_path / "job.lock",
    )
    first_value = conn.execute(
        """SELECT value_original FROM portfolio_valuation_snapshots
           WHERE scope_kind='account' AND scope_id='a' ORDER BY snapshot_version DESC LIMIT 1"""
    ).fetchone()[0]
    assert first_value == "1635.00"
    assert conn.execute(
        "SELECT COUNT(*) FROM portfolio_valuation_snapshots WHERE scope_kind='account' AND scope_id='bank'"
    ).fetchone()[0] == 0

    conn.execute(
        """INSERT INTO cash_account_snapshots(
               snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency,
               amount_chf,source,created_at
           ) VALUES('current-cash','a','confirmed','2026-07-01','10','CHF','10','synthetic',?)""",
        (NOW,),
    )
    conn.commit()
    second = run_daily_market_valuation(
        conn,
        as_of="2026-07-01",
        price_providers={"mock": quotes_for("2026-07-01")},
        fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}),
        lock_path=tmp_path / "job.lock",
    )
    assert first.run_id != second.run_id and second.idempotent is False
    assert conn.execute(
        """SELECT value_original FROM portfolio_valuation_snapshots
           WHERE scope_kind='account' AND scope_id='a' ORDER BY snapshot_version DESC LIMIT 1"""
    ).fetchone()[0] == "1645.00"
    lineage = conn.execute(
        """SELECT snapshot_version,supersedes_snapshot_id FROM portfolio_valuation_snapshots
           WHERE scope_kind='account' AND scope_id='a' ORDER BY snapshot_version"""
    ).fetchall()
    assert [row["snapshot_version"] for row in lineage] == [1, 2]
    assert lineage[1]["supersedes_snapshot_id"] is not None


def test_identical_partial_market_retry_does_not_replace_immutable_snapshots(tmp_path: Path) -> None:
    conn = database(policy=False)
    providers: dict[str, Any] = {"mock": quotes_for("2026-07-01", missing={"EUR.S"})}
    fx = Fx({("USD", "2026-07-01"): None})
    first = run_daily_market_valuation(
        conn, as_of="2026-07-01", price_providers=providers, fx_provider=fx,
        lock_path=tmp_path / "job.lock",
    )
    before = [tuple(row) for row in conn.execute(
        """SELECT snapshot_id,snapshot_version,supersedes_snapshot_id,value_original
           FROM portfolio_valuation_snapshots ORDER BY snapshot_id"""
    ).fetchall()]
    second = run_daily_market_valuation(
        conn, as_of="2026-07-01", price_providers=providers, fx_provider=fx,
        lock_path=tmp_path / "job.lock",
    )
    after = [tuple(row) for row in conn.execute(
        """SELECT snapshot_id,snapshot_version,supersedes_snapshot_id,value_original
           FROM portfolio_valuation_snapshots ORDER BY snapshot_id"""
    ).fetchall()]
    assert first.status == second.status == "partial"
    assert first.run_id == second.run_id
    assert before == after
    assert second.valuation_stored == 0


def run_portfolio_market_analytics_v1_scenarios(tmp_path: Path) -> None:
    """Run the Sprint-8 synthetic contract inside the existing integration gate count."""

    scenarios = (
        test_daily_job_uses_confirmed_chf_eur_usd_decimal_fx_and_unadjusted_close_idempotently,
        scenario_exact_date_cache_recovers_transient_provider_gap_without_rewriting_price,
        scenario_partial_provider_failure_unknown_isin_missing_fx_and_unconfirmed_record_are_excluded,
        scenario_weekend_uses_friday_without_staleness_and_benchmark_is_explicit_etf_proxy,
        scenario_market_holiday_uses_prior_business_day_fx_without_false_staleness,
        scenario_same_instrument_in_two_accounts_is_valued_without_overwrite,
        scenario_stale_prices_are_visible_but_not_counted_as_current_coverage,
        scenario_non_etf_benchmark_return_type_is_explicit_total_return,
        scenario_benchmark_rejects_currency_exchange_and_stale_quotes,
        scenario_benchmark_and_portfolio_align_on_same_chf_period_and_risk_short_history_is_suppressed,
        scenario_analytics_get_projection_is_read_only_and_scheduler_lock_blocks_parallel_run,
    )
    for index, scenario in enumerate(scenarios):
        scenario_path = tmp_path / str(index)
        scenario_path.mkdir(parents=True, exist_ok=True)
        scenario(scenario_path)