hl_indicators_server.py

# hl_indicators_server.py
"""
FastMCP server exposing Hyperliquid indicator tools.

This server provides a unified interface to compute common trading indicators
directly from Hyperliquid testnet market data via the `candles_snapshot` API.

Available tools:
- ema        → Exponential Moving Average
- macd       → Moving Average Convergence Divergence
- stoch_rsi  → Stochastic RSI
- adl        → Accumulation / Distribution Line
- obv        → On-Balance Volume
- atr_adx    → Average True Range / Directional Index / ADX
- bbands     → Bollinger Bands
- mfi        → Money Flow Index
- vwap       → Volume-Weighted Average Price
- volume     → Raw trading volume
- bundle     → Compute multiple indicators in one call

Run:
    python hl_indicators_server.py
"""

from __future__ import annotations
from typing import List, Optional, Literal, Dict, Any

from mcp.server.fastmcp import FastMCP
import hl_indicators as hi

Interval = Literal["1m", "5m", "15m", "1h", "4h", "1d"]

mcp = FastMCP("hl_indicators_server")


# ------------------ Health check ------------------ #

@mcp.tool()
async def ping() -> str:
    """Check if the MCP server is online and responding."""
    return "pong"


# ------------------ Indicator tools ------------------ #

@mcp.tool()
async def ema(
    name: str,
    interval: Interval = "1h",
    periods: Optional[List[int]] = None,
    lookback: Optional[int] = None,
    limit: int = 600,
) -> Dict[str, Any]:
    """
    Compute Exponential Moving Averages (EMA).

    Args:
        name: Coin name (e.g. "BTC", "ETH", "HYPE").
        interval: Candle interval ("1m", "5m", "15m", "1h", "4h", "1d").
        periods: List of EMA window lengths (e.g. [20, 200]).
        lookback: Optional shorthand for a single EMA (e.g. 36).
        limit: Number of candles to fetch from the API.

    Notes:
        - `limit` controls how many data points are retrieved; it should be at
          least 2–3× the largest EMA period for accurate results.
        - The function automatically uses Hyperliquid testnet data.

    Returns:
        A dictionary containing EMA series for each period and the most recent values.
    """
    if periods is None and lookback is not None:
        periods = [lookback]
    return hi.get_ema(name=name, periods=periods, interval=interval, limit=limit, testnet=False)


@mcp.tool()
async def macd(
    name: str,
    interval: Interval = "1h",
    fast: int = 12,
    slow: int = 26,
    signal: int = 9,
    limit: int = 600,
) -> Dict[str, Any]:
    """
    Compute the Moving Average Convergence Divergence (MACD).

    Args:
        name: Coin name (e.g. "BTC").
        interval: Candle interval.
        fast: Period for the fast EMA (default: 12).
        slow: Period for the slow EMA (default: 26).
        signal: Period for the MACD signal line (default: 9).
        limit: Number of candles to fetch.

    Returns:
        A dictionary with MACD line, signal line, histogram, and last computed values.
    """
    return hi.get_macd(name=name, fast=fast, slow=slow, signal=signal, interval=interval, limit=limit, testnet=False)


@mcp.tool()
async def stoch_rsi(
    name: str,
    interval: Interval = "1h",
    rsi_length: int = 14,
    stoch_length: int = 14,
    k_smooth: int = 3,
    d_smooth: int = 3,
    limit: int = 600,
) -> Dict[str, Any]:
    """
    Compute the Stochastic RSI oscillator (%K and %D).

    Args:
        name: Coin name.
        interval: Candle interval.
        rsi_length: Period for RSI computation (default: 14).
        stoch_length: Period for Stochastic window (default: 14).
        k_smooth: Smoothing factor for %K (default: 3).
        d_smooth: Smoothing factor for %D (default: 3).
        limit: Number of candles to fetch.

    Returns:
        A dictionary containing %K, %D, and the raw StochRSI values.
    """
    return hi.get_stoch_rsi(
        name=name,
        rsi_length=rsi_length,
        stoch_length=stoch_length,
        k_smooth=k_smooth,
        d_smooth=d_smooth,
        interval=interval,
        limit=limit,
        testnet=False,
    )


@mcp.tool()
async def adl(name: str, interval: Interval = "1h", limit: int = 600) -> Dict[str, Any]:
    """
    Compute the Accumulation/Distribution Line (ADL).

    Args:
        name: Coin name.
        interval: Candle interval.
        limit: Number of candles to fetch.

    Returns:
        A dictionary containing the ADL time series and the latest ADL value.
    """
    return hi.get_adl(name=name, interval=interval, limit=limit, testnet=False)


@mcp.tool()
async def obv(name: str, interval: Interval = "1h", limit: int = 600) -> Dict[str, Any]:
    """
    Compute the On-Balance Volume (OBV).

    Args:
        name: Coin name.
        interval: Candle interval.
        limit: Number of candles to fetch.

    Returns:
        OBV values accumulated over time and the latest OBV.
    """
    return hi.get_obv(name=name, interval=interval, limit=limit, testnet=False)


