kospi-kosdaq

Instructions

Args:
    fromdate (str): Start date for retrieval (YYYYMMDD)
    todate   (str): End date for retrieval (YYYYMMDD)
    ticker   (str): Stock ticker symbol

Returns:
    DataFrame with columns:
    - Volume (Sell/Buy/Net Buy)
    - Trading Value (Sell/Buy/Net Buy)
    Broken down by investor types (Financial Investment, Insurance, Trust, etc.)

Implementation Reference

• kospi_kosdaq_stock_server.py:369-427 (handler)

  The handler function for the 'get_stock_trading_volume' tool. It validates input dates and ticker, fetches trading volume data by investor type using pykrx.get_market_trading_volume_by_date, converts the DataFrame to a dictionary with dates as keys sorted in descending order, and handles errors.

  @mcp.tool()
  def get_stock_trading_volume(fromdate: Union[str, int], todate: Union[str, int], ticker: Union[str, int]) -> Dict[str, Any]:
      """Retrieves trading volume by investor type for a specific stock.
  
      Args:
          fromdate (str): Start date for retrieval (YYYYMMDD)
          todate   (str): End date for retrieval (YYYYMMDD)
          ticker   (str): Stock ticker symbol
  
      Returns:
          DataFrame with columns:
          - Volume (Sell/Buy/Net Buy)
          - Trading Value (Sell/Buy/Net Buy)
          Broken down by investor types (Financial Investment, Insurance, Trust, etc.)
      """
      # Validate and convert date format
      def validate_date(date_str: Union[str, int]) -> str:
          try:
              if isinstance(date_str, int):
                  date_str = str(date_str)
              if '-' in date_str:
                  parsed_date = datetime.strptime(date_str, '%Y-%m-%d')
                  return parsed_date.strftime('%Y%m%d')
              datetime.strptime(date_str, '%Y%m%d')
              return date_str
          except ValueError:
              raise ValueError(f"Date must be in YYYYMMDD format. Input value: {date_str}")
  
      def validate_ticker(ticker_str: Union[str, int]) -> str:
          if isinstance(ticker_str, int):
              return str(ticker_str)
          return ticker_str
  
      try:
          fromdate = validate_date(fromdate)
          todate = validate_date(todate)
          ticker = validate_ticker(ticker)
  
          logging.debug(f"Retrieving stock trading volume by investor type: {ticker}, {fromdate}-{todate}")
  
          # Call get_market_trading_volume_by_date
          df = get_market_trading_volume_by_date(fromdate, todate, ticker)
  
          # Convert DataFrame to dictionary
          result = df.to_dict(orient='index')
  
          # Convert datetime index to string and sort in reverse
          sorted_items = sorted(
              ((k.strftime('%Y-%m-%d'), v) for k, v in result.items()),
              reverse=True
          )
          result = dict(sorted_items)
  
          return result
  
      except Exception as e:
          error_message = f"Data retrieval failed: {str(e)}"
          logging.error(error_message)
          return {"error": error_message}