File size: 10,875 Bytes
9c400b9 d068566 9c400b9 6a7abb7 9c400b9 d068566 9c400b9 d068566 6a7abb7 9c400b9 6a7abb7 9c400b9 6a7abb7 9c400b9 6a7abb7 9c400b9 6a7abb7 9c400b9 6a7abb7 9c400b9 6a7abb7 9c400b9 6a7abb7 9c400b9 6a7abb7 9c400b9 2f3a7c5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 |
import os
from dotenv import load_dotenv
from src.databases.redis import REDIS_CACHED
from src.libs.constants import ONE_HOUR_IN_SECONDS, ONE_MINUTE_IN_SECONDS, ONE_MONTH_IN_SECONDS
from src.libs.logger import logger
from coingecko import CoinGeckoProClient, CoinGeckoDemoClient
load_dotenv()
redis_cache = REDIS_CACHED
class CoinGecko:
def __init__(self, pro_api: bool = False) -> None:
"""
Initializes the CoinGecko object.
Args:
pro_api (bool, optional): A boolean value indicating whether to use the Pro API. Defaults to False.
Returns:
None
Raises:
ValueError: If the `pro_api` argument is not a boolean value.
Note:
This method initializes the CoinGecko object by creating an instance of either CoinGeckoProClient or CoinGeckoDemoClient based on the `pro_api` argument. If `pro_api` is `True`, it uses the Pro API with the provided API key; otherwise, it uses the Demo API.
"""
if not isinstance(pro_api, bool):
raise ValueError("The `pro_api` argument must be a boolean value.")
if pro_api:
self.cgc = CoinGeckoProClient(api_key=os.getenv('COINGECKO_PRO_API_KEY'))
else:
self.cgc = CoinGeckoDemoClient(api_key=os.getenv('COINGECKO_DEMO_API_KEY'))
@redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
@logger.instrument()
def get_coin_price(self, ids: list[str], vs_currencies: list[str], cache_ttl: int = None) -> dict:
"""
Retrieves the current price of a list of coins in a specified currency.
Args:
ids (list[str]): A list of unique identifiers for the coins to retrieve prices for.
vs_currencies (list[str]): A list of currency codes to convert the coin prices to.
cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
Returns:
dict: A dictionary containing the current prices of the specified coins in the specified currencies.
Example:
To get the current prices of a list of coins in USD, you can call:
```python
coin_prices = coin_gecko.get_coin_price(ids=['bitcoin', 'ethereum'], vs_currencies=['usd'])
```
Note:
The `get_coin_price` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_MINUTE_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
"""
result = self.cgc.simple.get_price(ids=ids, vs_currencies=vs_currencies)
return result
@redis_cache(ttl=ONE_HOUR_IN_SECONDS)
@logger.instrument()
def get_coin_data(self, id: str, cache_ttl: int = None) -> dict:
"""
Retrieves detailed information about a specific coin.
Args:
id (str): The unique identifier of the coin.
cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
Returns:
dict: A dictionary containing the detailed information about the specified coin.
Example:
To get data for the coin with id 'bitcoin', you can call:
```python
coin_data = coin_gecko.get_coin_data(id='bitcoin')
```
Note:
The `get_coin_data` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_HOUR_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
"""
result = self.cgc.coins.get_id(id=id, localization=False, market_data=False, tickers=False, sparkline=False)
logger.debug(f"Result: {result}")
return result
@redis_cache(ttl=ONE_MONTH_IN_SECONDS)
@logger.instrument()
def get_coin_category_data(self, params: dict = None, cache_ttl: int = None) -> dict:
"""
Retrieves data for all available coin categories.
Args:
params (dict, optional): A dictionary containing optional parameters for the request. Defaults to None.
cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
Returns:
dict: A dictionary containing the data for all available coin categories.
Example:
To get data for all available coin categories, you can call:
```python
category_data = coin_gecko.get_coin_category_data()
```
Note:
The `get_coin_category_data` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_MONTH_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
"""
params = params or {
"order": "market_cap_desc",
}
result = self.cgc.categories.get_data(endpoint="coins/categories", params=params)
logger.debug(f"Result: {result}")
return { "categories": result}
@redis_cache(ttl=ONE_MONTH_IN_SECONDS)
@logger.instrument()
def get_exchanges_list(self, cache_ttl: int = None) -> dict:
"""
Retrieves a list of all available exchanges.
Args:
cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
Returns:
dict: A dictionary containing a list of all available exchanges.
Example:
To get a list of all available exchanges, you can call:
```python
exchange_list = coin_gecko.get_exchanges_list()
```
Note:
The `get_exchanges_list` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_MONTH_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
"""
result = self.cgc.exchanges.get_list()
logger.debug(f"Result: {result}")
return { 'exchanges': result }
@redis_cache(ttl=ONE_HOUR_IN_SECONDS)
@logger.instrument()
def get_exchange_data(self, id: str = None, cache_ttl: int = None) -> dict:
"""
Retrieves detailed information about a specific exchange.
Args:
id (str, optional): The unique identifier of the exchange. Defaults to None.
cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
Returns:
dict: A dictionary containing the detailed information about the specified exchange.
Raises:
ValueError: If the `id` is not provided and `None` is passed as the argument.
Example:
To get data for the exchange with id 'binance', you can call:
```python
exchange_data = coin_gecko.get_exchange_data(id='binance')
```
Note:
The `get_exchange_data` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_HOUR_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
"""
if id is None:
raise ValueError("Exchange ID must be provided.")
result = self.cgc.exchanges.get_id(id=id)
logger.debug(f"Result: {result}")
return result
@redis_cache(ttl=ONE_MONTH_IN_SECONDS)
@logger.instrument()
def get_coin_categories_list(self, cache_ttl: int = None) -> dict:
"""
Retrieves a list of all available coin categories.
Args:
cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
Returns:
dict: A dictionary containing a list of all available coin categories.
Example:
To get a list of all available coin categories, you can call:
```python
category_list = coin_gecko.get_coin_categories_list()
```
Note:
The `get_coin_categories_list` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_MONTH_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
"""
result = self.cgc.categories.get()
logger.debug(f"Result: {result}")
return { 'exchanges': result }
@redis_cache(ttl=ONE_HOUR_IN_SECONDS)
@logger.instrument()
def get_asset_platforms_list(self, cache_ttl: int = None) -> dict:
"""
Retrieves a list of all available asset platforms.
Args:
cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
Returns:
dict: A dictionary containing a list of all available asset platforms.
Example:
To get a list of all available asset platforms, you can call:
```python
asset_platforms_list = coin_gecko.get_asset_platforms_list()
```
Note:
The `get_asset_platforms_list` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_HOUR_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
"""
result = self.cgc.asset_platform.get()
logger.debug(f"Result: {result}")
return { 'exchanges': result }
@redis_cache(ttl=ONE_HOUR_IN_SECONDS)
@logger.instrument()
def get_trending_coin_list(self, cache_ttl: int = None) -> dict:
"""
Retrieves a list of query trending 15 coins, 7 nfts and 5 categories on CoinGecko in the last 24 hours.
Args:
cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
Returns:
dict: A dictionary containing a list of trending coins.
Example:
To get a list of trending coins, you can call:
```python
trending_coins = coin_gecko.get_trending_coin_list()
```
Note:
The `get_trending_coin_list` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_HOUR_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
"""
result = self.cgc.search.get_trending()
logger.debug(f"Result: {result}")
return { 'exchanges': result }
|