File size: 6,469 Bytes
d068566 383415a d068566 383415a d068566 383415a 1b4c441 383415a d068566 383415a d068566 383415a d068566 383415a 1b4c441 |
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 |
from phi.tools import Toolkit
from phi.utils.log import logger
from src.data_sources.coin_gecko import CoinGecko
from src.data_sources.cryptocompare import CryptoCompare
from src.data_sources.dexscreener import DexScreener
class CryptoDataTools(Toolkit):
def __init__(self):
super().__init__(name="crypto_data_tools")
self.register(self.get_coin_price)
self.register(self.get_coin_data)
self.register(self.get_exchange_data)
self.register(self.get_latest_crypto_news_data)
self.coingecko = CoinGecko()
self.crypto_compare = CryptoCompare()
self.dex_screener = DexScreener()
def get_coin_price(self, coin_id: str, vs_currency: str = "usd") -> str:
"""
Fetches the price data for a given cryptocurrency coin from CoinGecko and CryptoCompare.
Args:
coin_id (str): The unique identifier for the cryptocurrency coin.
vs_currency (str, optional): The currency to which the coin price will be compared. Defaults to "usd".
Returns:
str: A JSON string containing the price data from CoinGecko and CryptoCompare for the specified coin.
Raises:
Exception: If an error occurs while fetching the price data.
Example:
>>> get_coin_price("bitcoin", "eur")
"""
logger.info(f"Fetching price data for {coin_id} cryptocurrency coin from CoinGecko and CryptoCompare")
try:
coingecko_price = {}
crypto_compare_price = {}
coingecko_price_data = self.coingecko.get_coin_price(ids=[coin_id], vs_currencies=[vs_currency])
crypto_compare_price_data = self.crypto_compare.get_coin_price(ids=[coin_id], vs_currencies=[vs_currency])
logger.debug(f"coingecko_price_data: {coingecko_price_data}")
logger.debug(f"crypto_compare_price_data: {crypto_compare_price_data}")
if coin_id in coingecko_price_data:
coingecko_price[coin_id] = coingecko_price_data[coin_id]
else:
logger.warning(f"Warning: CoinGecko data for {coin_id} not found.")
if coin_id.upper() in crypto_compare_price_data:
crypto_compare_price[coin_id] = crypto_compare_price_data[coin_id.upper()]
else:
logger.warning(f"Warning: CryptoCompare data for {coin_id} not found.")
logger.debug(f"response {coingecko_price}")
logger.debug(f"response {crypto_compare_price}")
return f"{(coingecko_price, crypto_compare_price)}"
except Exception as e:
logger.warning(f"Failed to fetch price data for {coin_id}: {e}")
return f"Error: {e}"
def get_coin_data(self, coin_symbol: str) -> str:
"""
Fetches coin data for a given cryptocurrency coin from DexScreener and CryptoCompare.
Args:
coin_symbol (str): The symbol for the cryptocurrency coin.
Returns:
str: A JSON string containing the coin data from DexScreener and CryptoCompare for the specified coin.
Raises:
Exception: If an error occurs while fetching the coin data.
Example:
>>> get_coin_data("BTC")
"""
logger.info(f"Fetching coin data for {coin_symbol} cryptocurrency coin")
try:
dexscreener_coin_data = self.dex_screener.search(query=coin_symbol)
cryptocompare_coin_data = self.crypto_compare.get_overall_coin_data(symbol=coin_symbol)
logger.debug(f"dexscreener_coin_data: {dexscreener_coin_data}")
logger.debug(f"cryptocompare_coin_data: {cryptocompare_coin_data}")
return f"{(dexscreener_coin_data, cryptocompare_coin_data)}"
except Exception as e:
logger.warning(f"Failed to fetch coin data for {coin_symbol}: {e}")
return f"Error: {e}"
def get_exchange_data(self, exchange: str) -> str:
"""
Fetches exchange data for a given cryptocurrency exchange from CoinGecko and DexScreener.
Args:
exchange (str): The unique identifier for the cryptocurrency exchange.
Returns:
str: A JSON string containing the exchange data from CoinGecko and DexScreener for the specified exchange.
Raises:
Exception: If an error occurs while fetching the exchange data.
Example:
>>> get_exchange_data("binance")
"""
logger.info(f"Fetching exchange data for {exchange} cryptocurrency exchange")
try:
coin_gecko_exchange_data = self.coingecko.get_exchange_data(id=exchange)
dexscreener_exchange_data = self.dex_screener.search(query=exchange)
logger.debug(f"coingecko exchange data: {coin_gecko_exchange_data}")
logger.debug(f"dexscreener exchange data: {dexscreener_exchange_data}")
return f"{(coin_gecko_exchange_data, dexscreener_exchange_data)}"
except Exception as e:
logger.warning(f"Failed to fetch exchange data for {exchange}: {e}")
return f"Error: {e}"
def get_latest_crypto_news_data(self) -> str:
"""
Fetches the latest cryptocurrency news data from CoinGecko and CryptoCompare.
Returns:
str: A JSON string containing the latest cryptocurrency news data from CoinGecko and CryptoCompare.
Raises:
Exception: If an error occurs while fetching the latest cryptocurrency news data.
Example:
>>> get_latest_crypto_news_data()
"""
logger.info("Fetching latest cryptocurrency news data")
try:
coin_gecko_trending_coins = self.coingecko.get_trending_coin_list()
cryptocompare_news_categories = self.crypto_compare.get_news_categories()
cryptocompare_lates_news_articles = self.crypto_compare.get_latest_news_articles()
logger.debug(f"coingecko trending coins: {coin_gecko_trending_coins}")
logger.debug(f"cryptocompare news categories: {cryptocompare_news_categories}")
logger.debug(f"cryptocompare latest news articles: {cryptocompare_lates_news_articles}")
return f"{(coin_gecko_trending_coins, cryptocompare_news_categories, cryptocompare_lates_news_articles)}"
except Exception as e:
logger.warning(f"Failed to fetch latest cryptocurrency news data: {e}")
return f"Error: {e}"
|