Taiwan Stock Agent

Instructions

Implementation Reference

• tw_stock_agent/services/stock_service.py:278-357 (handler)

  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)}"
          )

• tw_stock_agent/tools/stock_tools.py:174-198 (handler)

  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)

• tw_stock_agent/models/stock_models.py:692-775 (schema)

  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"
          }