get_stock_industry

Instructions

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)