get_daily_trading_volume_by_collection
Retrieve daily trading volume data for top Ethereum NFT collections in a structured Markdown table format. Use this tool to analyze and monitor NFT market trends efficiently.
Instructions
Retrieve daily trading volume for top 5 Ethereum NFT collections.
Args:
limit (int, optional): Maximum number of rows to fetch from the query. Defaults to 1000.
Returns:
str: Markdown table of daily trading volumes by collection, or error message if the query fails.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No |
Implementation Reference
- main.py:46-64 (handler)The main handler function for the 'get_daily_trading_volume_by_collection' tool. It is decorated with @mcp.tool() for registration, fetches data from Dune Analytics query 5140422 using the helper function, processes it with pandas to create a pivoted table of daily trading volumes by NFT collection, sorts by date descending, and returns a markdown-formatted table or error string.def get_daily_trading_volume_by_collection(limit: int = 1000) -> str: """ Retrieve daily trading volume for top 5 Ethereum NFT collections. Args: limit (int, optional): Maximum number of rows to fetch from the query. Defaults to 1000. Returns: str: Markdown table of daily trading volumes by collection, or error message if the query fails. """ try: data = get_latest_result(5140422, limit=limit) df = pd.DataFrame(data) df["day"] = pd.to_datetime(df["day"]).dt.date pivot_df = df.pivot(index="day", columns="collection", values="daily_volume") pivot_df = pivot_df.sort_index(ascending=False) return pivot_df.to_markdown() except Exception as e: return str(e)
- main.py:21-44 (helper)Helper function used by the tool to fetch the latest results from a Dune Analytics query API endpoint, handling HTTP requests and extracting the 'rows' data.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