stock_zt_pool_em
Retrieve all limit-up stocks in China's A-share market (Shanghai and Shenzhen exchanges) for specific trading dates, helping investors identify stocks with maximum daily price increases.
Instructions
获取中国A股市场(上证、深证)的所有涨停股票
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| date | No | 交易日日期(可选)，默认为最近的交易日，格式: 20251231 | |
| limit | No | 返回数量(int,30-100) |
Implementation Reference
- mcp_aktools/__init__.py:241-260 (handler)The handler function decorated with @mcp.tool() that implements the stock_zt_pool_em tool. It fetches limit-up stocks data from akshare via cache, processes the dataframe by dropping columns, sorting by turnover, limiting rows, and returns a description with CSV output.@mcp.tool( title="A股涨停股池", description="获取中国A股市场(上证、深证)的所有涨停股票", ) def stock_zt_pool_em( date: str = Field("", description="交易日日期(可选)，默认为最近的交易日，格式: 20251231"), limit: int = Field(50, description="返回数量(int,30-100)", strict=False), ): if not date: date = recent_trade_date().strftime("%Y%m%d") dfs = ak_cache(ak.stock_zt_pool_em, date=date, ttl=1200) cnt = len(dfs) try: dfs.drop(columns=["序号", "流通市值", "总市值"], inplace=True) except Exception: pass dfs.sort_values("成交额", ascending=False, inplace=True) dfs = dfs.head(int(limit)) desc = f"共{cnt}只涨停股\n" return desc + dfs.to_csv(index=False, float_format="%.2f").strip()
- mcp_aktools/__init__.py:564-579 (helper)Helper function ak_cache used by the tool to cache akshare API calls, preventing repeated requests.def ak_cache(fun, *args, **kwargs) -> pd.DataFrame | None: key = kwargs.pop("key", None) if not key: key = f"{fun.__name__}-{args}-{kwargs}" ttl1 = kwargs.pop("ttl", 86400) ttl2 = kwargs.pop("ttl2", None) cache = CacheKey.init(key, ttl1, ttl2) all = cache.get() if all is None: try: _LOGGER.info("Request akshare: %s", [key, args, kwargs]) all = fun(*args, **kwargs) cache.set(all) except Exception as exc: _LOGGER.exception(str(exc)) return all
- mcp_aktools/__init__.py:229-238 (helper)Helper function recent_trade_date used to get the most recent trading date if none provided.def recent_trade_date(): now = datetime.now().date() dfs = ak_cache(ak.tool_trade_date_hist_sina, ttl=43200) if dfs is None: return now dfs.sort_values("trade_date", ascending=False, inplace=True) for d in dfs["trade_date"]: if d <= now: return d return now