calculate_wma
Compute the Weighted Moving Average for financial data analysis. This tool applies greater weight to recent data points to identify market trends and price movements.
Instructions
Calculate Weighted Moving Average (WMA).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| kwargs | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/mcp_talib/core/server.py:350-362 (handler)The MCP tool handler for 'calculate_wma', decorated with @mcp.tool(). It retrieves the 'wma' indicator from the registry, prepares market data, calls the indicator's calculate method, and returns the result.
@mcp.tool() async def calculate_wma(close: List[float], timeperiod: int = 30) -> Dict[str, Any]: try: indicator = registry.get_indicator("wma") if not indicator: raise ValueError("WMA indicator not found") market_data = MarketData(close=close) result = await indicator.calculate(market_data, {"timeperiod": timeperiod}) if result.success: return {"success": True, "values": result.values, "metadata": result.metadata} return {"success": False, "error": result.error_message} except Exception as e: return {"success": False, "error": str(e)} - Core implementation of the WMA calculation using TA-Lib's WMA function. This is the supporting utility called by the tool handler.
async def calculate(self, market_data: MarketData, options: Dict[str, Any] = None) -> IndicatorResult: if options is None: options = {} timeperiod = options.get("timeperiod", 30) close = np.asarray(market_data.close, dtype=float) try: out = ta.WMA(close, timeperiod=timeperiod) return IndicatorResult(indicator_name=self.name, success=True, values={"wma": out.tolist()}, metadata={"timeperiod": timeperiod, "input_points": len(close), "output_points": len(out)}) except Exception as e: return IndicatorResult(indicator_name=self.name, success=False, values={}, error_message=str(e)) - src/mcp_talib/indicators/__init__.py:43-43 (registration)Registers the WMAIndicator class in the global indicator registry, allowing the tool handler to retrieve it via registry.get_indicator('wma').
registry.register("wma", WMAIndicator) - Input schema definition for the WMA indicator, specifying expected parameters like close_prices and timeperiod.
def input_schema(self) -> Dict[str, Any]: return {"type": "object", "properties": {"close_prices": {"type": "array", "items": {"type": "number"}}, "timeperiod": {"type": "integer", "default": 30}}, "required": ["close_prices"]}