smart_volume_scanner
Scan cryptocurrency markets for assets with significant volume spikes and technical analysis conditions like RSI levels and price changes to identify trading opportunities.
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 | |
| min_volume_ratio | No | ||
| min_price_change | No | ||
| rsi_range | No | any | |
| limit | No |
Implementation Reference
- src/tradingview_mcp/server.py:1216-1272 (handler)The handler function for 'smart_volume_scanner' tool, including the @mcp.tool() registration decorator. Sanitizes inputs, calls volume_breakout_scanner for candidates, filters by RSI range, adds trading recommendations, and returns limited list of results.
@mcp.tool() 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] - Supporting helper tool 'volume_breakout_scanner' called by smart_volume_scanner to fetch initial volume breakout candidates for further filtering.
@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] - Utility function sanitize_exchange used in smart_volume_scanner to validate and normalize the exchange parameter.
def sanitize_exchange(ex: str, default: str = "kucoin") -> str: if not ex: return default exs = ex.strip().lower() return exs if exs in EXCHANGE_SCREENER else default - Input schema defined by function parameters type hints, defaults, and docstring Args section. Output is list[dict].
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) """