stock_prices
Retrieve historical stock prices and technical indicators for A-shares, Hong Kong, and US markets using stock symbols and market codes.
Instructions
根据股票代码和市场获取股票历史价格及技术指标, 不支持加密货币
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| symbol | Yes | 股票代码 | |
| market | No | 股票市场，仅支持: sh(上证), sz(深证), hk(港股), us(美股), 不支持加密货币 | sh |
| period | No | 周期，如: daily(日线), weekly(周线，不支持美股) | daily |
| limit | No | 返回数量(int) |
Implementation Reference
- mcp_aktools/__init__.py:75-108 (handler)Core implementation of the 'stock_prices' tool. Handles input parameters, determines start date, selects appropriate akshare function based on market, fetches data with caching, adds technical indicators, and returns formatted recent historical prices.def stock_prices( symbol: str = field_symbol, market: str = field_market, period: str = Field("daily", description="周期，如: daily(日线), weekly(周线，不支持美股)"), limit: int = Field(30, description="返回数量(int)", strict=False), ): if period == "weekly": delta = {"weeks": limit + 62} else: delta = {"days": limit + 62} start_date = (datetime.now() - timedelta(**delta)).strftime("%Y%m%d") markets = [ ["sh", ak.stock_zh_a_hist, {}], ["sz", ak.stock_zh_a_hist, {}], ["hk", ak.stock_hk_hist, {}], ["us", stock_us_daily, {}], ["sh", fund_etf_hist_sina, {"market": "sh"}], ["sz", fund_etf_hist_sina, {"market": "sz"}], ] for m in markets: if m[0] != market: continue kws = {"period": period, "start_date": start_date, **m[2]} dfs = ak_cache(m[1], symbol=symbol, ttl=3600, **kws) if dfs is None or dfs.empty: continue add_technical_indicators(dfs, dfs["收盘"], dfs["最低"], dfs["最高"]) columns = [ "日期", "开盘", "收盘", "最高", "最低", "成交量", "换手率", "MACD", "DIF", "DEA", "KDJ.K", "KDJ.D", "KDJ.J", "RSI", "BOLL.U", "BOLL.M", "BOLL.L", ] all = dfs.to_csv(columns=columns, index=False, float_format="%.2f").strip().split("\n") return "\n".join([all[0], *all[-limit:]]) return f"Not Found for {symbol}.{market}"
- mcp_aktools/__init__.py:71-74 (registration)Registers the stock_prices function as an MCP tool with title and description.@mcp.tool( title="获取股票历史价格", description="根据股票代码和市场获取股票历史价格及技术指标, 不支持加密货币", )
- mcp_aktools/__init__.py:76-80 (schema)Pydantic schema definitions for the tool inputs using Field for validation and descriptions.symbol: str = field_symbol, market: str = field_market, period: str = Field("daily", description="周期，如: daily(日线), weekly(周线，不支持美股)"), limit: int = Field(30, description="返回数量(int)", strict=False), ):
- mcp_aktools/__init__.py:581-611 (helper)Supporting function called by stock_prices to add technical indicators (MACD, KDJ, RSI, Bollinger Bands) to the dataframe.def add_technical_indicators(df, clos, lows, high): # 计算MACD指标 ema12 = clos.ewm(span=12, adjust=False).mean() ema26 = clos.ewm(span=26, adjust=False).mean() df["DIF"] = ema12 - ema26 df["DEA"] = df["DIF"].ewm(span=9, adjust=False).mean() df["MACD"] = (df["DIF"] - df["DEA"]) * 2 # 计算KDJ指标 low_min = lows.rolling(window=9, min_periods=1).min() high_max = high.rolling(window=9, min_periods=1).max() rsv = (clos - low_min) / (high_max - low_min) * 100 df["KDJ.K"] = rsv.ewm(com=2, adjust=False).mean() df["KDJ.D"] = df["KDJ.K"].ewm(com=2, adjust=False).mean() df["KDJ.J"] = 3 * df["KDJ.K"] - 2 * df["KDJ.D"] # 计算RSI指标 delta = clos.diff() gain = delta.where(delta > 0, 0) loss = -delta.where(delta < 0, 0) avg_gain = gain.rolling(window=14).mean() avg_loss = loss.rolling(window=14).mean() rs = avg_gain / avg_loss df["RSI"] = 100 - (100 / (1 + rs)) # 计算布林带指标 df["BOLL.M"] = clos.rolling(window=20).mean() std = clos.rolling(window=20).std() df["BOLL.U"] = df["BOLL.M"] + 2 * std df["BOLL.L"] = df["BOLL.M"] - 2 * std
- mcp_aktools/__init__.py:111-119 (helper)Helper function for US stock historical data, used by stock_prices for 'us' market.def stock_us_daily(symbol, start_date="2025-01-01", period="daily"): dfs = ak.stock_us_daily(symbol=symbol) if dfs is None or dfs.empty: return None dfs.rename(columns={"date": "日期", "open": "开盘", "close": "收盘", "high": "最高", "low": "最低", "volume": "成交量"}, inplace=True) dfs["换手率"] = None dfs.index = pd.to_datetime(dfs["日期"], errors="coerce") return dfs[start_date:"2222-01-01"]