from __future__ import annotations

from dataclasses import replace
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
import json
import time
from sqlite3 import Connection
from typing import Any
from urllib import error, parse, request

from fastapi import HTTPException

from jarvis_finance.api.schemas.market import ChartPoint, EquityCandlesResponse, MarketBatchUpdateResponse, MarketChartResponse, MarketQuoteResponse, MarketStatusResponse, QuoteRefreshRequest
from jarvis_finance.imports.common import utc_now
from jarvis_finance.market.providers import CoinGeckoClient, PriceQuote, store_crypto_price
from jarvis_finance.market_data.cache import get_crypto_chart_points, get_equity_chart_points, get_equity_intraday_candles, upsert_crypto_price_point, upsert_equity_intraday_candles, upsert_equity_price_point
from jarvis_finance.market_data.prices import EquityPriceQuote, equity_price_provider_by_name, exchange_matches, provider_capability, store_market_price


def _decimal_text(value: object | None) -> str | None:
    if value in (None, ""):
        return None
    try:
        return format(Decimal(str(value)), "f")
    except (InvalidOperation, ValueError):
        return None


def _quality_from_error(message: str | None, default: str = "missing") -> str:
    msg = (message or "").lower()
    if "rate_limited" in msg or "429" in msg:
        return "rate_limited"
    if "endpoint_restricted" in msg or "plan_restricted" in msg or "402" in msg:
        return "plan_restricted"
    if "auth_failed" in msg or "auth_error" in msg or "401" in msg or "403" in msg:
        return "auth_error"
    if "network_error" in msg:
        return "network_error"
    if "api_key_missing" in msg or "provider_symbol_missing" in msg or "missing" in msg:
        return "missing"
    if "unsupported" in msg:
        return "unsupported_pair"
    if "stale" in msg:
        return "stale"
    return default


def _chart_response(points, *, currency: str, provider_symbol: str | None = None, warnings: list[str] | None = None) -> MarketChartResponse:
    chart_points = [ChartPoint(timestamp=str(r["timestamp"]), price=str(r["price"]), currency=str(r["currency"] or currency), provider=r["provider"], quality_status=r["source_quality"]) for r in points]
    latest = chart_points[-1] if chart_points else None
    first = chart_points[0] if chart_points else None
    change_abs = None
    change_pct = None
    if latest and first:
        try:
            start = Decimal(first.price)
            end = Decimal(latest.price)
            change_abs = format(end - start, "f")
            change_pct = format(((end - start) / start * Decimal("100")), "f") if start else None
        except Exception:
            pass
    return MarketChartResponse(
        latest_price=latest.price if latest else None,
        currency=latest.currency if latest else currency,
        change_abs=change_abs,
        change_pct=change_pct,
        close=latest.price if latest else None,
        provider=latest.provider if latest else None,
        provider_symbol=provider_symbol,
        fetched_at=latest.timestamp if latest else None,
        quality_status=latest.quality_status if latest and latest.quality_status else ("fresh" if latest else "missing"),
        chart_points=chart_points,
        warnings=warnings or ([] if chart_points else ["Noch zu wenig Kursdaten"]),
    )


def get_market_status(conn: Connection) -> MarketStatusResponse:
    def scalar(sql: str, params: tuple = ()):
        row = conn.execute(sql, params).fetchone()
        return row[0] if row else None
    return MarketStatusResponse(
        equity_latest_update=scalar("SELECT MAX(fetched_at) FROM equity_price_points"),
        crypto_latest_update=scalar("SELECT MAX(fetched_at) FROM crypto_price_points"),
        equity_cached_points=int(scalar("SELECT COUNT(*) FROM equity_price_points") or 0),
        crypto_cached_points=int(scalar("SELECT COUNT(*) FROM crypto_price_points") or 0),
        mapped_equity_instruments=int(scalar("SELECT COUNT(DISTINCT instrument_id) FROM instrument_price_mappings WHERE mapping_status='mapped' AND provider_symbol IS NOT NULL") or 0),
        mapped_crypto_assets=int(scalar("SELECT COUNT(*) FROM crypto_assets WHERE is_active=1 AND coingecko_id IS NOT NULL AND coingecko_id!=''") or 0),
        render_provider_calls=False,
        warnings=[],
    )


