Improve crypto MCP: indicators, capabilities, regime, snapshot docs

README.md

@@ -41,9 +41,13 @@ docker compose down
 
 - `get_price`
 - `get_ohlcv`
-- `get_indicator`
+- `get_indicator` (supports: `rsi`, `ema`, `sma`, `macd`, `atr`, `bollinger`, `vwap` — each with optional params)
 - `get_market_snapshot`
 - `get_top_movers`
+- `get_capabilities`
+- `get_regime`
+
+`get_regime` returns a composite view (trend EMA/SMA regime, RSI + MACD momentum, ATR-based volatility, Bollinger bands, VWAP) for a given symbol/timeframe.
 
 ## Health
 

indicators/__init__.py

@@ -1,4 +1,7 @@
-"""Technical indicators."""
+"""Technical indicators and metadata for the crypto MCP."""
+
+from collections import OrderedDict
+from math import sqrt
 
 from errors import InsufficientDataError, UnsupportedIndicatorError
 
@@ -7,6 +10,18 @@ def _closes(candles: list) -> list[float]:
     return [float(c[4]) for c in candles]
 
 
+def _highs(candles: list) -> list[float]:
+    return [float(c[2]) for c in candles]
+
+
+def _lows(candles: list) -> list[float]:
+    return [float(c[3]) for c in candles]
+
+
+def _volumes(candles: list) -> list[float]:
+    return [float(c[5]) for c in candles]
+
+
 def _ema_series(values: list[float], period: int) -> list[float]:
     if len(values) < period:
         raise InsufficientDataError(f"EMA({period}) requires at least {period} candles, got {len(values)}")
@@ -60,16 +75,167 @@ def sma(candles: list, period: int = 20) -> float:
     return round(sum(closes[-period:]) / period, 6)
 
 
