Skip to content

Commit

Permalink
[0.0.7}
Browse files Browse the repository at this point in the history
kyegomez committed Jan 19, 2025
1 parent f802b30 commit 4495f1b
Showing 6 changed files with 508 additions and 4 deletions.
20 changes: 20 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
@@ -107,6 +107,26 @@ def send_alert(response: str):
send_alert("Mission-critical update from Swarms.")
```

---

## Dex Screener

This is a tool that allows you to fetch data from the Dex Screener API. It supports multiple chains and multiple tokens.

```python
from swarms_tools.finance.dex_screener import (
fetch_latest_token_boosts,
fetch_dex_screener_profiles,
)


fetch_dex_screener_profiles()
fetch_latest_token_boosts()

```

---


## Structs
The tool chainer enables the execution of multiple tools in a sequence, allowing for the aggregation of their results in either a parallel or sequential manner.
8 changes: 8 additions & 0 deletions dex_screener.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
from swarms_tools.finance.dex_screener import (
fetch_latest_token_boosts,
fetch_dex_screener_profiles,
)


fetch_dex_screener_profiles()
fetch_latest_token_boosts()
99 changes: 99 additions & 0 deletions examples/dex_screener_example.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
"""
Example usage of the DexScreener API client.
This example demonstrates how to use various endpoints of the DexScreener API
to fetch token information, search pairs, and handle responses.
"""

import asyncio
from loguru import logger

from swarms_tools.finance.dex_screener import DexScreenerAPI


async def main():
# Initialize the API client
dex_screener = DexScreenerAPI(timeout=30)

try:
# Example 1: Get latest token profiles
logger.info("Fetching latest token profiles...")
profiles = await dex_screener.get_latest_token_profiles()
logger.info(f"Found {len(profiles)} token profiles")

# Example 2: Search for a specific token pair
search_query = "USDT"
logger.info(
f"Searching for pairs matching '{search_query}'..."
)
pairs = await dex_screener.search_pairs(search_query)

# Print some information about the found pairs
for pair in pairs[:3]: # Show first 3 pairs
logger.info(
f"Found pair: {pair.base_token.symbol}/{pair.quote_token.symbol}"
)
logger.info(
f"Price: ${pair.price_usd if pair.price_usd else 'N/A'}"
)
if pair.liquidity:
logger.info(f"Liquidity: ${pair.liquidity.usd:,.2f}")
logger.info("---")

# Example 3: Get specific pair information
# Using Ethereum USDT/WETH pair as an example
chain_id = "ethereum"
pair_id = "0x0d4a11d5eeaac28ec3f61d100daf4d40471f1852" # USDT-WETH pair

logger.info(
f"Fetching specific pair info for {chain_id}/{pair_id}..."
)
pair_info = await dex_screener.get_pair(chain_id, pair_id)

if pair_info:
logger.info("Pair Details:")
logger.info(f"Base Token: {pair_info.base_token.symbol}")
logger.info(
f"Quote Token: {pair_info.quote_token.symbol}"
)
logger.info(
f"Price (USD): ${pair_info.price_usd if pair_info.price_usd else 'N/A'}"
)
if pair_info.liquidity:
logger.info(
f"Total Liquidity: ${pair_info.liquidity.usd:,.2f}"
)

# Example 4: Get token pairs for multiple addresses
token_addresses = [
"0xdac17f958d2ee523a2206206994597c13d831ec7", # USDT
"0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48", # USDC
]

logger.info("Fetching pairs for multiple tokens...")
token_pairs = await dex_screener.get_token_pairs(
chain_id, token_addresses
)

for pair in token_pairs[:3]: # Show first 3 pairs
logger.info(
f"Token Pair: {pair.base_token.symbol}/{pair.quote_token.symbol}"
)
logger.info(f"DEX: {pair.dex_id}")
logger.info(f"URL: {pair.url}")
logger.info("---")

except Exception as e:
logger.error(f"An error occurred: {str(e)}")

finally:
# Cleanup
del dex_screener


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

# Run the async example
asyncio.run(main())
7 changes: 3 additions & 4 deletions pyproject.toml
Original file line number Diff line number Diff line change
@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"

[tool.poetry]
name = "swarms-tools"
version = "0.0.5"
version = "0.0.7"
description = "Paper - Pytorch"
license = "MIT"
authors = ["Kye Gomez <kye@apac.ai>"]
@@ -24,10 +24,9 @@ classifiers = [
[tool.poetry.dependencies]
python = "^3.10"
yfinance = "*"
loguru = "^0.7.0"
httpx = "^0.25.0"
loguru = "*"
httpx = "*"
backoff = "*"
solana = "^0.30.2"


[tool.poetry.group.lint.dependencies]
10 changes: 10 additions & 0 deletions swarms_tools/finance/__init__.py
Original file line number Diff line number Diff line change
@@ -14,6 +14,12 @@
yahoo_finance_api,
)
from swarms_tools.finance.coin_market_cap import coinmarketcap_api
from swarms_tools.finance.dex_screener import (
DexScreenerAPI,
fetch_dex_screener_profiles,
fetch_latest_token_boosts,
fetch_solana_token_pairs,
)

__all__ = [
"fetch_stock_news",
@@ -26,4 +32,8 @@
"place_buy_order",
"place_sell_order",
"coinmarketcap_api",
"DexScreenerAPI",
"fetch_dex_screener_profiles",
"fetch_latest_token_boosts",
"fetch_solana_token_pairs",
]
368 changes: 368 additions & 0 deletions swarms_tools/finance/dex_screener.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,368 @@
"""
DexScreener API Client
This module provides a Python interface for the DexScreener API, allowing users to fetch
token profiles, pairs, and perform token-related searches.
"""

import httpx
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Union
from loguru import logger

from swarms_tools.utils.formatted_string import (
format_object_to_string,
)

# Constants
BASE_URL = "https://api.dexscreener.com"
PAIRS_RATE_LIMIT = 300 # requests per minute
PROFILES_RATE_LIMIT = 60 # requests per minute


@dataclass
class TokenInfo:
"""Token information data structure."""

address: str
name: str
symbol: str


@dataclass
class Liquidity:
"""Liquidity information data structure."""

usd: float
base: float
quote: float


@dataclass
class Website:
"""Website information data structure."""

url: str


@dataclass
class Social:
"""Social media information data structure."""

platform: str
handle: str


@dataclass
class TokenPairInfo:
"""Detailed token pair information data structure."""

chain_id: str
dex_id: str
url: str
pair_address: str
labels: Optional[List[str]]
base_token: TokenInfo
quote_token: TokenInfo
price_native: str
price_usd: Optional[str]
liquidity: Optional[Liquidity]
fdv: Optional[float]
market_cap: Optional[float]
pair_created_at: Optional[int]


class DexScreenerAPIError(Exception):
"""Base exception for DexScreener API errors."""

pass


class RateLimitExceeded(DexScreenerAPIError):
"""Raised when rate limit is exceeded."""

pass


class DexScreenerAPI:
"""
DexScreener API client for accessing token and pair information.
This class provides methods to interact with the DexScreener API endpoints
while handling rate limiting and error cases.
"""

def __init__(self, timeout: int = 10):
"""
Initialize the DexScreener API client.
Args:
timeout (int): Request timeout in seconds
"""
self.client = httpx.Client(timeout=timeout)
logger.info("DexScreener API client initialized")

def get_latest_token_profiles(self) -> Dict[str, Any]:
"""
Get the latest token profiles.
Returns:
Dict[str, Any]: Latest token profiles data
Raises:
DexScreenerAPIError: If the API request fails
RateLimitExceeded: If rate limit is exceeded
"""
try:
response = self.client.get(
f"{BASE_URL}/token-profiles/latest/v1"
)
response.raise_for_status()
logger.debug("Successfully fetched latest token profiles")
return response.json()
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
raise RateLimitExceeded(
"Token profiles rate limit exceeded"
)
logger.error(f"Failed to fetch token profiles: {str(e)}")
raise DexScreenerAPIError(f"API request failed: {str(e)}")

def get_latest_token_boosts(self) -> Dict[str, Any]:
"""
Get the latest token boosts.
Returns:
Dict[str, Any]: Latest token boosts data
Raises:
DexScreenerAPIError: If the API request fails
RateLimitExceeded: If rate limit is exceeded
"""
try:
response = self.client.get(
f"{BASE_URL}/token-boosts/latest/v1"
)
response.raise_for_status()
logger.debug("Successfully fetched latest token boosts")
return response.json()
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
raise RateLimitExceeded(
"Token boosts rate limit exceeded"
)
logger.error(f"Failed to fetch token boosts: {str(e)}")
raise DexScreenerAPIError(f"API request failed: {str(e)}")

def get_pair(self, chain_id: str, pair_id: str) -> TokenPairInfo:
"""
Get information about a specific trading pair.
Args:
chain_id (str): The blockchain network ID
pair_id (str): The trading pair ID
Returns:
TokenPairInfo: Detailed information about the trading pair
Raises:
DexScreenerAPIError: If the API request fails
RateLimitExceeded: If rate limit is exceeded
"""
try:
response = self.client.get(
f"{BASE_URL}/latest/dex/pairs/{chain_id}/{pair_id}"
)
response.raise_for_status()
data = response.json()

if not data.get("pairs"):
logger.warning(
f"No pair data found for {chain_id}/{pair_id}"
)
return None

pair = data["pairs"][0]
logger.debug(
f"Successfully fetched pair data for {chain_id}/{pair_id}"
)

return TokenPairInfo(
chain_id=pair["chainId"],
dex_id=pair["dexId"],
url=pair["url"],
pair_address=pair["pairAddress"],
labels=pair.get("labels"),
base_token=TokenInfo(**pair["baseToken"]),
quote_token=TokenInfo(**pair["quoteToken"]),
price_native=pair["priceNative"],
price_usd=pair.get("priceUsd"),
liquidity=(
Liquidity(**pair["liquidity"])
if pair.get("liquidity")
else None
),
fdv=pair.get("fdv"),
market_cap=pair.get("marketCap"),
pair_created_at=pair.get("pairCreatedAt"),
)
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
raise RateLimitExceeded("Pairs rate limit exceeded")
logger.error(f"Failed to fetch pair data: {str(e)}")
raise DexScreenerAPIError(f"API request failed: {str(e)}")

def search_pairs(self, query: str) -> List[TokenPairInfo]:
"""
Search for trading pairs matching the query.
Args:
query (str): Search query string
Returns:
List[TokenPairInfo]: List of matching trading pairs
Raises:
DexScreenerAPIError: If the API request fails
RateLimitExceeded: If rate limit is exceeded
"""
try:
response = self.client.get(
f"{BASE_URL}/latest/dex/search", params={"q": query}
)
response.raise_for_status()
data = response.json()

pairs = []
for pair in data.get("pairs", []):
pairs.append(
TokenPairInfo(
chain_id=pair["chainId"],
dex_id=pair["dexId"],
url=pair["url"],
pair_address=pair["pairAddress"],
labels=pair.get("labels"),
base_token=TokenInfo(**pair["baseToken"]),
quote_token=TokenInfo(**pair["quoteToken"]),
price_native=pair["priceNative"],
price_usd=pair.get("priceUsd"),
liquidity=(
Liquidity(**pair["liquidity"])
if pair.get("liquidity")
else None
),
fdv=pair.get("fdv"),
market_cap=pair.get("marketCap"),
pair_created_at=pair.get("pairCreatedAt"),
)
)

logger.debug(
f"Successfully searched pairs with query: {query}"
)
return pairs
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
raise RateLimitExceeded("Search rate limit exceeded")
logger.error(f"Failed to search pairs: {str(e)}")
raise DexScreenerAPIError(f"API request failed: {str(e)}")

def get_token_pairs(
self, chain_id: str, token_addresses: Union[str, List[str]]
) -> List[TokenPairInfo]:
"""
Get pairs information for one or multiple token addresses.
Args:
chain_id (str): The blockchain network ID
token_addresses (Union[str, List[str]]): Single token address or list of addresses
Returns:
List[TokenPairInfo]: List of token pair information
Raises:
DexScreenerAPIError: If the API request fails
RateLimitExceeded: If rate limit is exceeded
"""
if isinstance(token_addresses, list):
if len(token_addresses) > 30:
raise ValueError(
"Maximum 30 token addresses allowed per request"
)
addresses = ",".join(token_addresses)
else:
addresses = token_addresses

try:
response = self.client.get(
f"{BASE_URL}/tokens/v1/{chain_id}/{addresses}"
)
response.raise_for_status()
pairs = []

for pair in response.json():
pairs.append(
TokenPairInfo(
chain_id=pair["chainId"],
dex_id=pair["dexId"],
url=pair["url"],
pair_address=pair["pairAddress"],
labels=pair.get("labels"),
base_token=TokenInfo(**pair["baseToken"]),
quote_token=TokenInfo(**pair["quoteToken"]),
price_native=pair["priceNative"],
price_usd=pair.get("priceUsd"),
liquidity=(
Liquidity(**pair["liquidity"])
if pair.get("liquidity")
else None
),
fdv=pair.get("fdv"),
market_cap=pair.get("marketCap"),
pair_created_at=pair.get("pairCreatedAt"),
)
)

logger.debug(
f"Successfully fetched token pairs for {chain_id}/{addresses}"
)
return pairs
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
raise RateLimitExceeded(
"Token pairs rate limit exceeded"
)
logger.error(f"Failed to fetch token pairs: {str(e)}")
raise DexScreenerAPIError(f"API request failed: {str(e)}")

def __del__(self):
"""Cleanup method to close the HTTP client."""
if hasattr(self, "client"):
self.client.close()


# Example usage
def fetch_dex_screener_profiles():
dex_screener = DexScreenerAPI()
pairs = dex_screener.get_latest_token_profiles()
print(format_object_to_string(pairs))


def fetch_latest_token_boosts():
dex_screener = DexScreenerAPI()
pairs = dex_screener.get_latest_token_boosts()
print(format_object_to_string(pairs))


def fetch_solana_token_pairs(token_addresses: List[str]):
chain_id = "solana" # Replace with the actual chain ID for Solana
dex_screener = DexScreenerAPI()
pairs = dex_screener.get_token_pairs(chain_id, token_addresses)
print(format_object_to_string(pairs))


# # fetch_dex_screener_profiles()
# fetch_latest_token_boosts()
# # fetch_solana_token_pairs(
# # ["9pqmkBHY3FV2ayMAjtyrVp51jWCbDjFL7NEcPsNDpump"]
# # )

0 comments on commit 4495f1b

Please sign in to comment.