calculate_ht_trendline
Calculate Hilbert Transform Trendline for financial market analysis to identify trend direction and potential reversals in price data.
Instructions
Calculate Hilbert Transform Trendline.
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:180-192 (handler)MCP tool handler for 'calculate_ht_trendline'. Fetches 'ht_trendline' indicator from registry, creates MarketData from close prices, calls indicator.calculate(), and returns success/values or error.
@mcp.tool() async def calculate_ht_trendline(close: List[float]) -> Dict[str, Any]: try: indicator = registry.get_indicator("ht_trendline") if not indicator: raise ValueError("HT_TRENDLINE indicator not found") market_data = MarketData(close=close) result = await indicator.calculate(market_data, {}) 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)} - HTTrendlineIndicator class implementing the core logic: wraps TA-Lib's ta.HT_TRENDLINE on close prices, handles input/output as IndicatorResult with values['ht_trendline'] and metadata.
class HTTrendlineIndicator(BaseIndicator): def __init__(self): super().__init__(name="ht_trendline", description="Hilbert Transform - Instantaneous Trendline") @property def input_schema(self) -> Dict[str, Any]: return {"type": "object", "properties": {"close_prices": {"type": "array", "items": {"type": "number"}}}, "required": ["close_prices"]} async def calculate(self, market_data: MarketData, options: Dict[str, Any] = None) -> IndicatorResult: close = np.asarray(market_data.close, dtype=float) try: out = ta.HT_TRENDLINE(close) return IndicatorResult( indicator_name=self.name, success=True, values={"ht_trendline": out.tolist()}, metadata={"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)) - Input schema definition for the indicator, specifying 'close_prices' as required array of numbers.
@property def input_schema(self) -> Dict[str, Any]: return {"type": "object", "properties": {"close_prices": {"type": "array", "items": {"type": "number"}}}, "required": ["close_prices"]} - src/mcp_talib/indicators/__init__.py:31-31 (registration)Registration of HTTrendlineIndicator in the IndicatorRegistry under key 'ht_trendline', which is fetched by the handler.
registry.register("ht_trendline", HTTrendlineIndicator)