Huobi HTX API with Python: Complete Trader's Guide

HTX — rebranded from Huobi Global — remains one of the most liquid crypto exchanges in Asia, with deep order books across hundreds of spot and futures pairs. If you're building a trading bot, automating execution, or pulling market data for analysis, the HTX API is a serious option. Python makes it approachable even if you're not a full-time developer, and the official REST + WebSocket endpoints cover everything from order placement to account balance queries. Here's how to get set up and actually do something useful with it.

Setting Up Your HTX API Keys

Before writing a single line of Python, you need API credentials. Log into your HTX account, navigate to Account → API Management, and create a new key pair. HTX gives you an Access Key and a Secret Key — treat the secret like a password, it's shown only once. You'll also be asked to set IP whitelisting, which is strongly recommended for any trading key. If your bot runs on a VPS, whitelist only that server's IP. For read-only market data keys, IP restriction is optional but still good practice.

Never embed API keys directly in your source code. Store them in environment variables or a .env file that's excluded from version control. A leaked trading key on GitHub has cost traders real money.

pip install requests python-dotenv

# Create a .env file in your project root
echo 'HTX_ACCESS_KEY=your_access_key_here' >> .env
echo 'HTX_SECRET_KEY=your_secret_key_here' >> .env

Authentication and HMAC Signing

The HTX REST API uses HMAC-SHA256 signatures for all private endpoints. Every signed request includes your access key, a timestamp, and a signature built from the request parameters. This is where most beginners trip up — the signature calculation has to be exact or you'll get a 403. Here's a clean implementation that handles the signing for you:

import hmac
import hashlib
import base64
import urllib.parse
from datetime import datetime, timezone
import requests
import os
from dotenv import load_dotenv

load_dotenv()

ACCESS_KEY = os.getenv('HTX_ACCESS_KEY')
SECRET_KEY = os.getenv('HTX_SECRET_KEY')
BASE_URL = 'https://api.huobi.pro'

def sign_request(method, endpoint, params=None):
    """Generate HMAC-SHA256 signature for HTX API requests."""
    if params is None:
        params = {}

    timestamp = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S')

    sign_params = {
        'AccessKeyId': ACCESS_KEY,
        'SignatureMethod': 'HmacSHA256',
        'SignatureVersion': '2',
        'Timestamp': timestamp,
        **params
    }

    sorted_params = '&'.join(
        f'{k}={urllib.parse.quote(str(v), safe="")}'
        for k, v in sorted(sign_params.items())
    )

    host = 'api.huobi.pro'
    payload = f'{method}\n{host}\n{endpoint}\n{sorted_params}'
    digest = hmac.new(
        SECRET_KEY.encode('utf-8'),
        payload.encode('utf-8'),
        hashlib.sha256
    ).digest()

    signature = base64.b64encode(digest).decode()
    sign_params['Signature'] = signature

    return sign_params

def get_account_balance():
    """Fetch spot account balances."""
    # First get account ID
    params = sign_request('GET', '/v1/account/accounts')
    resp = requests.get(f'{BASE_URL}/v1/account/accounts', params=params)
    resp.raise_for_status()
    data = resp.json()

    if data['status'] != 'ok':
        raise Exception(f"API error: {data.get('err-msg', 'Unknown error')}")

    accounts = data['data']
    spot_account = next((a for a in accounts if a['type'] == 'spot'), None)

    if not spot_account:
        return []

    # Now fetch balances for that account
    account_id = spot_account['id']
    params = sign_request('GET', f'/v1/account/accounts/{account_id}/balance')
    resp = requests.get(
        f'{BASE_URL}/v1/account/accounts/{account_id}/balance',
        params=params
    )
    resp.raise_for_status()
    balance_data = resp.json()

    # Filter non-zero balances
    balances = [
        b for b in balance_data['data']['list']
        if float(b['balance']) > 0
    ]
    return balances

if __name__ == '__main__':
    try:
        balances = get_account_balance()
        for b in balances:
            print(f"{b['currency'].upper()}: {b['balance']} ({b['type']})")
    except requests.HTTPError as e:
        print(f'HTTP error: {e}')
    except Exception as e:
        print(f'Error: {e}')

Fetching Market Data and Order Books

Public market data endpoints don't require authentication — you can pull candlestick data, order books, and ticker prices without signing anything. This is useful for strategy research, backtesting signal logic, or building dashboards. Compared to what you'd get on Binance or OKX, the HTX market data API is similarly structured but uses slightly different parameter names, so don't assume you can copy-paste code from another exchange's integration.