def _instrument_mapping(conn: Connection, instrument_id: str):
    inst = conn.execute("SELECT instrument_id, provider_symbol, data_provider_primary, exchange, currency, is_active, instrument_status, valuation_policy FROM instruments WHERE instrument_id=?", (instrument_id,)).fetchone()
    if not inst:
        raise HTTPException(status_code=404, detail="Instrument not found")
    if not bool(inst["is_active"]) or str(inst["instrument_status"] or "active").lower() in {"inactive", "delisted", "suspended", "merged"}:
        return inst, None, ["Instrument ist nicht für automatische Kursaktualisierung aktiv"]
    if str(inst["valuation_policy"] or "").lower() == "exclude_from_auto_price_update":
        return inst, None, ["Instrument ist durch die Bewertungsrichtlinie von automatischen Kursaktualisierungen ausgeschlossen"]
    mapping = conn.execute("SELECT * FROM instrument_price_mappings WHERE instrument_id=? AND mapping_status='mapped' AND provider_symbol IS NOT NULL ORDER BY CASE provider WHEN 'fmp' THEN 1 WHEN 'finnhub' THEN 2 WHEN 'twelvedata' THEN 3 ELSE 4 END LIMIT 1", (instrument_id,)).fetchone()
    provider_symbol = (mapping["provider_symbol"] if mapping else inst["provider_symbol"]) if inst else None
    if not provider_symbol:
        return inst, None, ["Provider-Symbol fehlt"]
    return inst, mapping, []


def _effective_market_date(value: str | None = None) -> date:
    result = date.fromisoformat(value) if value else datetime.now(timezone.utc).date()
    while result.weekday() >= 5:
        result -= timedelta(days=1)
    return result


def _business_day_age(earlier: date, later: date) -> int:
    if earlier > later:
        return -1
    cursor = earlier
    age = 0
    while cursor < later:
        cursor += timedelta(days=1)
        if cursor.weekday() < 5:
            age += 1
    return age


def _has_fresh_price_for_target(conn: Connection, instrument_id: str, target: date) -> bool:
    mapping = conn.execute(
        """SELECT provider,provider_symbol,provider_market,upper(COALESCE(trading_currency,currency,'')) currency
             FROM instrument_price_mappings
            WHERE instrument_id=? AND mapping_status='mapped' AND provider_symbol IS NOT NULL
            ORDER BY CASE provider WHEN 'fmp' THEN 1 ELSE 2 END,updated_at DESC LIMIT 1""",
        (instrument_id,),
    ).fetchone()
    if not mapping:
        return False
    rows = conn.execute(
        """SELECT price_date,currency,provider,provider_symbol,provider_market FROM market_prices
             WHERE instrument_id=? AND price_date<=? AND close IS NOT NULL AND close!=''
               AND quality_status='fresh' AND error_message IS NULL
             ORDER BY price_date DESC,COALESCE(fetched_at,created_at) DESC""",
        (instrument_id, target.isoformat()),
    ).fetchall()
    for row in rows:
        actual_date = date.fromisoformat(str(row["price_date"])[:10])
        if not 0 <= _business_day_age(actual_date, target) <= 2:
            continue
        if str(row["provider_symbol"] or "").upper() != str(mapping["provider_symbol"] or "").upper():
            continue
        if str(row["currency"] or "").upper() != str(mapping["currency"] or "").upper():
            continue
        if not exchange_matches(mapping["provider_market"], row["provider_market"]):
            continue
        if str(row["provider"] or "").lower() != str(mapping["provider"] or "").lower():
            if str(row["provider"] or "").lower() != "yfinance":
                continue
        return True
    return False


def _quote_date(quote: EquityPriceQuote, fallback: date) -> date:
    if quote.price_timestamp:
        try:
            return datetime.fromisoformat(quote.price_timestamp.replace("Z", "+00:00")).date()
        except ValueError:
            try:
                return date.fromisoformat(quote.price_timestamp[:10])
            except ValueError:
                pass
    return fallback


