from __future__ import annotations

import fcntl
import json
from decimal import Decimal
from pathlib import Path
from sqlite3 import Connection

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.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.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,
            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,
            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,
            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',1,1,?)""",
        (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,mapping_status,confidence,created_at,updated_at)
               VALUES(?,?,?,?,?,'mock',?,'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 scenario_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_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 {"quote_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',1,1,?)""", (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 == 4
    assert conn.execute(
        "SELECT COUNT(*) FROM portfolio_valuation_snapshots WHERE scope_kind='instrument' AND scope_id='chf'"
    ).fetchone()[0] == 2
    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 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_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)
    assert analytics["performance"]["return_type"] == "etf_proxy"
    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"]


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

    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 run_portfolio_market_analytics_v1_scenarios(tmp_path: Path) -> None:
    """Run the Sprint-8 synthetic contract inside the existing integration gate count."""

    scenarios = (
        scenario_daily_job_uses_confirmed_chf_eur_usd_decimal_fx_and_unadjusted_close_idempotently,
        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_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)