get_best_four_points
Analyzes Taiwan stock market data to identify optimal entry and exit points for a specific stock using the Best Four Points method.
Instructions
Get Best Four Points analysis for a specific stock.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| stock_code | Yes |
Implementation Reference
- Core handler implementing the Best Four Points analysis using twstock.Stock to fetch historical data, twstock.BestFourPoint for analysis, computing buy_points, sell_points, and overall analysis. Includes caching, validation, circuit breaker, and comprehensive error handling.
@with_async_error_handling(operation="get_best_four_points") @with_retry(max_retries=2, base_delay=1.0) async def get_best_four_points(self, stock_code: str) -> dict[str, Any]: """ 獲取四大買賣點分析 Args: stock_code: 股票代號 Returns: 四大買賣點分析結果 Raises: InvalidStockCodeError: 股票代號格式錯誤 StockNotFoundError: 找不到股票 StockDataUnavailableError: 分析資料無法取得 """ # Track request for performance monitoring self._performance_monitor.record_request() # Validate stock code validated_code = StockCodeValidator.validate_stock_code(stock_code) cache_key = f"best_four_points_{validated_code}" cached_data = self._get_cached_data(cache_key) if cached_data: return cached_data try: # Use circuit breaker for external API calls stock = await self.circuit_breaker.acall( asyncio.to_thread, Stock, validated_code ) bfp = await self.circuit_breaker.acall( asyncio.to_thread, BestFourPoint, stock ) # Check if we have enough data for analysis if not stock.data or len(stock.data) < 20: # Need at least 20 days for meaningful analysis raise StockDataUnavailableError( stock_code=validated_code, data_type="analysis data", message="Insufficient historical data for Best Four Points analysis (minimum 20 days required)" ) data = { "stock_code": validated_code, "buy_points": bfp.best_four_point_to_buy(), "sell_points": bfp.best_four_point_to_sell(), "analysis": bfp.best_four_point(), "updated_at": datetime.now().isoformat(), "data_points": len(stock.data) } # 使用標籤進行分組管理 tags = ["best_four_points", f"stock_{validated_code}", "analysis"] self._set_cached_data(cache_key, data, self.cache_ttl['best_four_points'], tags) return data except (StockDataUnavailableError, StockNotFoundError): # Track error and re-raise specific stock errors self._performance_monitor.record_error() raise except Exception as e: # Track error and convert generic exceptions to appropriate stock errors self._performance_monitor.record_error() if "connection" in str(e).lower() or "timeout" in str(e).lower(): raise StockDataUnavailableError( stock_code=validated_code, data_type="analysis data", message=f"Unable to perform analysis due to network issues: {str(e)}" ) else: raise StockDataUnavailableError( stock_code=validated_code, data_type="analysis data", message=f"Failed to perform Best Four Points analysis: {str(e)}" ) - mcp_server.py:133-155 (registration)MCP tool registration using @mcp.tool(name="get_best_four_points") and the top-level handler get_best_four_points_tool that delegates to underlying implementation and formats response using BestFourPointsResponse schema.
@mcp.tool(name="get_best_four_points", description="Get Best Four Points analysis for a specific stock.", ) async def get_best_four_points_tool(stock_code: str) -> BestFourPointsResponse: """Get Best Four Points analysis for a specific stock.""" try: raw_data = await get_best_four_points(stock_code) # Extract clean data without _metadata for Pydantic model from tw_stock_agent.utils.mcp_error_handler import MCPResponseFormatter clean_data = MCPResponseFormatter.extract_metadata_for_model(raw_data) return BestFourPointsResponse(**clean_data) except TwStockAgentError as e: return BestFourPointsResponse( stock_code=stock_code, updated_at=e.context.timestamp.isoformat(), error=e.message ) except Exception as e: return BestFourPointsResponse( stock_code=stock_code, updated_at=datetime.now().isoformat(), error=f"Unexpected error: {str(e)}" ) - Intermediate handler with MCP error handling decorator, stock code validation, delegation to StockService core implementation, and response formatting.
@mcp_error_handler("get_best_four_points") async def get_best_four_points(stock_code: str) -> dict[str, Any]: """ 取得四大買賣點分析 Args: stock_code: 股票代號，例如：2330 Returns: 四大買賣點分析結果 Raises: InvalidStockCodeError: 股票代號格式錯誤 StockNotFoundError: 找不到股票 StockDataUnavailableError: 分析資料無法取得 """ # Validate request parameters validated_params = validate_stock_request(stock_code=stock_code) # Fetch analysis data result = await stock_service.get_best_four_points(validated_params["stock_code"]) # Format response for MCP with enhanced technical analysis structure return MCPResponseFormatter.format_technical_analysis_response(result) - Pydantic schema defining the response structure for Best Four Points analysis, including buy_signals, sell_signals, technical_analysis, with computed fields like signal_summary, validation, and Taiwan market-specific features.
class BestFourPointsResponse(BaseStockResponse): """Enhanced response model for Best Four Points analysis.""" buy_signals: List[TradingSignal] = Field( default_factory=list, description="Buy signals (買點)", alias="buyPoints" ) sell_signals: List[TradingSignal] = Field( default_factory=list, description="Sell signals (賣點)", alias="sellPoints" ) technical_analysis: Optional[Dict[str, Any]] = Field( None, description="Technical analysis results (技術分析結果)", alias="analysis" ) overall_recommendation: Optional[str] = Field( None, description="Overall trading recommendation", alias="overallRecommendation" ) risk_assessment: Optional[Dict[str, Any]] = Field( None, description="Risk assessment information", alias="riskAssessment" ) def __init__(self, **data): # Handle backward compatibility if 'buy_points' in data: data['buy_signals'] = data.pop('buy_points', []) if 'sell_points' in data: data['sell_signals'] = data.pop('sell_points', []) if 'analysis' in data: data['technical_analysis'] = data.pop('analysis') super().__init__(**data) self.metadata.data_type = "technical_analysis" self.metadata.record_count = len(self.buy_signals) + len(self.sell_signals) @field_validator("overall_recommendation") @classmethod def validate_recommendation(cls, v: Optional[str]) -> Optional[str]: """Validate overall recommendation.""" if v is None: return None valid_recommendations = { 'strong_buy', 'buy', 'hold', 'sell', 'strong_sell' } if v.lower() not in valid_recommendations: return 'hold' # Default to hold for invalid values return v.lower() @computed_field @property def signal_summary(self) -> Dict[str, Any]: """Summary of all trading signals.""" buy_count = len([s for s in self.buy_signals if s.signal_type in ['buy', 'strong_buy']]) sell_count = len([s for s in self.sell_signals if s.signal_type in ['sell', 'strong_sell']]) avg_buy_confidence = ( sum(s.confidence for s in self.buy_signals) / len(self.buy_signals) if self.buy_signals else 0 ) avg_sell_confidence = ( sum(s.confidence for s in self.sell_signals) / len(self.sell_signals) if self.sell_signals else 0 ) return { "total_signals": len(self.buy_signals) + len(self.sell_signals), "buy_signals_count": buy_count, "sell_signals_count": sell_count, "average_buy_confidence": round(avg_buy_confidence, 2), "average_sell_confidence": round(avg_sell_confidence, 2), "signal_bias": "bullish" if buy_count > sell_count else "bearish" if sell_count > buy_count else "neutral" }