Taiwan Stock Agent

Instructions

Implementation Reference

• mcp_server.py:157-179 (handler)

  MCP tool handler implementation for 'get_realtime_data'. This is the entry point decorated with @mcp.tool that handles the tool execution, error handling, and response formatting using RealtimeDataResponse model.

  @mcp.tool(name="get_realtime_data",
            description="Get real-time data for a specific stock.",
  )
  async def get_realtime_data_tool(stock_code: str) -> RealtimeDataResponse:
      """Get real-time data for a specific stock."""
      try:
          raw_data = await get_realtime_data(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 RealtimeDataResponse(**clean_data)
      except TwStockAgentError as e:
          return RealtimeDataResponse(
              stock_code=stock_code,
              updated_at=e.context.timestamp.isoformat(),
              error=e.message
          )
      except Exception as e:
          return RealtimeDataResponse(
              stock_code=stock_code,
              updated_at=datetime.now().isoformat(),
              error=f"Unexpected error: {str(e)}"
          )

• tw_stock_agent/models/stock_models.py:519-665 (schema)

  Pydantic model (schema) defining the structure and validation for the realtime data response, including fields like current_price, volume, bid/ask prices, with computed fields for analysis.

  class RealtimeDataResponse(BaseStockResponse):
      """Enhanced response model for realtime stock data."""
      
      company_name: Optional[str] = Field(
          None,
          description="Company name (股票名稱)",
          alias="name"
      )
      current_price: Optional[TWDAmount] = Field(
          None,
          description="Current price (現在價格)",
          alias="currentPrice"
      )
      opening_price: Optional[TWDAmount] = Field(
          None,
          description="Opening price (開盤價)",
          alias="open"
      )
      highest_price: Optional[TWDAmount] = Field(
          None,
          description="Highest price (最高價)",
          alias="high"
      )
      lowest_price: Optional[TWDAmount] = Field(
          None,
          description="Lowest price (最低價)",
          alias="low"
      )
      trading_volume: Optional[int] = Field(
          None,
          description="Trading volume (成交量)",
          alias="volume",
          ge=0
      )
      bid_price: Optional[TWDAmount] = Field(
          None,
          description="Best bid price (最佳買價)",
          alias="bidPrice"
      )
      ask_price: Optional[TWDAmount] = Field(
          None,
          description="Best ask price (最佳賣價)",
          alias="askPrice"
      )
      bid_size: Optional[int] = Field(
          None,
          description="Bid size (買張數)",
          alias="bidSize",
          ge=0
      )
      ask_size: Optional[int] = Field(
          None,
          description="Ask size (賣張數)",
          alias="askSize",
          ge=0
      )
      market_status: Optional[str] = Field(
          None,
          description="Market status (市場狀態)",
          alias="marketStatus"
      )
      
      def __init__(self, **data):
          # Handle backward compatibility
          for price_field in ['current_price', 'open', 'high', 'low', 'bid_price', 'ask_price']:
              field_name = price_field
              if price_field == 'open':
                  field_name = 'opening_price'
              elif price_field == 'high':
                  field_name = 'highest_price'
              elif price_field == 'low':
                  field_name = 'lowest_price'
              # current_price, bid_price, ask_price stay the same
              
              if price_field in data and data[price_field] is not None:
                  value = data[price_field]
                  if not isinstance(value, dict):
                      data[field_name] = {'amount': value}
                  if price_field != field_name:  # Only remove if field name changed
                      data.pop(price_field, None)
          
          super().__init__(**data)
          self.metadata.data_type = "realtime_data"
      
      @field_validator("market_status")
      @classmethod
      def validate_market_status(cls, v: Optional[str]) -> Optional[str]:
          """Validate market status."""
          if v is None:
              return None
          
          valid_statuses = {
              'open', 'closed', 'pre_market', 'after_hours', 
              'opening_auction', 'closing_auction', 'trading_halt'
          }
          
          v_lower = v.lower().replace(' ', '_')
          if v_lower not in valid_statuses:
              return 'unknown'
          
          return v_lower
      
      @computed_field
      @property
      def price_change_from_open(self) -> Optional[Dict[str, Any]]:
          """Calculate price change from opening."""
          if not self.current_price or not self.opening_price or self.opening_price.amount == 0:
              return None
          
          change_amount = self.current_price.amount - self.opening_price.amount
          change_percentage = float((change_amount / self.opening_price.amount) * 100)
          
          return {
              "change_amount": float(change_amount),
              "change_percentage": change_percentage,
              "is_positive": change_amount > 0,
              "formatted_change": f"{'+' if change_amount >= 0 else ''}{change_amount:.2f}"
          }
      
      @computed_field
      @property
      def bid_ask_spread(self) -> Optional[Dict[str, Any]]:
          """Calculate bid-ask spread."""
          if not self.bid_price or not self.ask_price:
              return None
          
          spread_amount = self.ask_price.amount - self.bid_price.amount
          mid_price = (self.ask_price.amount + self.bid_price.amount) / 2
          spread_percentage = float((spread_amount / mid_price) * 100) if mid_price > 0 else None
          
          return {
              "spread_amount": float(spread_amount),
              "spread_percentage": spread_percentage,
              "mid_price": float(mid_price)
          }
      
      @computed_field
      @property
      def is_actively_trading(self) -> bool:
          """Check if stock is actively trading."""
          return (
              self.market_status == 'open' and
              self.current_price is not None and
              self.trading_volume is not None and
              self.trading_volume > 0
          )

• tw_stock_agent/services/stock_service.py:358-453 (helper)

  Core helper function in StockService that fetches realtime stock data from twstock API, implements caching, circuit breaking, retry logic, and error handling.

  @with_async_error_handling(operation="get_realtime_data")
  @with_retry(max_retries=1, base_delay=0.5)  # Shorter retry for real-time data
  async def get_realtime_data(self, stock_code: str) -> dict[str, Any]:
      """
      獲取即時股票資訊
      
      Args:
          stock_code: 股票代號
          
      Returns:
          即時股票資訊
          
      Raises:
          InvalidStockCodeError: 股票代號格式錯誤
          StockNotFoundError: 找不到股票
          StockDataUnavailableError: 即時資料無法取得
          StockMarketClosedError: 股市休市
      """
      # Track request for performance monitoring
      self._performance_monitor.record_request()
      
      # Validate stock code
      validated_code = StockCodeValidator.validate_stock_code(stock_code)
      
      cache_key = f"realtime_{validated_code}"
      cached_data = self._get_cached_data(cache_key)
      if cached_data:
          return cached_data
      
      try:
          # Use circuit breaker for external API call
          realtime_data = await self.circuit_breaker.acall(
              asyncio.to_thread, twstock.realtime.get, validated_code
          )
          
          if not realtime_data:
              raise StockNotFoundError(
                  stock_code=validated_code,
                  message=f"No real-time data available for stock '{validated_code}'"
              )
          
          if realtime_data.get('success', False) is False:
              # Check if market is closed
              current_hour = datetime.now().hour
              if current_hour < 9 or current_hour > 14:  # Taiwan market hours: 9:00-13:30
                  from tw_stock_agent.exceptions import StockMarketClosedError
                  raise StockMarketClosedError(
                      stock_code=validated_code,
                      message="Taiwan stock market is closed. Trading hours: 9:00 AM - 1:30 PM (GMT+8)"
                  )
              else:
                  raise StockDataUnavailableError(
                      stock_code=validated_code,
                      data_type="real-time data",
                      message="Real-time data temporarily unavailable"
                  )
          
          realtime_info = realtime_data.get('realtime', {})
          stock_info = realtime_data.get('info', {})
          
          data = {
              "stock_code": validated_code,
              "name": stock_info.get('name'),
              "current_price": realtime_info.get('latest_trade_price'),
              "open": realtime_info.get('open'),
              "high": realtime_info.get('high'),
              "low": realtime_info.get('low'),
              "volume": realtime_info.get('accumulate_trade_volume'),
              "updated_at": datetime.now().isoformat(),
              "market_status": "open" if realtime_data.get('success') else "closed"
          }
          
          # 使用標籤進行分組管理
          tags = ["realtime", f"stock_{validated_code}", "live_data"]
          self._set_cached_data(cache_key, data, self.cache_ttl['realtime'], 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="real-time data",
                  message=f"Unable to fetch real-time data due to network issues: {str(e)}"
              )
          else:
              raise StockDataUnavailableError(
                  stock_code=validated_code,
                  data_type="real-time data",
                  message=f"Failed to fetch real-time data: {str(e)}"
              )

• tw_stock_agent/tools/stock_tools.py:199-223 (helper)

  Intermediate helper function that validates input parameters, delegates to StockService, and formats the response for MCP compatibility.

  @mcp_error_handler("get_realtime_data")
  async def get_realtime_data(stock_code: str) -> dict[str, Any]:
      """
      取得即時股票資訊
      
      Args:
          stock_code: 股票代號，例如：2330
          
      Returns:
          即時股票資訊
          
      Raises:
          InvalidStockCodeError: 股票代號格式錯誤
          StockNotFoundError: 找不到股票
          StockDataUnavailableError: 即時資料無法取得
          StockMarketClosedError: 股市休市
      """
      # Validate request parameters
      validated_params = validate_stock_request(stock_code=stock_code)
      
      # Fetch real-time data
      result = await stock_service.get_realtime_data(validated_params["stock_code"])
      
      # Format response for MCP with enhanced realtime structure
      return MCPResponseFormatter.format_realtime_data_response(result)