A股板块资金流

Instructions

Implementation Reference

• mcp_aktools/__init__.py:296-316 (registration)

  The @mcp.tool decorator registers 'stock_sector_fund_flow_rank' as an MCP tool with title 'A股板块资金流' and a description of the data it returns.

  @mcp.tool(
      title="A股板块资金流",
      description="获取中国A股市场(上证、深证)的行业资金流向数据",
  )
  def stock_sector_fund_flow_rank(
      days: str = Field("今日", description="天数，仅支持: {'今日','5日','10日'}，如果需要获取今日数据，请确保是交易日"),
      cate: str = Field("行业资金流", description="仅支持: {'行业资金流','概念资金流','地域资金流'}"),
  ):
      dfs = ak_cache(ak.stock_sector_fund_flow_rank, indicator=days, sector_type=cate, ttl=1200)
      if dfs is None:
          return "获取数据失败"
      try:
          dfs.sort_values("今日涨跌幅", ascending=False, inplace=True)
          dfs.drop(columns=["序号"], inplace=True)
      except Exception:
          pass
      try:
          dfs = pd.concat([dfs.head(20), dfs.tail(20)])
          return dfs.to_csv(index=False, float_format="%.2f").strip()
      except Exception as exc:
          return str(exc)

• mcp_aktools/__init__.py:300-316 (handler)

  The function 'stock_sector_fund_flow_rank' executes the tool logic: calls ak_cache wrapping ak.stock_sector_fund_flow_rank with indicator and sector_type parameters, sorts by '今日涨跌幅', drops the '序号' column, and returns top/bottom 20 rows as CSV.

  def stock_sector_fund_flow_rank(
      days: str = Field("今日", description="天数，仅支持: {'今日','5日','10日'}，如果需要获取今日数据，请确保是交易日"),
      cate: str = Field("行业资金流", description="仅支持: {'行业资金流','概念资金流','地域资金流'}"),
  ):
      dfs = ak_cache(ak.stock_sector_fund_flow_rank, indicator=days, sector_type=cate, ttl=1200)
      if dfs is None:
          return "获取数据失败"
      try:
          dfs.sort_values("今日涨跌幅", ascending=False, inplace=True)
          dfs.drop(columns=["序号"], inplace=True)
      except Exception:
          pass
      try:
          dfs = pd.concat([dfs.head(20), dfs.tail(20)])
          return dfs.to_csv(index=False, float_format="%.2f").strip()
      except Exception as exc:
          return str(exc)

• mcp_aktools/__init__.py:300-302 (schema)

  The function parameters define the input schema: 'days' (str, default '今日', values: {'今日','5日','10日'}) and 'cate' (str, default '行业资金流', values: {'行业资金流','概念资金流','地域资金流'}) with pydantic Field descriptions.

  def stock_sector_fund_flow_rank(
      days: str = Field("今日", description="天数，仅支持: {'今日','5日','10日'}，如果需要获取今日数据，请确保是交易日"),
      cate: str = Field("行业资金流", description="仅支持: {'行业资金流','概念资金流','地域资金流'}"),

• mcp_aktools/__init__.py:564-579 (helper)

  The 'ak_cache' helper function wraps akshare calls with a two-tier cache (in-memory + diskcache), keyed by function name and arguments, with configurable TTL.

  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