Taiwan Stock Agent

Instructions

Implementation Reference

• tw_stock_agent/tools/stock_tools.py:225-270 (handler)

  Core handler function implementing the market overview logic: fetches realtime data for TSMC (2330) as market proxy, handles errors, and formats response using MCPResponseFormatter.

  @mcp_error_handler("get_market_overview")
  async def get_market_overview() -> dict[str, Any]:
      """
      取得市場概況
      
      Returns:
          大盤指數、成交量、漲跌家數等資訊
          
      Raises:
          StockDataUnavailableError: 市場資料無法取得
      """
      try:
          # 使用台積電(2330)作為市場指標
          realtime_data = await stock_service.get_realtime_data("2330")
          
          # Check if we got valid data
          if "error" in realtime_data:
              from tw_stock_agent.exceptions import StockDataUnavailableError
              raise StockDataUnavailableError(
                  stock_code="market",
                  data_type="market overview",
                  message="Unable to fetch market overview data"
              )
          
          result = {
              "date": datetime.now().isoformat(),
              "taiex": realtime_data.get("current_price"),
              "volume": realtime_data.get("volume"),
              "updated_at": realtime_data.get("updated_at"),
              "market_status": realtime_data.get("market_status", "unknown"),
              "reference_stock": "2330"  # TSMC as market reference
          }
          
          # Format response for MCP with enhanced market overview structure
          return MCPResponseFormatter.format_market_overview_response(result)
          
      except TwStockAgentError:
          # Re-raise our custom errors
          raise
      except Exception as e:
          from tw_stock_agent.exceptions import StockDataUnavailableError
          raise StockDataUnavailableError(
              stock_code="market",
              data_type="market overview",
              message=f"Failed to fetch market overview: {str(e)}"
          )

• mcp_server.py:181-202 (registration)

  MCP tool registration using @mcp.tool decorator. Wrapper function get_market_overview_tool calls the core handler, processes response, and returns structured MarketOverviewResponse.

  @mcp.tool(name="get_market_overview",
            description="Get market overview information.",
  )
  async def get_market_overview_tool() -> MarketOverviewResponse:
      """Get market overview information."""
      try:
          raw_data = await get_market_overview()
          # 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 MarketOverviewResponse(**clean_data)
      except TwStockAgentError as e:
          return MarketOverviewResponse(
              date=e.context.timestamp.isoformat(),
              error=e.message
          )
      except Exception as e:
          from datetime import datetime
          return MarketOverviewResponse(
              date=datetime.now().isoformat(),
              error=f"Unexpected error: {str(e)}"
          )