def _validate_historical_quote(quote: EquityPriceQuote, *, mapping: Any, target: date, requested_provider: str) -> EquityPriceQuote:
    if quote.close is None or quote.close <= 0:
        return replace(quote, quality_status=_quality_from_error(quote.error_message, quote.quality_status))
    quote_date = _quote_date(quote, target)
    if quote_date > target:
        return replace(quote, close=None, quality_status="future_price_rejected", error_message="future_price_rejected")
    if _business_day_age(quote_date, target) > 2:
        return replace(quote, close=None, quality_status="stale", error_message="historical_price_too_old")
    expected_currency = str(mapping["trading_currency"] or mapping["currency"] or "").upper()
    actual_currency = str(quote.currency or "").upper()
    is_fallback = requested_provider == "auto" and str(quote.provider or "").lower() == "yfinance"
    if is_fallback:
        capability = provider_capability(str(quote.provider))
        if not capability.supports_historical_as_of:
            return replace(quote, close=None, quality_status="provider_not_historical", error_message="provider_not_historical")
        if not actual_currency or (expected_currency and actual_currency != expected_currency):
            return replace(quote, close=None, quality_status="currency_mismatch", error_message="currency_mismatch")
        if str(quote.provider_symbol or "").upper() != str(mapping["provider_symbol"] or "").upper():
            return replace(quote, close=None, quality_status="symbol_mismatch", error_message="symbol_mismatch")
        if not exchange_matches(str(mapping["provider_market"] or ""), quote.provider_market):
            return replace(quote, close=None, quality_status="exchange_mismatch", error_message="exchange_mismatch")
    elif actual_currency and expected_currency and actual_currency != expected_currency:
        return replace(quote, close=None, quality_status="currency_mismatch", error_message="currency_mismatch")
    return replace(quote, currency=actual_currency or expected_currency, price_timestamp=quote_date.isoformat(), quality_status="fresh")


def refresh_equity_quote(conn: Connection, instrument_id: str, req: QuoteRefreshRequest) -> MarketQuoteResponse:
    inst, mapping, warnings = _instrument_mapping(conn, instrument_id)
    if warnings:
        return MarketQuoteResponse(provider_symbol=None, currency=inst["currency"] if inst else None, quality_status="missing", warnings=warnings)
    mapping_data = dict(mapping) if mapping else {
        "provider": inst["data_provider_primary"] or "auto",
        "provider_symbol": inst["provider_symbol"],
        "provider_market": inst["exchange"],
        "trading_currency": inst["currency"],
        "currency": inst["currency"],
    }
    provider_symbol = str(mapping_data["provider_symbol"])
    provider_name = (req.provider or "auto").lower()
    target = _effective_market_date(req.price_date)
    quote = equity_price_provider_by_name(provider_name).get_price(provider_symbol, price_date=target.isoformat())
    quote = _validate_historical_quote(quote, mapping=mapping_data, target=target, requested_provider=provider_name)
    quality = quote.quality_status if quote.close is not None else _quality_from_error(quote.error_message, quote.quality_status)
    ts = quote.price_timestamp or target.isoformat()
    if not req.dry_run and quote.close is not None and quality == "fresh":
        store_market_price(
            conn,
            instrument_id=instrument_id,
            price_date=ts[:10],
            close=quote.close,
            currency=quote.currency or inst["currency"] or "CHF",
            provider=quote.provider,
            provider_symbol=quote.provider_symbol or provider_symbol,
            provider_market=quote.provider_market or mapping_data["provider_market"],
            price_timestamp=ts,
            adjusted_close=quote.adjusted_close,
            quality_status=quality,
            error_message=None,
        )
        upsert_equity_price_point(
            conn,
            instrument_id=instrument_id,
            timestamp=ts,
            price=quote.close,
            currency=quote.currency or inst["currency"] or "CHF",
            provider=quote.provider,
            provider_symbol=quote.provider_symbol or provider_symbol,
            interval=req.interval,
            source_quality=quality,
        )
        conn.commit()
    return MarketQuoteResponse(
        latest_price=_decimal_text(quote.close),
        currency=quote.currency or inst["currency"],
        close=_decimal_text(quote.close),
        provider=quote.provider,
        provider_symbol=quote.provider_symbol or provider_symbol,
        fetched_at=ts,
        quality_status=quality,
        warnings=warnings + ([quote.error_message] if quote.error_message else []),
    )


