bollinger_scan

Instructions

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)

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