paradex_trades

Instructions

Analyze actual market transactions to understand market sentiment and liquidity.

Use this tool when you need to:
- Detect large trades that might signal institutional activity
- Calculate average trade size during specific periods
- Identify buy/sell pressure imbalances
- Monitor execution prices vs. order book prices
- Understand market momentum through trade flow

Trade data provides insights into actual market activity versus just orders,
helping you understand how other participants are behaving.

Example use cases:
- Detecting large "whale" transactions that might influence price
- Analyzing trade sizes to gauge market participation
- Identifying periods of aggressive buying or selling
- Understanding trade frequency as an indicator of market interest
- Comparing executed prices to orderbook mid-price for market impact analysis

Implementation Reference

• src/mcp_paradex/tools/market.py:434-478 (handler)

  The async function `get_trades` decorated with @server.tool(name='paradex_trades') is the handler. It fetches trades from Paradex via `client.fetch_trades()`, validates them using the Trade model, and returns results with schema info.

  @server.tool(name="paradex_trades", annotations=ToolAnnotations(readOnlyHint=True))
  async def get_trades(
      market_id: Annotated[str, Field(description="Market symbol to get trades for.")],
      start_unix_ms: Annotated[int, Field(description="Start time in unix milliseconds.")],
      end_unix_ms: Annotated[int, Field(description="End time in unix milliseconds.")],
      ctx: Context = None,
  ) -> dict:
      """
      Analyze actual market transactions to understand market sentiment and liquidity.
  
      Use this tool when you need to:
      - Detect large trades that might signal institutional activity
      - Calculate average trade size during specific periods
      - Identify buy/sell pressure imbalances
      - Monitor execution prices vs. order book prices
      - Understand market momentum through trade flow
  
      Trade data provides insights into actual market activity versus just orders,
      helping you understand how other participants are behaving.
  
      Example use cases:
      - Detecting large "whale" transactions that might influence price
      - Analyzing trade sizes to gauge market participation
      - Identifying periods of aggressive buying or selling
      - Understanding trade frequency as an indicator of market interest
      - Comparing executed prices to orderbook mid-price for market impact analysis
      """
      try:
          # Get trades from Paradex
          client = await get_paradex_client()
          response = client.fetch_trades(
              params={"market": market_id, "start_at": start_unix_ms, "end_at": end_unix_ms}
          )
          if "error" in response:
              raise Exception(response["error"])
          trades = trade_adapter.validate_python(response["results"])
          results = {
              "description": Trade.__doc__.strip() if Trade.__doc__ else None,
              "fields": Trade.model_json_schema(),
              "results": trades,
          }
          return results
      except Exception as e:
          await ctx.error(f"Error fetching trades for {market_id}: {e!s}")
          raise e

• src/mcp_paradex/models.py:37-48 (schema)

  The `Trade` Pydantic model defines the schema for trade data (id, market, side, size, price, created_at, trade_type). Used via `trade_adapter = TypeAdapter(list[Trade])` to validate the API response.

  class Trade(BaseModel):
      """Trade model representing a completed trade on Paradex."""
  
      id: Annotated[str, Field(description="Unique Trade ID per TradeType")]
      market: Annotated[str, Field(description="Market for which trade was done")]
      side: Annotated[str, Field(description="Taker side")]
      size: Annotated[float, Field(description="Trade size")]
      price: Annotated[float, Field(description="Trade price")]
      created_at: Annotated[
          int, Field(description="Unix Millisecond timestamp at which trade was done")
      ]
      trade_type: Annotated[str, Field(description="Trade type, can be FILL or LIQUIDATION")]

• src/mcp_paradex/tools/__init__.py:1-12 (registration)

  The tools `__init__.py` imports the market module (which contains the decorated function), causing the @server.tool decorator to register 'paradex_trades' with the FastMCP server singleton.

  """
  Tools module for MCP Paradex.
  """
  
  from mcp_paradex.utils.config import config
  
  # Import tools modules to register them with the server
  from . import market, system, vaults
  
  # Only register auth-required tools when authenticated
  if config.is_configured():
      from . import account, orders

• src/mcp_paradex/server/server.py:78-82 (registration)

  The server singleton is created via `create_server()`, then `from mcp_paradex.tools import *` triggers the registration of all tools including 'paradex_trades'.

  server = create_server()
  
  from mcp_paradex.prompts import *
  from mcp_paradex.resources import *
  from mcp_paradex.tools import *

• src/mcp_paradex/utils/paradex_client.py:23-66 (helper)

  The `get_paradex_client()` helper function initializes and returns the ParadexApiClient singleton used by the trades handler to call `fetch_trades()`.

  async def get_paradex_client() -> ParadexApiClient:
      """
      Get or initialize the Paradex client.
  
      Returns:
          Paradex: The initialized Paradex client.
  
      Raises:
          ValueError: If the required configuration is not set.
      """
      global _paradex_client
  
      if _paradex_client is not None:
          return _paradex_client
  
      async with _client_lock:
          # Double-check in case another task initialized it while waiting
          if _paradex_client is not None:
              return _paradex_client
  
          logger.info("Initializing Paradex client env=%s", config.ENVIRONMENT)
          # retries=1 on the transport causes httpx to retry automatically on a fresh
          # connection when a pooled connection is stale (e.g. after a Lambda freeze).
          http_client = httpx.Client(
              transport=httpx.HTTPTransport(retries=1),
              timeout=httpx.Timeout(30.0),
          )
          _paradex_client = ParadexApiClient(
              env=config.ENVIRONMENT, logger=logger, http_client=http_client
          )
          logger.info("Paradex client api_url=%s", _paradex_client.api_url)
  
          if config.PARADEX_ACCOUNT_PRIVATE_KEY:
              logger.info("Authenticating Paradex client")
              response = _paradex_client.fetch_system_config()
              acc = ParadexAccount(
                  config=response,
                  l1_address="0x0000000000000000000000000000000000000000",
                  l2_private_key=config.PARADEX_ACCOUNT_PRIVATE_KEY,
              )
              _paradex_client.init_account(acc)
              logger.info("Paradex client authenticated account=%s", _paradex_client.account)
  
          return _paradex_client