crypto-mcp/services/__init__.py

"""Service layer."""

import time
from config import DEFAULT_OHLCV_LIMIT, MAX_OHLCV_LIMIT
from cache import get_cached_price, set_cached_price, get_cached_ohlcv, set_cached_ohlcv
import providers
import indicators as ind_module


async def get_price(symbol: str) -> dict:
    symbol = symbol.upper()
    cached = get_cached_price(symbol)
    if cached:
        return cached
    data = await providers.fetch_price(symbol)
    set_cached_price(symbol, data)
    return data


async def get_ohlcv(symbol: str, timeframe: str, limit: int = DEFAULT_OHLCV_LIMIT) -> dict:
    symbol = symbol.upper()
    limit = min(max(limit, 1), MAX_OHLCV_LIMIT)
    cached = get_cached_ohlcv(symbol, timeframe)
    if cached:
        result = dict(cached)
        result["candles"] = cached["candles"][-limit:]
        return result
    data = await providers.fetch_ohlcv(symbol, timeframe, limit=MAX_OHLCV_LIMIT)
    set_cached_ohlcv(symbol, timeframe, data)
    result = dict(data)
    result["candles"] = data["candles"][-limit:]
    return result


async def get_indicator(symbol: str, indicator: str, timeframe: str = "1h", params: dict = None, limit: int = 200) -> dict:
    params = params or {}
    symbol = symbol.upper()
    ohlcv_data = await get_ohlcv(symbol, timeframe, limit=limit)
    result = ind_module.compute_indicator(ohlcv_data["candles"], indicator, params)
    return {"symbol": symbol, "indicator": result["indicator"], "timeframe": timeframe, "value": result["value"], "timestamp": int(time.time())}


async def get_market_snapshot(symbol: str) -> dict:
    symbol = symbol.upper()
    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})]:
        try:
            snapshot[key] = ind_module.compute_indicator(candles, ind, params)["value"]
        except Exception:
            pass
    return snapshot


async def get_top_movers(limit: int = 10) -> dict:
    return await providers.fetch_top_movers(min(max(limit, 1), 50))