A股板块资金流

Instructions

Implementation Reference

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

  Tool registration via @mcp.tool decorator with title 'A股板块资金流' and description about A-share sector fund flow data

  @mcp.tool(
      title="A股板块资金流",
      description="获取中国A股市场(上证、深证)的行业资金流向数据",
  )

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

  Handler function that calls ak.stock_sector_fund_flow_rank via ak_cache, sorts by '今日涨跌幅', drops '序号' 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:564-579 (helper)

  ak_cache helper function that wraps akshare API calls with caching via CacheKey (TTL-based + disk-based 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

• mcp_aktools/cache.py:1-45 (helper)

  CacheKey class implementing dual-layer caching (TTLCache + diskcache) used by ak_cache

  import sys
  import pathlib
  import diskcache
  from cachetools import TTLCache
  
  
  class CacheKey:
      ALL: dict = {}
  
      def __init__(self, key, ttl=600, ttl2=None, maxsize=100):
          self.key = key
          self.ttl = ttl
          self.ttl2 = ttl2 or (ttl * 2)
          self.cache1 = TTLCache(maxsize=maxsize, ttl=ttl)
          self.cache2 = diskcache.Cache(self.get_cache_dir())
  
      @staticmethod
      def init(key, ttl=600, ttl2=None, maxsize=100):
          if key in CacheKey.ALL:
              return CacheKey.ALL[key]
          cache = CacheKey(key, ttl, ttl2, maxsize)
          return CacheKey.ALL.setdefault(key, cache)
  
      def get(self):
          try:
              return self.cache1[self.key]
          except KeyError:
              pass
          return self.cache2.get(self.key)
  
      def set(self, val):
          self.cache1[self.key] = val
          self.cache2.set(self.key, val, expire=self.ttl2)
          return val
  
      def delete(self):
          self.cache1.pop(self.key, None)
          self.cache2.delete(self.key)
  
      def get_cache_dir(self):
          home = pathlib.Path.home()
          name = __package__
          if sys.platform == "win32":
              return home / "AppData" / "Local" / "Cache" / name
          return home / ".cache" / name