Hummingbot 第14章：Hummingbot API 使用

Hummingbot 提供了丰富的 REST API 和 WebSocket 接口，方便用户通过外部程序控制交易机器人、获取市场数据和管理账户。本章详细介绍 API 的使用方法。

REST API 概述

Hummingbot 的 REST API 基于 HTTP 协议，默认监听 127.0.0.1:15871 端口。API 使用 JSON 格式进行数据交换。

启动 API 服务

在 conf_client.yml 中启用 API 服务：

# conf_client.yml
api_server_host: 127.0.0.1
api_server_port: 15871
api_server_enabled: true
api_key: "your-api-key-here"
api_secret: "your-api-secret-here"

API 基础 URL

http://127.0.0.1:15871

API 认证

Hummingbot API 使用 API Key + HMAC-SHA256 签名认证机制。

生成 API 凭证

# 在 Hummingbot 客户端中生成
>>> key
Enter new API key: your-api-key
Confirm new API key: your-api-key
API key has been created and saved.

认证请求头

每个请求需要包含以下头信息：

x-api-key: your-api-key
x-api-signature: HMAC-SHA256 签名
x-api-timestamp: 当前时间戳（毫秒）

Python 认证示例

import time
import hmac
import hashlib
import requests

class HummingbotClient:
    def __init__(self, api_key: str, api_secret: str, base_url: str = "http://127.0.0.1:15871"):
        self.api_key = api_key
        self.api_secret = api_secret
        self.base_url = base_url

    def _sign_request(self, params: dict = None) -> dict:
        timestamp = str(int(time.time() * 1000))
        raw = timestamp + "." + (json.dumps(params) if params else "")
        signature = hmac.new(
            self.api_secret.encode(),
            raw.encode(),
            hashlib.sha256
        ).hexdigest()
        return {
            "x-api-key": self.api_key,
            "x-api-signature": signature,
            "x-api-timestamp": timestamp
        }

    def _request(self, method: str, path: str, params: dict = None):
        headers = self._sign_request(params)
        url = self.base_url + path
        if method == "GET":
            return requests.get(url, headers=headers, params=params)
        elif method == "POST":
            return requests.post(url, headers=headers, json=params)

    def get_status(self) -> dict:
        return self._request("GET", "/status").json()

    def start_bot(self) -> dict:
        return self._request("POST", "/start").json()

    def stop_bot(self) -> dict:
        return self._request("POST", "/stop").json()

交易端点

启动和停止机器人

client = HummingbotClient("your-api-key", "your-api-secret")

# 启动机器人
response = client.start_bot()
print(response)
# {"status": "ok", "message": "Bot started successfully."}

# 停止机器人
response = client.stop_bot()
print(response)
# {"status": "ok", "message": "Bot stopped successfully."}

导入配置

# 导入策略配置
config_path = "/path/to/config.yml"
response = requests.post(
    "http://127.0.0.1:15871/config/import",
    headers=headers,
    json={"config_path": config_path}
)

查看当前配置

response = client._request("GET", "/config")
print(response.json())
# {
#     "strategy": "pure_market_making",
#     "exchange": "binance",
#     "trading_pair": "BTC-USDT",
#     ...
# }

市场数据

获取订单簿

response = requests.get(
    "http://127.0.0.1:15871/orderbook",
    headers=headers,
    params={"trading_pair": "BTC-USDT", "exchange": "binance", "depth": 10}
)
data = response.json()
print(f"Bids: {data['bids'][:3]}")
print(f"Asks: {data['asks'][:3]}")

获取 ticker

response = requests.get(
    "http://127.0.0.1:15871/ticker",
    headers=headers,
    params={"trading_pair": "BTC-USDT", "exchange": "binance"}
)
data = response.json()
# {
#     "bid": 50000.0,
#     "ask": 50001.0,
#     "last_price": 50000.5,
#     "volume_24h": 12345.6
# }

获取历史 K 线

response = requests.get(
    "http://127.0.0.1:15871/candles",
    headers=headers,
    params={
        "trading_pair": "BTC-USDT",
        "exchange": "binance",
        "interval": "1h",
        "limit": 100
    }
)
candles = response.json()

账户管理

查询余额

