Cerbero-Bite/src/cerbero_bite/gui/live_data.py

"""Live MCP fetch for the GUI (saldi exchange, FX rate).

The original architecture forbade the GUI from calling MCP services
(`docs/11-gui-streamlit.md`). For the "Saldi exchange" panel that
constraint is relaxed: the dashboard fetches balances on demand,
caches the result with Streamlit's TTL cache, and never holds the
async client open between renders. Every fetch is a one-shot:

* read endpoints + token from env (same path used by the CLI),
* spin up a short-lived ``httpx.AsyncClient``,
* query Deribit `get_account_summary` for both ``USDC`` and ``USDT``,
* query Hyperliquid `get_account_summary` (returns ``spot_usdc``,
  ``perps_equity`` etc.),
* query Macro `get_asset_price("EURUSD")` for FX,
* close the client and return a frozen dataclass to the page.

If a single exchange call fails the row is filled with ``error=...``
and the others are still rendered.
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from datetime import UTC, datetime
from decimal import Decimal
from typing import Any

import httpx

from cerbero_bite.clients._base import HttpToolClient
from cerbero_bite.clients.deribit import DeribitClient
from cerbero_bite.clients.hyperliquid import HyperliquidClient
from cerbero_bite.clients.macro import MacroClient
from cerbero_bite.config.mcp_endpoints import load_endpoints, load_token

__all__ = [
    "BalanceRow",
    "BalancesSnapshot",
    "fetch_balances_sync",
]


_DERIBIT_CURRENCIES = ("USDC", "USDT")


@dataclass(frozen=True)
class BalanceRow:
    """One row of the balances table."""

    exchange: str
    currency: str
    equity: Decimal | None
    available: Decimal | None
    unrealized_pnl: Decimal | None
    error: str | None = None


@dataclass(frozen=True)
class BalancesSnapshot:
    """Result of one fetch_balances call (rows + meta)."""

    rows: list[BalanceRow]
    eur_usd_rate: Decimal | None
    fetched_at: datetime
    fx_error: str | None = None

    def total_usd(self) -> Decimal:
        total = Decimal(0)
        for r in self.rows:
            if r.equity is not None:
                total += r.equity
        return total

    def total_eur(self) -> Decimal | None:
        if self.eur_usd_rate is None or self.eur_usd_rate <= 0:
            return None
        return self.total_usd() / self.eur_usd_rate


def _decimal_or_none(value: Any) -> Decimal | None:
    if value is None:
        return None
    try:
        return Decimal(str(value))
    except (ValueError, ArithmeticError):
        return None


def _resolve_token() -> str:
    """Read the MCP bearer token from the environment.

    The token is sourced from ``CERBERO_BITE_MCP_TOKEN``; on Cerbero MCP
    V2 the same single token decides whether the upstream environment
    is testnet or mainnet.
    """
    return load_token()


async def _fetch_deribit_currency(
    deribit: DeribitClient, currency: str
) -> BalanceRow:
    try:
        summary = await deribit.get_account_summary(currency=currency)
    except Exception as exc:
        return BalanceRow(
            exchange="deribit",
            currency=currency,
            equity=None,
            available=None,
            unrealized_pnl=None,
            error=f"{type(exc).__name__}: {exc}",
        )
    # Cerbero MCP V2 returns HTTP 200 with a soft ``error`` field when
    # the upstream Deribit call failed (e.g. invalid credentials). Treat
    # that as a row-level failure so the dashboard surfaces the cause
    # instead of showing a misleading equity=0.
    soft_error = summary.get("error")
    if soft_error:
        return BalanceRow(
            exchange="deribit",
            currency=currency,
            equity=None,
            available=None,
            unrealized_pnl=None,
            error=str(soft_error),
        )
    return BalanceRow(
        exchange="deribit",
        currency=currency,
        equity=_decimal_or_none(summary.get("equity")),
        available=_decimal_or_none(summary.get("available_funds")),
        unrealized_pnl=_decimal_or_none(summary.get("unrealized_pnl")),
    )


async def _fetch_hyperliquid(hl: HyperliquidClient) -> list[BalanceRow]:
    try:
        summary = await hl.get_account_summary()
    except Exception as exc:
        return [
            BalanceRow(
                exchange="hyperliquid",
                currency="USDC",
                equity=None,
                available=None,
                unrealized_pnl=None,
                error=f"{type(exc).__name__}: {exc}",
            )
        ]
    rows: list[BalanceRow] = [
        BalanceRow(
            exchange="hyperliquid",
            currency="USDC",
            equity=_decimal_or_none(summary.get("equity")),
            available=_decimal_or_none(summary.get("available_balance")),
            unrealized_pnl=_decimal_or_none(summary.get("unrealized_pnl")),
        )
    ]
    # Hyperliquid spot may also hold USDT; the MCP server exposes it
    # under spot_usdt when present. Add a row only if the field is there
    # so we don't render a confusing "0.00" against an asset the account
    # never held.
    spot_usdt = summary.get("spot_usdt")
    if spot_usdt is not None:
        rows.append(
            BalanceRow(
                exchange="hyperliquid",
                currency="USDT",
                equity=_decimal_or_none(spot_usdt),
                available=_decimal_or_none(spot_usdt),
                unrealized_pnl=Decimal(0),
            )
        )
    return rows


async def _fetch_balances_async(*, timeout_s: float = 8.0) -> BalancesSnapshot:
    endpoints = load_endpoints()
    token = _resolve_token()

    async with httpx.AsyncClient(timeout=timeout_s) as http_client:

        def _client(service: str) -> HttpToolClient:
            return HttpToolClient(
                service=service,
                base_url=endpoints.for_service(service),
                token=token,
                timeout_s=timeout_s,
                retry_max=1,
                client=http_client,
            )

        deribit = DeribitClient(_client("deribit"), _client("unified"))
        hl = HyperliquidClient(_client("hyperliquid"))
        macro = MacroClient(_client("macro"))

        deribit_results, hl_rows, (fx_value, fx_error) = await asyncio.gather(
            asyncio.gather(
                *(
                    _fetch_deribit_currency(deribit, cur)
                    for cur in _DERIBIT_CURRENCIES
                )
            ),
            _fetch_hyperliquid(hl),
            _fetch_eur_usd(macro),
        )
        deribit_rows = list(deribit_results)

    return BalancesSnapshot(
        rows=[*deribit_rows, *hl_rows],
        eur_usd_rate=fx_value,
        fetched_at=datetime.now(UTC),
        fx_error=fx_error,
    )


async def _fetch_eur_usd(
    macro: MacroClient,
) -> tuple[Decimal | None, str | None]:
    try:
        rate = await macro.eur_usd_rate()
    except Exception as exc:
        return None, f"{type(exc).__name__}: {exc}"
    return rate, None


def fetch_balances_sync(*, timeout_s: float = 8.0) -> BalancesSnapshot:
    """Sync wrapper for Streamlit pages (which run in a sync context)."""
    return asyncio.run(_fetch_balances_async(timeout_s=timeout_s))