美股关键指标
stock_indicators_usRetrieve key financial report indicators for US stocks. Input a stock symbol to get essential metrics from financial statements.
Instructions
获取美股市场的股票财务报告关键指标
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| symbol | Yes | 股票代码 |
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