def get_klines(symbol: str, period: str = '1min', size: int = 100):
    """
    Fetch OHLCV candlestick data.
    symbol: e.g. 'btcusdt'
    period: 1min, 5min, 15min, 30min, 60min, 4hour, 1day, 1week, 1mon
    size: number of candles, max 2000
    """
    endpoint = '/market/history/kline'
    params = {
        'symbol': symbol.lower(),
        'period': period,
        'size': size
    }

    resp = requests.get(f'{BASE_URL}{endpoint}', params=params)
    resp.raise_for_status()
    data = resp.json()

    if data['status'] != 'ok':
        raise Exception(f"Market data error: {data.get('err-msg')}")

    return data['data']  # list of OHLCV dicts


def get_orderbook(symbol: str, depth: int = 20):
    """Fetch current order book snapshot."""
    endpoint = '/market/depth'
    params = {
        'symbol': symbol.lower(),
        'type': f'step0',
        'depth': depth
    }

    resp = requests.get(f'{BASE_URL}{endpoint}', params=params)
    resp.raise_for_status()
    data = resp.json()

    tick = data['tick']
    return {
        'bids': tick['bids'][:depth],  # [price, qty] pairs
        'asks': tick['asks'][:depth],
        'ts': tick['ts']
    }


# Example usage
candles = get_klines('ethusdt', period='15min', size=50)
print(f"Latest ETH/USDT close: {candles[0]['close']}")

book = get_orderbook('btcusdt', depth=5)
print(f"BTC best bid: {book['bids'][0][0]}, best ask: {book['asks'][0][0]}")

One practical advantage of HTX's market data is the depth of altcoin coverage. For tokens that aren't listed on Binance or Bybit, HTX often has liquid markets, which makes it useful when you're building multi-exchange scanners or looking for arbitrage opportunities across platforms.

Placing and Managing Orders

Order placement is where authentication matters. HTX supports limit, market, stop-limit, and trailing stop orders through the REST API. The order endpoint is straightforward — you POST to /v1/order/orders/place with a signed payload containing the symbol, type, side, price, and quantity. Error handling here is critical because the exchange will reject malformed orders silently if you're not checking the response body properly.

def place_limit_order(account_id: str, symbol: str, side: str,
                      price: float, amount: float):
    """
    Place a limit order.
    side: 'buy-limit' or 'sell-limit'
    """
    endpoint = '/v1/order/orders/place'

    order_params = {
        'account-id': str(account_id),
        'symbol': symbol.lower(),
        'type': side,
        'amount': str(amount),
        'price': str(price),
        'source': 'spot-api'
    }

    sign_params = sign_request('POST', endpoint)

    resp = requests.post(
        f'{BASE_URL}{endpoint}',
        params=sign_params,
        json=order_params,
        headers={'Content-Type': 'application/json'}
    )
    resp.raise_for_status()
    result = resp.json()

    if result['status'] != 'ok':
        raise Exception(f"Order failed: {result.get('err-msg')} (code: {result.get('err-code')})")

    order_id = result['data']
    print(f"Order placed successfully. ID: {order_id}")
    return order_id


def cancel_order(order_id: str):
    """Cancel an open order by ID."""
    endpoint = f'/v1/order/orders/{order_id}/submitcancel'
    params = sign_request('POST', endpoint)

    resp = requests.post(f'{BASE_URL}{endpoint}', params=params)
    resp.raise_for_status()
    result = resp.json()

    if result['status'] != 'ok':
        raise Exception(f"Cancel failed: {result.get('err-msg')}")

    return result['data']  # returns order_id if successful


def get_order_status(order_id: str):
    """Check the current status of an order."""
    endpoint = f'/v1/order/orders/{order_id}'
    params = sign_request('GET', endpoint)

    resp = requests.get(f'{BASE_URL}{endpoint}', params=params)
    resp.raise_for_status()
    result = resp.json()

    if result['status'] != 'ok':
        raise Exception(f"Status check failed: {result.get('err-msg')}")

    order = result['data']
    print(f"Order {order_id}: {order['state']} | Filled: {order['field-amount']} / {order['amount']}")
    return order

HTX uses hyphenated parameter names like 'account-id' and 'field-amount' — not underscores. This differs from Binance and Bybit conventions. If you're porting code between exchanges, watch for this mismatch.

Real-Time Data with WebSocket Streams

For live trading bots, polling the REST API for price updates is too slow and will quickly hit rate limits. HTX's WebSocket API pushes market data in real time — trades, order book updates, candle ticks — and is essential for any strategy that needs sub-second reaction times. The connection requires handling a heartbeat (ping/pong) that HTX sends every 5 seconds, plus gzip decompression since all messages are compressed.

import asyncio
import json
import gzip
import websockets

