top_holders
Retrieve top ten shareholder data for Chinese stocks to analyze ownership structure and investment decisions using Xueqiu MCP's stock market API.
Instructions
获取十大股东数据
Args:
stock_code: 股票代码
circula: 只获取流通股,默认为1Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| stock_code | No | SZ000002 | |
| circula | No |
Implementation Reference
- main.py:205-214 (handler)The main handler function for the 'top_holders' MCP tool. It fetches top shareholders data for a given stock code using the pysnowball library (ball.top_holders), optionally filtered to circulating shares, processes the timestamps, and returns the data as a dictionary.
@mcp.tool() def top_holders(stock_code: str="SZ000002", circula: int = 1) -> dict: """获取十大股东数据 Args: stock_code: 股票代码 circula: 只获取流通股,默认为1 """ result = ball.top_holders(symbol=stock_code, circula=circula) return process_data(result) - main.py:34-61 (helper)Helper function used by the top_holders tool (and others) to process the raw data, primarily converting timestamps to readable datetime strings.
def process_data(data, process_config=None): """ 通用数据处理函数，可扩展添加各种数据处理操作 Args: data: 原始数据 process_config: 处理配置字典，用于指定要执行的处理操作 例如: {'convert_timestamps': True, 'other_process': params} Returns: 处理后的数据 """ if process_config is None: # 默认配置 process_config = { 'convert_timestamps': True } # 如果开启了时间戳转换 if process_config.get('convert_timestamps', True): data = convert_timestamps(data) # 在这里可以添加更多的数据处理逻辑 # 例如: # if 'format_numbers' in process_config: # data = format_numbers(data, **process_config['format_numbers']) return data - main.py:14-31 (helper)Supporting helper function called by process_data to recursively convert timestamp fields in the data to formatted datetime strings.
def convert_timestamps(data): """递归地将数据中的所有 timestamp 转换为 datetime 字符串""" if isinstance(data, dict): for key, value in list(data.items()): if key == 'timestamp' and isinstance(value, (int, float)) and value > 1000000000000: # 毫秒级时间戳 data[key] = datetime.datetime.fromtimestamp(value/1000).strftime('%Y-%m-%d %H:%M:%S') elif key == 'timestamp' and isinstance(value, (int, float)) and value > 1000000000: # 秒级时间戳 data[key] = datetime.datetime.fromtimestamp(value).strftime('%Y-%m-%d %H:%M:%S') elif key.endswith('_date') and isinstance(value, (int, float)) and value > 1000000000000: # 毫秒级时间戳 data[key] = datetime.datetime.fromtimestamp(value/1000).strftime('%Y-%m-%d %H:%M:%S') elif key.endswith('_date') and isinstance(value, (int, float)) and value > 1000000000: # 秒级时间戳 data[key] = datetime.datetime.fromtimestamp(value).strftime('%Y-%m-%d %H:%M:%S') elif isinstance(value, (dict, list)): data[key] = convert_timestamps(value) elif isinstance(data, list): for i, item in enumerate(data): data[i] = convert_timestamps(item) return data - main.py:508-510 (registration)The MCP server is run here, which registers and exposes all @mcp.tool() decorated functions including top_holders.
if __name__ == "__main__": # This code only runs when the file is executed directly mcp.run()