get_sz50_stocks
Fetch the constituent stocks of the SZSE 50 Index for a specific date. The tool returns a markdown table listing the stocks or provides an error message if data is unavailable.
Instructions
Fetches the constituent stocks of the SZSE 50 Index for a given date.
Args:
date: Optional. The date in 'YYYY-MM-DD' format. If None, uses the latest available date.
Returns:
Markdown table with SZSE 50 constituent stocks or an error message.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| date | No |
Implementation Reference
- src/tools/indices.py:52-69 (handler)The primary handler function for the 'get_sz50_stocks' MCP tool, decorated with @app.tool(). It delegates to a generic helper function to fetch data from the data source and format the output.@app.tool() def get_sz50_stocks(date: Optional[str] = None, limit: int = 250, format: str = "markdown") -> str: """ Fetches the constituent stocks of the SZSE 50 Index for a given date. Args: date: Optional. The date in 'YYYY-MM-DD' format. If None, uses the latest available date. Returns: Markdown table with SZSE 50 constituent stocks or an error message. """ return call_index_constituent_tool( "get_sz50_stocks", active_data_source.get_sz50_stocks, "SZSE 50", date, limit=limit, format=format )
- mcp_server.py:53-53 (registration)Invocation of the register_index_tools function during MCP app initialization, which registers the get_sz50_stocks tool among others.register_index_tools(app, active_data_source)
- src/tools/base.py:127-171 (helper)Generic helper function used by index constituent tools to fetch dataframe from data source method, handle exceptions, and format output as markdown table or other.def call_index_constituent_tool( tool_name: str, data_source_method: Callable, index_name: str, date: Optional[str] = None, *, limit: int = 250, format: str = "markdown", ) -> str: """ Helper function for index constituent tools Args: tool_name: Name of the tool for logging data_source_method: Method to call on the data source index_name: Name of the index (for logging) date: Optional date to query Returns: Markdown formatted string with results or error message """ log_msg = f"Tool '{tool_name}' called for date={date or 'latest'}" logger.info(log_msg) try: # Add date validation if desired df = data_source_method(date=date) logger.info( f"Successfully retrieved {index_name} constituents for {date or 'latest'}.") meta = {"index": index_name, "as_of": date or "latest"} return format_table_output(df, format=format, max_rows=limit, meta=meta) except NoDataFoundError as e: logger.warning(f"NoDataFoundError: {e}") return f"Error: {e}" except LoginError as e: logger.error(f"LoginError: {e}") return f"Error: Could not connect to data source. {e}" except DataSourceError as e: logger.error(f"DataSourceError: {e}") return f"Error: An error occurred while fetching data. {e}" except ValueError as e: logger.warning(f"ValueError: {e}") return f"Error: Invalid input parameter. {e}" except Exception as e: logger.exception(f"Unexpected Exception processing {tool_name}: {e}") return f"Error: An unexpected error occurred: {e}"
- src/baostock_data_source.py:570-573 (helper)Specific method in BaostockDataSource class that fetches SZSE 50 stock constituents using the Baostock query_sz50_stocks API via a shared helper.def get_sz50_stocks(self, date: Optional[str] = None) -> pd.DataFrame: """Fetches SZSE 50 index constituents using Baostock.""" return _fetch_index_constituent_data(bs.query_sz50_stocks, "SZSE 50", date)
- src/baostock_data_source.py:78-124 (helper)Shared private helper method in BaostockDataSource that performs the actual Baostock API query for index constituents, handles login, errors, and returns a pandas DataFrame.def _fetch_index_constituent_data( bs_query_func, index_name: str, date: Optional[str] = None ) -> pd.DataFrame: logger.info( f"Fetching {index_name} constituents for date={date or 'latest'}") try: with baostock_login_context(): # date is optional, defaults to latest rs = bs_query_func(date=date) if rs.error_code != '0': logger.error( f"Baostock API error ({index_name} Constituents) for date {date}: {rs.error_msg} (code: {rs.error_code})") if "no record found" in rs.error_msg.lower() or rs.error_code == '10002': raise NoDataFoundError( f"No {index_name} constituent data found for date {date}. Baostock msg: {rs.error_msg}") else: raise DataSourceError( f"Baostock API error fetching {index_name} constituents: {rs.error_msg} (code: {rs.error_code})") data_list = [] while rs.next(): data_list.append(rs.get_row_data()) if not data_list: logger.warning( f"No {index_name} constituent data found for date {date} (empty result set).") raise NoDataFoundError( f"No {index_name} constituent data found for date {date} (empty result set).") result_df = pd.DataFrame(data_list, columns=rs.fields) logger.info( f"Retrieved {len(result_df)} {index_name} constituents for date {date or 'latest'}.") return result_df except (LoginError, NoDataFoundError, DataSourceError, ValueError) as e: logger.warning( f"Caught known error fetching {index_name} constituents for date {date}: {type(e).__name__}") raise e except Exception as e: logger.exception( f"Unexpected error fetching {index_name} constituents for date {date}: {e}") raise DataSourceError( f"Unexpected error fetching {index_name} constituents for date {date}: {e}")