spb/crypto-trade-bot

1from __future__ import annotations23from decimal import Decimal4from typing import Literal56from pydantic import BaseModel, Field78# REST AssetPairs still reports v1 names (XBT/USD, XDG/USD); WebSocket v2 uses BTC/USD, DOGE/USD.9_V2_RENAMES = {"XBT": "BTC", "XDG": "DOGE"}101112def normalize_symbol(wsname: str) -> str:13    base, _, quote = wsname.partition("/")14    return f"{_V2_RENAMES.get(base, base)}/{_V2_RENAMES.get(quote, quote)}" if quote else wsname151617PairStatus = Literal["online", "cancel_only", "post_only", "limit_only", "reduce_only", "unknown"]181920class PairMeta(BaseModel):21    """Tradable pair metadata from /0/public/AssetPairs. Exchange capabilities are *discovered*, not assumed."""2223    pair_id: str = Field(description="Kraken internal key, e.g. XXBTZUSD")24    altname: str25    wsname: str = Field(description="WebSocket v2 symbol, e.g. BTC/USD")26    base: str27    quote: str28    status: PairStatus = "unknown"29    ordermin: Decimal = Decimal("0")30    costmin: Decimal = Decimal("0")31    tick_size: Decimal = Decimal("0")32    pair_decimals: int = 833    lot_decimals: int = 834    cost_decimals: int = 835    lot_multiplier: int = 136    leverage_buy: list[int] = Field(default_factory=list)37    leverage_sell: list[int] = Field(default_factory=list)38    long_position_limit: int | None = None39    short_position_limit: int | None = None40    aclass_base: str = "currency"41    aclass_quote: str = "currency"42    execution_venue: str | None = None4344    @classmethod45    def from_kraken(cls, key: str, d: dict) -> PairMeta:46        return cls(47            pair_id=key,48            altname=d.get("altname", key),49            wsname=normalize_symbol(d.get("wsname", d.get("altname", key))),50            base=d.get("base", ""),51            quote=d.get("quote", ""),52            status=d.get("status", "unknown") if d.get("status") in ("online", "cancel_only", "post_only", "limit_only", "reduce_only") else "unknown",53            ordermin=Decimal(str(d.get("ordermin", "0"))),54            costmin=Decimal(str(d.get("costmin", "0"))),55            tick_size=Decimal(str(d.get("tick_size", "0"))),56            pair_decimals=int(d.get("pair_decimals", 8)),57            lot_decimals=int(d.get("lot_decimals", 8)),58            cost_decimals=int(d.get("cost_decimals", 8)),59            lot_multiplier=int(d.get("lot_multiplier", 1)),60            leverage_buy=[int(x) for x in d.get("leverage_buy", [])],61            leverage_sell=[int(x) for x in d.get("leverage_sell", [])],62            long_position_limit=d.get("long_position_limit"),63            short_position_limit=d.get("short_position_limit"),64            aclass_base=d.get("aclass_base", "currency"),65            aclass_quote=d.get("aclass_quote", "currency"),66            execution_venue=d.get("execution_venue"),67        )686970class Candle(BaseModel):71    """OHLC bar. `ts` is the bar OPEN time (UTC seconds). The bar is only *known* at ts + interval*60."""7273    ts: int74    open: float75    high: float76    low: float77    close: float78    vwap: float79    volume: float80    count: int8182    @property83    def close_time(self) -> int:  # requires interval; see Candle.closed_at84        raise NotImplementedError8586    def closed_at(self, interval_min: int) -> int:87        return self.ts + interval_min * 60888990class Trade(BaseModel):91    ts: float92    price: float93    volume: float94    side: Literal["buy", "sell"]95    order_type: Literal["market", "limit"]96    trade_id: int | None = None979899class BookLevel(BaseModel):100    price: Decimal101    qty: Decimal102103104class Ticker(BaseModel):105    symbol: str106    bid: float107    bid_qty: float108    ask: float109    ask_qty: float110    last: float111    volume: float112    vwap: float113    low: float114    high: float115    change: float116    change_pct: float117    ts: float118119120class SystemStatus(BaseModel):121    status: str122    timestamp: str123