compute_indicators

Instructions

Implementation Reference

• coin_mcp/indicators.py:336-716 (handler)

  The main MCP tool handler function `compute_indicators` — an async function decorated with @mcp.tool() that takes OHLCV candles and computes requested technical indicators (RSI, MACD, Bollinger, EMA, SMA, ATR, ADX, Stochastic, OBV). Returns a dict with latest values, optional full series, meta, signal_summary, and disclaimer.

  @mcp.tool()
  async def compute_indicators(
      ohlcv: list[list[float]],
      indicators: list[str] = ["rsi", "macd", "bollinger", "ema", "sma", "atr"],
      rsi_period: int = 14,
      macd_fast: int = 12,
      macd_slow: int = 26,
      macd_signal: int = 9,
      bb_period: int = 20,
      bb_stddev: float = 2.0,
      ema_periods: list[int] = [12, 26, 50, 200],
      sma_periods: list[int] = [20, 50, 200],
      atr_period: int = 14,
      stoch_k_period: int = 14,
      stoch_d_period: int = 3,
      adx_period: int = 14,
      include_series: bool = False,
  ) -> dict:
      """Compute technical indicators on OHLCV candles you have already fetched.
  
      USE THIS WHEN: you have OHLCV data (from `get_exchange_ohlcv` for one
      specific exchange, or `get_aggregated_ohlc` for a CoinGecko cross-venue
      aggregate) and you want RSI / MACD / Bollinger / EMA / SMA / ATR / ADX /
      Stochastic / OBV without writing the math yourself.
  
      THIS TOOL DOES NOT FETCH DATA. The caller must provide candles. If you
      need candles first, call `get_exchange_ohlcv` (CCXT, per-venue, supports
      1m candles) or `get_aggregated_ohlc` (CoinGecko, market-aggregate, daily/
      hourly).
  
      THIS TOOL RETURNS OBSERVATIONS, NOT TRADING ADVICE. The `signal_summary`
      field describes what the indicators currently show (e.g. "RSI 72 —
      overbought"). It never recommends buying or selling.
  
      Input format:
          ohlcv: list of rows. Either
              6 columns: [timestamp_ms, open, high, low, close, volume]   (CCXT)
              5 columns: [timestamp_ms, open, high, low, close]           (CoinGecko aggregated_ohlc — no volume)
          Rows must be ordered oldest -> newest. With 5-column input, OBV is
          unavailable (returned as null with a `note`).
  
      Args:
          ohlcv: Candles, oldest first. 5 or 6 columns per row.
          indicators: Which indicators to compute. Any subset of
              ["rsi", "macd", "bollinger", "ema", "sma", "atr", "adx",
               "stochastic", "obv"]. Default omits adx/stochastic/obv to keep
              output compact; pass them explicitly to opt in.
          rsi_period: Lookback for Wilder's RSI. Default 14.
          macd_fast / macd_slow / macd_signal: MACD EMA periods. Defaults 12/26/9.
          bb_period / bb_stddev: Bollinger Bands lookback and stddev multiplier.
              Defaults 20 and 2.0.
          ema_periods: List of EMA lookbacks to compute. Default [12,26,50,200].
          sma_periods: List of SMA lookbacks to compute. Default [20,50,200].
          atr_period: Wilder ATR lookback. Default 14.
          stoch_k_period / stoch_d_period: Stochastic %K and %D periods.
              Defaults 14 and 3.
          adx_period: Wilder ADX lookback. Default 14.
          include_series: If True, return the full per-bar series for every
              indicator (suitable for charting). If False (default), return
              only the latest value per indicator — much smaller payload.
              When True and the input has more than `MAX_SERIES_RETURN` (1000)
              rows, each returned series is truncated to the last
              `MAX_SERIES_RETURN` entries and `truncated_series_to` is set on
              the response so the caller can tell.
  
      Bounds:
          The input is rejected with an error if it has more than
          `MAX_OHLCV_ROWS` (5000) rows — pass the most recent N bars or split
          into chunks. This guards against pathological / prompt-injected
          inputs whose ADX/Stochastic passes and JSON serialization would
          otherwise dominate runtime and memory.
  
      Returns:
          Dict with one key per requested indicator plus:
            - `meta`: { bar_count, has_volume, last_timestamp_ms, last_close }
            - `signal_summary`: human-readable interpretation per indicator
              (observations only — overbought / oversold / trend direction etc.)
          Per-indicator shape:
            - rsi:        { latest, period, series? }
            - macd:       { latest: { macd, signal, histogram }, params, series? }
            - bollinger:  { latest: { upper, middle, lower }, period, stddev,
                            percent_b, bandwidth, series? }
            - ema:        { latest: { "12": ..., "26": ... }, periods, series? }
            - sma:        { latest: { "20": ..., "50": ... }, periods, series? }
            - atr:        { latest, period, series? }
            - adx:        { latest: { adx, plus_di, minus_di }, period, series? }
            - stochastic: { latest: { k, d }, k_period, d_period, series? }
            - obv:        { latest, series? }   # null + note if no volume column
      """
      if not isinstance(ohlcv, list):
          return {"error": "ohlcv must be a list of [ts,o,h,l,c[,v]] rows"}
      n_rows = len(ohlcv)
      if n_rows == 0:
          return {"error": "ohlcv is empty"}
      if n_rows > MAX_OHLCV_ROWS:
          return {
              "error": (
                  f"ohlcv exceeds MAX_OHLCV_ROWS={MAX_OHLCV_ROWS}; received {n_rows} rows. "
                  "Truncate the input (most recent N bars) or split into chunks."
              ),
              "max_rows": MAX_OHLCV_ROWS,
          }
  
      # ----- parse rows -----
      width = len(ohlcv[0])
      if width not in (5, 6):
          return {
              "error": f"each row must have 5 or 6 columns, got {width}",
              "hint": "expected [ts, open, high, low, close] or [ts, open, high, low, close, volume]",
          }
      has_volume = width == 6
      try:
          timestamps = [int(row[0]) for row in ohlcv]
          opens = [float(row[1]) for row in ohlcv]
          highs = [float(row[2]) for row in ohlcv]
          lows = [float(row[3]) for row in ohlcv]
          closes = [float(row[4]) for row in ohlcv]
          volumes = [float(row[5]) for row in ohlcv] if has_volume else []
      except (ValueError, TypeError, IndexError) as e:
          return {"error": f"failed to parse ohlcv rows: {e}"}
  
      n = len(closes)
      requested = {s.lower().strip() for s in indicators}
      # When include_series is True, cap each returned series to the last
      # MAX_SERIES_RETURN entries to keep the JSON response reasonable.
      series_tail: int | None = (
          MAX_SERIES_RETURN if include_series and n > MAX_SERIES_RETURN else None
      )
      out: dict[str, Any] = {
          "meta": {
              "bar_count": n,
              "has_volume": has_volume,
              "last_timestamp_ms": timestamps[-1] if timestamps else None,
              "last_close": _round(closes[-1]) if closes else None,
          },
      }
      if series_tail is not None:
          out["truncated_series_to"] = series_tail
      summary: dict[str, str] = {}
  
      # ----- RSI -----
      if "rsi" in requested:
          rsi_series = _rsi(closes, rsi_period)
          latest = _last(rsi_series)
          out["rsi"] = {
              "latest": _round(latest, 4),
              "period": rsi_period,
              "series": _maybe_series(rsi_series, include_series, 4, series_tail),
          }
          if latest is None:
              summary["rsi"] = "insufficient data"
          elif latest >= 70:
              summary["rsi"] = f"{latest:.2f} — overbought (>70)"
          elif latest <= 30:
              summary["rsi"] = f"{latest:.2f} — oversold (<30)"
          else:
              summary["rsi"] = f"{latest:.2f} — neutral (30-70)"
  
      # ----- MACD -----
      if "macd" in requested:
          macd_line, signal_line, hist = _macd(closes, macd_fast, macd_slow, macd_signal)
          l_macd, l_sig, l_hist = _last(macd_line), _last(signal_line), _last(hist)
          out["macd"] = {
              "latest": {
                  "macd": _round(l_macd, 6),
                  "signal": _round(l_sig, 6),
                  "histogram": _round(l_hist, 6),
              },
              "params": {"fast": macd_fast, "slow": macd_slow, "signal": macd_signal},
              "series": (
                  {
                      "macd": _maybe_series(macd_line, True, 6, series_tail),
                      "signal": _maybe_series(signal_line, True, 6, series_tail),
                      "histogram": _maybe_series(hist, True, 6, series_tail),
                  }
                  if include_series
                  else None
              ),
          }
          if l_macd is None or l_sig is None or l_hist is None:
              summary["macd"] = "insufficient data"
          else:
              # Use a tiny tolerance so float noise around zero doesn't flip sides.
              tol = 1e-9 * max(abs(l_macd), abs(l_sig), 1.0)
              if l_macd > l_sig + tol:
                  direction, side = "bullish", "above"
              elif l_macd < l_sig - tol:
                  direction, side = "bearish", "below"
              else:
                  direction, side = "neutral", "at"
              summary["macd"] = (
                  f"{direction} — line {side} signal, histogram {l_hist:+.4f}"
              )
  
      # ----- Bollinger -----
      if "bollinger" in requested:
          upper, middle, lower = _bollinger(closes, bb_period, bb_stddev)
          l_up, l_mid, l_lo = _last(upper), _last(middle), _last(lower)
          last_close = closes[-1]
          percent_b = None
          bandwidth = None
          if l_up is not None and l_lo is not None and l_mid is not None:
              rng = l_up - l_lo
              if rng > 0:
                  percent_b = (last_close - l_lo) / rng
              if l_mid != 0:
                  bandwidth = (l_up - l_lo) / l_mid
          out["bollinger"] = {
              "latest": {
                  "upper": _round(l_up, 6),
                  "middle": _round(l_mid, 6),
                  "lower": _round(l_lo, 6),
              },
              "period": bb_period,
              "stddev": bb_stddev,
              "percent_b": _round(percent_b, 4),
              "bandwidth": _round(bandwidth, 6),
              "series": (
                  {
                      "upper": _maybe_series(upper, True, 6, series_tail),
                      "middle": _maybe_series(middle, True, 6, series_tail),
                      "lower": _maybe_series(lower, True, 6, series_tail),
                  }
                  if include_series
                  else None
              ),
          }
          if l_up is None or l_lo is None:
              summary["bollinger"] = "insufficient data"
          elif last_close >= l_up:
              summary["bollinger"] = f"price {last_close:.6g} at/above upper band ({l_up:.6g})"
          elif last_close <= l_lo:
              summary["bollinger"] = f"price {last_close:.6g} at/below lower band ({l_lo:.6g})"
          else:
              summary["bollinger"] = (
                  f"price {last_close:.6g} inside bands [{l_lo:.6g}, {l_up:.6g}]"
              )
  
      # ----- EMA -----
      ema_latest_map: dict[str, float | None] = {}
      if "ema" in requested:
          ema_series_map: dict[str, list[float | None] | None] = {}
          for p in ema_periods:
              s = _ema_series(closes, p)
              ema_latest_map[str(p)] = _round(_last(s), 6)
              ema_series_map[str(p)] = _maybe_series(s, include_series, 6, series_tail)
          out["ema"] = {
              "latest": ema_latest_map,
              "periods": list(ema_periods),
              "series": ema_series_map if include_series else None,
          }
  
      # ----- SMA -----
      if "sma" in requested:
          sma_latest: dict[str, float | None] = {}
          sma_series_map: dict[str, list[float | None] | None] = {}
          for p in sma_periods:
              s = _sma_series(closes, p)
              sma_latest[str(p)] = _round(_last(s), 6)
              sma_series_map[str(p)] = _maybe_series(s, include_series, 6, series_tail)
          out["sma"] = {
              "latest": sma_latest,
              "periods": list(sma_periods),
              "series": sma_series_map if include_series else None,
          }
  
      # ----- ATR -----
      if "atr" in requested:
          atr_series = _atr(highs, lows, closes, atr_period)
          latest_atr = _last(atr_series)
          out["atr"] = {
              "latest": _round(latest_atr, 6),
              "period": atr_period,
              "series": _maybe_series(atr_series, include_series, 6, series_tail),
          }
          if latest_atr is not None and closes[-1] != 0:
              pct = 100.0 * latest_atr / closes[-1]
              summary["atr"] = f"{latest_atr:.6g} ({pct:.2f}% of last close)"
          elif latest_atr is None:
              summary["atr"] = "insufficient data"
  
      # ----- ADX -----
      if "adx" in requested:
          plus_di, minus_di, adx_series = _adx(highs, lows, closes, adx_period)
          l_pdi, l_mdi, l_adx = _last(plus_di), _last(minus_di), _last(adx_series)
          out["adx"] = {
              "latest": {
                  "adx": _round(l_adx, 4),
                  "plus_di": _round(l_pdi, 4),
                  "minus_di": _round(l_mdi, 4),
              },
              "period": adx_period,
              "series": (
                  {
                      "adx": _maybe_series(adx_series, True, 4, series_tail),
                      "plus_di": _maybe_series(plus_di, True, 4, series_tail),
                      "minus_di": _maybe_series(minus_di, True, 4, series_tail),
                  }
                  if include_series
                  else None
              ),
          }
          if l_adx is None:
              summary["adx"] = "insufficient data"
          else:
              strength = (
                  "strong trend" if l_adx >= 25
                  else "weak/no trend" if l_adx < 20
                  else "developing trend"
              )
              dir_txt = ""
              if l_pdi is not None and l_mdi is not None:
                  dir_txt = " (+DI > -DI, bullish)" if l_pdi > l_mdi else " (-DI > +DI, bearish)"
              summary["adx"] = f"{l_adx:.2f} — {strength}{dir_txt}"
  
      # ----- Stochastic -----
      if "stochastic" in requested:
          k_series, d_series = _stochastic(highs, lows, closes, stoch_k_period, stoch_d_period)
          l_k, l_d = _last(k_series), _last(d_series)
          out["stochastic"] = {
              "latest": {"k": _round(l_k, 4), "d": _round(l_d, 4)},
              "k_period": stoch_k_period,
              "d_period": stoch_d_period,
              "series": (
                  {
                      "k": _maybe_series(k_series, True, 4, series_tail),
                      "d": _maybe_series(d_series, True, 4, series_tail),
                  }
                  if include_series
                  else None
              ),
          }
          if l_k is None:
              summary["stochastic"] = "insufficient data"
          elif l_k >= 80:
              summary["stochastic"] = f"%K {l_k:.2f} — overbought (>=80)"
          elif l_k <= 20:
              summary["stochastic"] = f"%K {l_k:.2f} — oversold (<=20)"
          else:
              summary["stochastic"] = f"%K {l_k:.2f} — neutral"
  
      # ----- OBV -----
      if "obv" in requested:
          if not has_volume:
              out["obv"] = {
                  "latest": None,
                  "series": None,
                  "note": (
                      "OBV requires a volume column; the input has 5 columns "
                      "(no volume — likely from get_aggregated_ohlc). Use "
                      "get_exchange_ohlcv for volume-bearing candles."
                  ),
              }
              summary["obv"] = "unavailable (no volume column)"
          else:
              obv_series = _obv(closes, volumes)
              latest_obv = obv_series[-1] if obv_series else None
              out["obv"] = {
                  "latest": _round(latest_obv, 4),
                  "series": _maybe_series(obv_series, include_series, 4, series_tail),
              }
  
      # ----- Trend summary (uses EMAs if available) -----
      e50 = ema_latest_map.get("50") if ema_latest_map else None
      e200 = ema_latest_map.get("200") if ema_latest_map else None
      last_close = closes[-1]
      if e50 is not None and e200 is not None:
          if last_close > e50 > e200:
              summary["trend"] = "uptrend (price > EMA50 > EMA200)"
          elif last_close < e50 < e200:
              summary["trend"] = "downtrend (price < EMA50 < EMA200)"
          else:
              summary["trend"] = (
                  f"mixed (close={last_close:.6g}, EMA50={e50:.6g}, EMA200={e200:.6g})"
              )
  
      out["signal_summary"] = summary
      out["disclaimer"] = (
          "These are observations on the supplied candles, not trading advice."
      )
      return out

