get-historical-ohlcv

Instructions

Implementation Reference

• src/server.py:288-305 (handler)

  Handler logic for the get-historical-ohlcv tool: parses arguments, fetches OHLCV data from the exchange using ccxt, formats it with helper, and returns formatted text content.

  elif name == "get-historical-ohlcv":
      symbol = arguments.get("symbol", "").upper()
      timeframe = arguments.get("timeframe", "1h")
      days_back = min(int(arguments.get("days_back", 7)), 30)
  
      # Calculate timestamps
      since = int((datetime.now() - timedelta(days=days_back)).timestamp() * 1000)
  
      # Fetch historical data
      ohlcv = await exchange.fetch_ohlcv(symbol, timeframe, since=since)
  
      formatted_data = format_ohlcv_data(ohlcv, timeframe)
      return [
          types.TextContent(
              type="text",
              text=f"Historical OHLCV data for {symbol} ({timeframe}) on {exchange_id.upper()}:\n\n{formatted_data}"
          )
      ]

• src/server.py:158-184 (registration)

  Registration of the get-historical-ohlcv tool in the list_tools handler, including description and input schema definition.

  types.Tool(
      name="get-historical-ohlcv",
      description="Get historical OHLCV (candlestick) data for a trading pair",
      inputSchema={
          "type": "object",
          "properties": {
              "symbol": {
                  "type": "string",
                  "description": "Trading pair symbol (e.g., BTC/USDT, ETH/USDT)",
              },
              "timeframe": {
                  "type": "string",
                  "description": "Timeframe for candlesticks (e.g., 1m, 5m, 15m, 1h, 4h, 1d)",
                  "enum": ["1m", "5m", "15m", "1h", "4h", "1d"],
                  "default": "1h"
              },
              "days_back": {
                  "type": "number",
                  "description": "Number of days of historical data to fetch (default: 7, max: 30)",
                  "default": 7,
                  "maximum": 30
              },
              "exchange": get_exchange_schema()
          },
          "required": ["symbol"],
      },
  ),

• src/server.py:162-183 (schema)

  Input schema definition for the get-historical-ohlcv tool, specifying parameters like symbol, timeframe, days_back, and exchange.

      "type": "object",
      "properties": {
          "symbol": {
              "type": "string",
              "description": "Trading pair symbol (e.g., BTC/USDT, ETH/USDT)",
          },
          "timeframe": {
              "type": "string",
              "description": "Timeframe for candlesticks (e.g., 1m, 5m, 15m, 1h, 4h, 1d)",
              "enum": ["1m", "5m", "15m", "1h", "4h", "1d"],
              "default": "1h"
          },
          "days_back": {
              "type": "number",
              "description": "Number of days of historical data to fetch (default: 7, max: 30)",
              "default": 7,
              "maximum": 30
          },
          "exchange": get_exchange_schema()
      },
      "required": ["symbol"],
  },

• src/server.py:69-97 (helper)

  Helper function used by the handler to format the raw OHLCV data into a human-readable string with timestamps, OHLCV values, and price changes.

  def format_ohlcv_data(ohlcv_data: List[List], timeframe: str) -> str:
      """Format OHLCV data into a readable string with price changes."""
      formatted_data = []
  
      for i, candle in enumerate(ohlcv_data):
          timestamp, open_price, high, low, close, volume = candle
  
          # Calculate price change from previous close if available
          price_change = ""
          if i > 0:
              prev_close = ohlcv_data[i-1][4]
              change_pct = ((close - prev_close) / prev_close) * 100
              price_change = f"Change: {change_pct:+.2f}%"
  
          # Format the candle data
          dt = datetime.fromtimestamp(timestamp/1000).strftime('%Y-%m-%d %H:%M:%S')
          candle_str = (
              f"Time: {dt}\n"
              f"Open: {open_price:.8f}\n"
              f"High: {high:.8f}\n"
              f"Low: {low:.8f}\n"
              f"Close: {close:.8f}\n"
              f"Volume: {volume:.2f}\n"
              f"{price_change}\n"
              "---"
          )
          formatted_data.append(candle_str)
  
      return "\n".join(formatted_data)