def refresh_equity_quotes_batch(conn: Connection, req: QuoteRefreshRequest) -> MarketBatchUpdateResponse:
    target_date = _effective_market_date(req.price_date)
    target = target_date.isoformat()
    all_rows = conn.execute(
        """
        SELECT DISTINCT i.instrument_id,i.name,i.ticker
        FROM instruments i
        JOIN instrument_price_mappings m
          ON m.instrument_id=i.instrument_id AND m.mapping_status='mapped'
        WHERE i.asset_class IN ('stock','equity','etf')
          AND i.is_active=1
          AND COALESCE(i.instrument_status,'active') NOT IN ('inactive','delisted','suspended','merged')
          AND COALESCE(i.valuation_policy,'')!='exclude_from_auto_price_update'
          AND m.provider_symbol IS NOT NULL AND m.provider_symbol!=''
        ORDER BY i.name
        """,
    ).fetchall()
    row_states = [(row, _has_fresh_price_for_target(conn, str(row["instrument_id"]), target_date)) for row in all_rows]
    row_states.sort(key=lambda item: (item[1], str(item[0]["name"] or "")))
    row_states = row_states[: max(1, min(int(req.limit or 100), 100))]
    updated = skipped = cached = processed = 0
    warnings: list[str] = []
    errors: list[str] = []
    item_results: list[dict[str, str | int | bool | None]] = []
    last_call_at = 0.0
    for row, has_fresh in row_states:
        if req.only_missing and has_fresh:
            cached += 1
            skipped += 1
            item_results.append({"instrument_id": row["instrument_id"], "ticker": row["ticker"], "status": "cached", "attempts": 0})
            continue
        elapsed = time.monotonic() - last_call_at
        if last_call_at and elapsed < req.pacing_seconds:
            time.sleep(req.pacing_seconds - elapsed)
        attempt = 0
        quote: MarketQuoteResponse | None = None
        while attempt <= req.max_retries:
            attempt += 1
            processed += 1
            last_call_at = time.monotonic()
            quote = refresh_equity_quote(conn, row["instrument_id"], req)
            if quote.latest_price is not None and quote.quality_status == "fresh":
                break
            if quote.quality_status not in {"rate_limited", "network_error"} or attempt > req.max_retries:
                break
            time.sleep(min(2 ** (attempt - 1), 4))
        assert quote is not None
        if quote.latest_price is not None and quote.quality_status == "fresh":
            updated += 1
        else:
            skipped += 1
            code = quote.warnings[0] if quote.warnings else quote.quality_status
            warnings.append(f"{row['ticker']}: {code}")
            errors.append(quote.quality_status)
        item_results.append({
            "instrument_id": row["instrument_id"],
            "ticker": row["ticker"],
            "status": quote.quality_status,
            "provider": quote.provider,
            "provider_symbol": quote.provider_symbol,
            "price_date": quote.fetched_at[:10] if quote.fetched_at else None,
            "currency": quote.currency,
            "attempts": attempt,
        })
    coverage_total = len(all_rows)
    valued = sum(_has_fresh_price_for_target(conn, str(row["instrument_id"]), target_date) for row in all_rows)
    if valued == coverage_total and coverage_total > 0 and not req.dry_run:
        try:
            from jarvis_finance.services.portfolio_analytics import run_daily_market_valuation

            valuation = run_daily_market_valuation(conn, as_of=target)
            if valuation.status != "complete":
                warnings.append("portfolio_valuation_partial")
        except RuntimeError as exc:
            warnings.append(str(exc) if str(exc) == "market_job_already_running" else "portfolio_valuation_failed")
    return MarketBatchUpdateResponse(
        action="equity_update_quotes",
        provider=req.provider,
        total=len(row_states),
        updated=updated,
        skipped=skipped,
        warnings=warnings,
        errors=errors,
        target_date=target,
        cached=cached,
        processed=processed,
        valued=valued,
        coverage_total=coverage_total,
        complete=coverage_total > 0 and valued == coverage_total,
        results=item_results,
        render_provider_calls=False,
    )

def get_equity_quote(conn: Connection, instrument_id: str) -> MarketQuoteResponse:
    inst, mapping, warnings = _instrument_mapping(conn, instrument_id)
    latest = conn.execute("SELECT * FROM market_prices WHERE instrument_id=? ORDER BY COALESCE(price_timestamp, created_at, price_date) DESC LIMIT 1", (instrument_id,)).fetchone()
    provider_symbol = (mapping["provider_symbol"] if mapping else inst["provider_symbol"]) if inst else None
    if not latest:
        return MarketQuoteResponse(currency=inst["currency"] if inst else None, provider_symbol=provider_symbol, quality_status="missing", warnings=warnings or ["Kurs fehlt"])
    return MarketQuoteResponse(latest_price=_decimal_text(latest["close"]), currency=latest["currency"], close=_decimal_text(latest["close"]), provider=latest["provider"], provider_symbol=provider_symbol, fetched_at=latest["price_timestamp"] or latest["created_at"], quality_status=latest["quality_status"] or "stale", warnings=warnings)