• coin_mcp/indicators.py:336-336 (registration)

  The tool is registered via `@mcp.tool()` decorator on the `compute_indicators` function. The `mcp` instance is imported from `coin_mcp.core` (line 22: `from .core import mcp`), which is a `FastMCP` instance created at line 58 of core.py.

  @mcp.tool()

• coin_mcp/indicators.py:36-48 (helper)

  `_sma_series` — helper computing a simple moving average series, aligned with input values; entries before the window are None.

  def _sma_series(values: list[float], period: int) -> list[float | None]:
      """Simple moving average. Returns a list aligned with `values`; entries
      before the window is full are None."""
      n = len(values)
      out: list[float | None] = [None] * n
      if period <= 0 or n < period:
          return out
      window_sum = sum(values[:period])
      out[period - 1] = window_sum / period
      for i in range(period, n):
          window_sum += values[i] - values[i - period]
          out[i] = window_sum / period
      return out

• coin_mcp/indicators.py:51-66 (helper)

  `_ema_series` — helper computing an exponential moving average series seeded with SMA of first `period` values (textbook convention).

  def _ema_series(values: list[float], period: int) -> list[float | None]:
      """Exponential moving average. Seeded with SMA over the first `period`
      values (textbook convention). Aligned with input; pre-seed entries None."""
      n = len(values)
      out: list[float | None] = [None] * n
      if period <= 0 or n < period:
          return out
      k = 2.0 / (period + 1)
      seed = sum(values[:period]) / period
      out[period - 1] = seed
      prev = seed
      for i in range(period, n):
          cur = values[i] * k + prev * (1 - k)
          out[i] = cur
          prev = cur
      return out

• coin_mcp/indicators.py:86-90 (helper)

  `_stddev` — population standard deviation helper used by Bollinger Bands.

  def _stddev(values: list[float], mean: float) -> float:
      """Population standard deviation (matches Bollinger Bands convention)."""
      if not values:
          return 0.0
      return math.sqrt(sum((v - mean) ** 2 for v in values) / len(values))