@mcp.tool()
async def atr_adx(name: str, interval: Interval = "1h", period: int = 14, limit: int = 600) -> Dict[str, Any]:
    """
    Compute volatility and directional indicators: ATR, +DI, -DI, and ADX.

    Args:
        name: Coin name.
        interval: Candle interval.
        period: Lookback for smoothing (default: 14).
        limit: Number of candles to fetch.

    Returns:
        A dictionary with ATR, +DI, -DI, and ADX values.
    """
    return hi.get_atr_adx(name=name, period=period, interval=interval, limit=limit, testnet=False)


@mcp.tool()
async def bbands(
    name: str,
    interval: Interval = "1h",
    period: int = 20,
    std_mult: float = 2.0,
    limit: int = 600,
) -> Dict[str, Any]:
    """
    Compute Bollinger Bands (basis, upper/lower bands, %b, bandwidth).

    Args:
        name: Coin name.
        interval: Candle interval.
        period: Window for SMA (default: 20).
        std_mult: Standard deviation multiplier (default: 2.0).
        limit: Number of candles to fetch.

    Returns:
        A dictionary with band series and the most recent band values.
    """
    return hi.get_bbands(name=name, period=period, std_mult=std_mult, interval=interval, limit=limit, testnet=False)


@mcp.tool()
async def mfi(name: str, interval: Interval = "1h", period: int = 14, limit: int = 600) -> Dict[str, Any]:
    """
    Compute the Money Flow Index (MFI), a volume-weighted momentum oscillator.

    Args:
        name: Coin name.
        interval: Candle interval.
        period: Rolling window (default: 14).
        limit: Number of candles to fetch.

    Returns:
        A dictionary containing MFI series and the most recent value.
    """
    return hi.get_mfi(name=name, period=period, interval=interval, limit=limit, testnet=False)


@mcp.tool()
async def vwap(name: str, interval: Interval = "1h", daily_reset: bool = False, limit: int = 600) -> Dict[str, Any]:
    """
    Compute the Volume-Weighted Average Price (VWAP).

    Args:
        name: Coin name.
        interval: Candle interval.
        daily_reset: If True, VWAP resets each trading day.
        limit: Number of candles to fetch.

    Returns:
        VWAP time series and the last computed VWAP value.
    """
    return hi.get_vwap(name=name, daily_reset=daily_reset, interval=interval, limit=limit, testnet=False)


@mcp.tool()
async def volume(name: str, interval: Interval = "1h", limit: int = 600) -> Dict[str, Any]:
    """
    Retrieve the raw trading volume per candle.

    Args:
        name: Coin name.
        interval: Candle interval.
        limit: Number of candles to fetch.

    Returns:
        Volume values for each candle and the latest volume.
    """
    return hi.get_volume(name=name, interval=interval, limit=limit, testnet=False)


@mcp.tool()
async def bundle(
    name: str,
    interval: Interval = "1h",
    limit: int = 600,
    include: Optional[List[str]] = None,
    ema_periods: Optional[List[int]] = None,
    macd_fast: int = 12,
    macd_slow: int = 26,
    macd_signal: int = 9,
    stoch_rsi_len: int = 14,
    stoch_len: int = 14,
    k_smooth: int = 3,
    d_smooth: int = 3,
    bb_period: int = 20,
    bb_std: float = 2.0,
    mfi_period: int = 14,
    vwap_daily_reset: bool = False,
) -> Dict[str, Any]:
    """
    Compute multiple indicators in a single request.

    Args:
        name: Coin name.
        interval: Candle interval.
        limit: Number of candles to fetch.
        include: List of indicators to include. Default includes all:
            ["ema","macd","stoch_rsi","adl","obv","atr_adx","bbands","mfi","vwap","volume"]
        ema_periods: EMA periods (default: [20, 200]).
        macd_fast / macd_slow / macd_signal: MACD configuration.
        stoch_rsi_len / stoch_len / k_smooth / d_smooth: StochRSI configuration.
        bb_period / bb_std: Bollinger Band configuration.
        mfi_period: Money Flow Index lookback.
        vwap_daily_reset: Whether VWAP resets daily.

    Returns:
        A combined dictionary with all requested indicators.
    """
    return hi.get_bundle(
        name=name,
        interval=interval,
        limit=limit,
        testnet=False,
        include=include or ("ema","macd","stoch_rsi","adl","obv","atr_adx","bbands","mfi","vwap","volume"),
        ema_periods=ema_periods,
        macd_fast=macd_fast,
        macd_slow=macd_slow,
        macd_signal=macd_signal,
        stoch_rsi_len=stoch_rsi_len,
        stoch_len=stoch_len,
        k_smooth=k_smooth,
        d_smooth=d_smooth,
        bb_period=bb_period,
        bb_std=bb_std,
        mfi_period=mfi_period,
        vwap_daily_reset=vwap_daily_reset,
    )


# ------------------ Entry point ------------------ #

if __name__ == "__main__":
    mcp.run(transport='stdio')