def get_equity_chart(conn: Connection, instrument_id: str, *, range: str = "1d", interval: str = "5m") -> MarketChartResponse:
    inst, mapping, warnings = _instrument_mapping(conn, instrument_id)
    points = get_equity_chart_points(conn, instrument_id, limit=390)
    provider_symbol = (mapping["provider_symbol"] if mapping else inst["provider_symbol"]) if inst else None
    return _chart_response(points, currency=inst["currency"] if inst else "CHF", provider_symbol=provider_symbol, warnings=warnings)


def _normalized_candle_params(range_key: str, interval_key: str) -> tuple[str, str]:
    allowed = {
        "1d": {"5m"},
        "5d": {"15m"},
        "1mo": {"1d"},
        "6mo": {"1d"},
        "ytd": {"1d"},
        "1y": {"1d"},
    }
    normalized_range = (range_key or "1d").lower()
    if normalized_range not in allowed:
        normalized_range = "1d"
    normalized_interval = (interval_key or "").lower()
    if normalized_interval not in allowed[normalized_range]:
        normalized_interval = next(iter(allowed[normalized_range]))
    return normalized_range, normalized_interval


def _candles_response_from_rows(rows: list[Any], *, instrument_id: str | None, symbol: str | None, range_key: str, interval_key: str, provider: str = "yfinance", quality_status: str = "fresh", warnings: list[str] | None = None) -> EquityCandlesResponse:
    candles: list[Any] = []
    volumes: list[int | None] = []
    currency = None
    exchange_timezone = None
    fetched_at = None
    for row in rows:
        item = dict(row) if hasattr(row, "keys") else row
        vol = item.get("volume")
        volume_text = _decimal_text(vol)
        candles.append({
            "time": str(item["timestamp"]),
            "open": _decimal_text(item["open"]) or "",
            "high": _decimal_text(item["high"]) or "",
            "low": _decimal_text(item["low"]) or "",
            "close": _decimal_text(item["close"]) or "",
            "volume": volume_text,
        })
        volumes.append(int(float(vol)) if vol not in (None, "") else None)
        currency = currency or item.get("currency")
        exchange_timezone = exchange_timezone or item.get("exchange_timezone")
        fetched_at = item.get("fetched_at") or fetched_at
    return EquityCandlesResponse(instrument_id=instrument_id, symbol=symbol, provider_symbol=symbol, range=range_key, interval=interval_key, provider=provider, quality_status=quality_status if candles else "missing", candles=candles, volume=volumes, currency=currency, exchange_timezone=exchange_timezone, fetched_at=fetched_at, warnings=warnings or ([] if candles else ["Zu wenig Kursdaten verfügbar."]))


def _yfinance_history(provider_symbol: str, *, range_key: str, interval_key: str) -> tuple[list[dict[str, Any]], str | None, str | None]:
    try:
        import yfinance as yf  # type: ignore[import-not-found]
    except Exception as exc:  # pragma: no cover - environment dependent
        raise RuntimeError("yfinance_not_installed") from exc
    ticker = yf.Ticker(provider_symbol)
    history = ticker.history(period=range_key, interval=interval_key, auto_adjust=False)
    if history is None or getattr(history, "empty", True):
        return [], None, None
    info = getattr(ticker, "fast_info", {}) or {}
    currency = None
    try:
        currency = info.get("currency") if hasattr(info, "get") else None
    except Exception:
        currency = None
    exchange_timezone = str(getattr(history.index, "tz", "") or "") or None
    candles: list[dict[str, Any]] = []
    for idx, row in history.iterrows():
        try:
            open_v = Decimal(str(row["Open"]))
            close_v = Decimal(str(row["Close"]))
            low_v = Decimal(str(row["Low"]))
            high_v = Decimal(str(row["High"]))
        except (InvalidOperation, KeyError, ValueError):
            continue
        if any(v.is_nan() for v in (open_v, close_v, low_v, high_v)):
            continue
        timestamp = idx.isoformat() if hasattr(idx, "isoformat") else str(idx)
        candles.append({"timestamp": timestamp, "open": open_v, "close": close_v, "low": low_v, "high": high_v, "volume": row.get("Volume"), "currency": currency, "exchange_timezone": exchange_timezone})
    return candles, currency, exchange_timezone