+def atr(candles: list, period: int = 14) -> float:
+    highs = _highs(candles)
+    lows = _lows(candles)
+    closes = _closes(candles)
+    if len(closes) < period + 1:
+        raise InsufficientDataError(f"ATR({period}) requires at least {period + 1} candles, got {len(closes)}")
+    true_ranges = []
+    for i in range(1, len(candles)):
+        high = highs[i]
+        low = lows[i]
+        prev_close = closes[i - 1]
+        tr = max(high - low, abs(high - prev_close), abs(low - prev_close))
+        true_ranges.append(tr)
+    first_atr = sum(true_ranges[:period]) / period
+    atr_val = first_atr
+    for tr in true_ranges[period:]:
+        atr_val = ((atr_val * (period - 1)) + tr) / period
+    return round(atr_val, 6)
+
+
+def bollinger_bands(candles: list, period: int = 20, multiplier: float = 2.0) -> dict:
+    closes = _closes(candles)
+    if len(closes) < period:
+        raise InsufficientDataError(f"Bollinger Bands require at least {period} candles, got {len(closes)}")
+    window = closes[-period:]
+    middle = sum(window) / period
+    variance = sum((price - middle) ** 2 for price in window) / period
+    std_dev = sqrt(variance)
+    upper = middle + multiplier * std_dev
+    lower = middle - multiplier * std_dev
+    return {"middle": round(middle, 6), "upper": round(upper, 6), "lower": round(lower, 6)}
+
+
+def vwap(candles: list, period: int | None = None) -> float:
+    if period is not None and period <= 0:
+        raise InsufficientDataError("VWAP period must be positive")
+    subset = candles[-period:] if period is not None else candles
+    if len(subset) < 1:
+        raise InsufficientDataError("VWAP requires at least 1 candle")
+    volumes = _volumes(subset)
+    if sum(volumes) == 0:
+        raise InsufficientDataError("VWAP requires non-zero volume")
+    typical_prices = [((float(c[2]) + float(c[3]) + float(c[4])) / 3) for c in subset]
+    pv = sum(tp * vol for tp, vol in zip(typical_prices, volumes))
+    total_vol = sum(volumes)
+    return round(pv / total_vol, 6)
+
+
+def _rsi_handler(candles: list, params: dict):
+    return rsi(candles, period=int(params.get("period", 14)))
+
+
+def _ema_handler(candles: list, params: dict):
+    return ema(candles, period=int(params.get("period", 20)))
+
+
+def _macd_handler(candles: list, params: dict):
+    return macd(
+        candles,
+        fast_period=int(params.get("fast_period", 12)),
+        slow_period=int(params.get("slow_period", 26)),
+        signal_period=int(params.get("signal_period", 9)),
+    )
+
+
+def _sma_handler(candles: list, params: dict):
+    return sma(candles, period=int(params.get("period", 20)))
+
+
+def _atr_handler(candles: list, params: dict):
+    return atr(candles, period=int(params.get("period", 14)))
+
+
+def _bollinger_handler(candles: list, params: dict):
+    return bollinger_bands(
+        candles,
+        period=int(params.get("period", 20)),
+        multiplier=float(params.get("multiplier", 2.0)),
+    )
+
+
+def _vwap_handler(candles: list, params: dict):
+    period = params.get("period")
+    if period is not None:
+        period = int(period)
+    return vwap(candles, period=period)
+
+
+SUPPORTED_INDICATORS = OrderedDict(
+    {
+        "rsi": {
+            "description": "Relative Strength Index (RSI) — momentum oscillator (0-100) derived from closing price gains/losses; higher values indicate stronger upward pressure.",
+            "handler": _rsi_handler,
+            "params": {"period": {"type": "integer", "default": 14, "min": 2}},
+            "value_type": "number",
+        },
+        "ema": {
+            "description": "Exponential Moving Average (EMA) — weighted moving average emphasizing recent closes to highlight near-term direction.",
+            "handler": _ema_handler,
+            "params": {"period": {"type": "integer", "default": 20, "min": 2}},
+            "value_type": "number",
+        },
+        "sma": {
+            "description": "Simple Moving Average (SMA) — unweighted rolling average of closes, useful for longer-term baselines (e.g., 200-period trend).",
+            "handler": _sma_handler,
+            "params": {"period": {"type": "integer", "default": 20, "min": 2}},
+            "value_type": "number",
+        },
+        "macd": {
+            "description": "Moving Average Convergence Divergence (MACD) — returns MACD, signal, and histogram values for spotting momentum shifts between fast/slow EMAs.",
+            "handler": _macd_handler,
+            "params": {
+                "fast_period": {"type": "integer", "default": 12, "min": 2},
+                "slow_period": {"type": "integer", "default": 26, "min": 3},
+                "signal_period": {"type": "integer", "default": 9, "min": 2},
+            },
+            "value_type": "object",
+        },
+        "atr": {
+            "description": "Average True Range (ATR) — classic volatility gauge derived from true range; higher values imply wider expected movement.",
+            "handler": _atr_handler,
+            "params": {"period": {"type": "integer", "default": 14, "min": 2}},
+            "value_type": "number",
+        },
+        "bollinger": {
+            "description": "Bollinger Bands — middle SMA with upper/lower bands offset by standard deviations; helpful for squeeze/mean-reversion checks.",
+            "handler": _bollinger_handler,
+            "params": {
+                "period": {"type": "integer", "default": 20, "min": 2},
+                "multiplier": {"type": "number", "default": 2.0, "min": 0.5},
+            },
+            "value_type": "object",
+        },
+        "vwap": {
+            "description": "Volume Weighted Average Price (VWAP) — rolling average price weighted by traded volume; useful as an intraday fair-value anchor.",
+            "handler": _vwap_handler,
+            "params": {"period": {"type": "integer", "default": None, "nullable": True}},
+            "value_type": "number",
+        },
+    }
+)
+
+
+def get_supported_indicators() -> list[dict]:
+    entries: list[dict] = []
+    for name, meta in SUPPORTED_INDICATORS.items():
+        entries.append(
+            {
+                "name": name,
+                "description": meta["description"],
+                "params": meta["params"],
+                "value_type": meta.get("value_type", "number"),
+            }
+        )
+    return entries
+
+
 def compute_indicator(candles: list, indicator: str, params: dict) -> dict:
     ind = indicator.lower()
