TradingView MCP Server

Instructions

Implementation Reference

• src/tradingview_mcp/server.py:315-336 (handler)

  The primary handler function for the 'bollinger_scan' tool. Registered via @mcp.tool() decorator. Sanitizes inputs, fetches data via helper, and returns formatted list of results.
  @mcp.tool() def bollinger_scan(exchange: str = "KUCOIN", timeframe: str = "4h", bbw_threshold: float = 0.04, limit: int = 50) -> list[dict]: """Scan for coins with low Bollinger Band Width (squeeze detection). Args: exchange: Exchange name like KUCOIN, BINANCE, BYBIT, etc. timeframe: One of 5m, 15m, 1h, 4h, 1D, 1W, 1M bbw_threshold: Maximum BBW value to filter (default 0.04) limit: Number of rows to return (max 100) """ exchange = sanitize_exchange(exchange, "KUCOIN") timeframe = sanitize_timeframe(timeframe, "4h") limit = max(1, min(limit, 100)) rows = _fetch_bollinger_analysis(exchange, timeframe=timeframe, bbw_filter=bbw_threshold, limit=limit) # Convert Row objects to dicts return [{ "symbol": row["symbol"], "changePercent": row["changePercent"], "indicators": dict(row["indicators"]) } for row in rows]
• src/tradingview_mcp/server.py:79-144 (helper)

  Core helper function that loads symbols, fetches TradingView analysis, computes metrics, applies BBW threshold filter, and returns sorted rows of results.
  def _fetch_bollinger_analysis(exchange: str, timeframe: str = "4h", limit: int = 50, bbw_filter: float = None) -> List[Row]: """Fetch analysis using tradingview_ta with bollinger band logic from the original screener.""" if not TRADINGVIEW_TA_AVAILABLE: raise RuntimeError("tradingview_ta is missing; run `uv sync`.") # Load symbols from coinlist files symbols = load_symbols(exchange) if not symbols: raise RuntimeError(f"No symbols found for exchange: {exchange}") # Limit symbols for performance symbols = symbols[:limit * 2] # Get more to filter later # Get screener type based on exchange screener = EXCHANGE_SCREENER.get(exchange, "crypto") try: analysis = get_multiple_analysis(screener=screener, interval=timeframe, symbols=symbols) except Exception as e: raise RuntimeError(f"Analysis failed: {str(e)}") rows: List[Row] = [] for key, value in analysis.items(): try: if value is None: continue indicators = value.indicators metrics = compute_metrics(indicators) if not metrics or metrics.get('bbw') is None: continue # Apply BBW filter if specified if bbw_filter is not None and (metrics['bbw'] >= bbw_filter or metrics['bbw'] <= 0): continue # Check if we have required indicators if not (indicators.get("EMA50") and indicators.get("RSI")): continue rows.append(Row( symbol=key, changePercent=metrics['change'], indicators=IndicatorMap( open=metrics.get('open'), close=metrics.get('price'), SMA20=indicators.get("SMA20"), BB_upper=indicators.get("BB.upper"), BB_lower=indicators.get("BB.lower"), EMA50=indicators.get("EMA50"), RSI=indicators.get("RSI"), volume=indicators.get("volume"), ) )) except (TypeError, ZeroDivisionError, KeyError): continue # Sort by change percentage in descending order (highest gainers first) rows.sort(key=lambda x: x["changePercent"], reverse=True) # Return the requested limit return rows[:limit]
• src/tradingview_mcp/core/services/indicators.py:41-63 (helper)

  Computes essential metrics for Bollinger Band analysis including price change, BBW (Bollinger Band Width), rating, and signal, crucial for filtering in bollinger_scan.
  def compute_metrics(indicators: Dict) -> Optional[Dict]: try: open_price = indicators["open"] close = indicators["close"] sma = indicators["SMA20"] bb_upper = indicators["BB.upper"] bb_lower = indicators["BB.lower"] bb_middle = sma change = compute_change(open_price, close) bbw = compute_bbw(sma, bb_upper, bb_lower) rating, signal = compute_bb_rating_signal(close, bb_upper, bb_middle, bb_lower) return { "price": round(close, 4), "change": round(change, 3), "bbw": round(bbw, 4) if bbw is not None else None, "rating": rating, "signal": signal, } except (KeyError, TypeError): return None
• src/tradingview_mcp/core/utils/validators.py:30-42 (helper)

  Input validation functions for timeframe and exchange parameters used in bollinger_scan to sanitize user inputs.
  def sanitize_timeframe(tf: str, default: str = "5m") -> str: if not tf: return default tfs = tf.strip() return tfs if tfs in ALLOWED_TIMEFRAMES else default 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