from __future__ import annotations

from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
from sqlite3 import Connection
from typing import Protocol
from urllib import error, parse, request
import json

from jarvis_finance.imports.common import stable_id, utc_now
from jarvis_finance.audit.log import record_audit_event
from jarvis_finance.market_data.catalog import _runtime_secret_value
from jarvis_finance.market_data.instruments import resolve_instrument_alerts, update_instrument_metadata
from jarvis_finance.quality.alerts import create_alert


@dataclass(frozen=True)
class EquityPriceQuote:
    provider_symbol: str
    currency: str
    close: Decimal | None
    provider: str = "mock"
    provider_market: str | None = None
    price_timestamp: str | None = None
    adjusted_close: Decimal | None = None
    quality_status: str = "fresh"
    error_message: str | None = None


@dataclass(frozen=True)
class ProviderCapability:
    supports_latest: bool
    supports_historical_as_of: bool
    supports_exchange_suffix: bool
    rate_limit_policy: str


@dataclass
class MarketPriceRefreshResult:
    asset_class: str
    total_mappings: int = 0
    updated_count: int = 0
    skipped_count: int = 0
    excluded_count: int = 0
    cached_count: int = 0
    stale_count: int = 0
    warning_count: int = 0
    error_count: int = 0
    dry_run: bool = False
    warnings: list[str] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)


class EquityPriceProvider(Protocol):
    name: str
    def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote: ...


class MockEquityPriceProvider:
    name = "mock"
    capability = ProviderCapability(True, True, True, "none")

    def __init__(self, prices: dict[str, Decimal | str | None], *, currency: str = "USD", timestamps: dict[str, str] | None = None) -> None:
        self.prices = {k: (Decimal(str(v)) if v is not None else None) for k, v in prices.items()}
        self.currency = currency.upper()
        self.timestamps = timestamps or {}

    def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote:
        if provider_symbol not in self.prices or self.prices[provider_symbol] is None:
            return EquityPriceQuote(provider_symbol=provider_symbol, currency=self.currency, close=None, quality_status="missing", error_message="price missing", provider=self.name)
        return EquityPriceQuote(provider_symbol=provider_symbol, currency=self.currency, close=self.prices[provider_symbol], price_timestamp=self.timestamps.get(provider_symbol), provider=self.name)