-    if ind == "rsi":
-        value = rsi(candles, period=int(params.get("period", 14)))
-    elif ind == "ema":
-        value = ema(candles, period=int(params.get("period", 20)))
-    elif ind == "macd":
-        value = macd(candles, fast_period=int(params.get("fast_period", 12)), slow_period=int(params.get("slow_period", 26)), signal_period=int(params.get("signal_period", 9)))
-    elif ind == "sma":
-        value = sma(candles, period=int(params.get("period", 20)))
-    else:
+    if ind not in SUPPORTED_INDICATORS:
         raise UnsupportedIndicatorError(f"Unsupported indicator: {indicator}")
+    handler = SUPPORTED_INDICATORS[ind]["handler"]
+    value = handler(candles, params)
     return {"indicator": ind, "value": value}

mcp_tools.py

@@ -1,9 +1,55 @@
 """MCP tool definitions."""
 
 MCP_TOOLS = [
-    {"name": "get_price", "description": "Get the current USD price of a cryptocurrency.", "inputSchema": {"type": "object", "properties": {"symbol": {"type": "string"}}, "required": ["symbol"]}},
-    {"name": "get_ohlcv", "description": "Get OHLCV candlestick data for a crypto asset.", "inputSchema": {"type": "object", "properties": {"symbol": {"type": "string"}, "timeframe": {"type": "string", "default": "1h"}, "limit": {"type": "integer", "default": 100}}, "required": ["symbol"]}},
-    {"name": "get_indicator", "description": "Compute a technical indicator for a crypto asset.", "inputSchema": {"type": "object", "properties": {"symbol": {"type": "string"}, "indicator": {"type": "string"}, "timeframe": {"type": "string", "default": "1h"}, "params": {"type": "object", "default": {}}}, "required": ["symbol", "indicator"]}},
-    {"name": "get_market_snapshot", "description": "Get a compact market snapshot for a crypto asset.", "inputSchema": {"type": "object", "properties": {"symbol": {"type": "string"}}, "required": ["symbol"]}},
-    {"name": "get_top_movers", "description": "Get top gaining and losing crypto assets by 24h % change.", "inputSchema": {"type": "object", "properties": {"limit": {"type": "integer", "default": 10}}, "required": []}},
+    {
+        "name": "get_price",
+        "description": "Return the latest USD spot price for a symbol (e.g., BTC, ETH). Uses Binance primary with CoinGecko fallback; response includes symbol, price, timestamp.",
+        "inputSchema": {"type": "object", "properties": {"symbol": {"type": "string"}}, "required": ["symbol"]},
+    },
+    {
+        "name": "get_ohlcv",
+        "description": "Fetch OHLCV candles for a symbol/timeframe (1m, 5m, 15m, 1h, 4h, 1d). Limit 1-500 candles; returned order is oldest→newest with [ts, open, high, low, close, volume] arrays.",
+        "inputSchema": {"type": "object", "properties": {"symbol": {"type": "string"}, "timeframe": {"type": "string", "default": "1h"}, "limit": {"type": "integer", "default": 100}}, "required": ["symbol"]},
+    },
+    {
+        "name": "get_indicator",
+        "description": "Compute a technical indicator for a symbol/timeframe. Supported names: rsi, ema, sma, macd, atr, bollinger, vwap. Provide params like period, fast/slow/signal periods (MACD) or multiplier (Bollinger).",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "symbol": {"type": "string"},
+                "indicator": {"type": "string"},
+                "timeframe": {"type": "string", "default": "1h"},
+                "params": {"type": "object", "default": {}},
+            },
+            "required": ["symbol", "indicator"],
+        },
+    },
+    {
+        "name": "get_market_snapshot",
+        "description": "Lightweight 1h snapshot with price, RSI14, EMA20/50/200, MACD histogram, ATR (absolute & %), Bollinger(20,2) bands, VWAP(48), and a derived trend_bias flag.",
+        "inputSchema": {"type": "object", "properties": {"symbol": {"type": "string"}}, "required": ["symbol"]},
+    },
+    {
+        "name": "get_top_movers",
+        "description": "List top 24h gainers/losers (limit 1-50). Each entry includes symbol and % change for quick leaderboards.",
+        "inputSchema": {"type": "object", "properties": {"limit": {"type": "integer", "default": 10}}, "required": []},
+    },
+    {
+        "name": "get_capabilities",
+        "description": "Describe the server surface: indicator catalog (with params/defaults) plus allowed OHLCV/indicator/regime timeframes.",
+        "inputSchema": {"type": "object", "properties": {}, "required": []},
+    },
+    {
+        "name": "get_regime",
+        "description": "Return a composite regime snapshot for a symbol/timeframe with trend (EMA20/50 + SMA200), momentum (RSI, MACD hist), volatility (ATR + %), Bollinger bands, VWAP, and bull/bear/range states.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "symbol": {"type": "string"},
+                "timeframe": {"type": "string", "default": "1h"},
+            },
+            "required": ["symbol"],
+        },
+    },
 ]

