TradingView MCP Server

Instructions

Implementation Reference

• src/tradingview_mcp/server.py:996-1093 (handler)

  The handler function for the 'volume_breakout_scanner' tool. It detects coins with significant volume increases relative to average (using volume.SMA20 or estimate) combined with minimum price change thresholds. Processes symbols in batches using tradingview_ta.get_multiple_analysis, calculates volume_ratio and price_change, filters matches, and returns sorted list of breakouts with indicators.
  @mcp.tool() def volume_breakout_scanner(exchange: str = "KUCOIN", timeframe: str = "15m", volume_multiplier: float = 2.0, price_change_min: float = 3.0, limit: int = 25) -> list[dict]: """Detect coins with volume breakout + price breakout. Args: exchange: Exchange name like KUCOIN, BINANCE, BYBIT, etc. timeframe: One of 5m, 15m, 1h, 4h, 1D, 1W, 1M volume_multiplier: How many times the volume should be above normal level (default 2.0) price_change_min: Minimum price change percentage (default 3.0) limit: Number of rows to return (max 50) """ exchange = sanitize_exchange(exchange, "KUCOIN") timeframe = sanitize_timeframe(timeframe, "15m") volume_multiplier = max(1.5, min(10.0, volume_multiplier)) price_change_min = max(1.0, min(20.0, price_change_min)) limit = max(1, min(limit, 50)) # Get symbols symbols = load_symbols(exchange) if not symbols: return [] screener = EXCHANGE_SCREENER.get(exchange, "crypto") volume_breakouts = [] # Process in batches batch_size = 100 for i in range(0, min(len(symbols), 500), batch_size): # Limit to 500 symbols for performance batch_symbols = symbols[i:i + batch_size] try: analysis = get_multiple_analysis(screener=screener, interval=timeframe, symbols=batch_symbols) except Exception: continue for symbol, data in analysis.items(): try: if not data or not hasattr(data, 'indicators'): continue indicators = data.indicators # Get required data volume = indicators.get('volume', 0) close = indicators.get('close', 0) open_price = indicators.get('open', 0) sma20_volume = indicators.get('volume.SMA20', 0) # 20-period volume average if not all([volume, close, open_price]) or volume <= 0: continue # Calculate price change % price_change = ((close - open_price) / open_price) * 100 if open_price > 0 else 0 # Volume ratio calculation # If SMA20 volume not available, use a simple heuristic if sma20_volume and sma20_volume > 0: volume_ratio = volume / sma20_volume else: # Estimate average volume as current volume / 2 (conservative) avg_volume_estimate = volume / 2 volume_ratio = volume / avg_volume_estimate if avg_volume_estimate > 0 else 1 # Check conditions if (abs(price_change) >= price_change_min and volume_ratio >= volume_multiplier): # Get additional indicators rsi = indicators.get('RSI', 50) bb_upper = indicators.get('BB.upper', 0) bb_lower = indicators.get('BB.lower', 0) # Volume strength score volume_strength = min(10, volume_ratio) # Cap at 10x volume_breakouts.append({ "symbol": symbol, "changePercent": price_change, "volume_ratio": round(volume_ratio, 2), "volume_strength": round(volume_strength, 1), "current_volume": volume, "breakout_type": "bullish" if price_change > 0 else "bearish", "indicators": { "close": close, "RSI": rsi, "BB_upper": bb_upper, "BB_lower": bb_lower, "volume": volume } }) except Exception: continue # Sort by volume strength first, then by price change volume_breakouts.sort(key=lambda x: (x["volume_strength"], abs(x["changePercent"])), reverse=True) return volume_breakouts[:limit]