from __future__ import annotations import httpx from app.market.base import MarketData from app.schemas import Candle BASE_URL = "https://api.exchange.coinbase.com" # Coinbase granularities are in seconds. GRANULARITIES: dict[str, int] = { "1m": 60, "5m": 300, "15m": 900, "1h": 3600, "6h": 21600, "1d": 86400, } def to_product_id(symbol: str) -> str: return symbol.replace("/", "-").upper() class CoinbaseMarketData(MarketData): """Public Coinbase Exchange 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, headers={"User-Agent": "crypto_trading/0.1"}, ) @staticmethod def _granularity(timeframe: str) -> int: try: return GRANULARITIES[timeframe] except KeyError: raise ValueError( f"unsupported timeframe '{timeframe}' for Coinbase; " f"use one of {sorted(GRANULARITIES)}" ) from None async def fetch_candles(self, symbol: str, timeframe: str, limit: int) -> list[Candle]: response = await self._client.get( f"/products/{to_product_id(symbol)}/candles", params={"granularity": self._granularity(timeframe)}, ) response.raise_for_status() # Coinbase returns newest-first rows: [time, low, high, open, close, volume]. rows = sorted(response.json(), key=lambda row: row[0]) candles = [ Candle( ts=int(row[0]) * 1000, open=float(row[3]), high=float(row[2]), low=float(row[1]), close=float(row[4]), volume=float(row[5]), ) for row in rows ] return candles[-limit:] async def fetch_price(self, symbol: str) -> float: response = await self._client.get(f"/products/{to_product_id(symbol)}/ticker") response.raise_for_status() return float(response.json()["price"]) async def close(self) -> None: await self._client.aclose()