Created
April 14, 2020 18:12
-
-
Save m-root/cf73d0103bb11eb7632f9b0c97dd6143 to your computer and use it in GitHub Desktop.
I really had a difficult time trying to come up with the authentication method for Primebit. I wouldn't want anyone else who is a developer to go through the same challenge. All the best. Happy Trading :-)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
''' | |
I really had a difficult time trying to come up with the authentication method for Primebit. I wouldn't want anyone | |
else who is a developer to go through the same challenge. | |
All the best. Happy Trading :-) | |
''' | |
from typing import Optional, Dict, Any, List | |
from requests import Request, Session, Response, request, exceptions | |
import hmac | |
import time | |
BASE_URL = 'https://app.primebit.com/api/v1' | |
class PrimebitPrivate(): | |
def __init__(self, accountID=None, key=None, secret=None): | |
self._session = Session() | |
self._account_ID = accountID | |
self._api_key = key | |
self._api_secret = secret | |
def signRequest(self, request: Request): | |
timestamp = int(time.time()) | |
prepared = request.prepare() | |
'''signature components order matters - should be {method}{body}{path}{query}{timestamp}''' | |
signature_payload = f'{prepared.method}{prepared.path_url}{timestamp}'.encode() | |
if prepared.body: | |
signature_payload += prepared.body | |
signature = hmac.new(self._api_secret.encode(), signature_payload, 'sha256').hexdigest() | |
request.headers['api-key'] = self._api_key | |
request.headers['authorization'] = signature | |
request.headers['timestamp'] = str(timestamp) | |
print(request.headers) | |
def request(self, method: str, path: str, **kwargs): | |
request = Request(method, BASE_URL + path, **kwargs) | |
print('---------------------------------') | |
print(method, BASE_URL + path) | |
self.signRequest(request) | |
response = self._session.send(request.prepare()) | |
print(response) | |
return self.processResponse(response) | |
def processResponse(self, response: Response) -> Any: | |
try: | |
data = response.json() | |
except ValueError: | |
response.raise_for_status() | |
raise | |
return data | |
def get(self, path: str, params: Optional[Dict[str, Any]] = None): | |
return self.request('GET', path, params=params) | |
def post(self, path: str, params: Optional[Dict[str, Any]] = None): | |
return self.request('POST', path, json=params) | |
def delete(self, path: str, params: Optional[Dict[str, Any]] = None): | |
return self.request('DELETE', path, json=params) | |
def getAllAccounts(self): | |
''' | |
GET | |
/trading/account | |
Get all user accounts | |
:param params: | |
:return: | |
''' | |
return self.get( | |
'/trading/account' | |
) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment