TradingView MCP Server

Instructions

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 }