from __future__ import annotations from typing import Any from .config import DB_PATH from .storage import last_candle, latest_candles, stats from .swissquote import SwissquoteClient client = SwissquoteClient() def get_capabilities() -> dict[str, Any]: return { "server": "metals-mcp", "transport": "fastmcp+sse", "tools": [ "get_price", "get_ohlcv", "get_indicator", "get_market_snapshot", "get_top_movers", "get_capabilities", "get_regime", ], "backend": "swissquote-poller+sqlite", "database": str(DB_PATH), } def get_price(symbol: str) -> dict[str, Any]: quote = client.fetch_quote(symbol) if not quote: return {"symbol": symbol, "price": None, "timestamp": None, "source": "swissquote", "status": "unavailable"} return {"symbol": symbol, "price": quote.mid, "timestamp": quote.timestamp, "source": "swissquote"} def get_ohlcv(symbol: str, timeframe: str = "5m", limit: int = 100) -> dict[str, Any]: candles = latest_candles(DB_PATH, symbol.upper(), timeframe, limit) return {"symbol": symbol.upper(), "timeframe": timeframe, "limit": limit, "candles": candles} def get_candles(symbol: str, timeframe: str = "5m", limit: int = 100) -> dict[str, Any]: return get_ohlcv(symbol, timeframe=timeframe, limit=limit) def get_last_candle(symbol: str, timeframe: str = "5m") -> dict[str, Any]: return {"symbol": symbol.upper(), "timeframe": timeframe, "candle": last_candle(DB_PATH, symbol.upper(), timeframe)} def get_indicator(symbol: str, indicator: str, timeframe: str = "5m", params: dict[str, Any] | None = None) -> dict[str, Any]: return {"symbol": symbol, "indicator": indicator, "timeframe": timeframe, "params": params or {}, "value": None, "status": "scaffolded"} def get_market_snapshot(symbol: str) -> dict[str, Any]: candle = last_candle(DB_PATH, symbol.upper(), "5m") price = candle["close"] if candle else None return {"symbol": symbol.upper(), "price": price, "trend_bias": "range", "status": "scaffolded"} def get_top_movers(limit: int = 10) -> dict[str, Any]: return {"limit": limit, "movers": [], "status": "scaffolded"} def get_regime(symbol: str, timeframe: str = "5m") -> dict[str, Any]: candles = latest_candles(DB_PATH, symbol.upper(), timeframe, 20) return {"symbol": symbol.upper(), "timeframe": timeframe, "candles": candles, "regime": None, "status": "scaffolded"} def get_health() -> dict[str, Any]: try: store = stats(DB_PATH) except Exception: store = {"ticks": 0, "candles": 0} return {"ok": True, "store": store}