response = client._request("GET", "/balances")
data = response.json()
for asset, info in data["balances"].items():
    print(f"{asset}: {info['total']} (可用: {info['available']})")

查看持仓

response = client._request("GET", "/positions")
positions = response.json()
# 返回当前所有持仓信息

查询历史交易

response = requests.get(
    "http://127.0.0.1:15871/history",
    headers=headers,
    params={"limit": 50, "page": 1}
)
history = response.json()

管理交易所连接

WebSocket 实时数据

连接 WebSocket

Hummingbot 的 WebSocket 端点与 REST API 相同端口：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    print(f"收到实时数据: {data}")

def on_error(ws, error):
    print(f"WebSocket 错误: {error}")

def on_close(ws, close_status_code, close_msg):
    print("WebSocket 连接关闭")

def on_open(ws):
    print("WebSocket 连接已建立")
    # 订阅订单簿更新
    ws.send(json.dumps({
        "type": "subscribe",
        "channel": "orderbook",
        "trading_pair": "BTC-USDT",
        "exchange": "binance"
    }))

ws = websocket.WebSocketApp(
    "ws://127.0.0.1:15871/ws",
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)
ws.run_forever()

支持的数据通道

API 客户端示例

完整的 Python 客户端

import time
import hmac
import hashlib
import json
import requests
from typing import Optional, Dict, Any

class HummingbotAPIClient:
    """Hummingbot API 完整客户端"""

    def __init__(self, api_key: str, api_secret: str,
                 base_url: str = "http://127.0.0.1:15871"):
        self.api_key = api_key
        self.api_secret = api_secret
        self.base_url = base_url

    def _sign(self, payload: str) -> str:
        return hmac.new(
            self.api_secret.encode(),
            payload.encode(),
            hashlib.sha256
        ).hexdigest()

    def _headers(self, params: dict = None) -> dict:
        timestamp = str(int(time.time() * 1e3))
        payload = timestamp + "."
        if params:
            payload += json.dumps(params)
        return {
            "x-api-key": self.api_key,
            "x-api-signature": self._sign(payload),
            "x-api-timestamp": timestamp,
            "Content-Type": "application/json"
        }

    def _get(self, path: str, params: dict = None) -> Dict[str, Any]:
        resp = requests.get(
            self.base_url + path,
            headers=self._headers(params),
            params=params
        )
        return self._handle(resp)

    def _post(self, path: str, data: dict = None) -> Dict[str, Any]:
        resp = requests.post(
            self.base_url + path,
            headers=self._headers(data),
            json=data
        )
        return self._handle(resp)

    def _handle(self, resp) -> Dict[str, Any]:
        if resp.status_code != 200:
            raise Exception(f"API Error: {resp.status_code} {resp.text}")
        return resp.json()

    # 机器人控制
    def start(self) -> dict:
        return self._post("/start")

    def stop(self) -> dict:
        return self._post("/stop")

    def status(self) -> dict:
        return self._get("/status")

    # 配置
    def get_config(self) -> dict:
        return self._get("/config")

    def import_config(self, path: str) -> dict:
        return self._post("/config/import", {"config_path": path})

    # 市场数据
    def orderbook(self, pair: str, exchange: str, depth: int = 10) -> dict:
        return self._get("/orderbook", {
            "trading_pair": pair,
            "exchange": exchange,
            "depth": depth
        })

    def ticker(self, pair: str, exchange: str) -> dict:
        return self._get("/ticker", {
            "trading_pair": pair,
            "exchange": exchange
        })

    def candles(self, pair: str, exchange: str, interval: str = "1h",
                limit: int = 100) -> dict:
        return self._get("/candles", {
            "trading_pair": pair,
            "exchange": exchange,
            "interval": interval,
            "limit": limit
        })

    # 账户
    def balances(self) -> dict:
        return self._get("/balances")

    def history(self, limit: int = 50, page: int = 1) -> dict:
        return self._get("/history", {"limit": limit, "page": page})

    def positions(self) -> dict:
        return self._get("/positions")


# 使用示例
if __name__ == "__main__":
    client = HummingbotAPIClient("my-key", "my-secret")
    print("机器人状态:", client.status())
    print("余额:", client.balances())
    print("订单簿:", client.orderbook("BTC-USDT", "binance", 5))

错误处理