smart_volume_scanner
Scan cryptocurrency markets for assets with significant volume spikes and technical analysis conditions to identify potential trading opportunities based on volume ratios, price changes, and RSI levels.
Instructions
Smart volume + technical analysis combination scanner.
Args:
exchange: Exchange name
min_volume_ratio: Minimum volume multiplier (default 2.0)
min_price_change: Minimum price change percentage (default 2.0)
rsi_range: "oversold" (<30), "overbought" (>70), "neutral" (30-70), "any"
limit: Number of results (max 30)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| exchange | No | KUCOIN | |
| limit | No | ||
| min_price_change | No | ||
| min_volume_ratio | No | ||
| rsi_range | No | any |
Implementation Reference
- src/tradingview_mcp/server.py:1217-1273 (handler)Core handler function implementing the smart_volume_scanner tool. It fetches volume breakouts, applies RSI filtering based on parameters, adds trading recommendations, and returns filtered results.def smart_volume_scanner(exchange: str = "KUCOIN", min_volume_ratio: float = 2.0, min_price_change: float = 2.0, rsi_range: str = "any", limit: int = 20) -> list[dict]: """Smart volume + technical analysis combination scanner. Args: exchange: Exchange name min_volume_ratio: Minimum volume multiplier (default 2.0) min_price_change: Minimum price change percentage (default 2.0) rsi_range: "oversold" (<30), "overbought" (>70), "neutral" (30-70), "any" limit: Number of results (max 30) """ exchange = sanitize_exchange(exchange, "KUCOIN") min_volume_ratio = max(1.2, min(10.0, min_volume_ratio)) min_price_change = max(0.5, min(20.0, min_price_change)) limit = max(1, min(limit, 30)) # Get volume breakouts first volume_breakouts = volume_breakout_scanner( exchange=exchange, volume_multiplier=min_volume_ratio, price_change_min=min_price_change, limit=limit * 2 # Get more to filter ) if not volume_breakouts: return [] # Apply RSI filter filtered_results = [] for coin in volume_breakouts: rsi = coin["indicators"].get("RSI", 50) if rsi_range == "oversold" and rsi >= 30: continue elif rsi_range == "overbought" and rsi <= 70: continue elif rsi_range == "neutral" and (rsi <= 30 or rsi >= 70): continue # "any" passes all # Add trading recommendation recommendation = "" if coin["changePercent"] > 0 and coin["volume_ratio"] >= 2.0: if rsi < 70: recommendation = "🚀 STRONG BUY" else: recommendation = "⚠️ OVERBOUGHT - CAUTION" elif coin["changePercent"] < 0 and coin["volume_ratio"] >= 2.0: if rsi > 30: recommendation = "📉 STRONG SELL" else: recommendation = "🛒 OVERSOLD - OPPORTUNITY?" coin["trading_recommendation"] = recommendation filtered_results.append(coin) return filtered_results[:limit]
- src/tradingview_mcp/server.py:1216-1216 (registration)MCP tool registration decorator for smart_volume_scanner.@mcp.tool()
- Supporting helper tool (volume_breakout_scanner) called internally by smart_volume_scanner to detect initial volume and price breakout candidates.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]