HTX_WS_URL = 'wss://api.huobi.pro/ws'

async def stream_trades(symbol: str):
    """
    Stream real-time trade data for a symbol.
    symbol: e.g. 'btcusdt'
    """
    async with websockets.connect(HTX_WS_URL) as ws:
        # Subscribe to trade feed
        sub_msg = {
            'sub': f'market.{symbol}.trade.detail',
            'id': f'trade_{symbol}'
        }
        await ws.send(json.dumps(sub_msg))
        print(f"Subscribed to {symbol} trade stream")

        async for raw_msg in ws:
            # HTX sends gzip-compressed binary frames
            try:
                msg = json.loads(gzip.decompress(raw_msg).decode('utf-8'))
            except Exception:
                continue

            # Handle heartbeat ping
            if 'ping' in msg:
                pong = {'pong': msg['ping']}
                await ws.send(json.dumps(pong))
                continue

            # Handle subscription confirmation
            if msg.get('status') == 'ok':
                print(f"Subscription confirmed: {msg.get('subbed')}")
                continue

            # Process trade data
            if 'tick' in msg:
                tick = msg['tick']
                for trade in tick.get('data', []):
                    direction = trade['direction']
                    price = trade['price']
                    qty = trade['amount']
                    print(f"{symbol.upper()} | {direction.upper()} | Price: {price} | Qty: {qty}")

async def main():
    await stream_trades('btcusdt')

if __name__ == '__main__':
    asyncio.run(main())

If you're building a signal-driven bot, pairing this WebSocket feed with a platform like VoiceOfChain makes sense — you get real-time trading signals telling you when to act, and your Python bot handles the actual execution on HTX. VoiceOfChain aggregates on-chain flow data and technical setups across multiple assets, so the combination of external signal generation and exchange API execution covers both the what and the how of automated trading.

Rate Limits, Error Codes, and Best Practices

HTX enforces rate limits per API key: typically 10 requests per second for private endpoints and higher for market data. Breaching this returns a 429 or a specific error code. Unlike OKX or Coinbase which provide rate limit headers in every response, HTX's limits are documented but not always reflected in headers, so you need to implement your own throttling. Here's a minimal retry wrapper that handles transient errors gracefully:

import time
from functools import wraps

def with_retry(max_retries=3, base_delay=1.0):
    """Decorator for retrying API calls with exponential backoff."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except requests.HTTPError as e:
                    status = e.response.status_code if e.response else 0
                    # Don't retry client errors (400, 401, 403)
                    if status in (400, 401, 403):
                        raise
                    last_error = e
                    delay = base_delay * (2 ** attempt)
                    print(f"Attempt {attempt + 1} failed ({status}), retrying in {delay}s...")
                    time.sleep(delay)
                except Exception as e:
                    last_error = e
                    delay = base_delay * (2 ** attempt)
                    time.sleep(delay)
            raise last_error
        return wrapper
    return decorator


# Common HTX error codes to handle:
HTX_ERRORS = {
    'account-frozen-balance-insufficient-error': 'Insufficient balance',
    'order-limitorder-price-error': 'Price outside allowed range',
    'order-limitorder-amount-min-error': 'Order size below minimum',
    'api-signature-not-valid': 'Invalid signature — check key/secret',
    'api-key-expired': 'API key has expired — regenerate in HTX account settings',
    'gate-way-banned': 'IP banned — too many requests'
}

@with_retry(max_retries=3, base_delay=0.5)
def safe_get_ticker(symbol: str):
    resp = requests.get(
        f'{BASE_URL}/market/detail/merged',
        params={'symbol': symbol.lower()}
    )
    resp.raise_for_status()
    data = resp.json()
    if data['status'] != 'ok':
        err_code = data.get('err-code', '')
        friendly = HTX_ERRORS.get(err_code, data.get('err-msg', 'Unknown error'))
        raise Exception(f"HTX API error [{err_code}]: {friendly}")
    return data['tick']

Frequently Asked Questions

Conclusion

The HTX API is production-capable for both simple price monitoring scripts and full algorithmic trading systems. The REST endpoints cover everything from account management to order execution, while the WebSocket API handles real-time feeds without the overhead of constant polling. The main friction points are the HMAC signature implementation and the non-standard parameter naming conventions — once you have a working signing function, the rest of the API follows logically. For traders who want to combine exchange connectivity with smarter entry signals, pairing your HTX Python integration with a real-time signal source like VoiceOfChain gives you both the intelligence layer and the execution layer. Whether you're trading BTC/USDT on HTX alongside Binance futures, or scanning for setups across HTX and OKX simultaneously, Python gives you the flexibility to build exactly the workflow you need.