Update crypto data toolkit

chatxbt-assistant.py

@@ -9,7 +9,7 @@ from phi.llm.openai import OpenAIChat
 from phi.tools.duckduckgo import DuckDuckGo
 from phi.tools.yfinance import YFinanceTools
 from src.databases.postgres import sqlalchemy_engine
-from src.tools.coin_data_toolkit import CryptoDataTools
+from src.tools.crypto_data_toolkit import CryptoDataTools
 from phi.storage.assistant.postgres import PgAssistantStorage
 from src.knowledge_bases.combined import knowledge_base
 

src/tools/{coin_data_toolkit.py → crypto_data_toolkit.py}

@@ -1,5 +1,3 @@
-import json
-
 from phi.tools import Toolkit
 from phi.utils.log import logger
 
@@ -13,6 +11,8 @@ class CryptoDataTools(Toolkit):
         
         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()
@@ -92,3 +92,60 @@ class CryptoDataTools(Toolkit):
         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}"