美股关键指标

Instructions

Implementation Reference

• mcp_aktools/__init__.py:197-200 (registration)

  Tool registration via @mcp.tool decorator with title and description

  @mcp.tool(
      title="美股关键指标",
      description="获取美股市场的股票财务报告关键指标",
  )

• mcp_aktools/__init__.py:201-206 (handler)

  Handler function: fetches US stock financial analysis indicators from akshare, converts to CSV, returns first 15 lines

  def stock_indicators_us(
      symbol: str = field_symbol,
  ):
      dfs = ak_cache(ak.stock_financial_us_analysis_indicator_em, symbol=symbol, indicator="单季报")
      keys = dfs.to_csv(index=False, float_format="%.3f").strip().split("\n")
      return "\n".join(keys[0:15])

• mcp_aktools/__init__.py:201-203 (schema)

  Input schema: single required parameter 'symbol' (string, stock code) using shared Field definition

  def stock_indicators_us(
      symbol: str = field_symbol,
  ):

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

  Helper caching function used to call akshare APIs with disk/memory cache layer

  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:7-46 (helper)

  Cache helper class providing TTL-based memory cache and persistent disk cache

  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