from __future__ import annotations

import json
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
from pathlib import Path
from sqlite3 import Connection

from jarvis_finance.config.settings import Settings
from jarvis_finance.crypto.holdings import calculate_crypto_holdings
from jarvis_finance.ledger.positions import Position, calculate_positions
from jarvis_finance.market.providers import is_price_stale
from jarvis_finance.services.portfolio_aggregation import (
    latest_audited_market_values,
    latest_official_postfinance_positions,
    post_snapshot_postfinance_quantity_deltas,
    postfinance_account_roles,
)

MONEY = Decimal("0.01")


def d(value: object, default: str = "0") -> Decimal:
    if value is None or str(value).strip() == "":
        return Decimal(default)
    try:
        return Decimal(str(value))
    except InvalidOperation:
        return Decimal(default)


def dec_text(value: Decimal | int | str | None, places: str | None = None) -> str:
    if value is None:
        return ""
    value = d(value)
    if places:
        value = value.quantize(Decimal(places), rounding=ROUND_HALF_UP)
    return format(value, "f")


def grouped_number(value: Decimal | int | str, places: str = "0.01") -> str:
    text = dec_text(value, places)
    whole, _, frac = text.partition(".")
    sign = "-" if whole.startswith("-") else ""
    whole = whole.lstrip("-")
    groups: list[str] = []
    while whole:
        groups.append(whole[-3:])
        whole = whole[:-3]
    grouped = "'".join(reversed(groups)) if groups else "0"
    return f"{sign}{grouped}{('.' + frac) if frac else ''}"


def chf_text(value: Decimal | int | str | None) -> str:
    if value is None:
        return "nicht bewertet"
    return f"CHF {grouped_number(value)}"


def date_text(value: str | None) -> str:
    if not value:
        return ""
    raw = str(value)
    if "T" in raw:
        return raw.split("T", 1)[0]
    return raw[:10]


def money_text(value: Decimal | None, currency: str = "CHF") -> str:
    if currency.upper() == "CHF":
        return chf_text(value)
    if value is None:
        return f"n/a {currency}"
    return f"{dec_text(value, '0.01')} {currency}"


STATUS_LABELS = {
    "missing_market_price": "Preis fehlt",
    "missing_provider_symbol": "Mapping fehlt",
    "missing_price": "Preis fehlt",
    "missing_fx": "FX-Kurs fehlt",
    "cost_basis_uncertain": "Einstand unvollständig",
    "incomplete": "Unvollständig",
    "snapshot_only": "Snapshot-only",
    "fresh": "Aktuell",
    "stale": "Veraltet",
    "missing": "Preis fehlt",
    "unknown": "Ungeklärt",
    "mapped": "Zugeordnet",
    "confirmed": "Bestätigt",
    "user_confirmed": "Bestätigt",
    "ok": "Aktuell",
    "warning": "Hinweis",
    "critical": "Kritisch",
    "kritisch": "Kritisch",
    "warnung": "Wichtig",
    "info": "Info",
    "active": "Aktiv",
    "inactive": "Inaktiv",
    "verified": "Verifiziert",
}


def translate_status(status: str | None) -> str:
    if not status:
        return "Ungeklärt"
    return STATUS_LABELS.get(str(status), str(status).replace("_", " ").capitalize())


def is_technical_note(text: str | None) -> bool:
    if not text:
        return False
    lowered = text.lower()
    markers = ["productive initial crypto snapshot import", "no wallet address from xlsx", "initial crypto snapshot import"]
    return any(marker in lowered for marker in markers)


def user_note(text: str | None) -> str:
    return "" if is_technical_note(text) else (text or "")


def visible_user_columns(row: dict[str, str]) -> list[str]:
    hidden = {"asset_id", "wallet_id", "instrument_id", "account_id", "entity_id", "rule_id", "source_id", "provider_symbol_status", "hedge_status", "dedup_key", "fingerprint", "raw_json"}
    return [k for k in row if not k.startswith("_") and k not in hidden]


def percent_text(value: Decimal | None) -> str:
    if value is None:
        return ""
    return f"{dec_text(value, '0.01')}%"


def price_status(price: dict[str, str] | None) -> str:
    if not price:
        return "missing"
    return price.get("quality_status") or "missing"


def _latest_price_time(conn: Connection, *, currency: str = "CHF") -> str:
    row = conn.execute(
        """
        SELECT COALESCE(MAX(provider_timestamp), MAX(fetched_at)) AS ts
        FROM crypto_prices WHERE price_currency=?
        """,
        (currency.upper(),),
    ).fetchone()
    return row["ts"] if row and row["ts"] else ""


def _crypto_value_rows(conn: Connection, *, price_max_age_seconds: int = 86_400) -> list[dict[str, object]]:
    holdings = calculate_crypto_holdings(conn)
    rows: list[dict[str, object]] = []
    for asset_id, holding in sorted(holdings.total_by_asset.items()):
        asset = conn.execute("SELECT * FROM crypto_assets WHERE asset_id=?", (asset_id,)).fetchone()
        if not asset:
            continue
        price = latest_crypto_price(conn, asset_id, "CHF", max_age_seconds=price_max_age_seconds)
        price_dec = d(price["price"]) if price and price["price"] else None
        value = holding.quantity * price_dec if price_dec is not None else None
        wallets = {wid for (wid, aid), wh in holdings.wallet_holdings.items() if aid == asset_id and wh.quantity != 0}
        rows.append({
            "asset_id": asset_id,
            "coin": asset["coin_name"],
            "symbol": asset["symbol"],
            "coingecko_id": asset["coingecko_id"] or "",
            "quantity": holding.quantity,
            "price": price_dec,
            "value": value,
            "wallet_count": len(wallets),
            "price_status": price_status(price),
            "last_price_update": (price["provider_timestamp"] or price["fetched_at"]) if price else "",
        })
    return rows


def rowdicts(rows) -> list[dict[str, str]]:
    return [{k: ("" if row[k] is None else str(row[k])) for k in row.keys()} for row in rows]


def db_exists(path: str | Path) -> bool:
    return Path(path).expanduser().exists()


def runtime_outside_repo(settings: Settings, repo_root: Path) -> bool:
    try:
        settings.runtime_paths.base_dir.resolve().relative_to(repo_root.resolve())
        return False
    except ValueError:
        return True


