top_gainers

Instructions

Args:
    exchange: Exchange name like KUCOIN, BINANCE, BYBIT, etc.
    timeframe: One of 5m, 15m, 1h, 4h, 1D, 1W, 1M
    limit: Number of rows to return (max 50)

Implementation Reference

• src/tradingview_mcp/server.py:274-293 (handler)

  Primary handler for the 'top_gainers' tool. Registered with @mcp.tool(). Sanitizes input parameters, invokes the _fetch_trending_analysis helper to retrieve data, sorts by gain percentage, and returns formatted list of top gainers with symbols, change percentages, and indicators.

  @mcp.tool()
  def top_gainers(exchange: str = "KUCOIN", timeframe: str = "15m", limit: int = 25) -> list[dict]:
      """Return top gainers for an exchange and timeframe using bollinger band analysis.
      
      Args:
          exchange: Exchange name like KUCOIN, BINANCE, BYBIT, etc.
          timeframe: One of 5m, 15m, 1h, 4h, 1D, 1W, 1M
          limit: Number of rows to return (max 50)
      """
      exchange = sanitize_exchange(exchange, "KUCOIN")
      timeframe = sanitize_timeframe(timeframe, "15m")
      limit = max(1, min(limit, 50))
      
      rows = _fetch_trending_analysis(exchange, timeframe=timeframe, limit=limit)
      # Convert Row objects to dicts properly
      return [{
          "symbol": row["symbol"],
          "changePercent": row["changePercent"], 
          "indicators": dict(row["indicators"])
      } for row in rows]

• src/tradingview_mcp/server.py:146-208 (helper)

  Key helper function implementing the data fetching and processing logic for top gainers. Loads exchange symbols, fetches TradingView analysis in batches, computes Bollinger Band metrics, filters, sorts by change percentage descending, and returns top limited rows in Row format. Called by top_gainers.

  def _fetch_trending_analysis(exchange: str, timeframe: str = "5m", filter_type: str = "", rating_filter: int = None, limit: int = 50) -> List[Row]:
      """Fetch trending coins analysis similar to the original app's trending endpoint."""
      if not TRADINGVIEW_TA_AVAILABLE:
          raise RuntimeError("tradingview_ta is missing; run `uv sync`.")
      
      symbols = load_symbols(exchange)
      if not symbols:
          raise RuntimeError(f"No symbols found for exchange: {exchange}")
      
      # Process symbols in batches due to TradingView API limits
      batch_size = 200  # Considering API limitations
      all_coins = []
      
      screener = EXCHANGE_SCREENER.get(exchange, "crypto")
      
      # Process symbols in batches
      for i in range(0, len(symbols), batch_size):
          batch_symbols = symbols[i:i + batch_size]
          
          try:
              analysis = get_multiple_analysis(screener=screener, interval=timeframe, symbols=batch_symbols)
          except Exception as e:
              continue  # If this batch fails, move to the next one
              
          # Process coins in this batch
          for key, value in analysis.items():
              try:
                  if value is None:
                      continue
                      
                  indicators = value.indicators
                  metrics = compute_metrics(indicators)
                  
                  if not metrics or metrics.get('bbw') is None:
                      continue
                  
                  # Apply rating filter if specified
                  if filter_type == "rating" and rating_filter is not None:
                      if metrics['rating'] != rating_filter:
                          continue
                  
                  all_coins.append(Row(
                      symbol=key,
                      changePercent=metrics['change'],
                      indicators=IndicatorMap(
                          open=metrics.get('open'),
                          close=metrics.get('price'),
                          SMA20=indicators.get("SMA20"),
                          BB_upper=indicators.get("BB.upper"),
                          BB_lower=indicators.get("BB.lower"),
                          EMA50=indicators.get("EMA50"),
                          RSI=indicators.get("RSI"),
                          volume=indicators.get("volume"),
                      )
                  ))
                  
              except (TypeError, ZeroDivisionError, KeyError):
                  continue
      
      # Sort all coins by change percentage
      all_coins.sort(key=lambda x: x["changePercent"], reverse=True)
      
      return all_coins[:limit]

• src/tradingview_mcp/server.py:39-42 (schema)

  TypedDict schema defining the structure of each row in the top_gainers response: symbol, changePercent, and indicators map.

  class Row(TypedDict):
  	symbol: str
  	changePercent: float
  	indicators: IndicatorMap

• src/tradingview_mcp/server.py:10-13 (helper)

  Imports for supporting helpers: compute_metrics for indicator calculations, load_symbols for exchange symbols, and validators for input sanitization and constants used in top_gainers implementation.

  from tradingview_mcp.core.services.indicators import compute_metrics
  from tradingview_mcp.core.services.coinlist import load_symbols
  from tradingview_mcp.core.utils.validators import sanitize_timeframe, sanitize_exchange, EXCHANGE_SCREENER, ALLOWED_TIMEFRAMES