server_fastmcp.py

@@ -17,31 +17,77 @@ mcp = FastMCP(
 )
 
 
-@mcp.tool(description="Get the current USD price of a cryptocurrency.")
+@mcp.tool(
+    description=(
+        "Return the latest USD spot price for a symbol (e.g., BTC, ETH). Uses Binance as primary data source "
+        "with CoinGecko fallback; payload includes symbol, price, and provider timestamp."
+    )
+)
 async def get_price(symbol: str):
     return await services.get_price(symbol)
 
 
-@mcp.tool(description="Get OHLCV candlestick data for a crypto asset.")
+@mcp.tool(
+    description=(
+        "Fetch OHLCV candles for a symbol/timeframe (1m, 5m, 15m, 1h, 4h, 1d). "
+        "Limit can range from 1-500 candles; candles are returned oldest→newest with [timestamp, open, high, low, close, volume]."
+    )
+)
 async def get_ohlcv(symbol: str, timeframe: str = "1h", limit: int = 100):
     return await services.get_ohlcv(symbol, timeframe, limit)
 
 
-@mcp.tool(description="Compute a technical indicator for a crypto asset.")
+@mcp.tool(
+    description=(
+        "Compute a technical indicator for a symbol/timeframe. Supported names: rsi, ema, sma, macd, atr, bollinger, vwap. "
+        "Pass params such as period, fast_period/slow_period/signal_period (MACD) or multiplier (Bollinger). Output echoes the indicator name and value object."
+    )
+)
 async def get_indicator(symbol: str, indicator: str, timeframe: str = "1h", params: dict | None = None):
     return await services.get_indicator(symbol, indicator, timeframe, params or {})
 
 
-@mcp.tool(description="Get a compact market snapshot for a crypto asset.")
+@mcp.tool(
+    description=(
+        "Return a lightweight 1h snapshot: price, RSI14, EMA20/50/200, MACD histogram, ATR (abs & %), Bollinger(20,2) bands, "
+        "rolling VWAP, and a simple trend_bias derived from EMA20 vs EMA50. Ideal for dashboards needing core stats."
+    )
+)
 async def get_market_snapshot(symbol: str):
     return await services.get_market_snapshot(symbol)
 
 
-@mcp.tool(description="Get top gaining and losing crypto assets by 24h % change.")
+@mcp.tool(
+    description=(
+        "List top 24h movers (gainers/losers) from the provider feed. Limit 1-50 entries; each item includes symbol, price change %, "
+        "and supporting metadata for quick leaderboard views."
+    )
+)
 async def get_top_movers(limit: int = 10):
     return await services.get_top_movers(limit)
 
 
+@mcp.tool(
+    description=(
+        "Describe what this server supports: indicator catalog (with params/defaults) plus allowed timeframes "
+        "for OHLCV/indicator/regime calls. Use this to self-discover valid inputs before invoking other tools."
+    )
+)
+async def get_capabilities():
+    return await services.get_capabilities()
+
+
+@mcp.tool(
+    description=(
+        "Return a composite regime snapshot that merges trend (EMA20/50, SMA200), momentum (RSI, MACD histogram), "
+        "volatility (ATR + % of price), Bollinger bands, and VWAP for the requested symbol/timeframe. "
+        "Fields include derived bull/bear/range states for quick downstream logic."
+    )
+)
+async def get_regime(symbol: str, timeframe: str = "1h"):
+    return await services.get_regime(symbol, timeframe)
+
+
 app = FastAPI(title="Crypto MCP Server")
 app.mount("/mcp", mcp.sse_app())
 

services/__init__.py