• tw_stock_agent/models/stock_models.py:797-962 (schema)

  Pydantic BaseModel schema for MarketOverviewResponse, defining fields, validators, computed fields, and metadata for the tool's response structure.

  class MarketOverviewResponse(BaseModel):
      """Enhanced response model for market overview."""
      
      model_config = ConfigDict(
          str_strip_whitespace=True,
          validate_assignment=True,
          populate_by_name=True,
          json_encoders={
              datetime: lambda v: v.isoformat(),
              Decimal: lambda v: float(v)
          }
      )
      
      trading_date: datetime = Field(
          ...,
          description="Trading date (交易日期)",
          alias="date"
      )
      taiex_index: Optional[MarketIndexData] = Field(
          None,
          description="TAIEX index data (台股加權指數)",
          alias="taiex"
      )
      total_volume: Optional[int] = Field(
          None,
          description="Total market volume (總成交量)",
          alias="volume",
          ge=0
      )
      total_turnover: Optional[TWDAmount] = Field(
          None,
          description="Total market turnover (總成交金額)",
          alias="turnover"
      )
      advancing_stocks: Optional[int] = Field(
          None,
          description="Number of advancing stocks (上漲家數)",
          ge=0
      )
      declining_stocks: Optional[int] = Field(
          None,
          description="Number of declining stocks (下跌家數)",
          ge=0
      )
      unchanged_stocks: Optional[int] = Field(
          None,
          description="Number of unchanged stocks (平盤家數)",
          ge=0
      )
      updated_at: datetime = Field(
          default_factory=lambda: datetime.now(TAIWAN_TZ),
          description="Last update timestamp (更新時間)",
          alias="updatedAt"
      )
      market_status: Optional[str] = Field(
          None,
          description="Current market status (市場狀態)"
      )
      reference_stock: Optional[str] = Field(
          None,
          description="Reference stock for market data"
      )
      error: Optional[str] = Field(
          None,
          description="Error message if any (錯誤訊息)"
      )
      metadata: ResponseMetadata = Field(
          default_factory=lambda: ResponseMetadata(data_type="market_overview")
      )
      
      def __init__(self, **data):
          # Handle backward compatibility
          if 'date' in data and isinstance(data['date'], str):
              try:
                  data['trading_date'] = datetime.fromisoformat(data['date'].replace('Z', '+00:00'))
                  data.pop('date', None)
              except ValueError:
                  try:
                      data['trading_date'] = datetime.strptime(data['date'], '%Y-%m-%d')
                      data.pop('date', None)
                  except ValueError:
                      pass
          
          # Handle TAIEX value conversion
          if 'taiex' in data and not isinstance(data['taiex'], dict):
              if data['taiex'] is not None:
                  data['taiex_index'] = {
                      'index_name': 'TAIEX',
                      'current_value': data['taiex']
                  }
              data.pop('taiex', None)
          
          super().__init__(**data)
      
      @field_validator("trading_date")
      @classmethod
      def validate_trading_date(cls, v: Union[str, datetime]) -> datetime:
          """Validate and convert trading date."""
          if isinstance(v, str):
              try:
                  if 'T' in v:
                      return datetime.fromisoformat(v.replace('Z', '+00:00')).astimezone(TAIWAN_TZ)
                  else:
                      return datetime.strptime(v, '%Y-%m-%d').replace(tzinfo=TAIWAN_TZ)
              except ValueError as e:
                  raise ValueError(f"Invalid date format: {v}") from e
          
          return v.astimezone(TAIWAN_TZ) if v.tzinfo else v.replace(tzinfo=TAIWAN_TZ)
      
      @field_validator("updated_at")
      @classmethod
      def validate_updated_at(cls, v: datetime) -> datetime:
          """Ensure updated_at is in Taiwan timezone."""
          if v.tzinfo is None:
              return v.replace(tzinfo=TAIWAN_TZ)
          return v.astimezone(TAIWAN_TZ)
      
      @model_validator(mode='after')
      def update_metadata(self) -> 'MarketOverviewResponse':
          """Update metadata based on response state."""
          if self.error:
              self.metadata.has_error = True
          return self
      
      @computed_field
      @property
      def total_trading_stocks(self) -> Optional[int]:
          """Calculate total number of stocks that traded."""
          counts = [self.advancing_stocks, self.declining_stocks, self.unchanged_stocks]
          valid_counts = [c for c in counts if c is not None]
          
          if not valid_counts:
              return None
          
          return sum(valid_counts)
      
      @computed_field
      @property
      def market_sentiment(self) -> Optional[str]:
          """Determine market sentiment based on advancing/declining stocks."""
          if not self.advancing_stocks or not self.declining_stocks:
              return None
          
          if self.advancing_stocks > self.declining_stocks * 1.5:
              return "very_bullish"
          elif self.advancing_stocks > self.declining_stocks:
              return "bullish"
          elif self.declining_stocks > self.advancing_stocks * 1.5:
              return "very_bearish"
          elif self.declining_stocks > self.advancing_stocks:
              return "bearish"
          else:
              return "neutral"
      
      @computed_field
      @property
      def is_trading_day(self) -> bool:
          """Check if the date is a trading day (weekday)."""
          return self.trading_date.weekday() < 5  # Monday = 0, Friday = 4
      
      @computed_field
      @property
      def formatted_timestamp(self) -> str:
          """ISO formatted timestamp string."""
          return self.updated_at.isoformat()