def get_equity_candles(conn: Connection, instrument_id: str, *, range: str = "1d", interval: str = "5m", refresh: bool = False) -> EquityCandlesResponse:
    range_key, interval_key = _normalized_candle_params(range, interval)
    inst, mapping, warnings = _instrument_mapping(conn, instrument_id)
    provider_symbol = (mapping["provider_symbol"] if mapping else inst["provider_symbol"]) if inst else None
    if not provider_symbol:
        return EquityCandlesResponse(instrument_id=instrument_id, symbol=None, provider_symbol=None, range=range_key, interval=interval_key, provider="yfinance", quality_status="missing_provider_symbol", currency=inst["currency"] if inst else None, warnings=warnings or ["Provider-Symbol fehlt"])
    if not refresh:
        cached = get_equity_intraday_candles(conn, instrument_id, range_key=range_key, interval_key=interval_key, max_age_minutes=10)
        if cached:
            return _candles_response_from_rows(cached, instrument_id=instrument_id, symbol=str(provider_symbol), range_key=range_key, interval_key=interval_key, warnings=warnings)
    try:
        candles, currency, exchange_timezone = _yfinance_history(str(provider_symbol), range_key=range_key, interval_key=interval_key)
    except Exception as exc:
        return EquityCandlesResponse(instrument_id=instrument_id, symbol=str(provider_symbol), provider_symbol=str(provider_symbol), range=range_key, interval=interval_key, provider="yfinance", quality_status="provider_error", currency=inst["currency"] if inst else None, warnings=[str(exc)])
    if not candles:
        return EquityCandlesResponse(instrument_id=instrument_id, symbol=str(provider_symbol), provider_symbol=str(provider_symbol), range=range_key, interval=interval_key, provider="yfinance", quality_status="missing", currency=currency or (inst["currency"] if inst else None), exchange_timezone=exchange_timezone, warnings=["Zu wenig Kursdaten verfügbar."])
    upsert_equity_intraday_candles(conn, instrument_id=instrument_id, provider="yfinance", provider_symbol=str(provider_symbol), range_key=range_key, interval_key=interval_key, candles=candles, currency=currency or (inst["currency"] if inst else None), exchange_timezone=exchange_timezone, quality_status="fresh")
    conn.commit()
    rows = get_equity_intraday_candles(conn, instrument_id, range_key=range_key, interval_key=interval_key, max_age_minutes=15)
    return _candles_response_from_rows(rows, instrument_id=instrument_id, symbol=str(provider_symbol), range_key=range_key, interval_key=interval_key, warnings=warnings)


def refresh_equity_fx(conn: Connection, instrument_id: str) -> dict[str, object]:
    inst = conn.execute("SELECT currency FROM instruments WHERE instrument_id=?", (instrument_id,)).fetchone()
    if not inst:
        raise HTTPException(status_code=404, detail="Instrument not found")
    currencies: set[str] = set()
    for value in [inst["currency"]]:
        if value:
            currencies.add(str(value).upper())
    latest_price = conn.execute("SELECT currency FROM market_prices WHERE instrument_id=? ORDER BY COALESCE(price_timestamp, created_at, price_date) DESC LIMIT 1", (instrument_id,)).fetchone()
    if latest_price and latest_price["currency"]:
        currencies.add(str(latest_price["currency"]).upper())
    tx_rows = conn.execute("SELECT transaction_id, currency_original, trade_date, fx_status, fx_rate_to_chf FROM transactions WHERE instrument_id=? AND COALESCE(is_voided,0)=0", (instrument_id,)).fetchall()
    for row in tx_rows:
        if row["currency_original"]:
            currencies.add(str(row["currency_original"]).upper())
    updated = 0
    warnings: list[str] = []
    from jarvis_finance.fx.providers import FrankfurterFxProvider, TwelveDataFxProvider
    from jarvis_finance.fx.rates import resolve_fx_rate_to_chf
    for cur in sorted(currencies):
        if cur == "CHF":
            continue
        latest_result = resolve_fx_rate_to_chf(conn, base_currency=cur, rate_date=None, providers=[FrankfurterFxProvider(), TwelveDataFxProvider()], persist=True, resolve_fixed=True)
        if latest_result.status == "ok":
            updated += 1
        elif latest_result.warning:
            warnings.append(f"{cur}: {latest_result.warning}")
    for row in tx_rows:
        cur = str(row["currency_original"] or "").upper()
        if cur == "CHF":
            if row["fx_status"] != "not_needed" or not row["fx_rate_to_chf"]:
                conn.execute("UPDATE transactions SET fx_rate_to_chf='1', fx_source='not_needed', fx_status='not_needed', updated_at=? WHERE transaction_id=?", (utc_now(), row["transaction_id"]))
                updated += 1
            continue
        if cur and (row["fx_status"] == "missing" or not row["fx_rate_to_chf"]):
            historical = resolve_fx_rate_to_chf(conn, base_currency=cur, rate_date=row["trade_date"], providers=[FrankfurterFxProvider(), TwelveDataFxProvider()], persist=True, resolve_fixed=True)
            if historical.status == "ok" and historical.rate is not None:
                conn.execute("UPDATE transactions SET fx_rate_to_chf=?, fx_source=?, fx_status='ok', updated_at=? WHERE transaction_id=?", (format(historical.rate, "f"), historical.source, utc_now(), row["transaction_id"]))
                updated += 1
            elif historical.warning:
                warnings.append(f"{cur} {row['trade_date']}: {historical.warning}")
    conn.commit()
    return {"action": "equity_fx_recheck", "instrument_id": instrument_id, "currencies": sorted(currencies), "updated": updated, "warnings": warnings[:10], "render_provider_calls": False}


