get_stock_trading_volume
Retrieve trading volume and value data by investor type for KOSPI/KOSDAQ stocks to analyze market participation patterns and investment flows.
Instructions
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.)Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| fromdate | Yes | ||
| todate | Yes | ||
| ticker | Yes |
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}