A股涨停股池
stock_zt_pool_emRetrieves all stocks that hit the daily price limit (涨停) in China's A-share markets, with optional date and count parameters.
Instructions
获取中国A股市场(上证、深证)的所有涨停股票
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| date | No | 交易日日期(可选)，默认为最近的交易日，格式: 20251231 | |
| limit | No | 返回数量(int,30-100) |
Implementation Reference
- mcp_aktools/__init__.py:245-260 (handler)The function that executes the stock_zt_pool_em tool logic: fetches the daily limit-up (涨停) stock pool from akshare, drops unnecessary columns, sorts by turnover amount, limits results, and returns as CSV text.
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:241-244 (registration)Registration of stock_zt_pool_em as an MCP tool using the @mcp.tool decorator with title and description metadata.
@mcp.tool( title="A股涨停股池", description="获取中国A股市场(上证、深证)的所有涨停股票", ) - mcp_aktools/__init__.py:246-248 (schema)Input parameter schema for stock_zt_pool_em: date (optional string) and limit (int, default 50, range 30-100).
date: str = Field("", description="交易日日期(可选)，默认为最近的交易日，格式: 20251231"), limit: int = Field(50, description="返回数量(int,30-100)", strict=False), ): - mcp_aktools/__init__.py:229-238 (helper)Helper function to find the most recent trading date, used when no date is provided to stock_zt_pool_em.
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 - mcp_aktools/__init__.py:564-579 (helper)Helper caching function that wraps akshare API calls with in-memory and disk caching (TTL-based), used to call ak.stock_zt_pool_em with a 1200-second TTL cache.
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