def _crypto_asset(conn: Connection, asset_id: str):
    asset = conn.execute("SELECT * FROM crypto_assets WHERE asset_id=?", (asset_id,)).fetchone()
    if not asset:
        raise HTTPException(status_code=404, detail="Crypto asset not found")
    return asset


class BinanceClient:
    base_url = "https://api.binance.com"

    def __init__(self, opener=None, timeout_seconds: float = 8.0) -> None:
        self.opener = opener or request.urlopen
        self.timeout_seconds = timeout_seconds

    def _json(self, path: str, params: dict[str, str]) -> object:
        url = self.base_url + path + "?" + parse.urlencode(params)
        try:
            with self.opener(url, timeout=self.timeout_seconds) as response:  # noqa: S310 - explicit user-triggered provider call
                return json.loads(response.read().decode("utf-8"))
        except error.HTTPError as exc:
            if exc.code == 429:
                raise RuntimeError("binance_rate_limited") from exc
            if exc.code == 400:
                raise RuntimeError("unsupported_pair") from exc
            raise RuntimeError("binance_provider_error") from exc
        except error.URLError as exc:
            raise RuntimeError("binance_network_error") from exc

    def ticker_24h(self, symbol: str) -> dict[str, object]:
        data = self._json("/api/v3/ticker/24hr", {"symbol": symbol.upper()})
        return data if isinstance(data, dict) else {}

    def klines(self, symbol: str, *, interval: str = "5m", limit: int = 288) -> list[list[object]]:
        data = self._json("/api/v3/klines", {"symbol": symbol.upper(), "interval": interval, "limit": str(limit)})
        return data if isinstance(data, list) else []


