get_weekly_trading_volume_by_dex
Fetch and analyze weekly trading volume across decentralized exchanges (DEXs) in a clear markdown pivot table, enabling easy comparison of DEX performance over time.
Instructions
Retrieve weekly trading volume by decentralized exchange (DEX).
This tool fetches weekly 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:129-150 (handler)The main handler function for 'get_weekly_trading_volume_by_dex', decorated with @mcp.tool() for registration in the MCP server. It fetches data from Dune query 4323, processes it into a pivot table using pandas, and returns a markdown table.@mcp.tool() def get_weekly_trading_volume_by_dex(limit: int = 1000) -> str: """ Retrieve weekly trading volume by decentralized exchange (DEX). This tool fetches weekly 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)Shared helper function called by the tool to retrieve the latest results from a Dune Analytics query using the configured API key.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