get_stock_industry
Retrieve industry classification data for specific stocks or all A-share stocks on a given date to analyze market sectors and investment opportunities.
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-32 (handler)The main handler function for the 'get_stock_industry' MCP tool, decorated with @app.tool() for registration, logs the call, and delegates to the use case layer.@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'}", )
- src/use_cases/indices.py:18-22 (helper)Use case helper that fetches industry data from data source, adds metadata, and formats output.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-569 (helper)Core data source implementation querying Baostock API for stock industry classification.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}")
- src/data_source_interface.py:188-189 (schema)Abstract method in FinancialDataSource interface defining the schema for get_stock_industry.def get_stock_industry(self, code: Optional[str] = None, date: Optional[str] = None) -> pd.DataFrame: pass
- mcp_server.py:53-53 (registration)Invocation of register_index_tools which defines and registers the get_stock_industry tool.register_index_tools(app, active_data_source)