Exploring trading and API integration use cases for WOO X API
This guide covers the basic approach to trading with the WOO X API through six common and practical integration and trading strategies. Each example is ready to use but requires traders to replace the placeholders with their own API key and API secret for authentication.
Strategy 1: Place a Limit Order (Buy or Sell)
The following example demonstrates how to submit a limit order to buy or sell a specific quantity of an asset at a defined price. It includes generating the required API signature for authentication and sending a POST request with order details.
python
import requests
import time
import hmac
import hashlib
import json
API_KEY = '<your_api_key>'
API_SECRET = '<your_api_secret>'
BASE_URL = 'https://api.woox.io'
def generate_signature(secret, method, endpoint, timestamp, body=''):
"""Generate HMAC SHA256 signature for WOO X API."""
if isinstance(body, dict):
body = json.dumps(body, separators=(',', ':'), sort_keys=True)
prehash = f"{timestamp}{method}{endpoint}{body}"
return hmac.new(secret.encode(), prehash.encode(), hashlib.sha256).hexdigest()
def place_limit_order(symbol, price, quantity, side):
endpoint = '/v3/trade/order'
url = BASE_URL + endpoint
timestamp = str(int(time.time() * 1000))
jsonbody = {
"symbol": symbol,
"price": price,
"quantity": quantity,
"type": "LIMIT",
"side": side
}
body = json.dumps(jsonbody)
signature = generate_signature(API_SECRET, 'POST', endpoint, timestamp, body)
headers = {
'x-api-key': API_KEY,
'x-api-timestamp': timestamp,
'x-api-signature': signature,
'Content-Type': 'application/json'
}
response = requests.post(url, headers=headers, data=body)
return response.json()
# Example usage:
result = place_limit_order('SPOT_BTC_USDT', '30000', '0.001', 'BUY')
print(result)
Strategy 2: Place a market order
This snippet shows how to execute a market order, which buys or sells immediately at the best available price. It also handles authentication and submits the order in a similar fashion to the limit order.
python
def place_market_order(symbol, quantity, side):
endpoint = '/v3/trade/order'
url = BASE_URL + endpoint
timestamp = str(int(time.time() * 1000))
jsonbody = {
"symbol": symbol,
"quantity": quantity,
"type": "MARKET",
"side": side
}
body = json.dumps(jsonbody)
signature = generate_signature(API_SECRET, 'POST', endpoint, timestamp, body)
headers = {
'x-api-key': API_KEY,
'x-api-timestamp': timestamp,
'x-api-signature': signature,
'Content-Type': 'application/json'
}
response = requests.post(url, headers=headers, data=body)
return response.json()
# Example usage:
result = place_market_order('SPOT_ETH_USDT', '0.05', 'SELL')
print(result)
Strategy 3: Cancel an order by order ID
This example illustrates how to cancel a previously placed order using its unique order ID. It sends a DELETE request with the proper signed headers to ensure the cancellation request is authenticated and processed.
python
def cancel_order(symbol, orderId):
endpoint = '/v3/trade/order'
url = BASE_URL + endpoint
timestamp = str(int(time.time() * 1000))
body = f"symbol={symbol}&order_id={orderId}"
signature = generate_signature(API_SECRET, 'DELETE', endpoint, timestamp, body)
headers = {
'x-api-key': API_KEY,
'x-api-timestamp': timestamp,
'x-api-signature': signature,
'Content-Type': 'application/x-www-form-urlencoded'
}
response = requests.delete(url, headers=headers, data=body)
return response.json()
# Example usage:
result = cancel_order('SPOT_BTC_USDT', '123456789')
print(result)
Strategy 4: Initializing local order book snapshot for accurate real-time market tracking
This code retrieves a snapshot of the current order book for the trading pair "SPOT_WOO_USDT" from the WOO X REST API, capturing the top 50 bid and ask levels.
It initializes a local order book with this data, creating a reliable and accurate baseline. This local snapshot allows the system to efficiently process real-time incremental updates, ensuring the order book remains consistent and up-to-date which is crucial for making timely and informed trading decisions.
python
import websocket
import json
import requests
# Define trading pair and depth
SYMBOL = "SPOT_WOO_USDT"
DEPTH = 50
# WebSocket API address
WS_PUBLIC_URL = "wss://wss.woox.io/v3/public"
# REST API address
REST_PUBLIC_URL = f"https://api.woox.io/v3/public/orderbook?symbol={SYMBOL}&maxLevel={DEPTH}"
# Store local order book snapshot
local_order_book = {'bids': {}, 'asks': {}}
snapshot_timestamp = None
def get_order_book_snapshot():
"""Get order book snapshot from REST API"""
print("Fetching order book snapshot...")
try:
response = requests.get(REST_PUBLIC_URL)
response.raise_for_status()
snapshot = response.json()
if not snapshot.get('success'):
print("Snapshot request failed:", snapshot.get('message'))
return None
print("Snapshot fetched successfully, timestamp:", snapshot.get('timestamp'))
global snapshot_timestamp
snapshot_timestamp = snapshot.get('timestamp')
bids = snapshot['data']['bids']
asks = snapshot['data']['asks']
for bid in bids:
price = float(bid['price'])
size = float(bid['quantity'])
local_order_book['bids'][price] = size
for ask in asks:
price = float(ask['price'])
size = float(ask['quantity'])
local_order_book['asks'][price] = size
print(
f"Local order book initialized successfully with {len(local_order_book['bids'])} bids and {len(local_order_book['asks'])} asks.")
except requests.exceptions.RequestException as e:
print(f"Failed to get snapshot: {e}")
return None
return snapshot
def print_order_book(book_type):
"""
Format and print the order book, sorted by price
"""
print(f"\n--- Local Order Book Changes ({book_type.upper()}) ---")
if book_type == 'bids':
# Bids are sorted by price in descending order
sorted_items = sorted(local_order_book[book_type].items(), reverse=True)
else:
# Asks are sorted by price in ascending order
sorted_items = sorted(local_order_book[book_type].items())
for price, size in sorted_items:
print(f"Price: {price}, Size: {size}")
print("-----------------------------------")
def on_message(ws, message):
"""Handle messages received from WebSocket"""
data = json.loads(message)
if data.get('topic', '').startswith('orderbookupdate'):
update_data = data['data']
update_timestamp = update_data.get('ts')
prev_update_timestamp = update_data.get('prevTs')
if prev_update_timestamp == snapshot_timestamp:
print(f"Starting to process real-time updates...")
for update_type in ['bids', 'asks']:
updates = update_data[update_type]
for price, size in updates:
price = float(price)
size = float(size)
if size == 0:
if price in local_order_book[update_type]:
del local_order_book[update_type][price]
else:
local_order_book[update_type][price] = size
print(f"Update {update_timestamp} processed, local order book updated.")
# Add logic to print the local order book
print_order_book('bids')
print_order_book('asks')
def on_error(ws, error):
print("WebSocket error:", error)
def on_close(ws, close_status_code, close_msg):
print("WebSocket closed")
def on_open(ws):
"""WebSocket connection successful, sending subscription message"""
print("WebSocket connection successful, sending subscription message...")
subscribe_msg = {
"id": "1",
"cmd": "SUBSCRIBE",
"params": [f"orderbookupdate@{SYMBOL}@{DEPTH}"]
}
ws.send(json.dumps(subscribe_msg))
def start_main_process():
"""Main program entry point"""
snapshot = get_order_book_snapshot()
if not snapshot:
return
ws = websocket.WebSocketApp(f"{WS_PUBLIC_URL}",
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close)
ws.run_forever()
if __name__ == "__main__":
start_main_process()
Strategy 5: Error handling example
This code wraps an API call with a retry mechanism that detects rate limit errors (HTTP 429). It employs exponential backoff — waiting progressively longer between retries — to comply with the server’s rate limits and avoid request failures.
python
import time
def safe_api_call(func, *args, **kwargs):
max_retries = 5
retry_count = 0
backoff = 1 # seconds
while retry_count < max_retries:
response = func(*args, **kwargs)
if isinstance(response, dict) and response.get('code') == -1003:
print(f"Rate limit reached. Retrying in {backoff} seconds...")
time.sleep(backoff)
backoff *= 2 # exponential backoff
retry_count += 1
else:
return response
raise Exception("Max retries reached due to rate limiting.")
# Example usage wrapping the limit order function:
result = safe_api_call(place_limit_order, 'SPOT_BTC_USDT', '30000', '0.001', 'BUY')
print(result)
Strategy 6: Reliable real-time trading updates with auto-renewed WebSocket connection
This code connects to WOO X's private WebSocket using a signed listen key to receive real-time execution reports.
It automatically refreshes the listen key before it expires, reconnecting and resubscribing to ensure uninterrupted, continuous access to private trading updates.
This approach enables reliable, seamless monitoring of order executions essential for automated trading without losing data due to key expiration.
python
import websocket
import json
from datetime import datetime, timezone, timedelta
import hmac, hashlib
import time
import requests
# API credentials and base URLs
BASE_URL = 'https://api.woox.io'
WS_PRIVATE_URL = "wss://wss.woox.io/v3/private"
API_KEY = '<your_api_key>'
API_SECRET = '<your_api_secret>'
# Refresh interval for listenKey (23.5 hours)
REFRESH_INTERVAL_SEC = 23 * 60 * 60 + 30 * 60
# --- Signature Generation Function ---
def generate_signature(secret, method, endpoint, timestamp, body=''):
"""Generate HMAC SHA256 signature for WOO X API."""
if isinstance(body, dict):
# Serialize JSON to a canonical string for consistent hashing
body = json.dumps(body, separators=(',', ':'), sort_keys=True)
prehash = f"{timestamp}{method}{endpoint}{body}"
print(f"Prehash string: {prehash}")
return hmac.new(secret.encode(), prehash.encode(), hashlib.sha256).hexdigest()
# --- Listen Key Retrieval Function ---
def get_listen_key():
"""Fetches a new listenKey and its expiration time from the WOO X API."""
endpoint = '/v3/account/listenKey'
# Generate timestamp and body
timestamp_ms = str(int(time.time() * 1000))
json_body = {
'type': 'WEBSOCKET'
}
body = json.dumps(json_body)
# Generate signature
signature = generate_signature(API_SECRET, 'POST', endpoint, timestamp_ms, body)
# Set up headers for the request
headers = {
'x-api-timestamp': timestamp_ms,
'x-api-key': API_KEY,
'x-api-signature': signature,
'Content-Type': 'application/json'
}
# Send the request
response = requests.post(BASE_URL + endpoint, headers=headers, data=body)
response.raise_for_status() # Raise an exception for bad status codes
listen_key_response = response.json()
listen_key = listen_key_response['data']['authKey']
expire_time = listen_key_response['data']['expiredTime']
# Print expiration time in UTC+8 and UTC
expire_time_utc = datetime.fromtimestamp(expire_time / 1000, tz=timezone.utc)
print(
f"✅ listen_key fetched. Expiration time (UTC): {expire_time_utc}")
return listen_key, expire_time
# --- Main WebSocket Management ---
def run_websocket_client():
"""Main function to manage the WebSocket connection and listenKey refresh."""
listen_key, expire_time = get_listen_key()
ws_url = f"{WS_PRIVATE_URL}?key={listen_key}"
# Connect to the WebSocket
print(f"Connecting to WebSocket at: {ws_url}")
ws = websocket.create_connection(ws_url)
# Subscribe to private topic
subscribe_msg = {
"id": "2",
"cmd": "SUBSCRIBE",
"params": ["executionreport"]
}
ws.send(json.dumps(subscribe_msg))
print("Subscribed to executionreport.")
last_refresh_time = time.time()
try:
while True:
# Check if listenKey is close to expiring
current_time_ms = int(time.time() * 1000)
if current_time_ms >= expire_time - 60000:
print("⏳ listen_key is about to expire. Refreshing...")
ws.close() # Close the old connection
listen_key, expire_time = get_listen_key()
# Open a new connection with the new key
ws_url = f"{WS_PRIVATE_URL}?key={listen_key}"
print(f"Connecting to new WebSocket at: {ws_url}")
ws = websocket.create_connection(ws_url)
# Resubscribe
ws.send(json.dumps(subscribe_msg))
print("Resubscribed to executionreport.")
# Receive and print messages
response = ws.recv()
print(f"Received message: {response}")
except websocket.WebSocketConnectionClosedException:
print("WebSocket connection closed unexpectedly. Exiting...")
except Exception as e:
print(f"An error occurred: {e}")
finally:
ws.close()
if __name__ == "__main__":
run_websocket_client()
Disclaimer
The information provided in this article is for general informational purposes only and does not constitute financial, investment, legal advice or professional advice of any kind. While we have made every effort to ensure that the information contained herein is accurate and up-to-date, we make no guarantees as to its completeness or accuracy. The content is based on information available at the time of writing and may be subject to change.
Cryptocurrencies involve significant risk and may not be suitable for all investors. The value of digital currencies can be extremely volatile, and you should carefully consider your investment objectives, level of experience, and risk appetite before participating in any staking or investment activities.
We strongly recommend that you seek independent advice from a qualified professional before making any investment or financial decisions related to cryptocurrencies or staking. We shall in NO case be liable for any loss or damage arising directly or indirectly from the use of or reliance on the information contained in this article.
