Skip to content

Commit

Permalink
[0.0.1]
Browse files Browse the repository at this point in the history
  • Loading branch information
kyegomez committed Jan 16, 2025
1 parent e897661 commit b3595e7
Show file tree
Hide file tree
Showing 17 changed files with 669 additions and 12 deletions.
2 changes: 2 additions & 0 deletions .env.example
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
EODHD_API_KEY=""
COINMARKETCAP_API_KEY=""
5 changes: 5 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -3,3 +3,8 @@
[![Join our Discord](https://img.shields.io/badge/Discord-Join%20our%20server-5865F2?style=for-the-badge&logo=discord&logoColor=white)](https://discord.gg/agora-999382051935506503) [![Subscribe on YouTube](https://img.shields.io/badge/YouTube-Subscribe-red?style=for-the-badge&logo=youtube&logoColor=white)](https://www.youtube.com/@kyegomez3242) [![Connect on LinkedIn](https://img.shields.io/badge/LinkedIn-Connect-blue?style=for-the-badge&logo=linkedin&logoColor=white)](https://www.linkedin.com/in/kye-g-38759a207/) [![Follow on X.com](https://img.shields.io/badge/X.com-Follow-1DA1F2?style=for-the-badge&logo=x&logoColor=white)](https://x.com/kyegomezb)


## Install

```bash
pip3 install -U swarms-tools
```
23 changes: 23 additions & 0 deletions coin_market_cap.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
from loguru import logger
from swarms_tools.financial_data.coin_market_cap import (
coinmarketcap_api,
)


if __name__ == "__main__":
# Set up logging
logger.add(
"coinmarketcap_api.log", rotation="500 MB", level="INFO"
)

# Example usage
single_coin = coinmarketcap_api(["Bitcoin"])
print("Single Coin Data:", single_coin)

multiple_coins = coinmarketcap_api(
["Bitcoin", "Ethereum", "Tether"]
)
print("Multiple Coins Data:", multiple_coins)

all_coins = coinmarketcap_api()
print("All Coins Data:", all_coins)
8 changes: 8 additions & 0 deletions coingecko_api.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
from swarms_tools.financial_data.coingecko_tool import (
coin_gecko_coin_api,
)


if __name__ == "__main__":
# Example: Fetch data for Bitcoin
print(coin_gecko_coin_api("bitcoin"))
3 changes: 3 additions & 0 deletions eodh_example.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
from swarms_tools.financial_data.eodh_api import fetch_stock_news

print(fetch_stock_news("AAPL"))
4 changes: 4 additions & 0 deletions htx_tool_example.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
from swarms_tools.financial_data.htx_tool import fetch_htx_data


print(fetch_htx_data("swarms"))
15 changes: 6 additions & 9 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -3,15 +3,15 @@ requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

[tool.poetry]
name = "paper"
name = "swarms-tools"
version = "0.0.1"
description = "Paper - Pytorch"
license = "MIT"
authors = ["Kye Gomez <[email protected]>"]
homepage = "https://github.com/kyegomez/paper"
documentation = "https://github.com/kyegomez/paper" # Add this if you have documentation.
homepage = "https://github.com/The-Swarm-Corporation/swarms-tools"
documentation = "https://github.com/The-Swarm-Corporation/swarms-tools" # Add this if you have documentation.
readme = "README.md" # Assuming you have a README.md
repository = "https://github.com/kyegomez/paper"
repository = "https://github.com/The-Swarm-Corporation/swarms-tools"
keywords = ["artificial intelligence", "deep learning", "optimizers", "Prompt Engineering"]
classifiers = [
"Development Status :: 4 - Beta",
Expand All @@ -23,11 +23,8 @@ classifiers = [

[tool.poetry.dependencies]
python = "^3.10"
swarms = "*"
zetascale = "*"

[tool.poetry.dev-dependencies]
# Add development dependencies here
loguru = "*"
requests = "*"


[tool.poetry.group.lint.dependencies]
Expand Down
5 changes: 2 additions & 3 deletions requirements.txt
Original file line number Diff line number Diff line change
@@ -1,3 +1,2 @@
torch
zetascale
swarms
requests
loguru
3 changes: 3 additions & 0 deletions swarms_tools/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
from dotenv import load_dotenv

load_dotenv()
4 changes: 4 additions & 0 deletions swarms_tools/financial_data/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
from swarms_tools.financial_data.eodh_api import fetch_stock_news
from swarms_tools.financial_data.htx_tool import fetch_htx_data

__all__ = ["fetch_stock_news", "fetch_htx_data"]
136 changes: 136 additions & 0 deletions swarms_tools/financial_data/coin_market_cap.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,136 @@
import os
from typing import Any, Dict, List, Optional

import requests
from loguru import logger
from swarms_tools.utils.formatted_string import (
format_object_to_string,
)


class CoinMarketCapAPI:
"""
A production-grade tool for fetching cryptocurrency data from CoinMarketCap's API.
"""

BASE_URL = "https://pro-api.coinmarketcap.com/v1"
API_KEY = os.getenv(
"COINMARKETCAP_API_KEY"
) # Replace with your actual API key

@staticmethod
@logger.catch
def fetch_coin_data(
coin_names: Optional[List[str]] = None,
) -> Dict[str, Any]:
"""
Fetch all possible data about one or more cryptocurrencies from CoinMarketCap.
Args:
coin_names (Optional[List[str]]): A list of coin names to fetch data for (e.g., ['Bitcoin', 'Ethereum']).
If None, fetches data for all available coins.
Returns:
Dict[str, Any]: A dictionary containing the fetched cryptocurrency data.
Raises:
ValueError: If the API response contains errors or if the coin names are invalid.
requests.RequestException: If the API request fails.
"""
endpoint = f"{CoinMarketCapAPI.BASE_URL}/cryptocurrency/listings/latest"
headers = {"X-CMC_PRO_API_KEY": CoinMarketCapAPI.API_KEY}
logger.info(
f"Fetching data from CoinMarketCap for coins: {coin_names or 'all available coins'}"
)

try:
response = requests.get(
endpoint, headers=headers, timeout=10
)
response.raise_for_status()
except requests.RequestException as e:
logger.error(
f"Failed to fetch data from CoinMarketCap API: {e}"
)
raise

data = response.json()
logger.debug(f"Raw data received: {data}")

if data.get("status", {}).get("error_code") != 0:
logger.error(
f"Error from CoinMarketCap API: {data['status'].get('error_message', 'Unknown error')}"
)
raise ValueError(
f"CoinMarketCap API error: {data['status'].get('error_message', 'Unknown error')}"
)

filtered_data = CoinMarketCapAPI._filter_data(
data["data"], coin_names
)
logger.info(f"Filtered data: {filtered_data}")
return filtered_data

@staticmethod
def _filter_data(
data: List[Dict[str, Any]], coin_names: Optional[List[str]]
) -> Dict[str, Any]:
"""
Filter raw API data for specific coin names or return all available data.
Args:
data (List[Dict[str, Any]]): The raw data from the CoinMarketCap API.
coin_names (Optional[List[str]]): A list of coin names to filter data for.
Returns:
Dict[str, Any]: A dictionary of filtered cryptocurrency data.
"""
if not coin_names:
return {coin["name"]: coin for coin in data}

filtered_data = {
coin["name"]: coin
for coin in data
if coin["name"].lower()
in {name.lower() for name in coin_names}
}
if not filtered_data:
logger.warning(
f"No data found for specified coin names: {coin_names}"
)
return filtered_data


def coinmarketcap_api(
coin_names: Optional[List[str]] = None,
) -> str:
"""
Fetch and display data for one or more cryptocurrencies using CoinMarketCap.
Args:
coin_names (Optional[List[str]]): A list of coin names to fetch data for.
Returns:
str: A str of fetched cryptocurrency data.
"""
try:
coin_data = CoinMarketCapAPI.fetch_coin_data(coin_names)
return format_object_to_string(coin_data)
except Exception as e:
logger.error(f"Error fetching data: {e}")
return {"error": str(e)}


# if __name__ == "__main__":
# # Set up logging
# logger.add("coinmarketcap_api.log", rotation="500 MB", level="INFO")

# # Example usage
# single_coin = coinmarketcap_api(["Bitcoin"])
# print("Single Coin Data:", single_coin)

# multiple_coins = coinmarketcap_api(["Bitcoin", "Ethereum", "Tether"])
# print("Multiple Coins Data:", multiple_coins)

# all_coins = coinmarketcap_api()
# print("All Coins Data:", all_coins)
128 changes: 128 additions & 0 deletions swarms_tools/financial_data/coingecko_tool.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,128 @@
from typing import Dict, Any
import requests
from loguru import logger
from swarms_tools.utils.formatted_string import (
format_object_to_string,
)


class CoinGeckoAPI:
"""
A production-grade tool for fetching cryptocurrency data from CoinGecko's API.
"""

BASE_URL = "https://api.coingecko.com/api/v3"

@staticmethod
@logger.catch
def fetch_coin_data(coin_id: str) -> Dict[str, Any]:
"""
Fetch all data about a cryptocurrency from CoinGecko.
Args:
coin_id (str): The unique ID of the cryptocurrency (e.g., 'bitcoin').
Returns:
Dict[str, Any]: A formatted dictionary containing the cryptocurrency data.
Raises:
ValueError: If the coin ID is invalid or data is unavailable.
requests.RequestException: If the API request fails.
"""
url = f"{CoinGeckoAPI.BASE_URL}/coins/{coin_id}"
logger.info(f"Fetching data for coin ID: {coin_id}")

try:
response = requests.get(url, timeout=10)
response.raise_for_status()
except requests.RequestException as e:
logger.error(
f"Failed to fetch data from CoinGecko API: {e}"
)
raise

data = response.json()
# logger.debug(f"Raw data received: {data}")

if "error" in data:
logger.error(f"Error from CoinGecko API: {data['error']}")
raise ValueError(f"CoinGecko API error: {data['error']}")

formatted_data = CoinGeckoAPI._format_coin_data(data)
# logger.info(f"Formatted data for coin ID {coin_id}: {formatted_data}")
return format_object_to_string(formatted_data)

@staticmethod
def _format_coin_data(data: Dict[str, Any]) -> Dict[str, Any]:
"""
Format raw cryptocurrency data into a structured dictionary.
Args:
data (Dict[str, Any]): Raw data from the CoinGecko API.
Returns:
Dict[str, Any]: Structured and formatted cryptocurrency data.
"""
return {
"id": data.get("id"),
"symbol": data.get("symbol"),
"name": data.get("name"),
"current_price": data.get("market_data", {})
.get("current_price", {})
.get("usd", "N/A"),
"market_cap": data.get("market_data", {})
.get("market_cap", {})
.get("usd", "N/A"),
"total_volume": data.get("market_data", {})
.get("total_volume", {})
.get("usd", "N/A"),
"high_24h": data.get("market_data", {})
.get("high_24h", {})
.get("usd", "N/A"),
"low_24h": data.get("market_data", {})
.get("low_24h", {})
.get("usd", "N/A"),
"price_change_percentage_24h": data.get(
"market_data", {}
).get("price_change_percentage_24h", "N/A"),
"circulating_supply": data.get("market_data", {}).get(
"circulating_supply", "N/A"
),
"total_supply": data.get("market_data", {}).get(
"total_supply", "N/A"
),
"max_supply": data.get("market_data", {}).get(
"max_supply", "N/A"
),
"last_updated": data.get("last_updated"),
"description": data.get("description", {}).get(
"en", "No description available."
),
"homepage": data.get("links", {}).get(
"homepage", ["No homepage available"]
)[0],
}


def coin_gecko_coin_api(coin: str) -> Dict[str, Any]:
"""
Fetch and display data for a specified cryptocurrency.
Args:
coin (str): The unique ID of the cryptocurrency (e.g., 'bitcoin').
Returns:
Dict[str, Any]: A formatted dictionary containing the cryptocurrency data.
"""
try:
coin_data = CoinGeckoAPI.fetch_coin_data(coin)
# print(f"Data for {coin}: {coin_data}")
return coin_data
except Exception as e:
logger.error(f"Error fetching data for {coin}: {e}")
return {"error": str(e)}


# if __name__ == "__main__":
# # Example: Fetch data for Bitcoin
# print(coin_gecko_coin_api("bitcoin"))
Loading

0 comments on commit b3595e7

Please sign in to comment.