get_trading_wallet_distribution
Analyze wallet distribution by trading volume on Pumpfun and Pumpswap, excluding bots. Provides a clear breakdown of volume tiers and wallet counts using data from Dune Analytics.
Instructions
Retrieve the distribution of wallets by trading volume on Pumpfun and Pumpswap, excluding bots.
This function queries Dune Analytics (query ID: 5239138) to fetch the number of wallets
grouped by trading volume tiers, formatted as a tabulated string.
Returns:
str: A tabulated string containing the volume tier and number of wallets in each tier,
or an empty string if the query fails.
Raises:
Exception: If the API request or data retrieval encounters an error.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- main.py:123-148 (handler)The handler function for the 'get_trading_wallet_distribution' tool. It is decorated with @mcp.tool() for registration and fetches the wallet distribution data from Dune Analytics query ID 5239138, processes it into a tabulated string using tabulate, or returns empty string on error.
@mcp.tool() def get_trading_wallet_distribution() -> str: """ Retrieve the distribution of wallets by trading volume on Pumpfun and Pumpswap, excluding bots. This function queries Dune Analytics (query ID: 5239138) to fetch the number of wallets grouped by trading volume tiers, formatted as a tabulated string. Returns: str: A tabulated string containing the volume tier and number of wallets in each tier, or an empty string if the query fails. Raises: Exception: If the API request or data retrieval encounters an error. """ try: data = get_latest_result(5239138) rows = [ [row["volume_tier"], row["num_wallets"]] for row in data ] headers = ["Volume Tier", "Number of Wallets"] return tabulate(rows, headers=headers) except: return "" - main.py:22-45 (helper)Supporting helper function used by the tool to fetch the latest execution results from a specified Dune Analytics query.
def get_latest_result(query_id: int, limit: int = 1000): """ Fetch the latest results from a Dune Analytics query. Args: query_id (int): The ID of the Dune query to fetch results from. limit (int, optional): Maximum number of rows to return. Defaults to 1000. Returns: list: A list of dictionaries containing the query results, or an empty list if the request fails. Raises: httpx.HTTPStatusError: If the API request fails due to a client or server error. """ url = f"{BASE_URL}/query/{query_id}/results" params = {"limit": limit} with httpx.Client() as client: response = client.get(url, params=params, headers=HEADERS, timeout=300) response.raise_for_status() data = response.json() result_data = data.get("result", {}).get("rows", []) return result_data