history_calculate
Analyze historical A-share stock performance by querying price data to calculate trends, indicators, and patterns without coding.
Instructions
对于给定的A股股票代码（其他市场的股票请不要使用此工具），有现成的历史股价数据，其数据结构如下：
名称 类型 描述
ts_code str 股票代码
trade_date str 交易日期
open float 开盘价
high float 最高价
low float 最低价
close float 收盘价
pre_close float 昨收价
change float 涨跌额
pct_chg float 涨跌幅
vol float 成交量 （手）
amount float 成交额 （千元）
你需要输入你想分析的股票代码以及你的问题。该工具将为你生成并执行相应的代码，并返回结果。 注意：
你无需编写任何代码——只需直接提问即可，例如：“过去一周涨了多少，有没有出现顶背离？”、“近期市场趋势如何？”、“MACD是否形成了金叉？”。
该工具只能基于上述数据结构中的数据回答问题，请勿提出需要超出该数据范围信息的问题。
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| code | Yes | A-share stock code (e.g. '600000' or '000001'). | |
| query | Yes | User question about the stock's historical performance. |
Input Schema (JSON Schema)
{
"properties": {
"code": {
"description": "A-share stock code (e.g. '600000' or '000001').",
"type": "string"
},
"query": {
"description": "User question about the stock's historical performance.",
"type": "string"
}
},
"required": [
"code",
"query"
],
"type": "object"
}
Implementation Reference
- Core handler function that normalizes the stock code, prompts an LLM to generate Python analysis code based on the query, executes the code using Tushare data, and sets the output.async def async_execute(self): """Generate and execute analysis code for the given stock code. The method normalizes the stock code to the Tushare format, calls the LLM to generate Python analysis code, and finally executes that code using ``exec_code``. """ code: str = self.input_dict["code"] # Normalize plain numeric codes into exchange-qualified codes. # Examples: '00'/'30' → 'SZ', '60'/'68' → 'SH', '92' → 'BJ'. if code[:2] in ["00", "30"]: code = f"{code}.SZ" elif code[:2] in ["60", "68"]: code = f"{code}.SH" elif code[:2] in ["92"]: code = f"{code}.BJ" query: str = self.input_dict["query"] import tushare as ts # Initialize the Tushare pro API using the token from environment. ts.pro_api(token=os.getenv("TUSHARE_API_TOKEN", "")) code_prompt: str = self.prompt_format( prompt_name="code_prompt", code=code, query=query, current_date=get_datetime(), example=self.get_prompt("code_example"), ) logger.info(f"code_prompt=\n{code_prompt}") messages = [Message(role=Role.USER, content=code_prompt)] def get_code(message: Message): """Extract Python code from the assistant response.""" return extract_content(message.content, language_tag="python") result_code = await self.llm.achat(messages=messages, callback_fn=get_code) logger.info(f"result_code=\n{result_code}") # Execute the generated Python code and set the execution result. self.set_output(exec_code(result_code))
- Defines the tool's input schema with 'code' (stock code) and 'query' (natural language question), and fetches description from prompt.def build_tool_call(self) -> ToolCall: """Describe the tool-call interface for this operator. Returns: ``ToolCall`` instance specifying required inputs and metadata. """ return ToolCall( **{ "description": self.get_prompt("tool_description"), "input_schema": { "code": { "type": "string", "description": "A-share stock code (e.g. '600000' or '000001').", "required": True, }, "query": { "type": "string", "description": "User question about the stock's historical performance.", "required": True, }, }, }, )
- finance_mcp/core/findata/history_calculate_op.py:21-31 (registration)Registers the HistoryCalculateOp class as an async tool operation using the FlowLLM context registry (@C.register_op()). The tool name 'history_calculate' is derived from the module/file name.@C.register_op() class HistoryCalculateOp(BaseAsyncToolOp): """Async op that lets an LLM write code for historical stock analysis. The op takes a stock code and a natural language question, then delegates to the LLM to produce executable Python code that fetches and analyses historical data obtained from Tushare. """ file_path = __file__
- finance_mcp/core/findata/__init__.py:14-20 (registration)Exports HistoryCalculateOp for use, making it available when importing from the findata package.from .history_calculate_op import HistoryCalculateOp from .tushare_client import TushareClient __all__ = [ "TushareClient", "HistoryCalculateOp", ]