akshare-one-mcp

Instructions

Implementation Reference

• src/akshare_one_mcp/server.py:14-152 (handler)

  The complete handler function for the MCP tool 'get_hist_data'. It is registered via @mcp.tool decorator. The function signature defines the input schema using Annotated and Field. It fetches historical data using ako.get_hist_data, optionally computes and adds technical indicators from a map, limits to recent N rows, and returns JSON.

  @mcp.tool
  def get_hist_data(
      symbol: Annotated[str, Field(description="Stock symbol/ticker (e.g. '000001')")],
      interval: Annotated[
          Literal["minute", "hour", "day", "week", "month", "year"],
          Field(description="Time interval"),
      ] = "day",
      interval_multiplier: Annotated[
          int, Field(description="Interval multiplier", ge=1)
      ] = 1,
      start_date: Annotated[
          str, Field(description="Start date in YYYY-MM-DD format")
      ] = "1970-01-01",
      end_date: Annotated[
          str, Field(description="End date in YYYY-MM-DD format")
      ] = "2030-12-31",
      adjust: Annotated[
          Literal["none", "qfq", "hfq"], Field(description="Adjustment type")
      ] = "none",
      source: Annotated[
          Literal["eastmoney", "eastmoney_direct", "sina"],
          Field(description="Data source"),
      ] = "eastmoney",
      indicators_list: Annotated[
          list[
              Literal[
                  "SMA",
                  "EMA",
                  "RSI",
                  "MACD",
                  "BOLL",
                  "STOCH",
                  "ATR",
                  "CCI",
                  "ADX",
                  "WILLR",
                  "AD",
                  "ADOSC",
                  "OBV",
                  "MOM",
                  "SAR",
                  "TSF",
                  "APO",
                  "AROON",
                  "AROONOSC",
                  "BOP",
                  "CMO",
                  "DX",
                  "MFI",
                  "MINUS_DI",
                  "MINUS_DM",
                  "PLUS_DI",
                  "PLUS_DM",
                  "PPO",
                  "ROC",
                  "ROCP",
                  "ROCR",
                  "ROCR100",
                  "TRIX",
                  "ULTOSC",
              ]
          ]
          | None,
          Field(description="Technical indicators to add"),
      ] = None,
      recent_n: Annotated[
          int | None, Field(description="Number of most recent records to return", ge=1)
      ] = 100,
  ) -> str:
      """Get historical stock market data. 'eastmoney_direct' support all A,B,H shares"""
      df = ako.get_hist_data(
          symbol=symbol,
          interval=interval,
          interval_multiplier=interval_multiplier,
          start_date=start_date,
          end_date=end_date,
          adjust=adjust,
          source=source,
      )
      if indicators_list:
          indicator_map = {
              "SMA": (indicators.get_sma, {"window": 20}),
              "EMA": (indicators.get_ema, {"window": 20}),
              "RSI": (indicators.get_rsi, {"window": 14}),
              "MACD": (indicators.get_macd, {"fast": 12, "slow": 26, "signal": 9}),
              "BOLL": (indicators.get_bollinger_bands, {"window": 20, "std": 2}),
              "STOCH": (
                  indicators.get_stoch,
                  {"window": 14, "smooth_d": 3, "smooth_k": 3},
              ),
              "ATR": (indicators.get_atr, {"window": 14}),
              "CCI": (indicators.get_cci, {"window": 14}),
              "ADX": (indicators.get_adx, {"window": 14}),
              "WILLR": (indicators.get_willr, {"window": 14}),
              "AD": (indicators.get_ad, {}),
              "ADOSC": (indicators.get_adosc, {"fast_period": 3, "slow_period": 10}),
              "OBV": (indicators.get_obv, {}),
              "MOM": (indicators.get_mom, {"window": 10}),
              "SAR": (indicators.get_sar, {"acceleration": 0.02, "maximum": 0.2}),
              "TSF": (indicators.get_tsf, {"window": 14}),
              "APO": (
                  indicators.get_apo,
                  {"fast_period": 12, "slow_period": 26, "ma_type": 0},
              ),
              "AROON": (indicators.get_aroon, {"window": 14}),
              "AROONOSC": (indicators.get_aroonosc, {"window": 14}),
              "BOP": (indicators.get_bop, {}),
              "CMO": (indicators.get_cmo, {"window": 14}),
              "DX": (indicators.get_dx, {"window": 14}),
              "MFI": (indicators.get_mfi, {"window": 14}),
              "MINUS_DI": (indicators.get_minus_di, {"window": 14}),
              "MINUS_DM": (indicators.get_minus_dm, {"window": 14}),
              "PLUS_DI": (indicators.get_plus_di, {"window": 14}),
              "PLUS_DM": (indicators.get_plus_dm, {"window": 14}),
              "PPO": (
                  indicators.get_ppo,
                  {"fast_period": 12, "slow_period": 26, "ma_type": 0},
              ),
              "ROC": (indicators.get_roc, {"window": 10}),
              "ROCP": (indicators.get_rocp, {"window": 10}),
              "ROCR": (indicators.get_rocr, {"window": 10}),
              "ROCR100": (indicators.get_rocr100, {"window": 10}),
              "TRIX": (indicators.get_trix, {"window": 30}),
              "ULTOSC": (
                  indicators.get_ultosc,
                  {"window1": 7, "window2": 14, "window3": 28},
              ),
          }
          temp = []
          for indicator in indicators_list:
              if indicator in indicator_map:
                  func, params = indicator_map[indicator]
                  indicator_df = func(df, **params)
                  temp.append(indicator_df)
          if temp:
              df = df.join(temp)
      if recent_n is not None:
          df = df.tail(recent_n)
      return df.to_json(orient="records") or "[]"