from __future__ import annotations

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

from jarvis_finance.imports.common import stable_id, utc_now
from jarvis_finance.quality.alerts import create_alert
from jarvis_finance.quality.data_quality import resolve_fixed_crypto_price_alerts


@dataclass(frozen=True)
class PriceQuote:
    coingecko_id: str
    currency: str
    price: Decimal | None
    provider: str = "CoinGecko"
    provider_timestamp: str | None = None
    quality_status: str = "fresh"
    error_message: str | None = None
    change_24h_pct: Decimal | None = None
    high_24h: Decimal | None = None
    low_24h: Decimal | None = None
    change_7d_pct: Decimal | None = None
    change_30d_pct: Decimal | None = None
    sparkline_7d: tuple[Decimal, ...] = ()
    market_cap_rank: int | None = None
    market_cap: Decimal | None = None
    volume_24h: Decimal | None = None


@dataclass
class PriceRefreshResult:
    currency: str
    total_assets: int = 0
    updated_count: int = 0
    skipped_count: int = 0
    cached_count: int = 0
    stale_count: int = 0
    warning_count: int = 0
    error_count: int = 0
    missing_local_price_count: int = 0
    dry_run: bool = False
    written_price_ids: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)

    @property
    def success_count(self) -> int:  # backward-compatible name used by older tests/callers
        return self.updated_count


class MarketDataProvider(Protocol):
    def get_crypto_price(self, coingecko_id: str, currency: str = "CHF") -> PriceQuote: ...


class BatchMarketDataProvider(MarketDataProvider, Protocol):
    def get_crypto_prices(self, coingecko_ids: Sequence[str], currency: str = "CHF") -> dict[str, PriceQuote]: ...


class ProviderRateLimitError(RuntimeError):
    pass


class ProviderFetchError(RuntimeError):
    pass


