from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta
import sqlite3
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from test_postfinance_portfolio_source import run_postfinance_source_contract
from test_portfolio_market_analytics_v1 import run_portfolio_market_analytics_v1_scenarios
from test_sprint9_fx_fmp_history import run_sprint9_fx_fmp_history_contract

from jarvis_finance.api.dependencies import get_db
from jarvis_finance.api.main import create_app
from jarvis_finance.services import portfolio_data
from jarvis_finance.services.portfolio_data import (
    build_portfolio_reconciliation,
    confirm_ingestion,
    ingestion_history,
    list_data_sources,
    preview_ingestion,
)
from jarvis_finance.services.portfolio_performance import build_portfolio_performance
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 MIGRATION_VERSION, apply_migrations

NOW = "2026-07-23T12:00:00+00:00"


def database(path: Path | None = None) -> sqlite3.Connection:
    conn = connect(path or ":memory:")
    apply_migrations(conn)
    seed_registry(conn)
    return conn


def seed_registry(conn: sqlite3.Connection) -> None:
    conn.execute("INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES('p','Synthetic','broker',?)", (NOW,))
    for account_id, name, account_type, role in (
        ("a", "AKB Haushaltskonto", "brokerage", "postfinance_etrading_depot"),
        ("b", "Raiffeisen Transferkonto", "brokerage", "postfinance_etrading_cash"),
        ("total", "Synthetic Totalwert", "robo_advisor", "crypto_portfolio"),
    ):
        conn.execute(
            """INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,performance_included,is_active,created_at)
               VALUES(?,?,?,?, 'CHF',0,1,?)""", (account_id, "p", name, account_type, NOW)
        )
        set_performance_scope_classification(
            conn,
            account_id=account_id,
            included=True,
            classification_role=role,
            source="synthetic_test_fixture",
            note="Explicit synthetic investment role; account names are intentionally irrelevant",
            classified_at=NOW,
        )
        set_performance_cashflow_coverage(
            conn,
            account_id=account_id,
            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 instruments(instrument_id,asset_class,name,ticker,isin,exchange,currency,is_active,created_at)
           VALUES('i','equity','Synthetic Asset','SYN','CH0000000001','SIX','CHF',1,?)""", (NOW,)
    )
    conn.execute(
        """INSERT INTO instruments(instrument_id,asset_class,name,ticker,isin,exchange,currency,is_active,created_at)
           VALUES('eur','equity','Synthetic EUR Asset','EURX','CH0000000002','SIX','EUR',1,?)""", (NOW,)
    )
    conn.commit()


def add_legacy(conn: sqlite3.Connection, snapshot_id: str, account: str, at: str, value: str, *, created: str = NOW, updated: str | None = None) -> None:
    conn.execute(
        """INSERT INTO account_value_snapshots(snapshot_id,account_id,valuation_date,total_value_chf,currency,source_type,quality_status,created_at,updated_at)
           VALUES(?,?,?,?,'CHF','manual_total_value','ok',?,?)""", (snapshot_id, account, at, value, created, updated)
    )
    conn.commit()


def add_cash(conn: sqlite3.Connection, snapshot_id: str, account: str, at: str, value: str) -> None:
    conn.execute(
        """INSERT INTO cash_account_snapshots(snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency,amount_chf,source,created_at)
           VALUES(?,?, 'reconciliation', ?,?,'CHF',?,'manual',?)""", (snapshot_id, account, at, value, value, NOW)
    )
    conn.commit()


def add_transaction(
    conn: sqlite3.Connection, transaction_id: str, kind: str, account: str, at: str,
    *, instrument: str | None = None, quantity: str | None = None, gross: str | None = None,
    net: str | None = None, row_hash: str | None = None, transfer_group: str | None = None,
) -> None:
    conn.execute(
        """INSERT INTO transactions(transaction_id,transaction_type,activity_kind,account_id,instrument_id,trade_date,
             event_timestamp,quantity,gross_amount_original,net_amount_original,currency_original,fx_rate_to_chf,
             fx_status,source_type,source_id,row_hash,is_confirmed,quality_status,internal_transfer_group_id,created_at)
           VALUES(?,?,?,?,?,?,?,?,?,?,'CHF','1','ok','fixture','fixture',?,1,'ok',?,?)""",
        (transaction_id, kind, kind, account, instrument, at, at, quantity, gross, net, row_hash or f"hash-{transaction_id}", transfer_group, NOW),
    )
    conn.commit()


def add_canonical_valuation(conn: sqlite3.Connection, snapshot_id: str, account: str, at: str, value: str) -> None:
    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,quality_status,reason_codes_json)
           VALUES(?, 'account', ?, ?, ?, 'CHF','CHF','1','original_to_base',?,'fixture',?,1,'complete','[]')""",
        (snapshot_id, account, account, value, at, NOW),
    )
    conn.commit()


