aster-info-mcp

Instructions

Implementation Reference

• main.py:688-768 (handler)

  The handler function for the 'get_aggregated_trades' tool. It fetches aggregated trade data from the Aster Finance API (/fapi/v1/aggTrades), processes the response using pandas to create a formatted Markdown table with columns aggTradeId, price, qty, firstTradeId, lastTradeId, time, and isBuyerMaker. It handles input parameters for filtering trades and includes error handling for API requests.
  @mcp.tool() async def get_aggregated_trades( symbol: str, fromId: Optional[int] = None, startTime: Optional[int] = None, endTime: Optional[int] = None, limit: Optional[int] = None ) -> str: """ Fetch aggregated trades data from Aster Finance API and return as Markdown table text. Parameters: symbol (str): Trading pair symbol (e.g., 'BTCUSDT', 'ETHUSDT'). Case-insensitive. fromId (Optional[int]): Aggregated trade ID to start from. If None, uses time-based query or most recent trades. startTime (Optional[int]): Start time in milliseconds since Unix epoch. If None, defaults to API behavior. endTime (Optional[int]): End time in milliseconds since Unix epoch. If None, defaults to API behavior. limit (Optional[int]): Number of aggregated trades to return (1 to 1000). If None, defaults to 500. Returns: str: Markdown table containing aggTradeId, price, qty, firstTradeId, lastTradeId, time, and isBuyerMaker. Raises: Exception: If the API request fails or data processing encounters an error. """ endpoint = "/fapi/v1/aggTrades" # Construct query parameters params = { "symbol": symbol.upper(), # Ensure symbol is uppercase (e.g., BTCUSDT) } if fromId is not None: params["fromId"] = fromId if startTime is not None: params["startTime"] = startTime if endTime is not None: params["endTime"] = endTime if limit is not None: params["limit"] = limit async with httpx.AsyncClient() as client: try: # Make GET request to the API response = await client.get(f"{BASE_URL}{endpoint}", params=params) response.raise_for_status() # Raise exception for 4xx/5xx errors # Parse JSON response trades_data = response.json() # Create pandas DataFrame with API response keys df = pd.DataFrame(trades_data, columns=["a", "p", "q", "f", "l", "T", "m"]) # Convert time to readable format df["T"] = pd.to_datetime(df["T"], unit="ms") # Select and rename columns for clarity df = df[["a", "p", "q", "f", "l", "T", "m"]] df = df.rename(columns={ "a": "aggTradeId", "p": "price", "q": "qty", "f": "firstTradeId", "l": "lastTradeId", "T": "time", "m": "isBuyerMaker" }) # Format numbers df["price"] = df["price"].astype(float).round(8) df["qty"] = df["qty"].astype(float).round(8) # Convert DataFrame to Markdown table markdown_table = df.to_markdown(index=False) return markdown_table except httpx.HTTPStatusError as e: # Handle HTTP errors (e.g., 400, 429) raise Exception(f"API request failed: {e.response.status_code} - {e.response.text}") except Exception as e: # Handle other errors (e.g., network issues, pandas errors) raise Exception(f"Error processing aggregated trades data: {str(e)}")