A股涨停股池

Instructions

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