stock_prices
Retrieve historical stock prices and technical indicators by entering a stock symbol and market. Supports major exchanges including Shanghai, Shenzhen, Hong Kong, and US markets.
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:71-108 (handler)The implementation of the 'stock_prices' tool. This is the handler function decorated with @mcp.tool(), which registers the tool and defines its schema via parameter annotations. It fetches historical stock price data using akshare for various markets (sh, sz, hk, us, etf), caches results, computes technical indicators (MACD, KDJ, RSI, BOLL), and returns the last 'limit' rows in CSV format.@mcp.tool( title="获取股票历史价格", description="根据股票代码和市场获取股票历史价格及技术指标, 不支持加密货币", ) 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)The @mcp.tool decorator registers the 'stock_prices' tool with FastMCP, specifying title and description. The schema is defined inline via Field descriptions in parameters.@mcp.tool( title="获取股票历史价格", description="根据股票代码和市场获取股票历史价格及技术指标, 不支持加密货币", )
- mcp_aktools/__init__.py:111-118 (helper)Helper function stock_us_daily used by stock_prices for US market data.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"]
- mcp_aktools/__init__.py:120-127 (helper)Helper function fund_etf_hist_sina used by stock_prices for ETF data in sh/sz markets.def fund_etf_hist_sina(symbol, market="sh", start_date="2025-01-01", period="daily"): dfs = ak.fund_etf_hist_sina(symbol=f"{market}{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"]
- mcp_aktools/__init__.py:581-611 (helper)Helper function to compute and add technical indicators used by stock_prices.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