get_stock_industry
Retrieve industry classification data for specific stocks or all A-share stocks on a given date to analyze market sectors and company categorization.
Instructions
Get industry classification for a specific stock or all stocks on a date.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| code | No | ||
| date | No | ||
| limit | No | ||
| format | No | markdown |
Implementation Reference
- src/tools/indices.py:24-31 (handler)The MCP tool handler for get_stock_industry, decorated with @app.tool(). Logs the invocation and delegates execution to the use case layer via run_tool_with_handling.
@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.""" logger.info(f"Tool 'get_stock_industry' called for code={code or 'all'}, date={date or 'latest'}") return run_tool_with_handling( lambda: fetch_stock_industry(active_data_source, code=code, date=date, limit=limit, format=format), context=f"get_stock_industry:{code or 'all'}", ) - mcp_server.py:53-53 (registration)Top-level registration call to register_index_tools, which includes the @app.tool() decorator for get_stock_industry.
register_index_tools(app, active_data_source) - src/use_cases/indices.py:18-22 (helper)Use case helper function that validates input, fetches raw data from the data source interface, and formats the output as a markdown table.
def fetch_stock_industry(data_source: FinancialDataSource, *, code: Optional[str], date: Optional[str], limit: int, format: str) -> str: validate_output_format(format) df = data_source.get_stock_industry(code=code, date=date) meta = {"code": code or "all", "as_of": date or "latest"} return format_table_output(df, format=format, max_rows=limit, meta=meta) - src/baostock_data_source.py:527-568 (helper)Concrete data source implementation that performs the actual Baostock API query for stock industry classification data.
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}")