124 lines
5.2 KiB
Python
124 lines
5.2 KiB
Python
import httpx
|
|
import logging
|
|
from typing import Dict, List, Optional, Any, Union
|
|
from app.config import COINCAP_API_BASE_URL, COINCAP_API_KEY
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class CoinCapClient:
|
|
"""Client for interacting with the CoinCap API v3"""
|
|
|
|
def __init__(self, api_key: str = COINCAP_API_KEY, base_url: str = COINCAP_API_BASE_URL):
|
|
self.api_key = api_key
|
|
self.base_url = base_url
|
|
|
|
async def _request(self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Dict:
|
|
"""Make a request to the CoinCap API
|
|
|
|
Args:
|
|
method: HTTP method (GET, POST, etc.)
|
|
endpoint: API endpoint (without base URL)
|
|
params: Query parameters
|
|
|
|
Returns:
|
|
API response as dictionary
|
|
"""
|
|
if params is None:
|
|
params = {}
|
|
|
|
# Add API key to all requests
|
|
params['apiKey'] = self.api_key
|
|
|
|
url = f"{self.base_url}{endpoint}"
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.request(method, url, params=params)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(f"HTTP error occurred: {e}")
|
|
if e.response.status_code == 404:
|
|
# Return empty data for 404 errors
|
|
return {"data": []}
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error in CoinCap API request: {e}")
|
|
raise
|
|
|
|
# Assets endpoints
|
|
async def get_assets(self, search: Optional[str] = None,
|
|
ids: Optional[str] = None,
|
|
limit: Optional[int] = None,
|
|
offset: Optional[int] = None) -> Dict:
|
|
"""Get list of assets with optional filters"""
|
|
params = {k: v for k, v in locals().items()
|
|
if k not in ['self'] and v is not None}
|
|
return await self._request("GET", "/assets", params)
|
|
|
|
async def get_asset(self, slug: str) -> Dict:
|
|
"""Get information about a specific asset by slug"""
|
|
return await self._request("GET", f"/assets/{slug}")
|
|
|
|
async def get_asset_markets(self, slug: str,
|
|
limit: Optional[int] = None,
|
|
offset: Optional[int] = None) -> Dict:
|
|
"""Get markets for a specific asset"""
|
|
params = {k: v for k, v in locals().items()
|
|
if k not in ['self', 'slug'] and v is not None}
|
|
return await self._request("GET", f"/assets/{slug}/markets", params)
|
|
|
|
async def get_asset_history(self, slug: str,
|
|
interval: str,
|
|
start: Optional[int] = None,
|
|
end: Optional[int] = None) -> Dict:
|
|
"""Get historical data for a specific asset
|
|
|
|
Args:
|
|
slug: Asset slug
|
|
interval: Time interval (m1, m5, m15, m30, h1, h2, h6, h12, d1)
|
|
start: Start time in milliseconds
|
|
end: End time in milliseconds
|
|
"""
|
|
params = {k: v for k, v in locals().items()
|
|
if k not in ['self', 'slug'] and v is not None}
|
|
return await self._request("GET", f"/assets/{slug}/history", params)
|
|
|
|
# Exchanges endpoints
|
|
async def get_exchanges(self, limit: Optional[int] = None,
|
|
offset: Optional[int] = None) -> Dict:
|
|
"""Get list of exchanges"""
|
|
params = {k: v for k, v in locals().items()
|
|
if k not in ['self'] and v is not None}
|
|
return await self._request("GET", "/exchanges", params)
|
|
|
|
async def get_exchange(self, exchange_id: str) -> Dict:
|
|
"""Get information about a specific exchange"""
|
|
return await self._request("GET", f"/exchanges/{exchange_id}")
|
|
|
|
# Markets endpoints
|
|
async def get_markets(self,
|
|
exchange_id: Optional[str] = None,
|
|
base_symbol: Optional[str] = None,
|
|
base_id: Optional[str] = None,
|
|
quote_symbol: Optional[str] = None,
|
|
quote_id: Optional[str] = None,
|
|
asset_symbol: Optional[str] = None,
|
|
asset_id: Optional[str] = None,
|
|
limit: Optional[int] = None,
|
|
offset: Optional[int] = None) -> Dict:
|
|
"""Get list of markets with optional filters"""
|
|
params = {k: v for k, v in locals().items()
|
|
if k not in ['self'] and v is not None}
|
|
return await self._request("GET", "/markets", params)
|
|
|
|
# Rates endpoints
|
|
async def get_rates(self, ids: Optional[str] = None) -> Dict:
|
|
"""Get conversion rates, optionally filtered by comma-separated slugs"""
|
|
params = {k: v for k, v in locals().items()
|
|
if k not in ['self'] and v is not None}
|
|
return await self._request("GET", "/rates", params)
|
|
|
|
async def get_rate(self, slug: str) -> Dict:
|
|
"""Get a specific conversion rate by slug"""
|
|
return await self._request("GET", f"/rates/{slug}") |