class FmpEquityPriceProvider:
    name = "fmp"
    capability = ProviderCapability(True, True, True, "sequential_retry_after_backoff")

    def __init__(self, *, api_key: str | None = None, timeout_seconds: float = 8.0) -> None:
        self.api_key = api_key or _runtime_secret_value(("FMP_API_KEY", "FINANCIAL_MODELING_PREP_API_KEY", "JARVIS_FMP_API_KEY"))
        self.timeout_seconds = timeout_seconds

    def _get_json(self, path: str, params: dict[str, str]) -> object:
        if not self.api_key:
            raise RuntimeError("fmp_api_key_missing")
        path = path if path.startswith("/") else "/" + path
        url = "https://financialmodelingprep.com" + path + "?" + parse.urlencode({**params, "apikey": self.api_key})
        req = request.Request(url, headers={"Accept": "application/json"})
        try:
            with request.urlopen(req, timeout=self.timeout_seconds) as response:  # noqa: S310 - explicit user-triggered provider lookup
                return json.loads(response.read().decode("utf-8"))
        except error.HTTPError as exc:
            if exc.code in {401, 403}:
                raise RuntimeError("fmp_auth_failed") from exc
            if exc.code == 402:
                raise RuntimeError("fmp_endpoint_restricted") from exc
            if exc.code == 429:
                raise RuntimeError("fmp_rate_limited") from exc
            raise RuntimeError("fmp_provider_error") from exc
        except error.URLError as exc:
            raise RuntimeError("fmp_network_error") from exc

    def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote:
        symbol = (provider_symbol or "").strip().upper()
        if not symbol:
            return EquityPriceQuote(provider_symbol=symbol, currency="", close=None, provider=self.name, quality_status="missing", error_message="provider_symbol_missing")
        try:
            if price_date and price_date != "latest":
                requested = date.fromisoformat(price_date)
                boundary = requested - timedelta(days=7)
                payload = self._get_json(
                    "/stable/historical-price-eod/full",
                    {"symbol": symbol, "from": boundary.isoformat(), "to": requested.isoformat()},
                )
            else:
                requested = None
                boundary = None
                payload = self._get_json("/stable/profile", {"symbol": symbol})
        except RuntimeError as exc:
            return EquityPriceQuote(provider_symbol=symbol, currency="", close=None, provider=self.name, quality_status="missing", error_message=str(exc))
        except ValueError:
            return EquityPriceQuote(provider_symbol=symbol, currency="", close=None, provider=self.name, quality_status="missing", error_message="price_missing")
        if requested is not None and boundary is not None:
            rows = payload if isinstance(payload, list) else []
            if isinstance(payload, dict):
                rows = payload.get("historical") or payload.get("data") or []
            eligible: list[tuple[date, dict[str, object]]] = []
            for item in rows if isinstance(rows, list) else []:
                if not isinstance(item, dict):
                    continue
                try:
                    row_date = date.fromisoformat(str(item.get("date") or "")[:10])
                except ValueError:
                    continue
                if boundary <= row_date <= requested:
                    eligible.append((row_date, item))
            for row_date, historical_row in sorted(eligible, key=lambda item: item[0], reverse=True):
                try:
                    close = Decimal(str(historical_row.get("close")))
                except (InvalidOperation, ValueError):
                    continue
                if close > 0:
                    currency = str(historical_row.get("currency") or "").upper()
                    return EquityPriceQuote(
                        provider_symbol=symbol,
                        currency=currency,
                        close=close,
                        provider=self.name,
                        price_timestamp=row_date.isoformat(),
                        quality_status="fresh",
                    )
            return EquityPriceQuote(provider_symbol=symbol, currency="", close=None, provider=self.name, quality_status="missing", error_message="price_missing")
        row = payload[0] if isinstance(payload, list) and payload and isinstance(payload[0], dict) else payload if isinstance(payload, dict) else {}
        if isinstance(row.get("data"), list) and row["data"] and isinstance(row["data"][0], dict):
            row = row["data"][0]
        price = row.get("price") or row.get("lastPrice")
        try:
            close = Decimal(str(price)) if price not in {None, ""} else None
        except (InvalidOperation, ValueError):
            close = None
        if close is None or close <= 0:
            return EquityPriceQuote(provider_symbol=symbol, currency="", close=None, provider=self.name, quality_status="missing", error_message="price_missing")
        currency = str(row.get("currency") or row.get("priceCurrency") or "").upper()
        return EquityPriceQuote(provider_symbol=symbol, currency=currency or "USD", close=close, provider=self.name, price_timestamp=utc_now(), quality_status="fresh")


class _HttpEquityPriceProvider:
    name = "provider"
    api_key_env: tuple[str, ...] = ()
    capability = ProviderCapability(True, False, False, "sequential_retry_after_backoff")

    def __init__(self, *, api_key: str | None = None, timeout_seconds: float = 8.0) -> None:
        self.api_key = api_key or _runtime_secret_value(self.api_key_env)
        self.timeout_seconds = timeout_seconds

    def _json_url(self, url: str, *, headers: dict[str, str] | None = None) -> object:
        req = request.Request(url, headers=headers or {"Accept": "application/json"})
        try:
            with request.urlopen(req, timeout=self.timeout_seconds) as response:  # noqa: S310 - explicit CLI/user provider lookup
                return json.loads(response.read().decode("utf-8"))
        except error.HTTPError as exc:
            if exc.code in {401, 403}:
                raise RuntimeError(f"{self.name}_auth_failed") from exc
            if exc.code == 402:
                raise RuntimeError(f"{self.name}_endpoint_restricted") from exc
            if exc.code == 429:
                raise RuntimeError(f"{self.name}_rate_limited") from exc
            raise RuntimeError(f"{self.name}_provider_error") from exc
        except error.URLError as exc:
            raise RuntimeError(f"{self.name}_network_error") from exc

    @staticmethod
    def _decimal_price(value: object) -> Decimal | None:
        try:
            price = Decimal(str(value)) if value not in {None, ""} else None
        except (InvalidOperation, ValueError):
            return None
        return price if price is not None and price > 0 else None

    def _missing(self, symbol: str, message: str) -> EquityPriceQuote:
        return EquityPriceQuote(provider_symbol=symbol, currency="", close=None, provider=self.name, quality_status="missing", error_message=message)