def latest_crypto_price(conn: Connection, asset_id: str, currency: str = "CHF", *, max_age_seconds: int = 86_400) -> dict[str, str] | None:
    row = conn.execute(
        """
        SELECT price, price_currency, fetched_at, provider_timestamp, quality_status, provider
        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()
    if not row:
        return None
    item = dict(row)
    ts = item.get("provider_timestamp") or item.get("fetched_at")
    if item.get("quality_status") == "fresh" and is_price_stale(ts, max_age_seconds=max_age_seconds):
        item["quality_status"] = "stale"
    return item


def get_crypto_summary_cards(conn: Connection, *, price_max_age_seconds: int = 86_400) -> dict[str, str]:
    rows = _crypto_value_rows(conn, price_max_age_seconds=price_max_age_seconds)
    holdings = calculate_crypto_holdings(conn)
    total_value = sum((r["value"] for r in rows if r["value"] is not None), Decimal("0"))
    missing = sum(1 for r in rows if r["price_status"] == "missing")
    stale = sum(1 for r in rows if r["price_status"] == "stale")
    wallets_with_holdings = {wid for (wid, _aid), holding in holdings.wallet_holdings.items() if holding.quantity != 0}
    status = "ok" if missing == 0 and stale == 0 else ("warning" if missing or stale else "ok")
    return {
        "total_value_chf": dec_text(total_value),
        "coin_count": str(len(rows)),
        "wallets_with_holdings": str(len(wallets_with_holdings)),
        "holding_count": str(sum(1 for h in holdings.wallet_holdings.values() if h.quantity != 0)),
        "coins_without_price": str(missing),
        "stale_price_count": str(stale),
        "last_price_update_at": _latest_price_time(conn),
        "data_quality_status": status,
    }


def _wallet_names_for_asset(conn: Connection, asset_id: str) -> list[str]:
    holdings = calculate_crypto_holdings(conn)
    names: list[str] = []
    for (wallet_id, aid), holding in holdings.wallet_holdings.items():
        if aid == asset_id and holding.quantity != 0:
            row = conn.execute("SELECT wallet_name FROM crypto_wallets WHERE wallet_id=?", (wallet_id,)).fetchone()
            if row:
                names.append(row["wallet_name"])
    return sorted(names)


def get_crypto_coin_summary(
    conn: Connection,
    *,
    admin_mode: bool = False,
    price_max_age_seconds: int = 86_400,
    sort_by: str = "Wert absteigend",
    search: str | None = None,
    status_filter: str | None = None,
    wallet_filter: str | None = None,
) -> list[dict[str, str]]:
    raw = _crypto_value_rows(conn, price_max_age_seconds=price_max_age_seconds)
    total = sum((r["value"] for r in raw if r["value"] is not None), Decimal("0"))
    out: list[dict[str, str]] = []
    for r in sorted(raw, key=lambda item: item["value"] or Decimal("-1"), reverse=True):
        value = r["value"]
        share = ((value / total) * Decimal("100")) if value is not None and total else None
        item = {
            "coin": str(r["coin"]),
            "symbol": str(r["symbol"]),
            "gesamtmenge": dec_text(r["quantity"]),
            "kurs_chf": chf_text(r["price"]) if r["price"] is not None else "nicht bewertet",
            "gesamtwert_chf": chf_text(value) if value is not None else "nicht bewertet",
            "anteil": percent_text(share),
            "wallets": str(r["wallet_count"]),
            "status": translate_status(str(r["price_status"])),
            "_sort_value_chf": dec_text(value or Decimal("0"), "0.01"),
            "_last_price_update": date_text(str(r["last_price_update"])),
            "_asset_id": str(r["asset_id"]),
            "_wallet_names": " | ".join(_wallet_names_for_asset(conn, str(r["asset_id"]))),
        }
        if admin_mode:
            item.update({"asset_id": str(r["asset_id"]), "coingecko_id": str(r["coingecko_id"]), "price_status": str(r["price_status"]), "last_price_update": str(r["last_price_update"])})
        out.append(item)
    if search:
        needle = search.lower()
        out = [row for row in out if needle in row["coin"].lower() or needle in row["symbol"].lower()]
    if status_filter and status_filter != "Alle":
        out = [row for row in out if row["status"] == status_filter]
    if wallet_filter and wallet_filter != "Alle":
        out = [row for row in out if wallet_filter in row.get("_wallet_names", "")]
    sorters = {
        "Wert absteigend": lambda row: (d(row["_sort_value_chf"]), row["coin"]),
        "Wert aufsteigend": lambda row: (d(row["_sort_value_chf"]), row["coin"]),
        "Name A–Z": lambda row: row["coin"].lower(),
        "Anteil": lambda row: (d(row["anteil"].replace("%", "")), row["coin"]),
        "Wallet-Anzahl": lambda row: (int(row["wallets"]), row["coin"]),
        "Status": lambda row: (row["status"], row["coin"]),
    }
    reverse = sort_by in {"Wert absteigend", "Anteil", "Wallet-Anzahl"}
    return sorted(out, key=sorters.get(sort_by, sorters["Wert absteigend"]), reverse=reverse)


def get_crypto_coin_wallet_details(conn: Connection, asset_id: str, *, price_max_age_seconds: int = 86_400, admin_mode: bool = False) -> list[dict[str, str]]:
    holdings = calculate_crypto_holdings(conn)
    price = latest_crypto_price(conn, asset_id, "CHF", max_age_seconds=price_max_age_seconds)
    price_dec = d(price["price"]) if price and price["price"] else None
    total_qty = sum((h.quantity for (wid, aid), h in holdings.wallet_holdings.items() if aid == asset_id), Decimal("0"))
    rows: list[dict[str, str]] = []
    for (wallet_id, aid), holding in sorted(holdings.wallet_holdings.items()):
        if aid != asset_id or holding.quantity == 0:
            continue
        wallet = conn.execute("SELECT wallet_name, wallet_type, platform_provider, notes FROM crypto_wallets WHERE wallet_id=?", (wallet_id,)).fetchone()
        value = holding.quantity * price_dec if price_dec is not None else None
        share = (holding.quantity / total_qty * Decimal("100")) if total_qty else None
        item = {
            "wallet": wallet["wallet_name"] if wallet else wallet_id,
            "typ": wallet["wallet_type"] if wallet else "",
            "menge": dec_text(holding.quantity),
            "wert_chf": chf_text(value) if value is not None else "nicht bewertet",
            "anteil": percent_text(share),
            "letzte_verifikation": date_text(holding.last_verified_at),
            "status": translate_status(holding.verification_status),
            "notiz": user_note(wallet["notes"] if wallet and wallet["notes"] else ""),
            "_wallet_id": wallet_id,
            "_asset_id": asset_id,
        }
        if admin_mode:
            item.update({"wallet_id": wallet_id, "asset_id": asset_id})
        rows.append(item)
    return rows


def get_wallet_overview_cards(conn: Connection, *, price_max_age_seconds: int = 86_400) -> dict[str, str]:
    rows = get_wallet_user_overview(conn, price_max_age_seconds=price_max_age_seconds)
    total_value = sum((d(r["_numeric_value_chf"]) for r in rows), Decimal("0"))
    missing_verification = sum(1 for r in rows if not r.get("last_verified_at"))
    return {"wallet_count": str(len(rows)), "total_value_chf": dec_text(total_value), "wallets_missing_verification": str(missing_verification)}


def get_wallet_user_overview(conn: Connection, *, price_max_age_seconds: int = 86_400, admin_mode: bool = False) -> list[dict[str, str]]:
    holdings = calculate_crypto_holdings(conn)
    rows: list[dict[str, str]] = []
    for wallet in conn.execute("SELECT * FROM crypto_wallets ORDER BY wallet_name").fetchall():
        value = Decimal("0")
        has_missing_price = False
        coins: set[str] = set()
        for (wallet_id, asset_id), holding in holdings.wallet_holdings.items():
            if wallet_id != wallet["wallet_id"] or holding.quantity == 0:
                continue
            coins.add(asset_id)
            price = latest_crypto_price(conn, asset_id, "CHF", max_age_seconds=price_max_age_seconds)
            if price and price.get("price"):
                value += holding.quantity * d(price["price"])
            else:
                has_missing_price = True
        item = {
            "wallet_name": wallet["wallet_name"],
            "wallet_type": wallet["wallet_type"],
            "provider": wallet["platform_provider"] or "",
            "total_value_chf": chf_text(value) if not has_missing_price else f"{chf_text(value)} + unbewertet",
            "coin_count": str(len(coins)),
            "last_verified_at": date_text(wallet["last_verified_at"]),
            "status": translate_status("active" if int(wallet["is_active"] or 0) else "inactive"),
            "actions": "Öffnen · Coin hinzufügen · Korrigieren · Transfer",
            "_numeric_value_chf": dec_text(value, "0.01"),
            "_wallet_id": wallet["wallet_id"],
        }
        if admin_mode:
            item["wallet_id"] = wallet["wallet_id"]
        rows.append(item)
    return rows


def get_wallet_coin_details(conn: Connection, wallet_id: str, *, price_max_age_seconds: int = 86_400, admin_mode: bool = False) -> list[dict[str, str]]:
    holdings = calculate_crypto_holdings(conn)
    rows: list[dict[str, str]] = []
    for (wid, asset_id), holding in sorted(holdings.wallet_holdings.items()):
        if wid != wallet_id or holding.quantity == 0:
            continue
        asset = conn.execute("SELECT coin_name, symbol FROM crypto_assets WHERE asset_id=?", (asset_id,)).fetchone()
        price = latest_crypto_price(conn, asset_id, "CHF", max_age_seconds=price_max_age_seconds)
        price_dec = d(price["price"]) if price and price["price"] else None
        value = holding.quantity * price_dec if price_dec is not None else None
        item = {
            "coin": asset["coin_name"] if asset else asset_id,
            "symbol": asset["symbol"] if asset else "",
            "menge": dec_text(holding.quantity),
            "wert_chf": chf_text(value) if value is not None else "nicht bewertet",
            "status": translate_status(price_status(price)),
            "actions": "Bestand korrigieren · Transfer aus Wallet · Audit anzeigen",
            "_asset_id": asset_id,
            "_wallet_id": wallet_id,
        }
        if admin_mode:
            item.update({"asset_id": asset_id, "wallet_id": wallet_id})
        rows.append(item)
    return rows


def get_crypto_coin_allocation_chart(conn: Connection, *, price_max_age_seconds: int = 86_400) -> list[dict[str, str]]:
    rows = _crypto_value_rows(conn, price_max_age_seconds=price_max_age_seconds)
    out = []
    for r in sorted(rows, key=lambda item: item["value"] or Decimal("-1"), reverse=True):
        if r["value"] is not None:
            out.append({"label": str(r["coin"]), "value_chf": dec_text(r["value"], "0.01")})
    return out


def get_crypto_wallet_allocation_chart(conn: Connection, *, price_max_age_seconds: int = 86_400) -> list[dict[str, str]]:
    rows = get_wallet_user_overview(conn, price_max_age_seconds=price_max_age_seconds)
    out = []
    for row in rows:
        value = row["_numeric_value_chf"]
        if d(value) > 0:
            out.append({"label": row["wallet_name"], "value_chf": dec_text(value, "0.01")})
    return sorted(out, key=lambda item: d(item["value_chf"]), reverse=True)


def get_crypto_coin_detail(conn: Connection, asset_id: str, *, price_max_age_seconds: int = 86_400) -> dict[str, object]:
    row = next((r for r in _crypto_value_rows(conn, price_max_age_seconds=price_max_age_seconds) if r["asset_id"] == asset_id), None)
    if not row:
        return {}
    all_value = sum((r["value"] for r in _crypto_value_rows(conn, price_max_age_seconds=price_max_age_seconds) if r["value"] is not None), Decimal("0"))
    value = row["value"]
    share = (value / all_value * Decimal("100")) if value is not None and all_value else None
    latest_price = latest_crypto_price(conn, asset_id, "CHF", max_age_seconds=price_max_age_seconds)
    return {
        "coin": str(row["coin"]),
        "symbol": str(row["symbol"]),
        "gesamtmenge": dec_text(row["quantity"]),
        "gesamtwert_chf": chf_text(value) if value is not None else "nicht bewertet",
        "kurs_chf": chf_text(row["price"]) if row["price"] is not None else "nicht bewertet",
        "anteil": percent_text(share),
        "preisstatus": translate_status(str(row["price_status"])),
        "letzte_preisaktualisierung": date_text(str(row["last_price_update"])),
        "preisquelle": str(latest_price.get("provider") or "unbekannt") if latest_price else "unbekannt",
        "cache_status": "lokaler Cache" if latest_price else "kein lokaler Preis",
        "coingecko_link": f"https://www.coingecko.com/en/coins/{row['coingecko_id']}" if row.get("coingecko_id") else "",
        "wallets": get_crypto_coin_wallet_details(conn, asset_id, price_max_age_seconds=price_max_age_seconds),
        "actions": ["Bestand hinzufügen", "Bestand korrigieren", "Transfer", "Bestand auf 0 setzen", "Audit / Verlauf anzeigen"],
    }


def get_wallet_detail(conn: Connection, wallet_id: str, *, price_max_age_seconds: int = 86_400) -> dict[str, object]:
    wallet = next((w for w in get_wallet_user_overview(conn, price_max_age_seconds=price_max_age_seconds, admin_mode=True) if w.get("wallet_id") == wallet_id), None)
    if not wallet:
        return {}
    return {
        "wallet_name": wallet["wallet_name"],
        "typ": wallet["wallet_type"],
        "provider": wallet["provider"],
        "gesamtwert_chf": wallet["total_value_chf"],
        "coins": get_wallet_coin_details(conn, wallet_id, price_max_age_seconds=price_max_age_seconds),
        "letzte_verifikation": wallet["last_verified_at"],
        "notizen": user_note(wallet.get("notes", "")),
        "status": wallet["status"],
        "actions": ["Coin in Wallet hinzufügen", "Bestand korrigieren", "Transfer aus Wallet", "Wallet verifizieren", "Wallet bearbeiten", "Wallet deaktivieren"],
    }


def get_crypto_overview(conn: Connection, *, price_max_age_seconds: int = 86_400) -> list[dict[str, str]]:
    holdings = calculate_crypto_holdings(conn)
    rows: list[dict[str, str]] = []
    for asset_id, holding in sorted(holdings.total_by_asset.items()):
        qty = holding.quantity
        asset = conn.execute("SELECT * FROM crypto_assets WHERE asset_id=?", (asset_id,)).fetchone()
        if not asset:
            continue
        price = latest_crypto_price(conn, asset_id, "CHF", max_age_seconds=price_max_age_seconds)
        price_dec = d(price["price"]) if price and price["price"] else None
        value = qty * price_dec if price_dec is not None else None
        rows.append({
            "asset_id": asset_id,
            "coin_name": asset["coin_name"],
            "symbol": asset["symbol"],
            "coingecko_id": asset["coingecko_id"] or "",
            "quantity": dec_text(qty),
            "latest_price_chf": dec_text(price_dec, "0.01") if price_dec is not None else "",
            "market_value_chf": dec_text(value, "0.01") if value is not None else "",
            "price_quality": price["quality_status"] if price else "missing",
            "fetched_at": price["fetched_at"] if price else "",
        })
    return rows


def get_crypto_by_wallet(conn: Connection, *, price_max_age_seconds: int = 86_400) -> list[dict[str, str]]:
    holdings = calculate_crypto_holdings(conn)
    rows: list[dict[str, str]] = []
    for (wallet_id, asset_id), holding in sorted(holdings.wallet_holdings.items()):
        wallet = conn.execute("SELECT wallet_name, wallet_type, platform_provider FROM crypto_wallets WHERE wallet_id=?", (wallet_id,)).fetchone()
        asset = conn.execute("SELECT coin_name, symbol, coingecko_id FROM crypto_assets WHERE asset_id=?", (asset_id,)).fetchone()
        price = latest_crypto_price(conn, asset_id, "CHF", max_age_seconds=price_max_age_seconds)
        price_dec = d(price["price"]) if price and price["price"] else None
        value = holding.quantity * price_dec if price_dec is not None else None
        rows.append({
            "wallet_id": wallet_id,
            "wallet_name": wallet["wallet_name"] if wallet else wallet_id,
            "wallet_type": wallet["wallet_type"] if wallet else "",
            "provider": wallet["platform_provider"] if wallet else "",
            "coin_name": asset["coin_name"] if asset else asset_id,
            "symbol": asset["symbol"] if asset else "",
            "coingecko_id": asset["coingecko_id"] if asset else "",
            "quantity": dec_text(holding.quantity),
            "market_value_chf": dec_text(value, "0.01") if value is not None else "",
            "wallet_value_chf": dec_text(value, "0.01") if value is not None else "",
            "price_quality": price["quality_status"] if price else "missing",
            "price_fetched_at": price["fetched_at"] if price else "",
            "verification_status": holding.verification_status,
            "last_verified_at": holding.last_verified_at or "",
        })
    return rows


def get_wallets(conn: Connection) -> list[dict[str, str]]:
    return rowdicts(conn.execute(
        """
        SELECT wallet_id, wallet_name, wallet_type, platform_provider AS provider,
               network_chain, last_verified_at, is_active, notes
        FROM crypto_wallets ORDER BY wallet_name
        """
    ).fetchall())


def get_latest_crypto_transactions(conn: Connection, limit: int = 20) -> list[dict[str, str]]:
    return rowdicts(conn.execute(
        """
        SELECT ct.transaction_datetime, ct.transaction_type, ca.symbol, ct.quantity, ct.fee_quantity,
               fw.wallet_name AS from_wallet, tw.wallet_name AS to_wallet, ct.currency_original,
               ct.amount_chf, ct.confirmation_status, ct.source
        FROM crypto_transactions ct
        LEFT JOIN crypto_assets ca ON ca.asset_id=ct.asset_id
        LEFT JOIN crypto_wallets fw ON fw.wallet_id=ct.from_wallet_id
        LEFT JOIN crypto_wallets tw ON tw.wallet_id=ct.to_wallet_id
        ORDER BY ct.transaction_datetime DESC, ct.created_at DESC LIMIT ?
        """, (limit,)
    ).fetchall())


def get_platform_overview(conn: Connection) -> list[dict[str, str]]:
    return rowdicts(conn.execute(
        """
        SELECT p.name AS platform, p.platform_type, a.account_name, a.account_type, a.currency,
               a.performance_included, a.is_active
        FROM accounts a JOIN platforms p ON p.platform_id=a.platform_id
        ORDER BY p.name, a.account_name
        """
    ).fetchall())


def get_cash_overview(conn: Connection) -> list[dict[str, str]]:
    # One shared cash read model for the API, Command Center, allocation and reports.
    # The local import avoids coupling the write-oriented cash service into module init.
    from jarvis_finance.services.cash_service import list_cash_positions

    return [
        {
            "account_id": position.account_id,
            "platform": position.platform,
            "account_name": position.account_label,
            "currency": position.currency,
            "amount_original": position.amount,
            "amount_chf": position.used_value_chf,
            "quality_status": "ok" if position.status != "Abgleich offen" else "partial",
        }
        for position in list_cash_positions(conn)
    ]


def get_positions(conn: Connection) -> list[dict[str, str]]:
    result = calculate_positions(conn, persist_alerts=False)
    official = latest_official_postfinance_positions(conn)
    post_snapshot_deltas = post_snapshot_postfinance_quantity_deltas(conn, official)
    audited = latest_audited_market_values(conn)
    roles = postfinance_account_roles(conn)
    trading_cash_id = roles.get("etrading_cash")
    official_instruments = {key[1] for key in official}
    for key in list(result.positions):
        if key[0] == trading_cash_id and key[1] in official_instruments:
            # Instrument-bearing settlement events live on trading cash, while the
            # authoritative holding belongs to the distinct depot role.
            result.positions.pop(key)
    for key, override in official.items():
        result.positions.setdefault(
            key, Position(account_id=override.account_id, instrument_id=override.instrument_id)
        )
    rows: list[dict[str, str]] = []
    for (account_id, instrument_id), pos in sorted(result.positions.items()):
        row = conn.execute(
            """
            SELECT a.account_name, p.name AS platform, i.name, i.ticker, i.isin, i.asset_class, i.currency,
                   i.hedge_status, i.instrument_status, i.valuation_policy, i.corporate_action_status,
                   mp.close AS latest_price, mp.currency AS latest_price_currency, mp.price_date AS latest_price_date,
                   mp.provider AS latest_price_provider
            FROM accounts a JOIN platforms p ON p.platform_id=a.platform_id
            JOIN instruments i ON i.instrument_id=?
            LEFT JOIN market_prices mp ON mp.market_price_id = (
                SELECT market_price_id FROM market_prices
                WHERE instrument_id=i.instrument_id AND quality_status IN ('fresh','ok')
                ORDER BY price_date DESC, created_at DESC LIMIT 1
            )
            WHERE a.account_id=?
            """,
            (instrument_id, account_id),
        ).fetchone()
        override = official.get((account_id, instrument_id))
        effective_quantity = (
            override.quantity + post_snapshot_deltas.get(instrument_id, Decimal("0"))
            if override
            else pos.quantity
        )
        if override and effective_quantity <= 0:
            continue
        market_value = audited.get((account_id, instrument_id))
        use_market_run = bool(
            market_value
            and (not override or market_value.as_of > override.snapshot_date)
        )
        if override and not use_market_run:
            quantity = effective_quantity
            market_price = override.market_price_original
            market_value_chf = (
                override.market_value_chf * effective_quantity / override.quantity
                if override.quantity
                else Decimal("0")
            )
            price_currency = override.price_currency
            price_date = override.valuation_at
            price_provider = "PostFinance official snapshot"
            valuation_source = "postfinance_official_import"
        elif use_market_run and market_value:
            quantity = effective_quantity
            market_price = market_value.close
            market_value_chf = (
                market_value.value_chf * effective_quantity / market_value.quantity
                if market_value.quantity
                else market_value.value_chf
            )
            price_currency = market_value.currency
            price_date = market_value.as_of
            price_provider = market_value.provider
            valuation_source = "audited_market_run"
        else:
            quantity = pos.quantity
            market_price = pos.market_price_original
            market_value_chf = pos.market_value_chf
            price_currency = row["latest_price_currency"] if row else ""
            price_date = date_text(row["latest_price_date"] if row else "")
            price_provider = row["latest_price_provider"] if row else ""
            valuation_source = "canonical_ledger_market"
        rows.append({
            "account_id": account_id,
            "instrument_id": instrument_id,
            "platform": row["platform"] if row else "",
            "account_name": row["account_name"] if row else account_id,
            "asset_class": row["asset_class"] if row else "",
            "name": row["name"] if row else instrument_id,
            "ticker": row["ticker"] if row else "",
            "isin": row["isin"] if row else "",
            "quantity": dec_text(quantity),
            "cost_basis_chf": dec_text(pos.cost_basis_chf, "0.01"),
            "market_price_original": dec_text(market_price, "0.01") if market_price is not None else "",
            "price_currency": price_currency,
            "price_date": price_date,
            "price_provider": price_provider,
            "market_value_chf": dec_text(market_value_chf, "0.01") if market_value_chf is not None else "",
            "price_status": "ok" if market_price is not None else "missing_market_price",
            "valuation_source": valuation_source,
            "valuation_as_of": price_date,
            "provider_symbol_status": "mapped" if conn.execute("SELECT 1 FROM instrument_price_mappings WHERE instrument_id=? AND mapping_status IN ('mapped','confirmed','user_confirmed') AND provider_symbol IS NOT NULL LIMIT 1", (instrument_id,)).fetchone() else "missing_provider_symbol",
            "hedge_status": row["hedge_status"] if row else pos.hedge_status,
            "instrument_status": row["instrument_status"] if row else pos.instrument_status,
            "valuation_policy": row["valuation_policy"] if row else pos.valuation_policy,
            "fx_status": "ok" if "missing_fx" not in pos.quality_warnings else "missing_fx",
            "market_price_status": "ok" if market_price is not None else "missing_market_price",
            "valuation_timestamp_status": pos.valuation_timestamp_status,
            "corporate_action_status": row["corporate_action_status"] if row else pos.corporate_action_status,
            "valuation_status": pos.valuation_status,
            "total_return_chf": dec_text(pos.total_return_chf, "0.01") if pos.total_return_chf is not None else "",
            "total_return_status": "complete" if pos.total_return_chf is not None else "incomplete",
            "quality_warnings": ",".join(pos.quality_warnings),
            "data_quality_status": pos.data_quality_status,
        })
    return rows


def get_portfolio_positions_user(conn: Connection) -> list[dict[str, str]]:
    rows: list[dict[str, str]] = []
    for row in get_positions(conn):
        if row.get("asset_class", "").lower() not in {"stock", "equity", "etf"}:
            continue
        rows.append({
            "name": row.get("name", ""),
            "ticker": row.get("ticker", ""),
            "isin": row.get("isin", ""),
            "depot": row.get("account_name", ""),
            "assettyp": "ETF" if row.get("asset_class", "").lower() == "etf" else "Aktie",
            "menge": row.get("quantity", ""),
            "kurs": row.get("market_price_original") or "Preis noch nicht verfügbar",
            "kurswaehrung": row.get("price_currency", ""),
            "marktwert_chf": chf_text(row.get("market_value_chf")) if row.get("market_value_chf") else "nicht bewertet",
            "einstand_chf": chf_text(row.get("cost_basis_chf")) if row.get("cost_basis_chf") else "unvollständig",
            "p_l_chf": chf_text(row.get("total_return_chf")) if row.get("total_return_chf") else "nicht berechenbar",
            "status": translate_status(row.get("price_status") or row.get("data_quality_status")),
        })
    return rows


def get_portfolio_asset_class_chart(conn: Connection) -> list[dict[str, str]]:
    summary = get_command_center_summary(conn)
    return [
        {"label": "Crypto", "value_chf": summary["crypto_total_chf"]},
        {"label": "Aktien/ETF", "value_chf": summary["equity_etf_total_chf"]},
        {"label": "Cash", "value_chf": summary["cash_total_chf"]},
    ]


def get_platform_value_chart(conn: Connection) -> list[dict[str, str]]:
    totals: dict[str, Decimal] = {}
    for row in get_cash_overview(conn):
        totals[row["platform"]] = totals.get(row["platform"], Decimal("0")) + d(row.get("amount_chf"))
    for row in get_positions(conn):
        totals[row["platform"]] = totals.get(row["platform"], Decimal("0")) + d(row.get("market_value_chf"))
    crypto_total = d(get_crypto_summary_cards(conn)["total_value_chf"])
    if crypto_total:
        totals["Crypto Wallets"] = totals.get("Crypto Wallets", Decimal("0")) + crypto_total
    return [{"label": k, "value_chf": dec_text(v, "0.01")} for k, v in sorted(totals.items(), key=lambda item: item[1], reverse=True)]


def get_ledger_transactions(conn: Connection, limit: int = 200) -> list[dict[str, str]]:
    return rowdicts(conn.execute(
        """
        SELECT t.trade_date, p.name AS platform, a.account_name, t.transaction_type,
               i.name AS instrument, i.ticker, t.quantity, t.gross_amount_original,
               t.net_amount_original, t.currency_original, t.fx_status, t.quality_status,
               t.source_type, t.source_id
        FROM transactions t
        LEFT JOIN accounts a ON a.account_id=t.account_id
        LEFT JOIN platforms p ON p.platform_id=a.platform_id
        LEFT JOIN instruments i ON i.instrument_id=t.instrument_id
        ORDER BY t.trade_date DESC, t.created_at DESC LIMIT ?
        """, (limit,)
    ).fetchall())


def get_watchlist(conn: Connection) -> list[dict[str, str]]:
    return rowdicts(conn.execute("SELECT name, asset_class, reason, status, target_entry_price, target_entry_currency, investment_case, bear_case, next_review_date FROM watchlist ORDER BY name").fetchall())


def resolve_alert_entity_label(conn: Connection, entity_type: str | None, entity_id: str | None) -> str:
    if not entity_id:
        return entity_type or "Portfolio"
    if entity_type == "instrument" or str(entity_id).startswith("instrument-"):
        row = conn.execute("SELECT name, ticker FROM instruments WHERE instrument_id=?", (entity_id,)).fetchone()
        if row:
            return f"{row['name']} ({row['ticker']})" if row["ticker"] else row["name"]
        return "Unbekanntes Instrument"
    if entity_type == "crypto_asset":
        row = conn.execute("SELECT coin_name, symbol FROM crypto_assets WHERE asset_id=?", (entity_id,)).fetchone()
        if row:
            return f"{row['coin_name']} ({row['symbol']})"
    if entity_type == "crypto_wallet":
        row = conn.execute("SELECT wallet_name FROM crypto_wallets WHERE wallet_id=?", (entity_id,)).fetchone()
        if row:
            return row["wallet_name"]
    return entity_type or "Portfolio"


def alert_title(rule_id: str, message: str = "", entity: str = "") -> str:
    labels = {
        "missing_fx": "FX-Kurs fehlt",
        "stale_fx": "FX-Kurs veraltet",
        "missing_market_price": "Marktpreis fehlt",
        "stale_market_price": "Marktpreis veraltet",
        "missing_provider_symbol": "Marktdaten-Zuordnung fehlt",
        "manual_fx_override_required": "Manueller FX-Kurs nötig",
        "historical_fx_unavailable": "Historischer FX-Kurs fehlt",
        "candidate_review_required": "Mapping muss geprüft werden",
        "cost_basis_uncertain": "Einstand unvollständig",
        "snapshot_only": "Nur Snapshot-Historie vorhanden",
        "missing_crypto_price": "Crypto-Preis fehlt",
        "crypto_price_missing": "Crypto-Preis fehlt",
    }
    title = labels.get(rule_id, translate_status(rule_id))
    return f"{title} für {entity}" if entity else title


def alert_description(rule_id: str, message: str = "") -> str:
    descriptions = {
        "missing_fx": "Ein historischer FX-Kurs fehlt. CHF-Performance ist daher unvollständig.",
        "stale_fx": "Ein FX-Kurs ist veraltet und sollte aktualisiert werden.",
        "missing_market_price": "Für diese Position fehlt ein lokaler Marktpreis. Bewertung bleibt unvollständig.",
        "stale_market_price": "Der lokale Marktpreis ist veraltet. Bitte Preisupdate ausführen.",
        "missing_provider_symbol": "Für diese Position fehlt oder fehlt noch die bestätigte Marktdaten-Zuordnung.",
        "candidate_review_required": "Es gibt Marktdaten-Kandidaten, die manuell bestätigt werden müssen.",
        "cost_basis_uncertain": "Die Kaufhistorie oder Einstandsdaten sind unvollständig.",
        "snapshot_only": "Diese Position basiert auf einem Snapshot und nicht auf kompletter Historie.",
    }
    return descriptions.get(rule_id, message.replace("_", " ") if message else "Bitte im Review-Bereich prüfen.")


def get_alert_summary_cards(conn: Connection) -> dict[str, str]:
    rows = conn.execute("SELECT priority, COUNT(*) AS c FROM alerts WHERE status='active' GROUP BY priority").fetchall()
    out = {"kritisch": "0", "wichtig": "0", "info": "0"}
    for row in rows:
        key = "kritisch" if row["priority"] == "kritisch" else ("info" if row["priority"] == "info" else "wichtig")
        out[key] = str(int(out[key]) + int(row["c"] or 0))
    return out


def get_alert_cards(conn: Connection, limit: int = 5) -> list[dict[str, str]]:
    rows = conn.execute(
        """
        SELECT alert_id, priority, category, entity_type, entity_id, rule_id, message, occurrence_count, last_seen_at
        FROM alerts WHERE status='active'
        ORDER BY CASE priority WHEN 'kritisch' THEN 0 WHEN 'warnung' THEN 1 ELSE 2 END, last_seen_at DESC, created_at DESC
        LIMIT ?
        """,
        (limit,),
    ).fetchall()
    cards = []
    for row in rows:
        affected = resolve_alert_entity_label(conn, row["entity_type"], row["entity_id"])
        cards.append({
            "_alert_id": row["alert_id"],
            "title": alert_title(row["rule_id"], row["message"], affected),
            "description": alert_description(row["rule_id"], row["message"]),
            "affected": affected,
            "severity": translate_status(row["priority"]),
            "last_seen": date_text(row["last_seen_at"]),
            "actions": "Prüfen · Ignorieren · Details",
        })
    return cards


def get_alerts(conn: Connection) -> list[dict[str, str]]:
    return rowdicts(conn.execute(
        """
        SELECT priority, status, category, entity_type, entity_id, rule_id, message,
               occurrence_count, last_seen_at, resolved_at, muted_until, fingerprint, dedup_key, created_at
        FROM alerts
        ORDER BY CASE priority WHEN 'kritisch' THEN 0 WHEN 'warnung' THEN 1 ELSE 2 END, status, created_at DESC
        """
    ).fetchall())


def get_audit_events(conn: Connection, limit: int = 200) -> list[dict[str, str]]:
    rows = conn.execute(
        """
        SELECT timestamp, source, action, entity_type, entity_id, old_values_json, new_values_json,
               confirmed, created_by, quality_status
        FROM audit_log ORDER BY timestamp DESC LIMIT ?
        """, (limit,)
    ).fetchall()
    out: list[dict[str, str]] = []
    for row in rows:
        item = {k: ("" if row[k] is None else str(row[k])) for k in row.keys()}
        for key in ("old_values_json", "new_values_json"):
            if item[key]:
                try:
                    item[key] = json.dumps(json.loads(item[key]), ensure_ascii=False, indent=2, sort_keys=True)
                except json.JSONDecodeError:
                    pass
        out.append(item)
    return out


def get_data_quality(conn: Connection) -> list[dict[str, str]]:
    items: list[dict[str, str]] = []
    checks = [
        ("fehlende FX", "SELECT COUNT(*) AS c FROM transactions WHERE fx_status='missing'", "kritisch"),
        ("fehlende/stale FX Alerts", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id IN ('missing_fx','stale_fx','historical_fx_unavailable','manual_override_required')", "kritisch"),
        ("historical FX unavailable", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='historical_fx_unavailable'", "kritisch"),
        ("manual override required", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='manual_override_required'", "kritisch"),
        ("fehlende Provider-Symbole", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='missing_provider_symbol'", "warnung"),
        ("provider lookup unavailable", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='provider_lookup_unavailable'", "warnung"),
        ("ambiguous_provider_mapping", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='ambiguous_provider_mapping'", "warnung"),
        ("multiple_listings_same_isin", "SELECT COUNT(*) AS c FROM (SELECT isin FROM instrument_price_mapping_candidates WHERE COALESCE(isin,'')<>'' AND review_status IN ('proposed','needs_manual_review') GROUP BY isin HAVING COUNT(DISTINCT COALESCE(candidate_exchange,'') || ':' || COALESCE(candidate_currency,''))>1)", "warnung"),
        ("currency_mismatch", "SELECT COUNT(*) AS c FROM instrument_price_mapping_candidates WHERE review_status IN ('proposed','needs_manual_review') AND risk_flags LIKE '%currency_mismatch%'", "warnung"),
        ("hedge_status_unknown", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='hedge_status_unknown'", "warnung"),
        ("instrument_status_unknown", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='instrument_status_unknown'", "warnung"),
        ("provider_symbol_missing", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='missing_provider_symbol'", "warnung"),
        ("candidate_review_required", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='candidate_review_required'", "warnung"),
        ("valuation_not_ready", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id IN ('missing_provider_symbol','missing_market_price','missing_fx','hedge_status_unknown','instrument_status_unknown','candidate_review_required')", "warnung"),
        ("instrument status unknown", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='instrument_status_unknown'", "warnung"),
        ("manual FX override required", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id IN ('manual_override_required','manual_fx_override_required')", "kritisch"),
        ("fehlende Market Prices", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='missing_market_price'", "warnung"),
        ("stale Market Prices", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='stale_market_price'", "warnung"),
        ("ambiguous Instrument Mapping", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='ambiguous_instrument_mapping'", "warnung"),
        ("price currency mismatch", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='price_currency_mismatch'", "warnung"),
        ("hedge status unknown", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='hedge_status_unknown'", "warnung"),
        ("stale valuation", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='stale_valuation'", "warnung"),
        ("corporate action suspected", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='corporate_action_suspected'", "warnung"),
        ("split/corporate action review", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='split_or_corporate_action_review_required'", "warnung"),
        ("delisted/suspended", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='delisted_or_suspended'", "warnung"),
        ("snapshot_only Alerts", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='snapshot_only'", "info"),
        ("cost_basis_uncertain Alerts", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='cost_basis_uncertain'", "warnung"),
        ("fehlende CoinGecko-ID", "SELECT COUNT(*) AS c FROM crypto_assets WHERE coingecko_id IS NULL OR coingecko_id='' ", "warnung"),
        ("fehlende Crypto-Preise", "SELECT COUNT(*) AS c FROM crypto_assets ca LEFT JOIN crypto_prices cp ON cp.asset_id=ca.asset_id WHERE cp.crypto_price_id IS NULL", "warnung"),
        ("fehlende ISIN", "SELECT COUNT(*) AS c FROM instrument_mappings WHERE asset_class IN ('equity','etf') AND (isin IS NULL OR isin='') AND mapping_status!='ignored'", "warnung"),
        ("Catalog Instrumente ohne ISIN", "SELECT COUNT(*) AS c FROM instrument_catalog_entries WHERE asset_class IN ('equity','stock','etf') AND (isin IS NULL OR isin='')", "warnung"),
        ("Catalog Ticker ohne Exchange", "SELECT COUNT(*) AS c FROM instrument_catalog_entries WHERE COALESCE(ticker,'')<>'' AND COALESCE(exchange,'')=''", "warnung"),
        ("Catalog mehrere Listings gleicher ISIN", "SELECT COUNT(*) AS c FROM (SELECT isin FROM instrument_catalog_entries WHERE COALESCE(isin,'')<>'' GROUP BY isin HAVING COUNT(DISTINCT COALESCE(exchange,'') || ':' || COALESCE(trading_currency,''))>1)", "info"),
        ("Manual Review needed", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id='manual_review_needed'", "warnung"),
        ("Mapping-Kandidaten unbestätigt", "SELECT COUNT(*) AS c FROM instrument_price_mapping_candidates WHERE review_status IN ('proposed','needs_manual_review')", "warnung"),
        ("Instrumente mit mehreren Mapping-Kandidaten", "SELECT COUNT(*) AS c FROM (SELECT instrument_id FROM instrument_price_mapping_candidates WHERE review_status IN ('proposed','needs_manual_review') GROUP BY instrument_id HAVING COUNT(*) > 1)", "warnung"),
        ("Valuation not ready", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id IN ('missing_provider_symbol','missing_market_price','missing_fx','hedge_status_unknown','instrument_status_unknown')", "warnung"),
        ("fehlende FX in Import-Reviews", "SELECT COUNT(*) AS c FROM instrument_mappings WHERE quality_flags_json LIKE '%missing_fx%' AND mapping_status!='ignored'", "warnung"),
        ("snapshot_only Positionen", "SELECT COUNT(*) AS c FROM instrument_mappings WHERE quality_flags_json LIKE '%snapshot_only%' AND mapping_status!='ignored'", "info"),
        ("cost_basis_uncertain", "SELECT COUNT(*) AS c FROM instrument_mappings WHERE quality_flags_json LIKE '%cost_basis_uncertain%' AND mapping_status!='ignored'", "warnung"),
        ("Instrument Import Kandidaten unbestätigt", "SELECT COUNT(*) AS c FROM instrument_import_candidates WHERE mapping_status NOT IN ('confirmed','rejected')", "warnung"),
        ("Instrument Import Kandidaten ohne ISIN", "SELECT COUNT(*) AS c FROM instrument_import_candidates WHERE COALESCE(proposed_isin, raw_isin, '')='' AND mapping_status!='rejected'", "warnung"),
        ("aktive Alerts", "SELECT COUNT(*) AS c FROM alerts WHERE status='active'", "warnung"),
        ("resolved Alerts", "SELECT COUNT(*) AS c FROM alerts WHERE status='resolved'", "ok"),
        ("alte aktive Alerts", "SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND last_seen_at < datetime('now', '-7 days')", "info"),
    ]
    for label, sql, severity in checks:
        row = conn.execute(sql).fetchone()
        count = int(row["c"] or 0) if row else 0
        sev = severity if count else "ok"
        items.append({"check": label, "count": str(count), "severity": sev})
    for flag_row in get_review_quality_flag_counts(conn):
        count = int(flag_row["open_items"] or 0)
        items.append({"check": f"Review Flag: {flag_row['quality_flag']}", "count": str(count), "severity": "warnung" if count else "ok"})
    for ready in get_review_readiness_counts(conn):
        count = int(ready["count"] or 0)
        sev = "ok" if ready["readiness_status"] in {"ready_for_import", "ignored"} else ("warnung" if count else "ok")
        items.append({"check": f"Review Readiness: {ready['readiness_status']}", "count": str(count), "severity": sev})
    for source in get_ready_for_import_by_source(conn):
        if int(source["total"] or 0) and int(source["ready_for_import"] or 0) == 0:
            items.append({"check": f"Quelle ohne ready_for_import: {source['source_platform']}", "count": source["total"], "severity": "warnung"})
    missing_snapshot = conn.execute("SELECT COUNT(*) AS c FROM broker_import_dry_runs WHERE snapshot_date_status='missing'").fetchone()
    count = int(missing_snapshot["c"] or 0) if missing_snapshot else 0
    items.append({"check": "Dry Runs ohne Snapshot Date", "count": str(count), "severity": "warnung" if count else "ok"})
    return items


def get_instrument_catalog_entries(conn: Connection, query: str | None = None, asset_class: str | None = None) -> list[dict[str, str]]:
    if query:
        from jarvis_finance.market_data.catalog import search_local_catalog
        return [r.__dict__ for r in search_local_catalog(conn, query, asset_class)]
    sql = """
        SELECT catalog_entry_id, asset_class, name, isin, ticker, exchange, trading_currency,
               instrument_currency, provider, provider_symbol, provider_market, hedge_status,
               instrument_status, valuation_policy, source, source_confidence, last_verified_at
        FROM instrument_catalog_entries
    """
    params: list[object] = []
    if asset_class:
        sql += " WHERE asset_class=?"
        params.append(asset_class.lower())
    sql += " ORDER BY name, exchange, trading_currency LIMIT 100"
    return rowdicts(conn.execute(sql, tuple(params)).fetchall())


def get_instrument_catalog_summary(conn: Connection) -> dict[str, str]:
    total = conn.execute("SELECT COUNT(*) AS c FROM instrument_catalog_entries").fetchone()["c"]
    multi = conn.execute("SELECT COUNT(*) AS c FROM (SELECT isin FROM instrument_catalog_entries WHERE COALESCE(isin,'')<>'' GROUP BY isin HAVING COUNT(DISTINCT COALESCE(exchange,'') || ':' || COALESCE(trading_currency,''))>1)").fetchone()["c"]
    no_provider = conn.execute("SELECT COUNT(*) AS c FROM instrument_catalog_entries WHERE COALESCE(provider_symbol,'')='' ").fetchone()["c"]
    manual = conn.execute("SELECT COUNT(*) AS c FROM alerts WHERE status='active' AND rule_id IN ('manual_review_needed','missing_provider_symbol','ambiguous_provider_mapping')").fetchone()["c"]
    return {"catalog_entries": str(total or 0), "multiple_listings": str(multi or 0), "without_provider_symbol": str(no_provider or 0), "manual_review_required": str(manual or 0)}


def get_instrument_mapping_review(conn: Connection) -> list[dict[str, str]]:
    rows = conn.execute(
        """
        SELECT i.instrument_id, i.isin, i.name, i.ticker, i.exchange, i.currency,
               i.hedge_status, i.instrument_status, i.valuation_policy,
               COALESCE(m.mapping_status, 'missing_provider_symbol') AS mapping_status,
               COALESCE(m.provider, '') AS provider,
               COALESCE(m.provider_symbol, '') AS provider_symbol,
               COUNT(c.candidate_id) AS candidate_count
        FROM instruments i
        JOIN transactions t ON t.instrument_id=i.instrument_id
        LEFT JOIN instrument_price_mappings m ON m.instrument_id=i.instrument_id AND m.mapping_status='mapped'
        LEFT JOIN instrument_price_mapping_candidates c ON c.instrument_id=i.instrument_id AND c.review_status IN ('proposed','needs_manual_review')
        WHERE t.transaction_type='initial_position_snapshot'
          AND t.source_type='broker_import_reviewed_snapshot'
          AND COALESCE(t.is_voided,0)=0
        GROUP BY i.instrument_id
        ORDER BY i.isin, i.name
        """
    ).fetchall()
    return rowdicts(rows)


def get_instrument_import_candidates(conn: Connection, status: str | None = None) -> list[dict[str, str]]:
    sql = """
        SELECT candidate_id, source_file, platform, account, source_row_number,
               raw_name, raw_ticker, raw_isin, raw_currency,
               CASE WHEN COALESCE(raw_quantity,'')<>'' THEN '<present>' ELSE '' END AS raw_quantity_present,
               raw_asset_class, extracted_by, extraction_confidence,
               proposed_name, proposed_ticker, proposed_isin, proposed_exchange,
               proposed_currency, proposed_asset_class, provider, provider_symbol,
               mapping_status, review_note, created_at, updated_at
        FROM instrument_import_candidates
    """
    params: list[object] = []
    if status:
        sql += " WHERE mapping_status=?"
        params.append(status)
    sql += " ORDER BY CASE mapping_status WHEN 'needs_manual_review' THEN 0 WHEN 'probable' THEN 1 WHEN 'exact_isin_match' THEN 2 WHEN 'confirmed' THEN 3 ELSE 4 END, platform, source_row_number"
    return rowdicts(conn.execute(sql, tuple(params)).fetchall())


def get_instrument_import_candidate_summary(conn: Connection) -> dict[str, str]:
    def c(where: str) -> str:
        row = conn.execute(f"SELECT COUNT(*) AS n FROM instrument_import_candidates WHERE {where}").fetchone()
        return str(row["n"] or 0)
    return {
        "candidates_total": c("1=1"),
        "with_isin": c("COALESCE(proposed_isin, raw_isin, '')<>''"),
        "without_isin": c("COALESCE(proposed_isin, raw_isin, '')=''"),
        "with_ticker": c("COALESCE(proposed_ticker, raw_ticker, '')<>''"),
        "with_currency": c("COALESCE(proposed_currency, raw_currency, '')<>''"),
        "exact_isin_match": c("mapping_status='exact_isin_match'"),
        "probable": c("mapping_status='probable'"),
        "needs_manual_review": c("mapping_status='needs_manual_review'"),
        "confirmed": c("mapping_status='confirmed'"),
        "blocked": c("mapping_status!='confirmed'"),
    }


def get_instrument_mapping_candidates(conn: Connection, instrument_id: str | None = None) -> list[dict[str, str]]:
    if instrument_id:
        rows = conn.execute(
            """
            SELECT candidate_id, instrument_id, isin, candidate_provider, candidate_provider_symbol,
                   candidate_exchange, candidate_currency, candidate_name, candidate_asset_class,
                   candidate_is_hedged, candidate_hedged_to_currency, candidate_hedge_status,
                   candidate_instrument_status, candidate_valuation_policy, confidence,
                   ranking_score, ranking_reason, risk_flags, recommended_action,
                   CASE
                       WHEN recommended_action='select_candidate' OR ranking_score >= 75 THEN 'wahrscheinlich passend'
                       WHEN recommended_action='reject_candidate' OR ranking_score < 40 THEN 'wahrscheinlich falsch'
                       ELSE 'unsicher'
                   END AS ranking_mark,
                   candidate_currency AS closing_currency,
                   evidence_source, evidence_note, review_status, created_at
            FROM instrument_price_mapping_candidates WHERE instrument_id=?
            ORDER BY ranking_score DESC, CASE confidence WHEN 'high' THEN 0 WHEN 'medium' THEN 1 ELSE 2 END, candidate_provider
            """,
            (instrument_id,),
        ).fetchall()
    else:
        rows = conn.execute(
            """
            SELECT candidate_id, instrument_id, isin, candidate_provider, candidate_provider_symbol,
                   candidate_exchange, candidate_currency, candidate_name, candidate_asset_class,
                   candidate_is_hedged, candidate_hedged_to_currency, candidate_hedge_status,
                   candidate_instrument_status, candidate_valuation_policy, confidence,
                   ranking_score, ranking_reason, risk_flags, recommended_action,
                   CASE
                       WHEN recommended_action='select_candidate' OR ranking_score >= 75 THEN 'wahrscheinlich passend'
                       WHEN recommended_action='reject_candidate' OR ranking_score < 40 THEN 'wahrscheinlich falsch'
                       ELSE 'unsicher'
                   END AS ranking_mark,
                   candidate_currency AS closing_currency,
                   evidence_source, evidence_note, review_status, created_at
            FROM instrument_price_mapping_candidates
            ORDER BY instrument_id, ranking_score DESC, CASE confidence WHEN 'high' THEN 0 WHEN 'medium' THEN 1 ELSE 2 END, candidate_provider
            """
        ).fetchall()
    return rowdicts(rows)


def get_fx_override_requirements(conn: Connection) -> list[dict[str, str]]:
    rows = conn.execute(
        """
        SELECT t.currency_original AS base_currency, 'CHF' AS quote_currency, t.trade_date AS rate_date,
               COUNT(DISTINCT t.instrument_id) AS instrument_count,
               CASE WHEN fr.fx_rate_id IS NULL THEN 'missing_fx' ELSE fr.quality_status END AS fx_status
        FROM transactions t
        LEFT JOIN fx_rates fr ON fr.base_currency=t.currency_original AND fr.quote_currency='CHF' AND fr.rate_date=t.trade_date
        WHERE t.transaction_type='initial_position_snapshot'
          AND t.source_type='broker_import_reviewed_snapshot'
          AND COALESCE(t.is_voided,0)=0
          AND t.currency_original IS NOT NULL
          AND t.currency_original <> 'CHF'
        GROUP BY t.currency_original, t.trade_date, fx_status
        ORDER BY t.currency_original, t.trade_date
        """
    ).fetchall()
    return rowdicts(rows)


def get_import_wizard_summary(conn: Connection) -> dict[str, str]:
    from jarvis_finance.imports.broker_mapping import build_import_wizard_summary

    return build_import_wizard_summary(conn)


def get_manual_review_items(conn: Connection) -> list[dict[str, str]]:
    from jarvis_finance.imports.broker_mapping import get_manual_review_queue

    return get_manual_review_queue(conn)


def get_broker_dry_runs(conn: Connection, limit: int = 20) -> list[dict[str, str]]:
    from jarvis_finance.imports.broker_mapping import get_dry_run_sessions

    return get_dry_run_sessions(conn, limit=limit)


def get_safe_mode_status(conn: Connection) -> dict[str, str]:
    backup = conn.execute("SELECT created_at FROM audit_log WHERE action LIKE '%backup%' ORDER BY created_at DESC LIMIT 1").fetchone()
    return {
        "db_mode": "Runtime",
        "write_mode": "Read-only",
        "last_backup_at": backup["created_at"] if backup else "nicht verfügbar",
        "git_safety_status": "zuletzt OK / CLI prüfen",
        "data_location": "Echte Daten liegen außerhalb Git",
    }


def get_broker_review_items(conn: Connection, limit: int = 100, **filters) -> list[dict[str, str]]:
    from jarvis_finance.imports.broker_mapping import get_broker_review_items as _items

    return _items(conn, limit=limit, **filters)


def get_review_readiness_counts(conn: Connection) -> list[dict[str, str]]:
    from jarvis_finance.imports.broker_mapping import get_review_readiness_counts as _counts

    return _counts(conn)


def get_ready_for_import_by_source(conn: Connection) -> list[dict[str, str]]:
    from jarvis_finance.imports.broker_mapping import get_ready_for_import_by_source as _counts

    return _counts(conn)


def get_review_quality_flag_counts(conn: Connection) -> list[dict[str, str]]:
    from jarvis_finance.imports.broker_mapping import aggregate_review_quality_flags

    return [{"quality_flag": k, "open_items": str(v)} for k, v in sorted(aggregate_review_quality_flags(conn).items())]


def equity_etf_user_label(summary: dict[str, str]) -> str:
    value = d(summary.get("equity_etf_total_chf"))
    return chf_text(value) if value > 0 else "noch nicht bewertet"


def equity_etf_is_user_ready(summary: dict[str, str]) -> bool:
    return d(summary.get("equity_etf_total_chf")) > 0 and summary.get("data_quality_status") == "ok"


def get_top_user_hints(conn: Connection, limit: int = 3) -> list[str]:
    hints: list[str] = []
    summary = get_command_center_summary(conn)
    if equity_etf_user_label(summary) == "noch nicht bewertet":
        hints.append("Aktien/ETF: Bewertung unvollständig")
    crypto = get_crypto_summary_cards(conn)
    if d(crypto.get("coins_without_price")) > 0:
        hints.append("Crypto-Preise fehlen für einzelne Coins")
    elif d(crypto.get("stale_price_count")) > 0:
        hints.append("Crypto-Preise teilweise veraltet")
    else:
        hints.append("Crypto-Preise aktuell")
    for alert in get_alert_cards(conn, limit=limit):
        text = alert["title"]
        if not any(marker in text for marker in ("asset_id", "wallet_id", "instrument-", "rule_id", "entity_id")):
            hints.append(text)
        if len(hints) >= limit:
            break
    return hints[:limit]


def compact_chart_rows(rows: list[dict[str, str]], *, limit: int = 10) -> list[dict[str, str]]:
    ordered = sorted(rows, key=lambda row: d(row.get("value_chf")), reverse=True)
    if len(ordered) <= limit:
        return ordered
    head = ordered[:limit]
    other = sum((d(row.get("value_chf")) for row in ordered[limit:]), Decimal("0"))
    if other > 0:
        head.append({"label": "Andere", "value_chf": dec_text(other, "0.01")})
    return head


def get_command_center_summary(conn: Connection) -> dict[str, str]:
    cash_rows = get_cash_overview(conn)
    crypto_summary = get_crypto_summary_cards(conn)
    positions = [r for r in get_positions(conn) if r.get("asset_class", "").lower() in {"stock", "equity", "etf"} and r.get("instrument_status", "").lower() != "inactive"]
    cash_total = sum(d(r["amount_chf"]) for r in cash_rows if r["amount_chf"])
    crypto_total = d(crypto_summary["total_value_chf"])
    stock_rows = [r for r in positions if r.get("asset_class", "").lower() in {"stock", "equity"}]
    etf_rows = [r for r in positions if r.get("asset_class", "").lower() == "etf"]
    stock_total = sum(d(r["market_value_chf"]) for r in stock_rows if r.get("market_value_chf"))
    etf_total = sum(d(r["market_value_chf"]) for r in etf_rows if r.get("market_value_chf"))
    equity_total = stock_total + etf_total
    missing_price_count = sum(1 for r in positions if not r.get("market_price_original"))
    missing_fx_count = sum(1 for r in positions if r.get("fx_status") == "missing_fx")
    incomplete_cost_count = sum(1 for r in positions if "cost_basis_uncertain" in (r.get("quality_warnings") or "") or not r.get("cost_basis_chf"))
    unvalued_count = missing_price_count + missing_fx_count
    latest_equity_price = max((r.get("price_date") or "" for r in positions), default="")
    alert_counts = conn.execute("SELECT priority, COUNT(*) AS c FROM alerts WHERE status='active' GROUP BY priority").fetchall()
    critical = sum(int(r["c"] or 0) for r in alert_counts if r["priority"] == "kritisch")
    quality = get_data_quality(conn)
    dq = "kritisch" if any(i["severity"] == "kritisch" and i["count"] != "0" for i in quality) else ("warnung" if any(i["severity"] == "warnung" and i["count"] != "0" for i in quality) else "ok")
    total = cash_total + crypto_total + equity_total
    return {
        "base_currency": "CHF",
        "cash_total_chf": dec_text(cash_total, "0.01"),
        "crypto_total_chf": dec_text(crypto_total, "0.01"),
        "stock_total_chf": dec_text(stock_total, "0.01"),
        "etf_total_chf": dec_text(etf_total, "0.01"),
        "equity_etf_total_chf": dec_text(equity_total, "0.01"),
        "equity_position_count": str(len(positions)),
        "unvalued_position_count": str(unvalued_count),
        "missing_price_count": str(missing_price_count),
        "missing_fx_count": str(missing_fx_count),
        "incomplete_cost_count": str(incomplete_cost_count),
        "total_portfolio_chf": dec_text(total, "0.01"),
        "total_value_chf": dec_text(total, "0.01"),
        "critical_alert_count": str(critical),
        "active_alerts": ", ".join(f"{r['priority']}: {r['c']}" for r in alert_counts) or "0",
        "data_quality_status": dq,
        "last_price_update_at": max([crypto_summary.get("last_price_update_at", ""), latest_equity_price]),
        "crypto_price_data_as_of": crypto_summary.get("last_price_update_at", ""),
        "equity_price_data_as_of": latest_equity_price,
    }