spb/crypto-trade-bot

1"""Credential preflight: discover what the configured keys can do without ever placing a live order."""23from __future__ import annotations45import time6from decimal import Decimal7from typing import Any89from ktrade.api.deps import AppState10from ktrade.kraken.rest import KrakenAPIError, KrakenREST111213async def _probe(rest: KrakenREST, name: str, coro) -> dict[str, Any]:  # noqa: ANN00114    try:15        await coro16        return {"ok": True}17    except KrakenAPIError as e:18        return {"ok": False, "permission_denied": e.is_permission, "error": str(e)}19    except Exception as e:  # noqa: BLE00120        return {"ok": False, "error": str(e)}212223async def run_preflight(st: AppState) -> dict[str, Any]:24    out: dict[str, Any] = {"checked_at": time.time(), "data_key": None, "trade_key": None}25    s = st.settings26    if s.has_data_key():27        rest = st.rest28        d: dict[str, Any] = {}29        d["balance"] = await _probe(rest, "balance", rest.balance())30        d["open_orders"] = await _probe(rest, "open_orders", rest.open_orders())31        d["trade_volume"] = await _probe(rest, "trade_volume", rest.trade_volume(["XXBTZUSD"]))32        d["ws_token"] = await _probe(rest, "ws_token", rest.websockets_token())33        # Trade-permission probe: validate-only order that could never execute (price = 10% of market, min size,34        # expires in 5 s). The data key SHOULD fail this with Permission denied (least privilege).35        d["trade_permission"] = await _trade_probe(rest)36        if d["trade_permission"].get("has_trade_permission"):37            d["warning"] = "The data/account key has order permission — least-privilege violated; create a read-only key for data."38        out["data_key"] = d39    if s.has_trade_key():40        trade_rest = KrakenREST(s.kraken_rest_url, s.kraken_trade_api_key, s.kraken_trade_api_secret, s.kraken_tier)41        try:42            out["trade_key"] = {"trade_permission": await _trade_probe(trade_rest), "note": "live mode is disabled (feature flag + UI activation required)"}43        finally:44            await trade_rest.aclose()45    return out464748async def _trade_probe(rest: KrakenREST) -> dict[str, Any]:49    try:50        tick = await rest.ticker(["XXBTZUSD"])51        last = Decimal(next(iter(tick.values()))["c"][0])52        price = (last * Decimal("0.10")).quantize(Decimal("0.1"))53        r = await rest.add_order(pair="XXBTZUSD", side="buy", ordertype="limit", volume="0.0001", price=str(price), timeinforce="GTC", validate=True)54        return {"has_trade_permission": True, "validated": r.get("descr")}55    except KrakenAPIError as e:56        if e.is_permission:57            return {"has_trade_permission": False, "error": str(e)}58        # any other validation error means the permission check passed59        return {"has_trade_permission": True, "error": str(e)}60    except Exception as e:  # noqa: BLE00161        return {"has_trade_permission": None, "error": str(e)}62