@@ -1,7 +1,7 @@
 """Service layer."""
 
 import time
-from config import DEFAULT_OHLCV_LIMIT, MAX_OHLCV_LIMIT
+from config import DEFAULT_OHLCV_LIMIT, MAX_OHLCV_LIMIT, TIMEFRAME_TO_BINANCE
 from cache import get_cached_price, set_cached_price, get_cached_ohlcv, set_cached_ohlcv
 import providers
 import indicators as ind_module
@@ -45,14 +45,175 @@ async def get_market_snapshot(symbol: str) -> dict:
     price_data = await get_price(symbol)
     ohlcv_data = await get_ohlcv(symbol, "1h", limit=200)
     candles = ohlcv_data["candles"]
-    snapshot = {"symbol": symbol, "price": price_data["price"], "rsi_1h": None, "ema_20_1h": None, "ema_50_1h": None, "timestamp": price_data["timestamp"]}
-    for key, ind, params in [("rsi_1h", "rsi", {"period": 14}), ("ema_20_1h", "ema", {"period": 20}), ("ema_50_1h", "ema", {"period": 50})]:
+    price = price_data["price"]
+    snapshot = {
+        "symbol": symbol,
+        "price": price,
+        "rsi_1h": None,
+        "ema_20_1h": None,
+        "ema_50_1h": None,
+        "ema_200_1h": None,
+        "macd_histogram_1h": None,
+        "atr_1h": None,
+        "atr_percent_1h": None,
+        "bollinger_1h": None,
+        "vwap_1h": None,
+        "trend_bias": None,
+        "timestamp": price_data["timestamp"],
+    }
+
+    def _compute(name: str, params: dict):
+        return ind_module.compute_indicator(candles, name, params)["value"]
+
+    for key, ind, params in [
+        ("rsi_1h", "rsi", {"period": 14}),
+        ("ema_20_1h", "ema", {"period": 20}),
+        ("ema_50_1h", "ema", {"period": 50}),
+        ("ema_200_1h", "sma", {"period": 200}),
+    ]:
         try:
-            snapshot[key] = ind_module.compute_indicator(candles, ind, params)["value"]
+            snapshot[key] = _compute(ind, params)
         except Exception:
-            pass
+            continue
+
+    try:
+        macd_vals = _compute("macd", {"fast_period": 12, "slow_period": 26, "signal_period": 9})
+        if isinstance(macd_vals, dict):
+            snapshot["macd_histogram_1h"] = macd_vals.get("histogram")
+    except Exception:
+        pass
+
+    try:
+        atr_val = _compute("atr", {"period": 14})
+        snapshot["atr_1h"] = atr_val
+        if atr_val is not None and price:
+            snapshot["atr_percent_1h"] = round((atr_val / price) * 100, 4)
+    except Exception:
+        pass
+
+    try:
+        snapshot["bollinger_1h"] = _compute("bollinger", {"period": 20, "multiplier": 2.0})
+    except Exception:
+        pass
+
+    try:
+        snapshot["vwap_1h"] = _compute("vwap", {"period": 48})
+    except Exception:
+        pass
+
+    ema_fast = snapshot.get("ema_20_1h")
+    ema_slow = snapshot.get("ema_50_1h")
+    if ema_fast is not None and ema_slow not in (None, 0):
+        delta = (ema_fast - ema_slow) / ema_slow
+        if delta > 0.002:
+            snapshot["trend_bias"] = "bull"
+        elif delta < -0.002:
+            snapshot["trend_bias"] = "bear"
+        else:
+            snapshot["trend_bias"] = "range"
     return snapshot
 
 
 async def get_top_movers(limit: int = 10) -> dict:
     return await providers.fetch_top_movers(min(max(limit, 1), 50))
