get_forecast_report
Generate earnings forecast reports for A-share stocks within specified date ranges to analyze company performance projections.
Instructions
Earnings forecast report within date range.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| code | Yes | ||
| start_date | Yes | ||
| end_date | Yes | ||
| limit | No | ||
| format | No | markdown |
Implementation Reference
- src/tools/financial_reports.py:89-97 (handler)The main MCP tool handler for 'get_forecast_report', decorated with @app.tool(). It invokes the fetch_forecast_report helper via run_tool_with_handling.@app.tool() def get_forecast_report(code: str, start_date: str, end_date: str, limit: int = 250, format: str = "markdown") -> str: """Earnings forecast report within date range.""" return run_tool_with_handling( lambda: fetch_forecast_report( active_data_source, code=code, start_date=start_date, end_date=end_date, limit=limit, format=format ), context=f"get_forecast_report:{code}:{start_date}-{end_date}", )
- Helper function that calls data_source.get_forecast_report(), formats the DataFrame output using format_table_output, and handles validation.def fetch_forecast_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_forecast_report(code=code, start_date=start_date, end_date=end_date) meta = {"code": code, "start_date": start_date, "end_date": end_date, "dataset": "Forecast"} return format_table_output(df, format=format, max_rows=limit, meta=meta)
- src/baostock_data_source.py:481-525 (helper)Concrete implementation of get_forecast_report in BaostockDataSource, using bs.query_forecast_report to fetch data from Baostock API, with full error handling and pagination support.def get_forecast_report(self, code: str, start_date: str, end_date: str) -> pd.DataFrame: """Fetches performance forecast reports (业绩预告) using Baostock.""" logger.info( f"Fetching Performance Forecast Report for {code} ({start_date} to {end_date})") try: with baostock_login_context(): rs = bs.query_forecast_report( code=code, start_date=start_date, end_date=end_date) # Note: Baostock docs mention pagination for this, but the Python API doesn't seem to expose it directly. # We fetch all available pages in the loop below. if rs.error_code != '0': logger.error( f"Baostock API error (Forecast) 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 forecast report found for {code} in range {start_date}-{end_date}. Baostock msg: {rs.error_msg}") else: raise DataSourceError( f"Baostock API error fetching performance forecast report: {rs.error_msg} (code: {rs.error_code})") data_list = [] while rs.next(): # Loop should handle pagination implicitly if rs manages it data_list.append(rs.get_row_data()) if not data_list: logger.warning( f"No performance forecast report found for {code} in range {start_date}-{end_date} (empty result set).") raise NoDataFoundError( f"No performance forecast 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 forecast report records for {code}.") return result_df except (LoginError, NoDataFoundError, DataSourceError, ValueError) as e: logger.warning( f"Caught known error fetching performance forecast report for {code}: {type(e).__name__}") raise e except Exception as e: logger.exception( f"Unexpected error fetching performance forecast report for {code}: {e}") raise DataSourceError( f"Unexpected error fetching performance forecast report for {code}: {e}")