| 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}" | |