stock_zt_pool_strong_em
Retrieve data on strong-performing stock pools in China's A-share markets (Shanghai, Shenzhen) to identify potential investment opportunities.
Instructions
获取中国A股市场(上证、深证)的强势股池数据
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| date | No | 交易日日期(可选)，默认为最近的交易日，格式: 20251231 | |
| limit | No | 返回数量(int,30-100) |
Implementation Reference
- mcp_aktools/__init__.py:263-280 (handler)The handler function for the 'stock_zt_pool_strong_em' MCP tool. It is registered via the @mcp.tool decorator, defines input schema via Pydantic Fields, fetches data from akshare using caching, processes the dataframe (drops columns, sorts by transaction amount, limits rows), and returns formatted CSV output.@mcp.tool( title="A股强势股池", description="获取中国A股市场(上证、深证)的强势股池数据", ) def stock_zt_pool_strong_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_strong_em, date=date, ttl=1200) try: dfs.drop(columns=["序号", "流通市值", "总市值"], inplace=True) except Exception: pass dfs.sort_values("成交额", ascending=False, inplace=True) dfs = dfs.head(int(limit)) return dfs.to_csv(index=False, float_format="%.2f").strip()
- mcp_aktools/__init__.py:564-579 (helper)Helper function that implements caching for akshare API calls using CacheKey, with configurable TTL. Used in the handler to cache the ak.stock_zt_pool_strong_em data fetch.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 to determine the most recent trading date, used in the handler when date parameter is not 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