[TOOL TABLE]

.env.example

@@ -1,3 +1,16 @@
+# EODHD API Key
 EODHD_API_KEY=""
+
+# CoinMarketCap API Key
 COINMARKETCAP_API_KEY=""
-TELEGRAM_API_KEY=""
+
+# Telegram API Key
+TELEGRAM_API_KEY=""
+
+# Helius API Key
+HELIUS_API_KEY=""
+
+# OKX API Credentials
+OKX_API_SECRET=""
+OKX_PASSPHRASE=""
+OKX_API_KEY=""

README.md

@@ -47,6 +47,22 @@ swarms-tools/
 
 ## 💼 Use Cases
 
+
+
+## Finance
+
+Explore our diverse range of financial tools, designed to streamline your operations. If you need a tool not listed, feel free to submit an issue or accelerate integration by contributing a pull request with your tool of choice.
+
+| **Tool Name**             | **Function**             | **Description**                                                                 |
+|---------------------------|--------------------------|---------------------------------------------------------------------------------|
+| `fetch_stock_news`        | `fetch_stock_news`       | Fetches the latest stock news and updates.                                     |
+| `fetch_htx_data`          | `fetch_htx_data`         | Retrieves financial data from the HTX platform.                                |
+| `yahoo_finance_api`       | `yahoo_finance_api`      | Fetches comprehensive stock data from Yahoo Finance, including prices and trends. |
+| `coin_gecko_coin_api`     | `coin_gecko_coin_api`    | Fetches cryptocurrency data from CoinGecko, including market and price information. |
+| `helius_api_tool`         | `helius_api_tool`        | Retrieves blockchain account, transaction, or token data using the Helius API. |
+| `okx_api_tool`            | `okx_api_tool`           | Fetches detailed cryptocurrency data for coins from the OKX exchange.         |
+
+
 ### Financial Data Retrieval
 Enable precise and actionable financial insights:
 

examples/eodh_example.py → eodh_example.py

@@ -1,3 +1,3 @@
 from swarms_tools.finance.eodh_api import fetch_stock_news
 
-print(fetch_stock_news("AAPL"))
+print(fetch_stock_news("ETH"))

swarms_tools/__init__.py

@@ -3,3 +3,4 @@
 load_dotenv()
 
 from swarms_tools.finance import *
+from swarms_tools.structs import *

swarms_tools/finance/__init__.py

@@ -7,11 +7,13 @@
     coin_gecko_coin_api,
 )
 from swarms_tools.finance.helius_api import helius_api_tool
+from swarms_tools.finance.okx_tool import okx_api_tool
 
 __all__ = [
     "fetch_stock_news",
     "fetch_htx_data",
     "yahoo_finance_api",
     "coin_gecko_coin_api",
-    "helius_api_tool"
+    "helius_api_tool",
+    "okx_api_tool",
 ]

swarms_tools/finance/helius_api.py

@@ -36,11 +36,15 @@ def fetch_account_data(account: str) -> Dict[str, Any]:
             response = requests.get(endpoint, timeout=10)
             response.raise_for_status()
         except requests.RequestException as e:
-            logger.error(f"Failed to fetch account data from Helius API: {e}")
+            logger.error(
+                f"Failed to fetch account data from Helius API: {e}"
+            )
             raise
 
         data = response.json()
-        logger.debug(f"Raw data received for account {account}: {data}")
+        logger.debug(
+            f"Raw data received for account {account}: {data}"
+        )
 
         if "error" in data:
             logger.error(f"Error from Helius API: {data['error']}")
@@ -65,17 +69,23 @@ def fetch_transaction_data(tx_signature: str) -> Dict[str, Any]:
             requests.RequestException: If the API request fails.
         """
         endpoint = f"{HeliusAPI.BASE_URL}/transactions/{tx_signature}?api-key={HeliusAPI.API_KEY}"
-        logger.info(f"Fetching transaction data for signature: {tx_signature}")
+        logger.info(
+            f"Fetching transaction data for signature: {tx_signature}"
+        )
 
         try:
             response = requests.get(endpoint, timeout=10)
             response.raise_for_status()
         except requests.RequestException as e:
-            logger.error(f"Failed to fetch transaction data from Helius API: {e}")
+            logger.error(
+                f"Failed to fetch transaction data from Helius API: {e}"
+            )
             raise
 
         data = response.json()
-        logger.debug(f"Raw data received for transaction {tx_signature}: {data}")
+        logger.debug(
+            f"Raw data received for transaction {tx_signature}: {data}"
+        )
 
         if "error" in data:
             logger.error(f"Error from Helius API: {data['error']}")
@@ -100,17 +110,23 @@ def fetch_token_data(mint_address: str) -> Dict[str, Any]:
             requests.RequestException: If the API request fails.
         """
         endpoint = f"{HeliusAPI.BASE_URL}/tokens/{mint_address}?api-key={HeliusAPI.API_KEY}"
-        logger.info(f"Fetching token data for mint address: {mint_address}")
+        logger.info(
+            f"Fetching token data for mint address: {mint_address}"
+        )
 
         try:
             response = requests.get(endpoint, timeout=10)
             response.raise_for_status()
         except requests.RequestException as e:
-            logger.error(f"Failed to fetch token data from Helius API: {e}")
+            logger.error(
+                f"Failed to fetch token data from Helius API: {e}"
+            )
             raise
 
         data = response.json()
-        logger.debug(f"Raw data received for token {mint_address}: {data}")
+        logger.debug(
+            f"Raw data received for token {mint_address}: {data}"
+        )
 
         if "error" in data:
             logger.error(f"Error from Helius API: {data['error']}")