+
+
+async def get_capabilities() -> dict:
+    timeframe_descriptions = {
+        "1m": "1-minute candles — ultra-fast scalping / heartbeat data (highest volatility, shortest TTL)",
+        "5m": "5-minute candles — intraday momentum and micro-structure",
+        "15m": "15-minute candles — short-term swings, aligns with many bot cycles",
+        "1h": "1-hour candles — default for balanced trend/momentum reads",
+        "4h": "4-hour candles — swing/position traders",
+        "1d": "1-day candles — macro trend / higher timeframe context",
+    }
+    timeframes = []
+    for tf, interval in TIMEFRAME_TO_BINANCE.items():
+        timeframes.append(
+            {
+                "timeframe": tf,
+                "provider_interval": interval,
+                "description": timeframe_descriptions.get(tf, ""),
+            }
+        )
+    return {
+        "description": "Supported technical indicators (with params/defaults) and the timeframes you can request via get_ohlcv / get_indicator / get_regime.",
+        "indicators": ind_module.get_supported_indicators(),
+        "timeframes": timeframes,
+    }
+
+
+async def get_regime(symbol: str, timeframe: str = "1h", limit: int = 200) -> dict:
+    symbol = symbol.upper()
+    ohlcv_data = await get_ohlcv(symbol, timeframe, limit=limit)
+    candles = ohlcv_data["candles"]
+    close_price = float(candles[-1][4]) if candles else None
+    timestamp = int(time.time())
+
+    def _compute(name: str, params: dict | None = None):
+        try:
+            return ind_module.compute_indicator(candles, name, params or {})["value"]
+        except Exception:
+            return None
+
+    ema_fast = _compute("ema", {"period": 20})
+    ema_slow = _compute("ema", {"period": 50})
+    sma_long = _compute("sma", {"period": 200})
+    atr_val = _compute("atr", {"period": 14})
+    boll = _compute("bollinger", {"period": 20, "multiplier": 2.0})
+    vwap_val = _compute("vwap", {"period": 48})
+    rsi_val = _compute("rsi", {"period": 14})
+    macd_vals = _compute("macd", {"fast_period": 12, "slow_period": 26, "signal_period": 9})
+
+    regime_delta = None
+    if ema_fast is not None and ema_slow is not None and ema_slow != 0:
+        regime_delta = (ema_fast - ema_slow) / ema_slow
+    if regime_delta is None:
+        trend_state = "unknown"
+    elif regime_delta > 0.002:
+        trend_state = "bull"
+    elif regime_delta < -0.002:
+        trend_state = "bear"
+    else:
+        trend_state = "range"
+
+    if rsi_val is None:
+        momentum_state = "unknown"
+    elif rsi_val >= 60:
+        momentum_state = "bull"
+    elif rsi_val <= 40:
+        momentum_state = "bear"
+    else:
+        momentum_state = "neutral"
+
+    atr_percent = None
+    if atr_val is not None and close_price:
+        atr_percent = round((atr_val / close_price) * 100, 4)
+
+    macd_hist = macd_vals.get("histogram") if isinstance(macd_vals, dict) else None
+
+    return {
+        "symbol": symbol,
+        "timeframe": timeframe,
+        "timestamp": timestamp,
+        "price": close_price,
+        "trend": {
+            "ema_fast": ema_fast,
+            "ema_slow": ema_slow,
+            "sma_long": sma_long,
+            "state": trend_state,
+        },
+        "momentum": {
+            "rsi": rsi_val,
+            "macd_histogram": macd_hist,
+            "state": momentum_state,
+        },
+        "volatility": {
+            "atr": atr_val,
+            "atr_percent": atr_percent,
+        },
+        "bands": {
+            "bollinger": boll,
+        },
+        "vwap": vwap_val,
+    }

tests.py

@@ -145,9 +145,19 @@ class TestDispatcher:
         assert result["indicator"] == "macd"
         assert "histogram" in result["value"]
 
+    def test_bollinger(self):
+        result = compute_indicator(CANDLES, "bollinger", {"period": 20, "multiplier": 2})
+        assert result["indicator"] == "bollinger"
+        assert {"upper", "middle", "lower"}.issubset(result["value"].keys())
+
+    def test_vwap(self):
+        result = compute_indicator(CANDLES, "vwap", {})
+        assert result["indicator"] == "vwap"
+        assert isinstance(result["value"], float)
+
     def test_unsupported(self):
         with pytest.raises(UnsupportedIndicatorError):
-            compute_indicator(CANDLES, "bollinger", {})
+            compute_indicator(CANDLES, "madeup", {})
 
     def test_case_insensitive(self):
         result = compute_indicator(CANDLES, "RSI", {"period": 14})