get_profit_data
Retrieve quarterly profitability data for A-share stocks to analyze financial performance and track earnings trends over specific periods.
Instructions
Quarterly profitability data.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| code | Yes | ||
| year | Yes | ||
| quarter | Yes | ||
| limit | No | ||
| format | No | markdown |
Implementation Reference
- src/tools/financial_reports.py:31-37 (handler)MCP tool handler for 'get_profit_data'. Decorated with @app.tool(), it invokes the use case via run_tool_with_handling for caching/error handling.@app.tool() def get_profit_data(code: str, year: str, quarter: int, limit: int = 250, format: str = "markdown") -> str: """Quarterly profitability data.""" return run_tool_with_handling( lambda: fetch_profit_data(active_data_source, code=code, year=year, quarter=quarter, limit=limit, format=format), context=f"get_profit_data:{code}:{year}Q{quarter}", )
- mcp_server.py:51-58 (registration)Main server file where register_financial_report_tools is called to register the financial reports tools, including 'get_profit_data'.register_stock_market_tools(app, active_data_source) register_financial_report_tools(app, active_data_source) register_index_tools(app, active_data_source) register_market_overview_tools(app, active_data_source) register_macroeconomic_tools(app, active_data_source) register_date_utils_tools(app, active_data_source) register_analysis_tools(app, active_data_source) register_helpers_tools(app)
- Use case implementation: input validation, data fetching from datasource, and output formatting.def fetch_profit_data(data_source: FinancialDataSource, *, code: str, year: str, quarter: int, limit: int, format: str) -> str: validate_year(year) validate_quarter(quarter) validate_output_format(format) df = data_source.get_profit_data(code=code, year=year, quarter=quarter) return _format_financial_df(df, code=code, year=year, quarter=quarter, dataset="Profitability", format=format, limit=limit)
- src/baostock_data_source.py:413-415 (helper)Concrete data source method implementing get_profit_data by calling Baostock API via shared helper.def get_profit_data(self, code: str, year: str, quarter: int) -> pd.DataFrame: """Fetches quarterly profitability data using Baostock.""" return _fetch_financial_data(bs.query_profit_data, "Profitability", code, year, quarter)
- src/baostock_data_source.py:26-64 (helper)Shared helper function in data source for fetching financial data from Baostock, handling login, errors, and DataFrame construction.def _fetch_financial_data( bs_query_func, data_type_name: str, code: str, year: str, quarter: int ) -> pd.DataFrame: logger.info( f"Fetching {data_type_name} data for {code}, year={year}, quarter={quarter}") try: with baostock_login_context(): # Assuming all these functions take code, year, quarter rs = bs_query_func(code=code, year=year, quarter=quarter) if rs.error_code != '0': logger.error( f"Baostock API error ({data_type_name}) 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 {data_type_name} data found for {code}, {year}Q{quarter}. Baostock msg: {rs.error_msg}") else: raise DataSourceError( f"Baostock API error fetching {data_type_name} 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 {data_type_name} data found for {code}, {year}Q{quarter} (empty result set from Baostock).") raise NoDataFoundError( f"No {data_type_name} data found for {code}, {year}Q{quarter} (empty result set).") result_df = pd.DataFrame(data_list, columns=rs.fields) logger.info( f"Retrieved {len(result_df)} {data_type_name} records for {code}, {year}Q{quarter}.") return result_df