from __future__ import annotations import httpx from app.market.base import MarketData from app.schemas import Candle BASE_URL = "https://api.kraken.com" # Kraken intervals are in minutes; map our timeframe strings to them. INTERVALS: dict[str, int] = { "1m": 1, "5m": 5, "15m": 15, "30m": 30, "1h": 60, "4h": 240, "1d": 1440, "1w": 10080, } class KrakenMarketData(MarketData): """Public Kraken spot market data over REST (no API key required).""" def __init__(self, base_url: str = BASE_URL, timeout: float = 10.0) -> None: self._client = httpx.AsyncClient(base_url=base_url, timeout=timeout) @staticmethod def _interval(timeframe: str) -> int: try: return INTERVALS[timeframe] except KeyError: raise ValueError( f"unsupported timeframe '{timeframe}' for Kraken; use one of {sorted(INTERVALS)}" ) from None @staticmethod def _result(payload: dict) -> dict: errors = payload.get("error") or [] if errors: raise ValueError(f"kraken error: {errors}") return payload.get("result", {}) async def fetch_candles(self, symbol: str, timeframe: str, limit: int) -> list[Candle]: response = await self._client.get( "/0/public/OHLC", params={"pair": symbol, "interval": self._interval(timeframe)}, ) response.raise_for_status() result = self._result(response.json()) rows = next((value for key, value in result.items() if key != "last"), []) candles = [ Candle( ts=int(row[0]) * 1000, open=float(row[1]), high=float(row[2]), low=float(row[3]), close=float(row[4]), volume=float(row[6]), ) for row in rows ] return candles[-limit:] async def fetch_price(self, symbol: str) -> float: response = await self._client.get("/0/public/Ticker", params={"pair": symbol}) response.raise_for_status() result = self._result(response.json()) ticker = next(iter(result.values())) return float(ticker["c"][0]) async def close(self) -> None: await self._client.aclose()