get_performance_express_report

Instructions

Implementation Reference

• src/tools/financial_reports.py:79-87 (handler)

  MCP tool handler for get_performance_express_report. Decorated with @app.tool(), validates inputs via run_tool_with_handling, delegates to use case fetch_performance_express_report.

  @app.tool()
  def get_performance_express_report(code: str, start_date: str, end_date: str, limit: int = 250, format: str = "markdown") -> str:
      """Performance express report within date range."""
      return run_tool_with_handling(
          lambda: fetch_performance_express_report(
              active_data_source, code=code, start_date=start_date, end_date=end_date, limit=limit, format=format
          ),
          context=f"get_performance_express_report:{code}:{start_date}-{end_date}",
      )

• mcp_server.py:52-52 (registration)

  Registration call for financial reports tools, including get_performance_express_report, in the main MCP server setup.

  register_financial_report_tools(app, active_data_source)

• src/use_cases/financial_reports.py:70-74 (helper)

  Use case helper that fetches data from FinancialDataSource interface, applies formatting and validation.

  def fetch_performance_express_report(data_source: FinancialDataSource, *, code: str, start_date: str, end_date: str, limit: int, format: str) -> str:
      validate_output_format(format)
      df = data_source.get_performance_express_report(code=code, start_date=start_date, end_date=end_date)
      meta = {"code": code, "start_date": start_date, "end_date": end_date, "dataset": "Performance Express"}
      return format_table_output(df, format=format, max_rows=limit, meta=meta)

• src/data_source_interface.py:158-160 (schema)

  Abstract method definition in FinancialDataSource interface, defining the expected input/output signature for the core data fetching method.

  @abstractmethod
  def get_performance_express_report(self, code: str, start_date: str, end_date: str) -> pd.DataFrame:
      pass

• src/baostock_data_source.py:437-480 (helper)

  Concrete implementation in BaostockDataSource using bs.query_performance_express_report to fetch raw data from Baostock API.

  def get_performance_express_report(self, code: str, start_date: str, end_date: str) -> pd.DataFrame:
      """Fetches performance express reports (业绩快报) using Baostock."""
      logger.info(
          f"Fetching Performance Express Report for {code} ({start_date} to {end_date})")
      try:
          with baostock_login_context():
              rs = bs.query_performance_express_report(
                  code=code, start_date=start_date, end_date=end_date)
  
              if rs.error_code != '0':
                  logger.error(
                      f"Baostock API error (Perf Express) for {code}: {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 performance express report found for {code} in range {start_date}-{end_date}. Baostock msg: {rs.error_msg}")
                  else:
                      raise DataSourceError(
                          f"Baostock API error fetching performance express report: {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 performance express report found for {code} in range {start_date}-{end_date} (empty result set).")
                  raise NoDataFoundError(
                      f"No performance express report found for {code} in range {start_date}-{end_date} (empty result set).")
  
              result_df = pd.DataFrame(data_list, columns=rs.fields)
              logger.info(
                  f"Retrieved {len(result_df)} performance express report records for {code}.")
              return result_df
  
      except (LoginError, NoDataFoundError, DataSourceError, ValueError) as e:
          logger.warning(
              f"Caught known error fetching performance express report for {code}: {type(e).__name__}")
          raise e
      except Exception as e:
          logger.exception(
              f"Unexpected error fetching performance express report for {code}: {e}")
          raise DataSourceError(
              f"Unexpected error fetching performance express report for {code}: {e}")