class FinnhubEquityPriceProvider(_HttpEquityPriceProvider):
    name = "finnhub"
    api_key_env = ("FINNHUB_API_KEY", "JARVIS_FINNHUB_API_KEY")

    def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote:
        symbol = (provider_symbol or "").strip().upper()
        if not symbol:
            return self._missing(symbol, "provider_symbol_missing")
        if not self.api_key:
            return self._missing(symbol, "finnhub_api_key_missing")
        try:
            payload = self._json_url("https://finnhub.io/api/v1/quote?" + parse.urlencode({"symbol": symbol, "token": self.api_key}))
            profile = self._json_url("https://finnhub.io/api/v1/stock/profile2?" + parse.urlencode({"symbol": symbol, "token": self.api_key}))
        except RuntimeError as exc:
            return self._missing(symbol, str(exc))
        row = payload if isinstance(payload, dict) else {}
        price = self._decimal_price(row.get("c") or row.get("pc"))
        if price is None:
            return self._missing(symbol, "price_missing")
        currency = str(profile.get("currency") or "USD").upper() if isinstance(profile, dict) else "USD"
        return EquityPriceQuote(provider_symbol=symbol, currency=currency or "USD", close=price, provider=self.name, price_timestamp=utc_now(), quality_status="fresh")


class TwelveDataEquityPriceProvider(_HttpEquityPriceProvider):
    name = "twelvedata"
    api_key_env = ("TWELVEDATA_API_KEY", "TWELVE_DATA_API_KEY", "JARVIS_TWELVEDATA_API_KEY", "JARVIS_TWELVE_DATA_API_KEY")

    def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote:
        symbol = (provider_symbol or "").strip().upper()
        if not symbol:
            return self._missing(symbol, "provider_symbol_missing")
        if not self.api_key:
            return self._missing(symbol, "twelvedata_api_key_missing")
        try:
            payload = self._json_url("https://api.twelvedata.com/quote?" + parse.urlencode({"symbol": symbol, "apikey": self.api_key}))
        except RuntimeError as exc:
            return self._missing(symbol, str(exc))
        row = payload if isinstance(payload, dict) else {}
        if str(row.get("status") or "").lower() == "error":
            return self._missing(symbol, "twelvedata_provider_error")
        price = self._decimal_price(row.get("close") or row.get("previous_close") or row.get("price"))
        if price is None:
            return self._missing(symbol, "price_missing")
        currency = str(row.get("currency") or "USD").upper()
        ts = str(row.get("datetime") or row.get("timestamp") or utc_now())
        return EquityPriceQuote(provider_symbol=symbol, currency=currency or "USD", close=price, provider=self.name, price_timestamp=ts, quality_status="fresh")


class MassiveEquityPriceProvider(_HttpEquityPriceProvider):
    name = "massive"
    api_key_env = ("MASSIVE_API_KEY", "JARVIS_MASSIVE_API_KEY")

    def _massive_json(self, path: str) -> object:
        if not self.api_key:
            raise RuntimeError("massive_api_key_missing")
        base = "https://api.massive.com" + path
        try:
            return self._json_url(base, headers={"Accept": "application/json", "Authorization": f"Bearer {self.api_key}"})
        except RuntimeError as exc:
            if "auth_failed" not in str(exc):
                raise
            sep = "&" if "?" in base else "?"
            return self._json_url(base + sep + parse.urlencode({"apiKey": self.api_key}), headers={"Accept": "application/json"})

    def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote:
        symbol = (provider_symbol or "").strip().upper()
        if not symbol:
            return self._missing(symbol, "provider_symbol_missing")
        try:
            payload = self._massive_json(f"/v2/aggs/ticker/{parse.quote(symbol)}/prev")
        except RuntimeError as exc:
            return self._missing(symbol, str(exc))
        results = payload.get("results", []) if isinstance(payload, dict) else []
        row = results[0] if results and isinstance(results[0], dict) else {}
        price = self._decimal_price(row.get("c") or row.get("vw"))
        if price is None:
            return self._missing(symbol, "price_missing")
        return EquityPriceQuote(provider_symbol=symbol, currency="USD", close=price, provider=self.name, price_timestamp=utc_now(), quality_status="fresh")