def refresh_crypto_quote(conn: Connection, asset_id: str, req: QuoteRefreshRequest) -> MarketQuoteResponse:
    asset = _crypto_asset(conn, asset_id)
    provider = (req.provider or "coingecko").lower()
    if provider == "binance":
        symbol = asset["binance_symbol"] if "binance_symbol" in asset.keys() else None
        if not symbol:
            return MarketQuoteResponse(currency="USD", provider="binance", quality_status="unsupported_pair", warnings=["Binance-Symbol fehlt"])
        try:
            ticker = BinanceClient().ticker_24h(symbol)
            price = Decimal(str(ticker.get("lastPrice")))
            ts = utc_now()
            if not req.dry_run:
                upsert_crypto_price_point(conn, asset_id=asset_id, timestamp=ts, price=price, currency="USD", provider="binance", provider_symbol=symbol, interval=req.interval, source_quality="fresh")
                conn.commit()
            return MarketQuoteResponse(latest_price=format(price, "f"), currency="USD", change_abs=_decimal_text(ticker.get("priceChange")), change_pct=_decimal_text(ticker.get("priceChangePercent")), high=_decimal_text(ticker.get("highPrice")), low=_decimal_text(ticker.get("lowPrice")), close=format(price, "f"), volume=_decimal_text(ticker.get("volume")), provider="binance", provider_symbol=symbol, fetched_at=ts, quality_status="fresh")
        except Exception as exc:
            return MarketQuoteResponse(currency="USD", provider="binance", provider_symbol=symbol, quality_status=_quality_from_error(str(exc), "provider_error"), warnings=[str(exc)])
    if not asset["coingecko_id"]:
        return MarketQuoteResponse(currency=req.currency.upper(), provider="CoinGecko", quality_status="missing", warnings=["CoinGecko-ID fehlt"])
    quote: PriceQuote = CoinGeckoClient().get_crypto_price(asset["coingecko_id"], req.currency)
    quality = quote.quality_status if quote.price is not None else _quality_from_error(quote.error_message, quote.quality_status)
    ts = quote.provider_timestamp or utc_now()
    if not req.dry_run:
        store_crypto_price(conn, asset_id=asset_id, quote=quote)
        if quote.price is not None:
            upsert_crypto_price_point(conn, asset_id=asset_id, timestamp=ts, price=quote.price, currency=quote.currency, provider=quote.provider, provider_symbol=quote.coingecko_id, interval=req.interval, source_quality=quality)
            conn.commit()
    return MarketQuoteResponse(latest_price=_decimal_text(quote.price), currency=quote.currency, close=_decimal_text(quote.price), provider=quote.provider, provider_symbol=quote.coingecko_id, fetched_at=ts, quality_status=quality, warnings=[quote.error_message] if quote.error_message else [])


def refresh_crypto_quotes_batch(conn: Connection, req: QuoteRefreshRequest) -> MarketBatchUpdateResponse:
    provider = (req.provider or "binance").lower()
    if provider == "binance":
        sql = "SELECT asset_id FROM crypto_assets WHERE is_active=1 AND binance_symbol IS NOT NULL AND binance_symbol!='' ORDER BY symbol LIMIT ?"
    else:
        sql = "SELECT asset_id FROM crypto_assets WHERE is_active=1 AND coingecko_id IS NOT NULL AND coingecko_id!='' ORDER BY symbol LIMIT ?"
    rows = conn.execute(sql, (max(1, min(int(req.limit or 20), 100)),)).fetchall()
    updated = skipped = 0
    warnings: list[str] = []
    errors: list[str] = []
    for row in rows:
        quote = refresh_crypto_quote(conn, row["asset_id"], req)
        if quote.quality_status in {"fresh", "delayed", "stale"} and quote.latest_price is not None:
            updated += 1
        else:
            skipped += 1
            warnings.extend(quote.warnings)
            if quote.quality_status in {"rate_limited", "provider_error"}:
                errors.append(quote.quality_status)
    return MarketBatchUpdateResponse(action="crypto_update_live_stats", provider=provider, total=len(rows), updated=updated, skipped=skipped, warnings=warnings[:10], errors=errors[:10], render_provider_calls=False)


def get_crypto_quote(conn: Connection, asset_id: str, *, currency: str = "CHF") -> MarketQuoteResponse:
    asset = _crypto_asset(conn, asset_id)
    latest = conn.execute("SELECT * FROM crypto_prices WHERE asset_id=? AND price_currency=? ORDER BY COALESCE(provider_timestamp, fetched_at) DESC LIMIT 1", (asset_id, currency.upper())).fetchone()
    if not latest:
        return MarketQuoteResponse(currency=currency.upper(), provider_symbol=asset["coingecko_id"], quality_status="missing", warnings=["Kurs fehlt"])
    return MarketQuoteResponse(latest_price=_decimal_text(latest["price"]), currency=latest["price_currency"], close=_decimal_text(latest["price"]), provider=latest["provider"], provider_symbol=latest["coingecko_id"], fetched_at=latest["provider_timestamp"] or latest["fetched_at"], quality_status=latest["quality_status"] or "stale")


def get_crypto_chart(conn: Connection, asset_id: str, *, range: str = "1d", interval: str = "5m", currency: str = "CHF") -> MarketChartResponse:
    asset = _crypto_asset(conn, asset_id)
    points = get_crypto_chart_points(conn, asset_id, currency=currency.upper(), limit=390)
    return _chart_response(points, currency=currency.upper(), provider_symbol=asset["coingecko_id"])