class CoinGeckoClient:
    provider_key = "coingecko"

    def __init__(self, *, base_url: str = "https://api.coingecko.com/api/v3", max_retries: int = 2, initial_backoff_seconds: float = 1.0, max_backoff_seconds: float = 8.0, opener=None, sleeper=time.sleep) -> None:
        self.base_url = base_url.rstrip("/")
        self.max_retries = max_retries
        self.initial_backoff_seconds = initial_backoff_seconds
        self.max_backoff_seconds = max_backoff_seconds
        self.opener = opener or urllib.request.urlopen
        self.sleeper = sleeper

    def get_crypto_price(self, coingecko_id: str, currency: str = "CHF") -> PriceQuote:
        return self.get_crypto_prices([coingecko_id], currency).get(
            coingecko_id,
            PriceQuote(coingecko_id, currency.upper(), None, quality_status="missing", error_message="price missing"),
        )

    def get_crypto_prices(self, coingecko_ids: Sequence[str], currency: str = "CHF") -> dict[str, PriceQuote]:
        ids = [cid for cid in dict.fromkeys(coingecko_ids) if cid]
        cur = currency.lower()
        if not ids:
            return {}
        query = urllib.parse.urlencode({"ids": ",".join(ids), "vs_currencies": cur, "include_last_updated_at": "true"})
        url = f"{self.base_url}/simple/price?{query}"
        delay = self.initial_backoff_seconds
        last_error: str | None = None
        for attempt in range(self.max_retries + 1):
            try:
                with self.opener(url, timeout=20) as resp:
                    payload = json.loads(resp.read().decode("utf-8"))
                return {coingecko_id: self._quote_from_payload(coingecko_id, currency, payload.get(coingecko_id) or {}) for coingecko_id in ids}
            except urllib.error.HTTPError as exc:
                last_error = f"HTTP {exc.code}"
                if exc.code == 429 and attempt < self.max_retries:
                    self.sleeper(min(delay, self.max_backoff_seconds))
                    delay *= 2
                    continue
                status = "stale" if exc.code == 429 else "error"
                return {cid: PriceQuote(cid, currency.upper(), None, quality_status=status, error_message=last_error) for cid in ids}
            except Exception as exc:  # network skeleton must not crash dashboard jobs
                last_error = str(exc)
                if attempt < self.max_retries:
                    self.sleeper(min(delay, self.max_backoff_seconds))
                    delay *= 2
                    continue
                return {cid: PriceQuote(cid, currency.upper(), None, quality_status="error", error_message=last_error) for cid in ids}
        return {cid: PriceQuote(cid, currency.upper(), None, quality_status="error", error_message=last_error) for cid in ids}

    def get_crypto_market_bundle(
        self,
        coingecko_ids: Sequence[str],
    ) -> dict[str, dict[str, PriceQuote]]:
        """Load canonical CHF valuation and USD/24h details in one refresh cycle."""
        ids = [cid for cid in dict.fromkeys(coingecko_ids) if cid]
        if not ids:
            return {}
        chf = self.get_crypto_prices(ids, "CHF")
        query = urllib.parse.urlencode(
            {
                "vs_currency": "usd",
                "ids": ",".join(ids),
                "order": "market_cap_desc",
                "per_page": str(min(len(ids), 250)),
                "page": "1",
                "sparkline": "true",
                "price_change_percentage": "24h,7d,30d",
            }
        )
        url = f"{self.base_url}/coins/markets?{query}"
        delay = self.initial_backoff_seconds
        last_error: str | None = None
        usd: dict[str, PriceQuote] = {}
        for attempt in range(self.max_retries + 1):
            try:
                with self.opener(url, timeout=20) as resp:
                    payload = json.loads(resp.read().decode("utf-8"))
                rows = payload if isinstance(payload, list) else []
                usd = {
                    str(row.get("id")): self._market_quote_from_payload(str(row.get("id")), row)
                    for row in rows
                    if isinstance(row, dict) and row.get("id")
                }
                break
            except urllib.error.HTTPError as exc:
                last_error = f"HTTP {exc.code}"
                if exc.code == 429 and attempt < self.max_retries:
                    self.sleeper(min(delay, self.max_backoff_seconds))
                    delay *= 2
                    continue
                break
            except Exception as exc:  # provider boundary; keep per-coin fallback below
                last_error = str(exc)
                if attempt < self.max_retries:
                    self.sleeper(min(delay, self.max_backoff_seconds))
                    delay *= 2
                    continue
                break
        return {
            cid: {
                "CHF": chf.get(cid, PriceQuote(cid, "CHF", None, quality_status="missing", error_message="price missing")),
                "USD": usd.get(cid, PriceQuote(cid, "USD", None, quality_status="error" if last_error else "missing", error_message=last_error or "price missing")),
            }
            for cid in ids
        }

    @staticmethod
    def _quote_from_payload(coingecko_id: str, currency: str, data: dict) -> PriceQuote:
        cur = currency.lower()
        value = data.get(cur)
        if value is None:
            return PriceQuote(coingecko_id, currency.upper(), None, quality_status="missing", error_message="price missing")
        try:
            price = Decimal(str(value))
        except InvalidOperation:
            return PriceQuote(coingecko_id, currency.upper(), None, quality_status="error", error_message="invalid price")
        ts = data.get("last_updated_at")
        provider_ts = None
        if ts:
            try:
                provider_ts = datetime.fromtimestamp(int(ts), tz=timezone.utc).isoformat()
            except (TypeError, ValueError):
                provider_ts = str(ts)
        return PriceQuote(coingecko_id, currency.upper(), price, provider_timestamp=provider_ts)

    @staticmethod
    def _market_quote_from_payload(coingecko_id: str, data: dict) -> PriceQuote:
        try:
            price = Decimal(str(data["current_price"]))
        except (KeyError, InvalidOperation, TypeError):
            return PriceQuote(coingecko_id, "USD", None, quality_status="missing", error_message="price missing")
        values: dict[str, Decimal | None] = {}
        for key in (
            "price_change_percentage_24h",
            "price_change_percentage_7d_in_currency",
            "price_change_percentage_30d_in_currency",
            "high_24h",
            "low_24h",
            "market_cap",
            "total_volume",
        ):
            try:
                values[key] = Decimal(str(data[key])) if data.get(key) is not None else None
            except InvalidOperation:
                values[key] = None
        sparkline: list[Decimal] = []
        raw_sparkline = data.get("sparkline_in_7d")
        if isinstance(raw_sparkline, dict) and isinstance(raw_sparkline.get("price"), list):
            for value in raw_sparkline["price"]:
                try:
                    sparkline.append(Decimal(str(value)))
                except InvalidOperation:
                    continue
        rank = data.get("market_cap_rank")
        return PriceQuote(
            coingecko_id,
            "USD",
            price,
            provider_timestamp=str(data.get("last_updated") or "") or None,
            change_24h_pct=values["price_change_percentage_24h"],
            high_24h=values["high_24h"],
            low_24h=values["low_24h"],
            change_7d_pct=values["price_change_percentage_7d_in_currency"],
            change_30d_pct=values["price_change_percentage_30d_in_currency"],
            sparkline_7d=tuple(sparkline),
            market_cap_rank=int(rank) if isinstance(rank, (int, float)) else None,
            market_cap=values["market_cap"],
            volume_24h=values["total_volume"],
        )