class EodhdEquityPriceProvider(_HttpEquityPriceProvider):
    name = "eodhd"
    api_key_env = ("EODHD_API_KEY", "JARVIS_EODHD_API_KEY")

    def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote:
        symbol = (provider_symbol or "").strip().upper()
        if not symbol:
            return self._missing(symbol, "provider_symbol_missing")
        if not self.api_key:
            return self._missing(symbol, "eodhd_api_key_missing")
        eod_symbol = symbol if "." in symbol else f"{symbol}.US"
        try:
            payload = self._json_url("https://eodhd.com/api/real-time/" + parse.quote(eod_symbol) + "?" + parse.urlencode({"api_token": self.api_key, "fmt": "json"}))
        except RuntimeError as exc:
            return self._missing(symbol, str(exc))
        row = payload if isinstance(payload, dict) else {}
        price = self._decimal_price(row.get("close") or row.get("previousClose"))
        if price is None:
            return self._missing(symbol, "price_missing")
        return EquityPriceQuote(provider_symbol=symbol, currency=str(row.get("currency") or "USD").upper(), close=price, provider=self.name, price_timestamp=utc_now(), quality_status="fresh")


class YFinanceEquityPriceProvider:
    """Historical Yahoo adapter used only as an explicit, capability-checked fallback."""

    name = "yfinance"
    capability = ProviderCapability(True, True, True, "sequential_fixed_pacing")

    @staticmethod
    def _fast_info_value(info: object, key: str) -> str | None:
        try:
            getter = getattr(info, "get", None)
            value = getter(key) if callable(getter) else None
        except Exception:
            value = None
        return str(value).upper() if value not in {None, ""} else None

    def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote:
        symbol = (provider_symbol or "").strip().upper()
        if not symbol:
            return EquityPriceQuote(provider_symbol=symbol, currency="", close=None, provider=self.name, quality_status="missing", error_message="provider_symbol_missing")
        try:
            import yfinance as yf  # type: ignore[import-not-found]

            ticker = yf.Ticker(symbol)
            info = ticker.fast_info
            currency = self._fast_info_value(info, "currency") or ""
            market = self._fast_info_value(info, "exchange")
            if price_date and price_date != "latest":
                requested = date.fromisoformat(price_date)
                start = requested - timedelta(days=7)
                end = requested + timedelta(days=1)
                history = ticker.history(start=start.isoformat(), end=end.isoformat(), auto_adjust=False)
            else:
                requested = None
                history = ticker.history(period="7d", interval="1d", auto_adjust=False)
            if history is None or getattr(history, "empty", True):
                return EquityPriceQuote(provider_symbol=symbol, currency=currency, close=None, provider=self.name, provider_market=market, quality_status="missing", error_message="price_missing")
            eligible: list[tuple[date, Decimal]] = []
            for index, row in history.iterrows():
                try:
                    row_date = date.fromisoformat(str(index)[:10])
                    close = Decimal(str(row["Close"]))
                except (InvalidOperation, KeyError, TypeError, ValueError):
                    continue
                if close.is_nan() or close <= 0 or (requested is not None and row_date > requested):
                    continue
                eligible.append((row_date, close))
            if not eligible:
                return EquityPriceQuote(provider_symbol=symbol, currency=currency, close=None, provider=self.name, provider_market=market, quality_status="missing", error_message="price_missing")
            row_date, close = max(eligible, key=lambda item: item[0])
            timestamp = row_date.isoformat() if requested is not None else datetime.now(timezone.utc).isoformat()
            return EquityPriceQuote(provider_symbol=symbol, currency=currency, close=close, provider=self.name, provider_market=market, price_timestamp=timestamp, quality_status="fresh")
        except ValueError:
            return EquityPriceQuote(provider_symbol=symbol, currency="", close=None, provider=self.name, quality_status="missing", error_message="price_date_invalid")
        except Exception:
            return EquityPriceQuote(provider_symbol=symbol, currency="", close=None, provider=self.name, quality_status="missing", error_message="yfinance_provider_error")


class CompositeEquityPriceProvider:
    name = "auto"
    capability = ProviderCapability(True, True, True, "sequential_capability_filtered")

    def __init__(self, providers: list[EquityPriceProvider] | None = None) -> None:
        self.providers = providers or [FmpEquityPriceProvider(), YFinanceEquityPriceProvider(), TwelveDataEquityPriceProvider(), FinnhubEquityPriceProvider(), MassiveEquityPriceProvider()]

    def get_price(self, provider_symbol: str, *, price_date: str | None = None) -> EquityPriceQuote:
        last_quote: EquityPriceQuote | None = None
        for provider in self.providers:
            capability = getattr(provider, "capability", ProviderCapability(True, False, False, "unknown"))
            if price_date and price_date != "latest" and not capability.supports_historical_as_of:
                continue
            quote = provider.get_price(provider_symbol, price_date=price_date)
            if quote.close is not None and quote.quality_status == "fresh":
                return quote
            last_quote = quote
        return last_quote or EquityPriceQuote(provider_symbol=provider_symbol, currency="", close=None, provider=self.name, quality_status="missing", error_message="price_missing")


