To authenticate a WebSocket connection to wss://ws.bitvavo.com/v2/, you must send an authenticate action containing a signature created via HMAC-SHA256.
Signature Generation Process:
- Concatenate the
timestamp (Unix milliseconds), the HTTP method (e.g., 'GET'), and the path (e.g., '/v2/websocket') without delimiters. - Encode this concatenated string using HMAC-SHA256 with your API secret as the key.
- The resulting signature must be hex-encoded.
Authentication Message Parameters:
action: Must be authenticate.key: Your Bitvavo API key.signature: The hex-encoded HMAC-SHA256 signature.timestamp: The Unix timestamp in milliseconds used to create the signature.window (optional): Time window in milliseconds for request validity. Default is 10000, maximum is 60000.
import hashlib
import hmac
import json
import websocket
import time
class BitvavoWebSocketClient:
def __init__(self, api_key: str, api_secret: str, access_window: int = 10000):
self.api_key = api_key
self.api_secret = api_secret
self.access_window = access_window
self.endpoint = 'wss://ws.bitvavo.com/v2/'
self.request_id = 0
self.ws = None
def authenticate(self):
timestamp = int(time.time() * 1000)
body = {
'key': self.api_key,
'signature': self.create_signature(timestamp),
'timestamp': timestamp,
'window': self.access_window,
}
self.call_action(action='authenticate', body=body)
def create_signature(self, timestamp: int):
# Concatenate timestamp + method + path
string = str(timestamp) + 'GET' + '/v2/websocket'
signature = hmac.new(self.api_secret.encode('utf-8'), string.encode('utf-8'), hashlib.sha256).hexdigest()
return signature
def call_action(self, action: str, body: dict):
request_id = self.request_id
self.request_id += 1
body['action'] = action
body['requestId'] = request_id
self.send_message(body)
def send_message(self, message: dict):
if self.ws is None:
self.ws = websocket.create_connection(self.endpoint)
self.ws.send(json.dumps(message))
