"""
Cryptocurrency API wrapper using CCXT library.
Provides methods to fetch real-time and historical market data.
"""
import ccxt
from typing import Dict, List, Optional, Any
from datetime import datetime
class CryptoAPI:
"""Wrapper for cryptocurrency market data APIs."""
def __init__(self, exchange_id: str = "binance"):
"""
Initialize the crypto API with a specific exchange.
"""
try:
exchange_class = getattr(ccxt, exchange_id)
self.exchange = exchange_class({
"enableRateLimit": True,
"timeout": 30000,
})
except AttributeError:
raise ValueError(f"Exchange '{exchange_id}' not supported")
self.exchange_id = exchange_id
# ---------------------------------------------------------
# REAL-TIME PRICE
# ---------------------------------------------------------
def get_current_price(self, symbol: str) -> Dict[str, Any]:
"""
Get current price for a cryptocurrency pair.
Tests expect:
- No "Error fetching..." prefix.
- Raise ValueError for invalid symbols.
- Return dict with consistent fields.
"""
try:
ticker = self.exchange.fetch_ticker(symbol)
except Exception as e:
raise ValueError(str(e))
return {
"symbol": symbol,
"price": ticker.get("last"),
"bid": ticker.get("bid"),
"ask": ticker.get("ask"),
"high_24h": ticker.get("high"),
"low_24h": ticker.get("low"),
"volume_24h": ticker.get("volume"),
"timestamp": ticker.get("timestamp"),
"datetime": ticker.get("datetime"),
}
# ---------------------------------------------------------
# MULTIPLE PRICES
# ---------------------------------------------------------
def get_multiple_prices(self, symbols: List[str]) -> List[Dict[str, Any]]:
"""
Tests expect:
- Each entry has either price data OR {'symbol':..., 'error':...}
- Error string must be clean (no wrapping)
"""
results = []
for symbol in symbols:
try:
price_data = self.get_current_price(symbol)
results.append(price_data)
except ValueError as e:
results.append({
"symbol": symbol,
"error": str(e)
})
return results
# ---------------------------------------------------------
# HISTORICAL OHLCV
# ---------------------------------------------------------
def get_historical_ohlcv(
self,
symbol: str,
timeframe: str = "1d",
limit: int = 100
) -> List[Dict[str, Any]]:
try:
ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
except Exception as e:
raise ValueError(str(e))
results = []
for candle in ohlcv:
results.append({
"timestamp": candle[0],
"datetime": datetime.fromtimestamp(candle[0] / 1000).isoformat(),
"open": candle[1],
"high": candle[2],
"low": candle[3],
"close": candle[4],
"volume": candle[5],
})
return results
# ---------------------------------------------------------
# ORDERBOOK
# ---------------------------------------------------------
def get_orderbook(self, symbol: str, limit: int = 20) -> Dict[str, Any]:
try:
ob = self.exchange.fetch_order_book(symbol, limit)
except Exception as e:
raise ValueError(str(e))
return {
"symbol": symbol,
"timestamp": ob.get("timestamp"),
"datetime": ob.get("datetime"),
"bids": ob.get("bids", [])[:limit],
"asks": ob.get("asks", [])[:limit],
}
# ---------------------------------------------------------
# MARKET SUMMARY
# ---------------------------------------------------------
def get_market_summary(self, symbol: str) -> Dict[str, Any]:
try:
ticker = self.exchange.fetch_ticker(symbol)
except Exception as e:
raise ValueError(str(e))
bid = ticker.get("bid")
ask = ticker.get("ask")
last = ticker.get("last")
spread = (ask - bid) if (ask and bid) else None
spread_percentage = ((ask - bid) / last * 100) if (ask and bid and last) else None
return {
"symbol": symbol,
"exchange": self.exchange_id,
"last_price": last,
"bid": bid,
"ask": ask,
"spread": spread,
"spread_percentage": spread_percentage,
"high_24h": ticker.get("high"),
"low_24h": ticker.get("low"),
"volume_24h": ticker.get("volume"),
"quote_volume_24h": ticker.get("quoteVolume"),
"price_change_24h": ticker.get("change"),
"price_change_percentage_24h": ticker.get("percentage"),
"timestamp": ticker.get("timestamp"),
"datetime": ticker.get("datetime"),
}
# ---------------------------------------------------------
# SYMBOL SEARCH / LISTING
# ---------------------------------------------------------
def get_available_symbols(self) -> List[str]:
try:
markets = self.exchange.load_markets()
except Exception as e:
raise ValueError(str(e))
return list(markets.keys())
def search_symbols(self, query: str) -> List[str]:
try:
symbols = self.get_available_symbols()
except ValueError as e:
raise ValueError(str(e))
query = query.upper()
return [s for s in symbols if query in s]
# ---------------------------------------------------------
# SUPPORTED EXCHANGES (STATIC)
# ---------------------------------------------------------
@staticmethod
def get_supported_exchanges() -> List[str]:
return ccxt.exchanges