def request(source: str = "legacy_account_values") -> dict[str, object]:
    return {"source_key": source, "scope_kind": "portfolio", "account_id": None, "period_from": "2025-01-01", "period_to": "2025-12-31", "data_cutoff": NOW}


def confirm_payload(preview: dict[str, object]) -> dict[str, object]:
    return {
        **{key: preview[key] for key in ("source_key", "scope_kind", "account_id", "period_from", "period_to", "data_cutoff")},
        **{key: preview[key] for key in ("preview_id", "confirmation_id", "preview_created_at", "source_revision", "input_fingerprint", "payload_hash")},
        "confirm": True,
    }


def test_migrations_41_to_43_are_additive_and_ingestion_history_is_immutable(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    conn = database()
    assert MIGRATION_VERSION == 53
    assert conn.execute("SELECT MAX(version) FROM schema_migrations").fetchone()[0] == MIGRATION_VERSION
    assert {row[1] for row in conn.execute("PRAGMA table_info(instrument_price_mappings)")} >= {
        "source_symbol", "source_venue", "source_currency",
    }
    assert conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='portfolio_ingestion_batches'").fetchone()
    add_legacy(conn, "legacy", "a", "2025-12-31", "1000")
    result = confirm_ingestion(conn, confirm_payload(preview_ingestion(conn, request())))
    with pytest.raises(sqlite3.DatabaseError, match="immutable"):
        conn.execute("UPDATE portfolio_ingestion_batches SET status='confirmed' WHERE batch_id=?", (result["batch_id"],))
    with pytest.raises(sqlite3.DatabaseError, match="cannot be deleted"):
        conn.execute("DELETE FROM portfolio_ingestion_items WHERE batch_id=?", (result["batch_id"],))
    with pytest.raises(sqlite3.DatabaseError, match="audit is immutable"):
        conn.execute("UPDATE audit_log SET action='changed' WHERE entity_id=?", (result["batch_id"],))
    with pytest.raises(sqlite3.DatabaseError, match="audit cannot be deleted"):
        conn.execute("DELETE FROM audit_log WHERE entity_id=?", (result["batch_id"],))
    run_portfolio_market_analytics_v1_scenarios(tmp_path / "market-analytics")
    run_sprint9_fx_fmp_history_contract(tmp_path / "sprint9-history", monkeypatch)


def test_preview_is_storage_free_and_classifies_new_duplicate_blocked_and_unchanged() -> None:
    conn = database()
    add_legacy(conn, "one", "a", "2025-12-31", "1000")
    add_legacy(conn, "duplicate", "a", "2025-12-31", "1000")
    add_legacy(conn, "blocked", "b", "2025-12-31", "900", updated=NOW)
    before = conn.total_changes
    preview = preview_ingestion(conn, request())
    assert conn.total_changes == before
    assert preview["counts"] == {"new": 1, "unchanged": 0, "duplicate": 1, "ambiguous": 0, "blocked": 1, "versioned": 0, "discovered": 3}
    assert preview["quality_impact"]["status"] == "partial"
    assert len(preview["payload_hash"]) == 64
    assert all("/" not in item["source_record_ref"] for item in preview["items"])
    confirmed = confirm_ingestion(conn, confirm_payload(preview))
    assert confirmed["written_records"] == 1
    assert conn.execute("SELECT COUNT(*) FROM portfolio_ingestion_items WHERE batch_id=?", (confirmed["batch_id"],)).fetchone()[0] == 3


def test_confirm_is_atomic_audited_idempotent_and_second_fresh_preview_is_unchanged() -> None:
    conn = database()
    add_legacy(conn, "legacy", "a", "2025-12-31", "1000")
    preview = preview_ingestion(conn, request())
    payload = confirm_payload(preview)
    first = confirm_ingestion(conn, payload)
    second = confirm_ingestion(conn, payload)
    assert first["idempotent"] is False and first["written_records"] == 1
    assert second == {"batch_id": first["batch_id"], "audit_id": first["audit_id"], "idempotent": True}
    assert conn.execute("SELECT COUNT(*) FROM portfolio_valuation_snapshots WHERE source='ingestion:legacy_account_values'").fetchone()[0] == 1
    assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity_id=?", (first["batch_id"],)).fetchone()[0] == 1
    fresh = preview_ingestion(conn, request())
    assert fresh["counts"]["unchanged"] == 1 and fresh["counts"]["new"] == 0
    run_postfinance_source_contract()


def test_confirmation_identity_rejects_changed_payload_and_stale_source_revision() -> None:
    conn = database()
    add_legacy(conn, "legacy", "a", "2025-12-31", "1000")
    preview = preview_ingestion(conn, request())
    payload = confirm_payload(preview)
    confirm_ingestion(conn, payload)
    changed = dict(payload)
    changed["period_from"] = "2025-02-01"
    with pytest.raises(ValueError, match="verändertem Inhalt"):
        confirm_ingestion(conn, changed)
    preview2 = preview_ingestion(conn, request())
    add_legacy(conn, "later", "b", "2025-12-31", "500", created="2026-01-01T00:00:00+00:00")
    with pytest.raises(ValueError, match="veraltet|Ausgangsrevision"):
        confirm_ingestion(conn, confirm_payload(preview2))


def test_expired_preview_is_rejected() -> None:
    conn = database()
    add_legacy(conn, "legacy", "a", "2025-12-31", "1000")
    preview = preview_ingestion(conn, request(), preview_created_at="2020-01-01T00:00:00+00:00")
    with pytest.raises(ValueError, match="abgelaufen"):
        confirm_ingestion(conn, confirm_payload(preview))


def test_snapshot_correction_creates_new_immutable_version() -> None:
    conn = database()
    add_legacy(conn, "v1", "a", "2025-12-31", "1000", created="2026-01-01T00:00:00+00:00")
    confirm_ingestion(conn, confirm_payload(preview_ingestion(conn, request())))
    add_legacy(conn, "v2", "a", "2025-12-31", "1010", created="2026-02-01T00:00:00+00:00")
    preview = preview_ingestion(conn, request())
    assert preview["counts"]["versioned"] == 1
    confirm_ingestion(conn, confirm_payload(preview))
    rows = conn.execute("SELECT snapshot_version,supersedes_snapshot_id,value_original FROM portfolio_valuation_snapshots WHERE source='ingestion:legacy_account_values' ORDER BY snapshot_version").fetchall()
    assert [(row["snapshot_version"], row["value_original"]) for row in rows] == [(1, "1000"), (2, "1010")]
    assert rows[1]["supersedes_snapshot_id"] is not None


def test_canonical_activity_preview_handles_duplicates_ambiguity_and_unsupported_records() -> None:
    conn = database()
    add_transaction(conn, "buy1", "buy", "a", "2025-01-01", instrument="i", quantity="1", gross="100", row_hash="same")
    with pytest.raises(sqlite3.IntegrityError, match="transactions.row_hash"):
        add_transaction(conn, "buy2", "buy", "a", "2025-01-01", instrument="i", quantity="1", gross="100", row_hash="same")
    add_transaction(conn, "ambiguous", "buy", "a", "2025-01-02", quantity="1", gross="100")
    add_transaction(conn, "blocked", "option_exercise", "a", "2025-01-03", instrument="i", quantity="1")
    preview = preview_ingestion(conn, request("canonical_transactions"))
    assert preview["counts"]["unchanged"] == 1
    assert preview["counts"]["duplicate"] == 0
    assert preview["counts"]["ambiguous"] == 1
    assert preview["counts"]["blocked"] == 1
    assert preview["planned_activities"] == []


def test_parallel_confirmation_produces_one_batch_and_one_snapshot(tmp_path: Path) -> None:
    path = tmp_path / "parallel.sqlite3"
    conn = database(path)
    add_legacy(conn, "legacy", "a", "2025-12-31", "1000")
    preview = preview_ingestion(conn, request())
    payload = confirm_payload(preview)
    conn.close()

    def execute() -> dict[str, object]:
        local = connect(path)
        try:
            return confirm_ingestion(local, payload)
        finally:
            local.close()

    with ThreadPoolExecutor(max_workers=2) as pool:
        results = list(pool.map(lambda _: execute(), range(2)))
    verify = connect(path)
    assert sum(not bool(result["idempotent"]) for result in results) == 1
    assert verify.execute("SELECT COUNT(*) FROM portfolio_ingestion_batches").fetchone()[0] == 1
    assert verify.execute("SELECT COUNT(*) FROM portfolio_valuation_snapshots").fetchone()[0] == 1


def test_source_status_and_history_are_redacted_and_paginated() -> None:
    conn = database()
    add_legacy(conn, "legacy", "a", "2025-12-31", "1000")
    confirm_ingestion(conn, confirm_payload(preview_ingestion(conn, request())))
    sources = list_data_sources(conn)
    assert {item["source_key"] for item in sources["sources"]} >= {"canonical_transactions", "legacy_account_values", "reported_positions", "market_prices", "fx_rates"}
    history = ingestion_history(conn, limit=10, offset=0)
    assert history["total"] == 1
    assert history["items"][0]["batch_id"].startswith("batch-")
    assert "source_filename" not in str(history)


def seed_matched_reconciliation(conn: sqlite3.Connection) -> None:
    add_transaction(conn, "buy", "buy", "a", "2025-01-01", instrument="i", quantity="10", gross="800")
    conn.execute(
        """INSERT INTO positions_snapshot(position_snapshot_id,snapshot_date,account_id,platform_id,instrument_id,quantity,market_value_chf,data_quality_status,created_at)
           VALUES('pos','2025-12-31','a','p','i','10','1000','ok',?)""", (NOW,)
    )
    conn.execute(
        """INSERT INTO market_prices(market_price_id,instrument_id,price_date,close,currency,provider,quality_status,created_at)
           VALUES('price','i','2025-12-31','100','CHF','fixture','ok',?)""", (NOW,)
    )
    add_cash(conn, "cash", "a", "2025-12-31", "100")
    add_legacy(conn, "total-a", "a", "2025-12-31", "1100")
    add_legacy(conn, "total-only", "total", "2025-12-31", "5000")
    add_canonical_valuation(conn, "canonical-total", "total", "2025-12-31", "5000")
    conn.commit()


def test_reconciliation_matches_quantity_value_cash_and_total_value_account_without_holdings() -> None:
    conn = database()
    seed_matched_reconciliation(conn)
    result = build_portfolio_reconciliation(conn, as_of="2025-12-31", data_cutoff=NOW)
    instrument = next(item for item in result["differences"] if item["instrument_label"] == "Synthetic Asset")
    assert instrument["quantity"] == {"ledger": "10", "reported": "10", "difference": "0", "status": "matched"}
    assert instrument["valuation"]["status"] == "matched"
    total_only = next(item for item in result["account_totals"] if item["total_value_only"])
    assert total_only["status"] == "matched"
    assert total_only["positions_value"] is None
    assert result["coverage"]["activities"]["ratio"] == "1"
    assert len(result["input_fingerprint"]) == 64
    repeated = build_portfolio_reconciliation(conn, as_of="2025-12-31", data_cutoff=NOW)
    assert repeated["input_fingerprint"] == result["input_fingerprint"]


def test_reconciliation_distinguishes_tolerance_mismatch_missing_fx_and_cutoff() -> None:
    conn = database()
    seed_matched_reconciliation(conn)
    conn.execute("UPDATE positions_snapshot SET market_value_chf='1000.005' WHERE position_snapshot_id='pos'")
    inside = build_portfolio_reconciliation(conn, as_of="2025-12-31", data_cutoff=NOW)
    assert inside["differences"][0]["valuation"]["status"] == "within_tolerance"
    conn.execute("UPDATE positions_snapshot SET quantity='9',market_value_chf='900' WHERE position_snapshot_id='pos'")
    mismatch = build_portfolio_reconciliation(conn, as_of="2025-12-31", data_cutoff=NOW)
    assert "quantity_mismatch" in mismatch["differences"][0]["reason_codes"]
    assert "value_mismatch" in mismatch["differences"][0]["reason_codes"]
    conn.execute("UPDATE positions_snapshot SET instrument_id='eur',quantity='1',market_value_chf='100' WHERE position_snapshot_id='pos'")
    conn.execute("INSERT INTO transactions(transaction_id,transaction_type,activity_kind,account_id,instrument_id,trade_date,quantity,gross_amount_original,currency_original,source_type,source_id,row_hash,quality_status,created_at) VALUES('eur-buy','buy','buy','a','eur','2025-01-01','1','90','EUR','fixture','fixture','eur-hash','ok',?)", (NOW,))
    conn.execute("INSERT INTO market_prices(market_price_id,instrument_id,price_date,close,currency,provider,quality_status,created_at) VALUES('eur-price','eur','2025-12-31','100','EUR','fixture','ok',?)", (NOW,))
    missing_fx = build_portfolio_reconciliation(conn, as_of="2025-12-31", data_cutoff=NOW)
    assert "missing_fx" in next(item for item in missing_fx["differences"] if item["instrument_label"] == "Synthetic EUR Asset")["reason_codes"]
    conn.execute("UPDATE positions_snapshot SET snapshot_date='2025-12-30' WHERE position_snapshot_id='pos'")
    cutoff = build_portfolio_reconciliation(conn, as_of="2025-12-31", data_cutoff=NOW)
    assert "cutoff_mismatch" in next(item for item in cutoff["differences"] if item["instrument_label"] == "Synthetic EUR Asset")["reason_codes"]


def test_scope_aware_raiffeisen_akb_transfer_is_internal_for_portfolio_and_external_for_account() -> None:
    conn = database()
    add_transaction(conn, "transfer-out", "internal_transfer", "a", "2025-06-01T00:00:00+00:00", net="-100", transfer_group="raiffeisen-akb")
    add_transaction(conn, "transfer-in", "internal_transfer", "b", "2025-06-01T00:00:00+00:00", net="100", transfer_group="raiffeisen-akb")
    for account, values in {"a": ("1000", "1000", "900"), "b": ("1000", "1000", "1100")}.items():
        for suffix, at, value in zip(("open", "flow", "close"), ("2025-01-01", "2025-06-01T00:00:00+00:00", "2025-12-31"), values):
            add_canonical_valuation(conn, f"{account}-{suffix}", account, at, value)
    portfolio = build_portfolio_performance(conn, from_date="2025-01-01", to_date="2025-12-31", data_cutoff=NOW)
    account = build_portfolio_performance(conn, from_date="2025-01-01", to_date="2025-12-31", account_id="a", data_cutoff=NOW)
    assert portfolio["external_cashflows"] == []
    assert account["external_cashflows"] == [{"at": "2025-06-01T00:00:00+00:00", "kind": "external_withdrawal", "amount": "-100"}]


def test_api_openapi_write_security_and_read_routes() -> None:
    conn = database()
    seed_matched_reconciliation(conn)
    app = create_app(write_mode="test")
    app.dependency_overrides[get_db] = lambda: conn
    client = TestClient(app)
    assert client.get("/api/portfolio/data-sources").status_code == 200
    assert client.get("/api/portfolio/reconciliation?as_of=2025-12-31&data_cutoff=2026-07-23T12%3A00%3A00%2B00%3A00").status_code == 200
    preview_response = client.post("/api/portfolio/ingestion/preview", json=request())
    assert preview_response.status_code == 200
    confirm_response = client.post("/api/portfolio/ingestion/confirm", json=confirm_payload(preview_response.json()))
    assert confirm_response.status_code == 200
    history_response = client.get("/api/portfolio/ingestion/history?limit=20&offset=0")
    assert history_response.status_code == 200
    assert history_response.json()["total"] == 1
    paths = app.openapi()["paths"]
    assert set(paths["/api/portfolio/data-sources"]) == {"get"}
    assert set(paths["/api/portfolio/ingestion/preview"]) == {"post"}
    assert set(paths["/api/portfolio/ingestion/confirm"]) == {"post"}
    assert set(paths["/api/portfolio/ingestion/history"]) == {"get"}
    assert set(paths["/api/portfolio/reconciliation"]) == {"get"}
    assert paths["/api/portfolio/ingestion/confirm"]["post"]["responses"]["409"]["description"] == "Abgelaufene oder veränderte Preview"
    blocked_app = create_app(write_mode="disabled")
    blocked_app.dependency_overrides[get_db] = lambda: conn
    blocked = TestClient(blocked_app, client=("100.64.0.10", 50000))
    assert blocked.post("/api/portfolio/ingestion/preview", json=request()).status_code == 200
    assert blocked.post("/api/portfolio/ingestion/confirm", json={}).status_code == 403


def test_invalid_parameters_fail_closed() -> None:
    conn = database()
    with pytest.raises(ValueError, match="Datenquelle"):
        preview_ingestion(conn, {**request(), "source_key": "bank_scraper"})
    with pytest.raises(ValueError, match="Zeitraum"):
        preview_ingestion(conn, {**request(), "period_from": "2026-01-01"})
    with pytest.raises(ValueError, match="eindeutige Zeitzone"):
        preview_ingestion(conn, {**request(), "data_cutoff": "2026-07-23T12:00:00"})
    with pytest.raises(ValueError, match="ausschliesslich CHF"):
        build_portfolio_reconciliation(conn, as_of="2025-12-31", base_currency="EUR")
    with pytest.raises(ValueError, match="Decimal-Text"):
        build_portfolio_reconciliation(conn, as_of="2025-12-31", absolute_tolerance=0.1)  # type: ignore[arg-type]


def test_multiple_same_day_values_receive_sequential_versions_in_one_batch() -> None:
    conn = database()
    add_legacy(conn, "v1", "a", "2025-12-31", "1000", created="2026-01-01T00:00:00+00:00")
    add_legacy(conn, "v2", "a", "2025-12-31", "1010", created="2026-02-01T00:00:00+00:00")
    preview = preview_ingestion(conn, request())
    assert preview["counts"]["new"] == 1
    assert preview["counts"]["versioned"] == 1
    result = confirm_ingestion(conn, confirm_payload(preview))
    assert result["written_records"] == 2
    rows = conn.execute(
        """SELECT snapshot_id,snapshot_version,supersedes_snapshot_id,value_original
           FROM portfolio_valuation_snapshots ORDER BY snapshot_version"""
    ).fetchall()
    assert [(row["snapshot_version"], row["value_original"]) for row in rows] == [(1, "1000"), (2, "1010")]
    assert rows[1]["supersedes_snapshot_id"] == rows[0]["snapshot_id"]


def test_preview_timestamp_is_hash_bound_and_expired_idempotent_retry_still_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:
    conn = database()
    add_legacy(conn, "legacy", "a", "2025-12-31", "1000")
    preview = preview_ingestion(conn, request())
    payload = confirm_payload(preview)
    first = confirm_ingestion(conn, payload)
    created = datetime.fromisoformat(str(preview["preview_created_at"]))
    monkeypatch.setattr(portfolio_data, "_now", lambda: created + timedelta(hours=1))
    retry = confirm_ingestion(conn, payload)
    assert retry == {"batch_id": first["batch_id"], "audit_id": first["audit_id"], "idempotent": True}
    tampered = dict(payload)
    tampered["preview_created_at"] = (created + timedelta(days=1)).isoformat()
    with pytest.raises(ValueError, match="verändertem Inhalt"):
        confirm_ingestion(conn, tampered)


def test_unconfirmed_transactions_are_excluded_from_ingestion_and_reconciliation() -> None:
    conn = database()
    add_transaction(conn, "confirmed", "buy", "a", "2025-01-01", instrument="i", quantity="10", gross="800")
    conn.execute(
        """INSERT INTO transactions(transaction_id,transaction_type,activity_kind,account_id,instrument_id,trade_date,
             quantity,gross_amount_original,currency_original,source_type,source_id,row_hash,is_confirmed,quality_status,created_at)
           VALUES('unconfirmed','buy','buy','a','i','2025-01-02','5','400','CHF','fixture','fixture','unconfirmed-hash',0,'ok',?)""",
        (NOW,),
    )
    conn.execute(
        """INSERT INTO positions_snapshot(position_snapshot_id,snapshot_date,account_id,platform_id,instrument_id,quantity,market_value_chf,data_quality_status,created_at)
           VALUES('pos','2025-12-31','a','p','i','10','1000','ok',?)""",
        (NOW,),
    )
    conn.execute(
        """INSERT INTO market_prices(market_price_id,instrument_id,price_date,close,currency,provider,quality_status,created_at)
           VALUES('price','i','2025-12-31','100','CHF','fixture','ok',?)""",
        (NOW,),
    )
    conn.commit()
    preview = preview_ingestion(conn, request("canonical_transactions"))
    assert preview["counts"]["discovered"] == 1
    reconciliation = build_portfolio_reconciliation(conn, as_of="2025-12-31", data_cutoff=NOW)
    item = reconciliation["differences"][0]
    assert item["quantity"]["ledger"] == "10"
    assert item["quantity"]["status"] == "matched"


def test_stale_position_price_and_fx_inputs_are_not_comparable_and_reduce_coverage() -> None:
    conn = database()
    seed_matched_reconciliation(conn)
    conn.execute("UPDATE positions_snapshot SET snapshot_date='2025-12-30' WHERE position_snapshot_id='pos'")
    conn.execute("UPDATE market_prices SET price_date='2025-12-30' WHERE market_price_id='price'")
    stale = build_portfolio_reconciliation(conn, as_of="2025-12-31", data_cutoff=NOW)
    item = next(item for item in stale["differences"] if item["instrument_label"] == "Synthetic Asset")
    assert item["quantity"]["status"] == "not_comparable"
    assert item["valuation"]["status"] == "not_comparable"
    assert {"stale_snapshot", "stale_price"} <= set(item["reason_codes"])
    assert stale["status"] == "not_comparable"
    assert stale["coverage"]["prices"]["covered"] == 0

    conn.execute("UPDATE positions_snapshot SET instrument_id='eur',snapshot_date='2025-12-31',quantity='1',market_value_chf='100' WHERE position_snapshot_id='pos'")
    conn.execute("DELETE FROM transactions WHERE transaction_id='buy'")
    add_transaction(conn, "eur-buy", "buy", "a", "2025-01-01", instrument="eur", quantity="1", gross="90")
    conn.execute("DELETE FROM market_prices")
    conn.execute("INSERT INTO market_prices(market_price_id,instrument_id,price_date,close,currency,provider,quality_status,created_at) VALUES('eur-price','eur','2025-12-31','100','EUR','fixture','ok',?)", (NOW,))
    conn.execute("INSERT INTO fx_rates(fx_rate_id,base_currency,quote_currency,rate_date,rate,provider,rate_type,quality_status,created_at) VALUES('eur-fx','EUR','CHF','2025-12-30','1','fixture','close','ok',?)", (NOW,))
    conn.commit()
    stale_fx = build_portfolio_reconciliation(conn, as_of="2025-12-31", data_cutoff=NOW)
    eur = next(item for item in stale_fx["differences"] if item["instrument_label"] == "Synthetic EUR Asset")
    assert eur["valuation"]["status"] == "not_comparable"
    assert "stale_fx" in eur["reason_codes"]
    assert stale_fx["coverage"]["fx"]["covered"] == 0