kospi-kosdaq

Instructions

Implementation Reference

• kospi_kosdaq_stock_server.py:430-504 (handler)

  The primary handler for the get_index_ohlcv tool. Decorated with @mcp.tool() for registration in FastMCP. Validates input dates, ticker, and frequency; fetches OHLCV data using pykrx.get_index_ohlcv_by_date; converts to sorted dictionary by date (descending); handles exceptions.
  @mcp.tool() def get_index_ohlcv(fromdate: Union[str, int], todate: Union[str, int], ticker: Union[str, int], freq: str = 'd') -> \ Dict[str, Any]: """Retrieves OHLCV data for a specific index. Args: fromdate (str): Start date for retrieval (YYYYMMDD) todate (str): End date for retrieval (YYYYMMDD) ticker (str): Index ticker symbol (e.g., 1001 for KOSPI, 2001 for KOSDAQ) freq (str, optional): d - daily / m - monthly / y - yearly. Defaults to 'd'. Returns: DataFrame: >> get_index_ohlcv("20210101", "20210130", "1001") Open High Low Close Volume Trading Value Date 2021-01-04 2874.50 2946.54 2869.11 2944.45 1026510465 25011393960858 2021-01-05 2943.67 2990.57 2921.84 2990.57 1519911750 26548380179493 2021-01-06 2993.34 3027.16 2961.37 2968.21 1793418534 29909396443430 2021-01-07 2980.75 3055.28 2980.75 3031.68 1524654500 27182807334912 2021-01-08 3040.11 3161.11 3040.11 3152.18 1297903388 40909490005818 """ # 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 def validate_freq(freq_str: str) -> str: valid_freqs = ['d', 'm', 'y'] if freq_str not in valid_freqs: raise ValueError(f"Frequency must be one of {valid_freqs}. Input value: {freq_str}") return freq_str try: fromdate = validate_date(fromdate) todate = validate_date(todate) ticker = validate_ticker(ticker) freq = validate_freq(freq) logging.debug(f"Retrieving index OHLCV data: {ticker}, {fromdate}-{todate}, freq={freq}") # Call get_index_ohlcv_by_date # Note: name_display is set to False to match the pattern of other functions df = get_index_ohlcv_by_date(fromdate, todate, ticker, freq=freq, name_display=False) # 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}
• kospi_kosdaq_stock_server.py:454-476 (schema)

  Input validation functions used within the handler for schema enforcement: validate_date (ensures YYYYMMDD format), validate_ticker (converts to str), validate_freq (checks 'd','m','y'). Also type hints on function parameters provide schema info.
  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 def validate_freq(freq_str: str) -> str: valid_freqs = ['d', 'm', 'y'] if freq_str not in valid_freqs: raise ValueError(f"Frequency must be one of {valid_freqs}. Input value: {freq_str}") return freq_str
• kospi_kosdaq_stock_server.py:430-430 (registration)

  The @mcp.tool() decorator registers the get_index_ohlcv function as an MCP tool in the FastMCP server.
  @mcp.tool()
• kospi_kosdaq_stock_server.py:487-487 (helper)

  Invocation of the underlying pykrx library function get_index_ohlcv_by_date, which provides the core data retrieval logic.
  df = get_index_ohlcv_by_date(fromdate, todate, ticker, freq=freq, name_display=False)