WSB Analyst MCP Server

Instructions

Fetch top trending stock tickers from ApeWisdom, filtered by valid NASDAQ symbols.

Args:
    num_stocks: Number of top stocks to consider based on upvotes and mentions. Defaults to 20.
    filter: ApeWisdom filter category (e.g., 'wallstreetbets', 'all'). Defaults to 'wallstreetbets'.

Returns:
    A list of valid, trending tickers.

Implementation Reference

• server.py:400-462 (handler)

  The primary handler function decorated with @mcp.tool() for automatic registration. It fetches top trending tickers from the ApeWisdom API based on upvotes and mentions, collects potential tickers, and validates them against NASDAQ listed symbols using the filter_valid_tickers helper. Returns a list of unique valid tickers.

  @mcp.tool()
  async def get_top_trending_tickers(num_stocks: int = 20, filter: str = "wallstreetbets") -> list[str]:
      """
      Fetch top trending stock tickers from ApeWisdom, filtered by valid NASDAQ symbols.
  
      Args:
          num_stocks: Number of top stocks to consider based on upvotes and mentions. Defaults to 20.
          filter: ApeWisdom filter category (e.g., 'wallstreetbets', 'all'). Defaults to 'wallstreetbets'.
  
      Returns:
          A list of valid, trending tickers.
      """
      api_url = f"https://apewisdom.io/api/v1.0/filter/{filter}"
      logger.info(f"Fetching trending tickers from {api_url}")
  
      async with httpx.AsyncClient() as client:
          try:
              response = await client.get(api_url)
              response.raise_for_status() # Raise exception for bad status codes
              data = response.json()
              logger.info(f"Successfully fetched data from ApeWisdom for filter '{filter}'")
          except httpx.RequestError as e:
              logger.error(f"Error fetching from ApeWisdom: {e}")
              return []
          except httpx.HTTPStatusError as e:
               logger.error(f"ApeWisdom API returned error {e.response.status_code}: {e}")
               return []
          except Exception as e: # Catch potential JSON decoding errors etc.
              logger.error(f"An unexpected error occurred fetching/parsing ApeWisdom data: {e}")
              return []
  
      if "results" not in data or not isinstance(data["results"], list):
           logger.error("Invalid data structure received from ApeWisdom API.")
           return []
  
      potential_tickers = set()
      metrics = ["upvotes", "mentions"]
      for metric in metrics:
          # Sort by the current metric and take top N
          # Add try-except for robustness against missing keys or non-numeric values
          try:
              sorted_stocks = sorted(
                  [stock for stock in data["results"] if isinstance(stock.get(metric), (int, float))],
                  key=lambda item: item.get(metric, 0),
                  reverse=True
              )
              for stock in sorted_stocks[:num_stocks]:
                   if "ticker" in stock:
                       potential_tickers.add(stock["ticker"])
          except Exception as e:
               logger.warning(f"Could not process metric '{metric}' due to error: {e}")
  
      if not potential_tickers:
          logger.info("No potential tickers found after processing ApeWisdom results.")
          return []
  
      logger.info(f"Found {len(potential_tickers)} potential tickers, filtering...")
  
      # Filter valid tickers
      valid_tickers = await filter_valid_tickers(list(potential_tickers))
      logger.info(f"Filtered down to {len(valid_tickers)} valid trending tickers.")
  
      return valid_tickers

• server.py:81-115 (helper)

  Helper function used by get_top_trending_tickers to validate potential tickers against official NASDAQ listed symbols by fetching and parsing nasdaqlisted.txt and otherlisted.txt files, returning only valid uppercase tickers.

  async def filter_valid_tickers(tickers: list[str]) -> list[str]:
      """Filter out invalid tickers using NASDAQ's symbol list."""
      async with httpx.AsyncClient() as client:
          try:
              nasdaq_response = await client.get("https://www.nasdaqtrader.com/dynamic/SymDir/nasdaqlisted.txt")
              other_response = await client.get("https://www.nasdaqtrader.com/dynamic/SymDir/otherlisted.txt")
              nasdaq_response.raise_for_status()
              other_response.raise_for_status()
          except httpx.RequestError as e:
              logger.error(f"Error fetching NASDAQ ticker lists: {e}")
              return list(set(ticker.upper() for ticker in tickers))
          except httpx.HTTPStatusError as e:
               logger.error(f"Error fetching NASDAQ ticker lists (status {e.response.status_code}): {e}")
               return list(set(ticker.upper() for ticker in tickers))
  
          valid_symbols = set()
          try:
              for line in nasdaq_response.text.splitlines()[1:-1]:  # Skip header and footer
                  parts = line.split('|')
                  if len(parts) > 0:
                      symbol = parts[0]
                      valid_symbols.add(symbol)
  
              for line in other_response.text.splitlines()[1:-1]:  # Skip header and footer
                   parts = line.split('|')
                   if len(parts) > 0:
                      symbol = parts[0]
                      valid_symbols.add(symbol)
          except Exception as e:
              logger.error(f"Error parsing NASDAQ ticker lists: {e}")
              return list(set(ticker.upper() for ticker in tickers))
  
      unique_tickers = set(ticker.upper() for ticker in tickers)
      return [ticker for ticker in unique_tickers if ticker in valid_symbols]