calculate_kama
Compute the Kaufman Adaptive Moving Average (KAMA) to analyze financial market trends by adapting to price volatility, enabling technical analysis of price data.
Instructions
Calculate Kaufman Adaptive Moving Average (KAMA).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| kwargs | Yes |
Implementation Reference
- src/mcp_talib/core/server.py:194-206 (handler)MCP tool handler for 'calculate_kama'. Retrieves the 'kama' indicator from the registry, creates MarketData, calls its calculate method, and returns the result.@mcp.tool() async def calculate_kama(close: List[float], timeperiod: int = 10) -> Dict[str, Any]: try: indicator = registry.get_indicator("kama") if not indicator: raise ValueError("KAMA 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)}
- KAMAIndicator class implementing the core logic using talib.KAMA. Includes input schema property and the calculate method that performs the computation.class KAMAIndicator(BaseIndicator): def __init__(self): super().__init__(name="kama", description="Kaufman Adaptive Moving Average (KAMA)") @property def input_schema(self) -> Dict[str, Any]: return {"type": "object", "properties": {"close_prices": {"type": "array", "items": {"type": "number"}}, "timeperiod": {"type": "integer", "default": 10}}, "required": ["close_prices"]} async def calculate(self, market_data: MarketData, options: Dict[str, Any] = None) -> IndicatorResult: if options is None: options = {} timeperiod = options.get("timeperiod", 10) close = np.asarray(market_data.close, dtype=float) try: out = ta.KAMA(close, timeperiod=timeperiod) return IndicatorResult(indicator_name=self.name, success=True, values={"kama": 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:32-32 (registration)Registers the KAMAIndicator class in the indicator registry under the name 'kama'.registry.register("kama", KAMAIndicator)
- src/mcp_talib/indicators/__init__.py:11-11 (registration)Imports the KAMAIndicator class required for registration.from .kama import KAMAIndicator