get_stock_industry
Retrieve industry classification for a specific stock or all stocks on a chosen date. Input stock code and date to generate a markdown table with detailed industry data.
Instructions
Fetches industry classification for a specific stock or all stocks on a given date.
Args:
code: Optional. The stock code (e.g., 'sh.600000'). If None, fetches for all stocks.
date: Optional. The date in 'YYYY-MM-DD' format. If None, uses the latest available date.
Returns:
Markdown table with industry classification data or an error message.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| code | No | ||
| date | No |
Implementation Reference
- src/tools/indices.py:25-51 (handler)MCP tool handler for 'get_stock_industry'. Decorated with @app.tool(), it invokes the data source method and formats the resulting DataFrame as a markdown table or handles errors.@app.tool() def get_stock_industry(code: Optional[str] = None, date: Optional[str] = None, limit: int = 250, format: str = "markdown") -> str: """ Get industry classification for a specific stock or all stocks on a date. Args: code: Optional stock code in Baostock format (e.g., 'sh.600000'). If None, returns all. date: Optional 'YYYY-MM-DD'. If None, uses the latest available date. Returns: Markdown table with industry data or an error message. """ log_msg = f"Tool 'get_stock_industry' called for code={code or 'all'}, date={date or 'latest'}" logger.info(log_msg) try: # Add date validation if desired df = active_data_source.get_stock_industry(code=code, date=date) logger.info( f"Successfully retrieved industry data for {code or 'all'}, {date or 'latest'}.") meta = {"code": code or "all", "as_of": date or "latest"} return format_table_output(df, format=format, max_rows=limit, meta=meta) except Exception as e: logger.exception( f"Exception processing get_stock_industry: {e}") return f"Error: An unexpected error occurred: {e}"
- src/baostock_data_source.py:527-568 (helper)Helper method in BaostockDataSource class that performs the actual API query to Baostock for stock industry classification data and returns a pandas DataFrame.def get_stock_industry(self, code: Optional[str] = None, date: Optional[str] = None) -> pd.DataFrame: """Fetches industry classification using Baostock.""" log_msg = f"Fetching industry data for code={code or 'all'}, date={date or 'latest'}" logger.info(log_msg) try: with baostock_login_context(): rs = bs.query_stock_industry(code=code, date=date) if rs.error_code != '0': logger.error( f"Baostock API error (Industry) for {code}, {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 industry data found for {code}, {date}. Baostock msg: {rs.error_msg}") else: raise DataSourceError( f"Baostock API error fetching industry data: {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 industry data found for {code}, {date} (empty result set).") raise NoDataFoundError( f"No industry data found for {code}, {date} (empty result set).") result_df = pd.DataFrame(data_list, columns=rs.fields) logger.info( f"Retrieved {len(result_df)} industry records for {code or 'all'}, {date or 'latest'}.") return result_df except (LoginError, NoDataFoundError, DataSourceError, ValueError) as e: logger.warning( f"Caught known error fetching industry data for {code}, {date}: {type(e).__name__}") raise e except Exception as e: logger.exception( f"Unexpected error fetching industry data for {code}, {date}: {e}") raise DataSourceError( f"Unexpected error fetching industry data for {code}, {date}: {e}")
- mcp_server.py:53-53 (registration)Invocation of register_index_tools during MCP app initialization, which registers the get_stock_industry tool (among others) to the FastMCP app instance.register_index_tools(app, active_data_source)
- mcp_server.py:15-15 (registration)Import of the register_index_tools function required for tool registration.from src.tools.indices import register_index_tools