TradingView MCP Server

Instructions

Args:
    exchange: Exchange name (BINANCE, KUCOIN, etc.)
    timeframe: Time interval (5m, 15m, 1h, 4h)
    pattern_type: "bullish" (growing candles) or "bearish" (shrinking candles)
    candle_count: Number of consecutive candles to check (2-5)
    min_growth: Minimum growth percentage for each candle
    limit: Maximum number of results to return

Returns:
    List of coins with consecutive candle patterns

Implementation Reference

• src/tradingview_mcp/server.py:497-676 (handler)

  The primary handler for the 'consecutive_candles_scan' tool. Decorated with @mcp.tool() for automatic registration in FastMCP. Implements pattern scanning by analyzing current candle body/range ratio, price change, RSI momentum, volume, and position relative to SMAs as proxy for consecutive candles since historical data unavailable. Sanitizes inputs, fetches analysis via tradingview_ta.get_multiple_analysis, filters and scores coins, sorts by strength.

  @mcp.tool()
  def consecutive_candles_scan(
      exchange: str = "KUCOIN",
      timeframe: str = "15m",
      pattern_type: str = "bullish",
      candle_count: int = 3,
      min_growth: float = 2.0,
      limit: int = 20
  ) -> dict:
      """Scan for coins with consecutive growing/shrinking candles pattern.
      
      Args:
          exchange: Exchange name (BINANCE, KUCOIN, etc.)
          timeframe: Time interval (5m, 15m, 1h, 4h)
          pattern_type: "bullish" (growing candles) or "bearish" (shrinking candles)
          candle_count: Number of consecutive candles to check (2-5)
          min_growth: Minimum growth percentage for each candle
          limit: Maximum number of results to return
      
      Returns:
          List of coins with consecutive candle patterns
      """
      try:
          exchange = sanitize_exchange(exchange, "KUCOIN")
          timeframe = sanitize_timeframe(timeframe, "15m")
          candle_count = max(2, min(5, candle_count))
          min_growth = max(0.5, min(20.0, min_growth))
          limit = max(1, min(50, limit))
          
          # Get symbols for the exchange
          symbols = load_symbols(exchange)
          if not symbols:
              return {
                  "error": f"No symbols found for exchange: {exchange}",
                  "exchange": exchange,
                  "timeframe": timeframe
              }
          
          # Limit symbols for performance (we need historical data)
          symbols = symbols[:min(limit * 3, 200)]
          
          # We need to get data from multiple timeframes to analyze candle progression
          # For now, we'll use current timeframe data and simulate pattern detection
          screener = EXCHANGE_SCREENER.get(exchange, "crypto")
          
          try:
              analysis = get_multiple_analysis(
                  screener=screener,
                  interval=timeframe,
                  symbols=symbols
              )
              
              pattern_coins = []
              
              for symbol, data in analysis.items():
                  if data is None:
                      continue
                      
                  try:
                      indicators = data.indicators
                      
                      # Calculate current candle metrics
                      open_price = indicators.get("open")
                      close_price = indicators.get("close")
                      high_price = indicators.get("high") 
                      low_price = indicators.get("low")
                      volume = indicators.get("volume", 0)
                      
                      if not all([open_price, close_price, high_price, low_price]):
                          continue
                      
                      # Calculate current candle body size and change
                      current_change = ((close_price - open_price) / open_price) * 100
                      candle_body = abs(close_price - open_price)
                      candle_range = high_price - low_price
                      body_to_range_ratio = candle_body / candle_range if candle_range > 0 else 0
                      
                      # For consecutive pattern, we'll use available indicators to simulate
                      # In a real implementation, we'd need historical OHLC data
                      
                      # Use RSI and price momentum as proxy for consecutive pattern
                      rsi = indicators.get("RSI", 50)
                      sma20 = indicators.get("SMA20", close_price)
                      ema50 = indicators.get("EMA50", close_price)
                      
                      # Calculate momentum indicators
                      price_above_sma = close_price > sma20
                      price_above_ema = close_price > ema50
                      strong_momentum = abs(current_change) >= min_growth
                      
                      # Pattern detection logic
                      pattern_detected = False
                      pattern_strength = 0
                      
                      if pattern_type == "bullish":
                          # Bullish pattern: price rising, good momentum, strong candle body
                          conditions = [
                              current_change > min_growth,  # Current candle is bullish
                              body_to_range_ratio > 0.6,    # Strong candle body
                              price_above_sma,              # Above short MA
                              rsi > 45 and rsi < 80,        # RSI in momentum range
                              volume > 1000                 # Decent volume
                          ]
                          
                          pattern_strength = sum(conditions)
                          pattern_detected = pattern_strength >= 3
                          
                      elif pattern_type == "bearish":
                          # Bearish pattern: price falling, bearish momentum
                          conditions = [
                              current_change < -min_growth,  # Current candle is bearish
                              body_to_range_ratio > 0.6,     # Strong candle body
                              not price_above_sma,           # Below short MA
                              rsi < 55 and rsi > 20,         # RSI in bearish range
                              volume > 1000                  # Decent volume
                          ]
                          
                          pattern_strength = sum(conditions)
                          pattern_detected = pattern_strength >= 3
                      
                      if pattern_detected:
                          # Calculate additional metrics
                          metrics = compute_metrics(indicators)
                          
                          coin_data = {
                              "symbol": symbol,
                              "price": round(close_price, 6),
                              "current_change": round(current_change, 3),
                              "candle_body_ratio": round(body_to_range_ratio, 3),
                              "pattern_strength": pattern_strength,
                              "volume": volume,
                              "bollinger_rating": metrics.get('rating', 0) if metrics else 0,
                              "rsi": round(rsi, 2),
                              "price_levels": {
                                  "open": round(open_price, 6),
                                  "high": round(high_price, 6), 
                                  "low": round(low_price, 6),
                                  "close": round(close_price, 6)
                              },
                              "momentum_signals": {
                                  "above_sma20": price_above_sma,
                                  "above_ema50": price_above_ema,
                                  "strong_volume": volume > 5000
                              }
                          }
                          
                          pattern_coins.append(coin_data)
                          
                  except Exception as e:
                      continue
              
              # Sort by pattern strength and current change
              if pattern_type == "bullish":
                  pattern_coins.sort(key=lambda x: (x['pattern_strength'], x['current_change']), reverse=True)
              else:
                  pattern_coins.sort(key=lambda x: (x['pattern_strength'], -x['current_change']), reverse=True)
              
              return {
                  "exchange": exchange,
                  "timeframe": timeframe,
                  "pattern_type": pattern_type,
                  "candle_count": candle_count,
                  "min_growth": min_growth,
                  "total_found": len(pattern_coins),
                  "data": pattern_coins[:limit]
              }
              
          except Exception as e:
              return {
                  "error": f"Pattern analysis failed: {str(e)}",
                  "exchange": exchange,
                  "timeframe": timeframe
              }
              
      except Exception as e:
          return {
              "error": f"Consecutive candles scan failed: {str(e)}",
              "exchange": exchange,
              "timeframe": timeframe
          }