get_monthly_trading_volume_by_dex
Retrieve monthly trading volume data for decentralized exchanges (DEXs) in a markdown-formatted pivot table with dates as index and DEX projects as columns.
Instructions
Retrieve monthly trading volume by decentralized exchange (DEX).
This tool fetches monthly trading volume data for various DEXs from a Dune Analytics query and returns it in a markdown-formatted pivot table, with dates as the index and DEX projects as columns.
Args:
limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.
Returns:
str: A markdown-formatted pivot table of trading volume data, or an error message if the query fails.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No |
Implementation Reference
- main.py:152-173 (handler)The main handler function decorated with @mcp.tool(), which registers and implements the tool logic. It fetches monthly trading volume data from Dune Analytics query ID 4323, processes it into a pivot table using pandas, sorts by date descending, and returns a markdown-formatted table or error message.
@mcp.tool() def get_monthly_trading_volume_by_dex(limit: int = 1000) -> str: """ Retrieve monthly trading volume by decentralized exchange (DEX). This tool fetches monthly trading volume data for various DEXs from a Dune Analytics query and returns it in a markdown-formatted pivot table, with dates as the index and DEX projects as columns. Args: limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000. Returns: str: A markdown-formatted pivot table of trading volume data, or an error message if the query fails. """ try: data = get_latest_result(4323, limit=limit) df = pd.DataFrame(data) df["date"] = pd.to_datetime(df["_col1"]).dt.date pivot_df = df.pivot(index="date", columns="project", values="usd_volume") pivot_df = pivot_df.sort_index(ascending=False) return pivot_df.to_markdown() except Exception as e: return str(e) - main.py:21-43 (helper)Helper function used by the tool to fetch the latest results from a specified Dune Analytics query using the Dune API.
def get_latest_result(query_id: int, limit: int = 1000) -> list: """ 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