get_alpha_wallets
Identify top profitable wallets on Pumpfun and Pumpswap by analyzing trading activity and realized profits over the past 30 days using Dune Analytics data.
Instructions
Retrieve the top profitable wallets on Pumpfun and Pumpswap for the last 30 days.
This function queries Dune Analytics (query ID: 4032586) to fetch a ranked list of wallets
based on their realized profit over the past 30 days, formatted as a tabulated string.
Args:
limit (int, optional): Maximum number of wallets to return. Defaults to 100.
Returns:
str: A tabulated string containing the rank, wallet address, realized profit (in USD),
and last transaction timestamp for each wallet, 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 |
|---|---|---|---|
| limit | No |
Implementation Reference
- main.py:65-92 (handler)The handler function for the 'get_alpha_wallets' tool. It uses @mcp.tool() decorator for registration and fetches top profitable wallets from Dune Analytics query ID 4032586, formatting the results as a tabulated string.
@mcp.tool() def get_alpha_wallets(limit: int = 100) -> str: """ Retrieve the top profitable wallets on Pumpfun and Pumpswap for the last 30 days. This function queries Dune Analytics (query ID: 4032586) to fetch a ranked list of wallets based on their realized profit over the past 30 days, formatted as a tabulated string. Args: limit (int, optional): Maximum number of wallets to return. Defaults to 100. Returns: str: A tabulated string containing the rank, wallet address, realized profit (in USD), and last transaction timestamp for each wallet, 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(4032586, limit) rows = [ [row["rank"], row["wallet"], f'${row["realized_profit"]:.0f}', row["last_tx"]] for row in data ] headers = ["Rank", "Wallet", "Realized Profit", "Last Tx"] return tabulate(rows, headers=headers) except: return "" - main.py:22-44 (helper)Helper function to fetch latest results from a specified Dune Analytics query, used by get_alpha_wallets and other tools.
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