@@ -141,9 +157,13 @@ def helius_api_tool(action: str, identifier: str) -> Dict[str, Any]:
         elif action == "token":
             return HeliusAPI.fetch_token_data(identifier)
         else:
-            raise ValueError(f"Invalid action: {action}. Must be 'account', 'transaction', or 'token'.")
+            raise ValueError(
+                f"Invalid action: {action}. Must be 'account', 'transaction', or 'token'."
+            )
     except Exception as e:
-        logger.error(f"Error performing action '{action}' with identifier '{identifier}': {e}")
+        logger.error(
+            f"Error performing action '{action}' with identifier '{identifier}': {e}"
+        )
         return {"error": str(e)}
 
 

swarms_tools/finance/okx_tool.py

@@ -0,0 +1,145 @@
+import os
+from typing import List, Dict, Any, Optional
+import requests
+from loguru import logger
+from swarms_tools.utils.formatted_string import (
+    format_object_to_string,
+)
+
+
+class OKXAPI:
+    """
+    A production-grade tool for interacting with the OKX API to fetch coin data.
+    """
+
+    BASE_URL = "https://www.okx.com/api/v5"
+    API_KEY = os.getenv("OKX_API_KEY")
+    API_SECRET = os.getenv(
+        "OKX_API_SECRET"
+    )  # Fetch API secret from environment variable
+    PASSPHRASE = os.getenv(
+        "OKX_PASSPHRASE"
+    )  # Fetch passphrase from environment variable
+
+    @staticmethod
+    @logger.catch
+    def fetch_coin_data(
+        coin_symbols: Optional[List[str]] = None,
+    ) -> Dict[str, Any]:
+        """
+        Fetch all possible data about one or more coins from OKX.
+
+        Args:
+            coin_symbols (Optional[List[str]]): A list of coin symbols (e.g., ['BTC-USDT', 'ETH-USDT']).
+                                                 If None, fetches data for all available coins.
+
+        Returns:
+            Dict[str, Any]: A dictionary containing the formatted coin data.
+
+        Raises:
+            ValueError: If the API response contains errors or the coin symbols are invalid.
+            requests.RequestException: If the API request fails.
+        """
+        endpoint = f"{OKXAPI.BASE_URL}/market/tickers"
+        params = {"instType": "SPOT"}
+        logger.info(
+            f"Fetching coin data for: {coin_symbols or 'all available coins'}"
+        )
+
+        try:
+            response = requests.get(
+                endpoint, params=params, timeout=10
+            )
+            response.raise_for_status()
+        except requests.RequestException as e:
+            logger.error(
+                f"Failed to fetch coin data from OKX API: {e}"
+            )
+            raise
+
+        data = response.json()
+        logger.debug(f"Raw data received: {data}")
+
+        if data.get("code") != "0":
+            logger.error(
+                f"Error from OKX API: {data.get('msg', 'Unknown error')}"
+            )
+            raise ValueError(
+                f"OKX API error: {data.get('msg', 'Unknown error')}"
+            )
+
+        filtered_data = OKXAPI._filter_data(
+            data["data"], coin_symbols
+        )
+        logger.info(f"Filtered data: {filtered_data}")
+        return filtered_data
+
+    @staticmethod
+    def _filter_data(
+        data: List[Dict[str, Any]], coin_symbols: Optional[List[str]]
+    ) -> Dict[str, Any]:
+        """
+        Filter raw API data for specific coin symbols or return all available data.
+
+        Args:
+            data (List[Dict[str, Any]]): The raw data from the OKX API.
+            coin_symbols (Optional[List[str]]): A list of coin symbols to filter data for.
+
+        Returns:
+            Dict[str, Any]: A dictionary of filtered coin data.
+        """
+        if not coin_symbols:
+            return {coin["instId"]: coin for coin in data}
+
+        filtered_data = {
+            coin["instId"]: coin
+            for coin in data
+            if coin["instId"] in coin_symbols
+        }
+        if not filtered_data:
+            logger.warning(
+                f"No data found for specified coin symbols: {coin_symbols}"
+            )
+        return filtered_data
+
+
+def okx_api_tool(coin_symbols: Optional[List[str]] = None) -> str:
+    """
+    Fetch and display data for one or more coins using the OKX API.
+
+    Args:
+        coin_symbols (Optional[List[str]]): A list of coin symbols to fetch data for.
+
+    Returns:
+       String: A string containing the fetched coin data.
+    """
+    try:
+        coin_data = OKXAPI.fetch_coin_data(coin_symbols)
+        return format_object_to_string(coin_data)
+    except ValueError as ve:
+        logger.error(f"ValueError occurred: {ve}")
+        return {"error": str(ve)}
+    except Exception as e:
+        logger.error(f"Unexpected error occurred: {e}")
+        return {"error": str(e)}
+
+
+# if __name__ == "__main__":
+#     # Set up logging
+#     logger.add("okx_api_tool.log", rotation="500 MB", level="INFO")
+
+#     # Example usage
+#     try:
+#         # Fetch data for a single coin
+#         single_coin = okx_api_tool(["BTC-USDT"])
+#         print("Single Coin Data:", single_coin)
+
+#         # Fetch data for multiple coins
+#         multiple_coins = okx_api_tool(["BTC-USDT", "ETH-USDT"])
+#         print("Multiple Coins Data:", multiple_coins)
+
+#         # Fetch data for all coins
+#         all_coins = okx_api_tool()
+#         print("All Coins Data:", all_coins)
+#     except Exception as e:
+#         logger.error(f"Error in OKX API tool: {e}")