BTC Markets Connection Client
This is part of the BTC Markets API Evaluation, and I’ll create a generic connection client for BTC Markets API. This client will handle the authentication and https request and reply.
API Key Generation
I generated an API Key, and Private Key, in the following “API Key” page.
Authentication
The Auth class will handle the authentication http headers,
- BM-AUTH-APIKEY
- BM-AUTH-SIGNATURE
- BM-AUTH-TIMESTAMP
The api key and private key are loaded from the local file. And the __sign_msg() is private method for the Auth class. The result header will be generated in the build_headers() method.
class Auth(SpecBase): def __init__(self, **kwargs): super().__init__(**{ 'apiKeyFile': os.path.expanduser('~/.bm_api_key'), 'apiPrivateKeyFile': os.path.expanduser('~/.bm_api_private_key'), **kwargs, })
def onSpecUpdated(self): super().onSpecUpdated() self.updateSpec({ 'api_key': Utils.get_key_from_file(self.spec.get('apiKeyFile')), 'api_private_key': base64.b64decode(Utils.get_key_from_file(self.spec.get('apiPrivateKeyFile'))), }, notifyUpdate=False)
def __sign_msg(self, msg): try: return base64.b64encode( hmac.new( self.spec.get('api_private_key'), msg.encode('utf-8'), digestmod=hashlib.sha512 ).digest() ).decode('utf-8') except Exception as e: print(f' S (Warning) exception on sign: {repr(e)}')
def build_headers(self, httpMethod, apiPath, payload=None): ts = str(int(time.time() * 1000)) msg = ''.join([ httpMethod, apiPath, ts, *([payload] if payload else []), ]) return { 'BM-AUTH-APIKEY': self.spec.get('api_key'), 'BM-AUTH-SIGNATURE': self.__sign_msg(msg), 'BM-AUTH-TIMESTAMP': ts, } |
I tested with simple data like below, and it seems to be working.
def test_build_headers(self): payload = json.dumps({'type': 'TransactionReport', 'format': 'json'}) headers = self.obj.build_headers('GET', '/v3/reports', payload) pprint.PrettyPrinter().pprint(headers) # {'BM-AUTH-APIKEY': '9014817e-adcc-b439-a754-36773578fdac', # 'BM-AUTH-SIGNATURE': # 'I5ruzMbhJdoiO/wtxmExFcY17wkCDHO6iqsBA9R2Pe86P58hMW9UJo8S # hCf07Tc7k+riXvionwA0FJsFv3Xxtg==', # 'BM-AUTH-TIMESTAMP': '1757844183305'} |
Client
The Client implementation has the make_api_call(), which is a generic and accepting parameter for each API call.
The build_headers() will add more HTTP headers, on top of the Auth headers, like below.
class Client(Auth): def __init__(self, **kwargs): super().__init__(**{ **kwargs, })
def build_headers(self, httpMethod, apiPath, strData): return { **super().build_headers(httpMethod, apiPath, strData), **{ "Accept": "application/json", "Accept-Charset": "UTF-8", "Content-Type": "application/json", }, } |
The make_api_call() implementation has the details for API calls. I referenced a lot of the sample implementations suggested by the company here.
- The caller should provide the HTTP Method, API Path, any Query string, and the Json Data if required
- The return value of the function is also in Json format
class Client(Auth): def make_api_call(self, httpMethod, apiPath, strQuery=None, jsonData=None): httpMethod = httpMethod.upper() strData = json.dumps(jsonData) if jsonData else '' headers = self.build_headers(httpMethod, apiPath, strData) url = ''.join([ STR.BaseUrl.value, apiPath, *([ '?', strQuery, ] if strQuery else []), ]) try: request = Request(**{ 'url': url, **({'data': strData} if strData else {}), 'headers': headers, 'method': httpMethod, }) response = urlopen(request, **({ 'data': bytes(strData, encoding='utf-8'), } if strData and httpMethod in ['POST', 'PUT'] else {}) ) jsonRes = json.loads(str(response.read(), 'utf-8')) return jsonRes except Exception as e: jsonErr = json.loads(e.read()) if hasattr(e, 'code'): jsonErr['statusCode'] = e.code return jsonErr |
List Live Markets
The first API I tried is the list of live markets. The API Document for this is like below.
- It uses http GET method
- API path is “/v3/markets”
- And no Query String or Json Data required
It is interpreted in the following way to Client.make_api_call().
class ClientTest(unittest.TestCase): def test_list_active_markets(self): jsonRes = self.obj.make_api_call('GET', '/v3/markets') pprint.PrettyPrinter().pprint(jsonRes) |
And the actual list I get from the API is like below, too many, so I cut the result.
[{'amountDecimals': '8', 'baseAssetName': 'AAVE', 'marketId': 'AAVE-AUD', 'maxOrderAmount': '1000000', 'minOrderAmount': '0.005', 'priceDecimals': '2', 'quoteAssetName': 'AUD', 'status': 'Online'}, .. {'amountDecimals': '8', 'baseAssetName': 'BTC', 'marketId': 'BTC-AUD', 'maxOrderAmount': '1000000', 'minOrderAmount': '0.00001', 'priceDecimals': '2', 'quoteAssetName': 'AUD', 'status': 'Online'}, .. |
Trade Info
I tried this to see the API failure. I don’t have any trade yet, but I asked for the info of trade 123. The API should return a error, in Json format.
class ClientTest(unittest.TestCase): def test_get_trade_info(self): jsonRes = self.obj.make_api_call('GET', '/v3/trades/123') pprint.PrettyPrinter().pprint(jsonRes) |
And the result is as expected.
{'code': 'TradeNotFound', 'message': 'trade not found', 'statusCode': 404} |