def equity_price_provider_by_name(name: str) -> EquityPriceProvider:
    key = (name or "").lower().replace("-", "")
    providers = {
        "auto": CompositeEquityPriceProvider,
        "fmp": FmpEquityPriceProvider,
        "finnhub": FinnhubEquityPriceProvider,
        "twelvedata": TwelveDataEquityPriceProvider,
        "massive": MassiveEquityPriceProvider,
        "eodhd": EodhdEquityPriceProvider,
        "yfinance": YFinanceEquityPriceProvider,
    }
    if key not in providers:
        raise ValueError("unknown equity price provider")
    return providers[key]()


def provider_capability(name: str) -> ProviderCapability:
    provider = equity_price_provider_by_name(name)
    return getattr(provider, "capability", ProviderCapability(True, False, False, "unknown"))


def exchange_matches(expected: str | None, actual: str | None) -> bool:
    expected_key = (expected or "").strip().upper()
    actual_key = (actual or "").strip().upper()
    if not expected_key or not actual_key:
        return False
    aliases = {
        "NASDAQ": {"NASDAQ", "NMS", "NGM", "NCM"},
        "NYSE": {"NYSE", "NYQ"},
        "CBOE": {"CBOE", "BTS"},
        "LSE": {"LSE", "LON"},
        "FSX": {"FSX", "FRA"},
        "SIX": {"SIX", "SWX", "XSWX"},
    }
    return actual_key in aliases.get(expected_key, {expected_key})


def _resolve_market_alerts(conn: Connection, *, instrument_id: str) -> int:
    return resolve_instrument_alerts(conn, instrument_id=instrument_id, rule_ids=["missing_market_price", "stale_market_price"])


def _previous_price(conn: Connection, *, instrument_id: str, price_date: str, provider: str):
    return conn.execute(
        """
        SELECT * FROM market_prices
        WHERE instrument_id=? AND provider=? AND price_date<? AND close IS NOT NULL AND close!=''
        ORDER BY price_date DESC, created_at DESC LIMIT 1
        """,
        (instrument_id, provider, price_date),
    ).fetchone()


def _corporate_action_status(conn: Connection, *, instrument_id: str, price_date: str, provider: str, close: Decimal | None) -> str:
    if close is None or close <= 0:
        return "not_checked"
    prev = _previous_price(conn, instrument_id=instrument_id, price_date=price_date, provider=provider)
    if not prev or not prev["close"]:
        return "not_checked"
    old = Decimal(str(prev["close"]))
    if old <= 0:
        return "not_checked"
    change = abs((close - old) / old)
    if change > Decimal("0.25"):
        create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="corporate_action_suspected", message="Large local price move detected; corporate action review required before high-confidence valuation.", evidence={"previous_price_date": prev["price_date"], "price_date": price_date}, fingerprint="corporate_action_suspected")
        create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="split_or_corporate_action_review_required", message="Split/corporate-action review required; no automatic split correction is applied.", evidence={"previous_price_date": prev["price_date"], "price_date": price_date}, fingerprint="split_or_corporate_action_review_required")
        update_instrument_metadata(conn, instrument_id=instrument_id, corporate_action_status="suspected", split_or_corporate_action_review_required=True, note="Automatic C2 heuristic detected >25% local price move; review required.")
        return "suspected"
    return "none_known"