def _parse_dt(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None
    return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)


def is_price_stale(timestamp: str | None, *, max_age_seconds: int = 86_400) -> bool:
    dt = _parse_dt(timestamp)
    if dt is None:
        return True
    return (datetime.now(timezone.utc) - dt).total_seconds() > max_age_seconds


def latest_crypto_price_row(conn: Connection, asset_id: str, currency: str = "CHF"):
    return conn.execute(
        """
        SELECT * FROM crypto_prices
        WHERE asset_id=? AND price_currency=?
        ORDER BY COALESCE(provider_timestamp, fetched_at, '') DESC, fetched_at DESC
        LIMIT 1
        """,
        (asset_id, currency.upper()),
    ).fetchone()


def store_crypto_price(conn: Connection, *, asset_id: str, quote: PriceQuote) -> str:
    now = utc_now()
    price_text = format(quote.price, "f") if quote.price is not None else ""
    existing = conn.execute(
        """SELECT crypto_price_id FROM crypto_prices
             WHERE asset_id=? AND coingecko_id=? AND price_currency=? AND provider=?
               AND provider_timestamp IS ? AND price=? AND quality_status=?
               AND COALESCE(error_message,'')=COALESCE(?,'')
             ORDER BY fetched_at DESC LIMIT 1""",
        (
            asset_id,
            quote.coingecko_id,
            quote.currency.upper(),
            quote.provider,
            quote.provider_timestamp,
            price_text,
            quote.quality_status,
            quote.error_message,
        ),
    ).fetchone()
    if existing:
        # Retrieval freshness is mutable request context; economic payload remains one row.
        conn.execute(
            "UPDATE crypto_prices SET fetched_at=? WHERE crypto_price_id=?",
            (now, existing["crypto_price_id"]),
        )
        if quote.quality_status in {"missing", "stale", "error", "conflict"}:
            create_alert(
                conn,
                priority="warnung",
                category="market_data",
                entity_type="crypto_asset",
                entity_id=asset_id,
                rule_id=f"crypto_price_{quote.quality_status}",
                message=f"Crypto price quality is {quote.quality_status}.",
                evidence={
                    "coingecko_id": quote.coingecko_id,
                    "currency": quote.currency,
                    "error": quote.error_message,
                },
            )
        conn.commit()
        return str(existing["crypto_price_id"])
    price_id = stable_id("cryptoprice", asset_id, quote.coingecko_id, quote.currency, quote.provider, now)
    conn.execute(
        """
        INSERT INTO crypto_prices(crypto_price_id, asset_id, coingecko_id, price_currency, price, provider, provider_timestamp, fetched_at, quality_status, error_message)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (price_id, asset_id, quote.coingecko_id, quote.currency.upper(), price_text, quote.provider, quote.provider_timestamp, now, quote.quality_status, quote.error_message),
    )
    if quote.quality_status in {"missing", "stale", "error", "conflict"}:
        create_alert(conn, priority="warnung", category="market_data", entity_type="crypto_asset", entity_id=asset_id, rule_id=f"crypto_price_{quote.quality_status}", message=f"Crypto price quality is {quote.quality_status}.", evidence={"coingecko_id": quote.coingecko_id, "currency": quote.currency, "error": quote.error_message})
    conn.commit()
    return price_id


def _create_missing_local_price_alert(conn: Connection, *, asset_id: str, symbol: str | None, coingecko_id: str | None, currency: str) -> None:
    create_alert(
        conn,
        priority="warnung",
        category="market_data",
        entity_type="crypto_asset",
        entity_id=asset_id,
        rule_id="crypto_price_missing_local",
        message="Crypto asset has no fresh local price after refresh.",
        evidence={"symbol": symbol, "coingecko_id": coingecko_id, "currency": currency},
        fingerprint=f"crypto_price_missing_local:{currency}",
    )


def _has_fresh_local_price(conn: Connection, asset_id: str, currency: str, max_age_seconds: int) -> bool:
    latest = latest_crypto_price_row(conn, asset_id, currency)
    return bool(latest and latest["quality_status"] == "fresh" and not is_price_stale(latest["fetched_at"], max_age_seconds=max_age_seconds))


def _filter_assets(
    conn: Connection,
    *,
    currency: str,
    max_age_seconds: int,
    only_missing: bool,
    only_stale: bool,
    only_symbol: str | None,
    limit: int | None,
    asset_ids: Sequence[str] | None,
):
    assets = conn.execute("SELECT asset_id, coingecko_id, symbol FROM crypto_assets WHERE is_active=1 ORDER BY symbol").fetchall()
    if asset_ids is not None:
        wanted_ids = {str(asset_id) for asset_id in asset_ids}
        assets = [asset for asset in assets if str(asset["asset_id"]) in wanted_ids]
    filtered = []
    wanted_symbol = only_symbol.upper() if only_symbol else None
    for asset in assets:
        if wanted_symbol and (asset["symbol"] or "").upper() != wanted_symbol:
            continue
        latest = latest_crypto_price_row(conn, asset["asset_id"], currency)
        has_any = latest is not None and latest["price"] not in (None, "") and latest["quality_status"] == "fresh"
        is_stale = bool(latest and latest["quality_status"] == "fresh" and is_price_stale(latest["fetched_at"], max_age_seconds=max_age_seconds))
        if only_missing and has_any:
            continue
        if only_stale and not is_stale:
            continue
        filtered.append(asset)
        if limit is not None and len(filtered) >= limit:
            break
    return assets, filtered


def _provider_get_batch(provider: MarketDataProvider, coingecko_ids: Sequence[str], currency: str) -> dict[str, PriceQuote]:
    if hasattr(provider, "get_crypto_prices"):
        return provider.get_crypto_prices(coingecko_ids, currency)  # type: ignore[attr-defined]
    return {cid: provider.get_crypto_price(cid, currency) for cid in coingecko_ids}


def refresh_crypto_prices(
    conn: Connection,
    *,
    provider: MarketDataProvider,
    currency: str = "CHF",
    max_age_seconds: int = 3600,
    only_missing: bool = False,
    only_stale: bool = False,
    only_symbol: str | None = None,
    limit: int | None = None,
    dry_run: bool = False,
    sleep_seconds: float = 0.0,
    batch_size: int = 100,
    asset_ids: Sequence[str] | None = None,
) -> PriceRefreshResult:
    currency = currency.upper()
    result = PriceRefreshResult(currency=currency, dry_run=dry_run)
    all_assets, assets = _filter_assets(
        conn,
        currency=currency,
        max_age_seconds=max_age_seconds,
        only_missing=only_missing,
        only_stale=only_stale,
        only_symbol=only_symbol,
        limit=limit,
        asset_ids=asset_ids,
    )
    result.total_assets = len(all_assets)
    if only_missing or only_stale or only_symbol or limit is not None:
        result.skipped_count += max(0, len(all_assets) - len(assets))

    request_assets = []
    for asset in assets:
        asset_id = asset["asset_id"]
        if not asset["coingecko_id"]:
            if not dry_run:
                create_alert(conn, priority="warnung", category="crypto", entity_type="crypto_asset", entity_id=asset_id, rule_id="missing_coingecko_id", message="Crypto asset has no CoinGecko ID; price refresh skipped.", evidence={"symbol": asset["symbol"]}, fingerprint="missing_coingecko_id")
                _create_missing_local_price_alert(conn, asset_id=asset_id, symbol=asset["symbol"], coingecko_id=None, currency=currency)
            result.skipped_count += 1
            result.warning_count += 1
            result.warnings.append(f"{asset['symbol']}: missing_coingecko_id")
            continue
        latest = latest_crypto_price_row(conn, asset_id, currency)
        if latest and latest["quality_status"] == "fresh" and not is_price_stale(latest["fetched_at"], max_age_seconds=max_age_seconds):
            result.cached_count += 1
            continue
        request_assets.append(asset)

    refreshed_fresh_asset_ids: set[str] = set()
    for start in range(0, len(request_assets), max(1, batch_size)):
        if start and sleep_seconds > 0:
            time.sleep(sleep_seconds)
        chunk = request_assets[start : start + max(1, batch_size)]
        ids = [asset["coingecko_id"] for asset in chunk]
        try:
            quotes = _provider_get_batch(provider, ids, currency)
        except Exception as exc:
            quotes = {cid: PriceQuote(cid, currency, None, quality_status="error", error_message=str(exc)) for cid in ids}
        for asset in chunk:
            quote = quotes.get(asset["coingecko_id"]) or PriceQuote(asset["coingecko_id"], currency, None, quality_status="missing", error_message="price missing")
            observation_count_before = conn.execute(
                "SELECT COUNT(*) FROM crypto_prices WHERE asset_id=? AND price_currency=?",
                (asset["asset_id"], currency),
            ).fetchone()[0]
            if not dry_run:
                price_id = store_crypto_price(conn, asset_id=asset["asset_id"], quote=quote)
                result.written_price_ids.append(price_id)
            observation_created = dry_run or conn.execute(
                "SELECT COUNT(*) FROM crypto_prices WHERE asset_id=? AND price_currency=?",
                (asset["asset_id"], currency),
            ).fetchone()[0] > observation_count_before
            if quote.price is not None and quote.quality_status == "fresh":
                if observation_created:
                    result.updated_count += 1
                else:
                    result.cached_count += 1
                refreshed_fresh_asset_ids.add(asset["asset_id"])
                if not dry_run:
                    resolve_fixed_crypto_price_alerts(conn, currency=currency, max_age_seconds=max_age_seconds)
            elif quote.quality_status in {"missing", "stale"}:
                result.warning_count += 1
                result.warnings.append(f"{asset['symbol']}: {quote.quality_status}")
            else:
                result.error_count += 1
                result.errors.append(f"{asset['symbol']}: {quote.error_message or quote.quality_status}")
            if quote.quality_status == "stale":
                result.stale_count += 1
            if (quote.price is None or quote.quality_status != "fresh") and not dry_run:
                _create_missing_local_price_alert(conn, asset_id=asset["asset_id"], symbol=asset["symbol"], coingecko_id=asset["coingecko_id"], currency=currency)

    if not dry_run:
        for asset in assets:
            if asset["asset_id"] in refreshed_fresh_asset_ids:
                continue
            if not _has_fresh_local_price(conn, asset["asset_id"], currency, max_age_seconds):
                result.missing_local_price_count += 1
                _create_missing_local_price_alert(conn, asset_id=asset["asset_id"], symbol=asset["symbol"], coingecko_id=asset["coingecko_id"], currency=currency)
        conn.commit()
    else:
        # Dry-run reports current known gaps plus simulated failed quotes without mutating alerts/prices.
        for asset in assets:
            if not _has_fresh_local_price(conn, asset["asset_id"], currency, max_age_seconds):
                result.missing_local_price_count += 1
    return result