get_sz50_stocks
Retrieve the current constituents of the SZSE 50 index, providing stock data for China's A-share market in your preferred format.
Instructions
SZSE 50 constituents.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| date | No | ||
| limit | No | ||
| format | No | markdown |
Implementation Reference
- src/tools/indices.py:33-39 (handler)MCP tool handler and registration for 'get_sz50_stocks'. Delegates to use case 'fetch_index_constituents' via run_tool_with_handling for execution and error handling.@app.tool() def get_sz50_stocks(date: Optional[str] = None, limit: int = 250, format: str = "markdown") -> str: """SZSE 50 constituents.""" return run_tool_with_handling( lambda: fetch_index_constituents(active_data_source, index="sz50", date=date, limit=limit, format=format), context="get_sz50_stocks", )
- src/use_cases/indices.py:25-35 (helper)Use case helper 'fetch_index_constituents' that handles sz50 index by calling data_source.get_sz50_stocks(), validates input, and formats the DataFrame output.def fetch_index_constituents(data_source: FinancialDataSource, *, index: str, date: Optional[str], limit: int, format: str) -> str: validate_output_format(format) key = validate_index_key(index, INDEX_MAP) if key == "hs300": df = data_source.get_hs300_stocks(date=date) elif key == "sz50": df = data_source.get_sz50_stocks(date=date) else: df = data_source.get_zz500_stocks(date=date) meta = {"index": key, "as_of": date or "latest"} return format_table_output(df, format=format, max_rows=limit, meta=meta)
- src/baostock_data_source.py:570-573 (helper)Data source implementation of get_sz50_stocks in BaostockDataSource, which fetches constituents using Baostock's query_sz50_stocks API via shared helper.def get_sz50_stocks(self, date: Optional[str] = None) -> pd.DataFrame: """Fetches SZSE 50 index constituents using Baostock.""" return _fetch_index_constituent_data(bs.query_sz50_stocks, "SZSE 50", date)
- src/data_source_interface.py:196-197 (schema)Abstract method definition in FinancialDataSource interface specifying the get_sz50_stocks signature.def get_sz50_stocks(self, date: Optional[str] = None) -> pd.DataFrame: pass
- src/baostock_data_source.py:78-123 (helper)Shared helper function _fetch_index_constituent_data used by get_sz50_stocks to query Baostock API, handle errors, and return DataFrame.def _fetch_index_constituent_data( bs_query_func, index_name: str, date: Optional[str] = None ) -> pd.DataFrame: logger.info( f"Fetching {index_name} constituents for date={date or 'latest'}") try: with baostock_login_context(): # date is optional, defaults to latest rs = bs_query_func(date=date) if rs.error_code != '0': logger.error( f"Baostock API error ({index_name} Constituents) for date {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 {index_name} constituent data found for date {date}. Baostock msg: {rs.error_msg}") else: raise DataSourceError( f"Baostock API error fetching {index_name} constituents: {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 {index_name} constituent data found for date {date} (empty result set).") raise NoDataFoundError( f"No {index_name} constituent data found for date {date} (empty result set).") result_df = pd.DataFrame(data_list, columns=rs.fields) logger.info( f"Retrieved {len(result_df)} {index_name} constituents for date {date or 'latest'}.") return result_df except (LoginError, NoDataFoundError, DataSourceError, ValueError) as e: logger.warning( f"Caught known error fetching {index_name} constituents for date {date}: {type(e).__name__}") raise e except Exception as e: logger.exception( f"Unexpected error fetching {index_name} constituents for date {date}: {e}") raise DataSourceError( f"Unexpected error fetching {index_name} constituents for date {date}: {e}")