def store_market_price(
    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
    existing = conn.execute(
        "SELECT market_price_id, close, currency, quality_status FROM market_prices WHERE instrument_id=? AND price_date=? AND provider=?",
        (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)
    conn.execute(
        """
        INSERT INTO market_prices(
            market_price_id, instrument_id, price_date, price_timestamp, close, adjusted_close,
            currency, provider, provider_symbol, quality_status, created_at, provider_market, error_message,
            corporate_action_status, fetched_at, price_type, run_id
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(instrument_id, price_date, provider) DO UPDATE SET
            price_timestamp=excluded.price_timestamp,
            close=excluded.close,
            adjusted_close=excluded.adjusted_close,
            currency=excluded.currency,
            provider_symbol=excluded.provider_symbol,
            quality_status=excluded.quality_status,
            created_at=excluded.created_at,
            provider_market=excluded.provider_market,
            error_message=excluded.error_message,
            corporate_action_status=excluded.corporate_action_status,
            fetched_at=excluded.fetched_at,
            price_type=excluded.price_type,
            run_id=excluded.run_id
        """,
        (market_price_id, instrument_id, price_date, price_timestamp, format(close, "f") if close is not None else "", format(adjusted_close, "f") if adjusted_close is not None else None, currency.upper(), provider, provider_symbol, quality_status, now, provider_market, error_message, corp_status, fetched, price_type, run_id),
    )
    if existing and (
        str(existing["close"] or "") != (format(close, "f") if close is not None else "")
        or str(existing["currency"] or "") != currency.upper()
    ):
        record_audit_event(
            conn,
            source="daily_market_fx_v1",
            action="market_price_provider_correction",
            entity_type="market_price",
            entity_id=str(existing["market_price_id"]),
            old_values={"close": existing["close"], "currency": existing["currency"], "quality_status": existing["quality_status"]},
            new_values={"close": format(close, "f") if close is not None else None, "currency": currency.upper(), "quality_status": quality_status, "run_id": run_id},
            confirmed=True,
            created_by="system",
        )
    if quality_status in {"missing", "stale", "error", "conflict"} or close is None:
        rule = "stale_market_price" if quality_status == "stale" else "missing_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


def refresh_market_prices(
    conn: Connection,
    *,
    provider: EquityPriceProvider,
    asset_class: str,
    price_date: str | None = None,
    only_missing: bool = False,
    only_stale: bool = False,
    only_isin: str | None = None,
    limit: int | None = None,
    dry_run: bool = False,
) -> MarketPriceRefreshResult:
    asset = asset_class.lower()
    result = MarketPriceRefreshResult(asset_class=asset, dry_run=dry_run)
    query = """
        SELECT m.*, i.asset_class, i.isin, i.instrument_status AS current_instrument_status, i.valuation_policy AS current_valuation_policy FROM instrument_price_mappings m
        JOIN instruments i ON i.instrument_id=m.instrument_id
        WHERE LOWER(i.asset_class)=? AND m.mapping_status='mapped' AND m.provider_symbol IS NOT NULL
        ORDER BY i.isin, m.provider_symbol
    """
    mappings = conn.execute(query, (asset,)).fetchall()
    if only_isin:
        mappings = [m for m in mappings if (m["isin"] or "").upper() == only_isin.upper()]
    if limit is not None:
        mappings = mappings[:limit]
    result.total_mappings = len(mappings)
    effective_date = price_date or utc_now()[:10]
    for mapping in mappings:
        if mapping["current_instrument_status"] in {"delisted", "suspended", "merged", "inactive"} or mapping["current_valuation_policy"] == "exclude_from_auto_price_update":
            result.excluded_count += 1
            result.skipped_count += 1
            if not dry_run:
                create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=mapping["instrument_id"], rule_id="delisted_or_suspended", message="Instrument excluded from automatic market-price update.", evidence={"instrument_status": mapping["current_instrument_status"], "valuation_policy": mapping["current_valuation_policy"]}, fingerprint=f"excluded:{mapping['current_instrument_status']}:{mapping['current_valuation_policy']}")
            continue
        latest = conn.execute("SELECT * FROM market_prices WHERE instrument_id=? AND quality_status='fresh' ORDER BY price_date DESC, created_at DESC LIMIT 1", (mapping["instrument_id"],)).fetchone()
        if only_missing and latest:
            result.cached_count += 1
            continue
        if only_stale and latest and latest["price_date"] >= effective_date:
            result.cached_count += 1
            continue
        quote = provider.get_price(mapping["provider_symbol"], price_date=effective_date)
        status = quote.quality_status
        if not dry_run:
            store_market_price(conn, instrument_id=mapping["instrument_id"], price_date=effective_date, close=quote.close, currency=quote.currency or mapping["currency"], provider=quote.provider or provider.name, provider_symbol=mapping["provider_symbol"], provider_market=mapping["provider_market"], price_timestamp=quote.price_timestamp, adjusted_close=quote.adjusted_close, quality_status=status, error_message=quote.error_message)
        if quote.close is not None and status == "fresh":
            result.updated_count += 1
        elif status in {"missing", "stale"}:
            result.warning_count += 1
            result.warnings.append(mapping["instrument_id"])
            if status == "stale":
                result.stale_count += 1
        else:
            result.error_count += 1
            result.errors.append(mapping["instrument_id"])
    if not dry_run:
        conn.commit()
    return result