QuoineのAPIをpythonでラッピングしといたよ!
かぴぱら(@kapipara180)です。
界隈でにわかに注目を浴びているQuoineのAPIをpythonから使えるようにラッピングするライブラリを作ったので公開しますね。
開発はpybitflyerをベースにさせていただきました。この場を借りて製作者様にお礼申し上げます!
取り急ぎテキストで置いておきます。gitに挙げろよという批判はあると思いますが、いったんこれで勘弁してください。笑
2020/1/17: スナフキンさんがさらに改良してくれたコードがGitにあるのでそちらをご利用ください。
https://github.com/Snufkin0866/pyliquid
あまり使わないライブラリなので「jwt」をインストールする人が多いと思いますが、pip install jwtはNGです。pip intall PyJWTが正解です。詳しくは以下のページをご覧ください。
# -*- coding: utf-8 -*-
# created by @kapipara180
import json
import requests
import time
import hmac
import hashlib
import urllib
import jwt
from exception import AuthException
class API(object):
def __init__(self, api_key=None, api_secret=None):
self.api_url = "https://api.quoine.com"
self.api_key = api_key
self.api_secret = api_secret
def request(self, endpoint, method="GET", params=None):
url = self.api_url + endpoint
body = ""
auth_header = None
if method == "POST":
body = json.dumps(params)
elif method == "PUT":
body = json.dumps(params)
else:
if params:
body = "?" + urllib.parse.urlencode(params)
if self.api_key and self.api_secret:
nonce = str(round(time.time()))
api_secret = str.encode(self.api_secret)
token_id = self.api_key
auth_payload = {
"path": endpoint,
"nonce": nonce,
"token_id": token_id
}
#signature = hmac.new(api_secret, auth_payload, hashlib.sha256).hexdigest()
signature = jwt.encode(auth_payload, api_secret, 'HS256')
auth_header = {
"X-Quoine-API-Version": "2",
"X-Quoine-Auth": signature,
"Content-Type": "application/json"
}
try:
with requests.Session() as s:
if auth_header:
s.headers.update(auth_header)
if method == "GET":
response = s.get(url, params=params)
elif method == "POST":
response = s.post(url, data=json.dumps(params))
else: # put
response = s.put(url, data=json.dumps(params))
except requests.RequestException as e:
print(e)
raise e
content = ""
if len(response.content) > 0:
content = json.loads(response.content.decode("utf-8"))
return content
"""HTTP Public API"""
def getproducts(self, **params): # test済
"""
Get the list of all available products.
"""
endpoint = "/products"
return self.request(endpoint, params=params)
def getaproduct(self, id=1): # test済
"""
PARAMETERS:
Parameters Optional? Description
id Product ID
"""
endpoint = "/products/" + str(id)
return self.request(endpoint)
def getorderbook(self, id=1): # test済
"""
PARAMETERS:
Parameters Optional? Description
id Product ID
full yes 1 to get all price levels (default is 20 each side)
FORMAT
Each price level follows: [price, amount]
"""
endpoint = "/products/" + str(id) + "/price_levels"
return self.request(endpoint)
def getexecutions(self, **params): # test済
"""
Get a list of recent executions from a product (Executions are sorted in DESCENDING order - Latest first)
Parameters Optional? Description
product_id Product ID
limit yes How many executions should be returned. Must be <= 1000. Default is 20
page yes From what page the executions should be returned, e.g if limit=20 and page=2, the response would start from the 21st execution. Default is 1
or
Get a list of executions after a particular time (Executions are sorted in ASCENDING order)
Parameters Optional? Description
currency_pair_code e.g. BTCJPY
timestamp Only show executions at or after this timestamp (Unix timestamps in seconds)
limit yes How many executions should be returned. Must be <= 1000. Default is 20
"""
endpoint = "/executions"
return self.request(endpoint, params=params)
def getinterestrates(self, **params): # 済
"""
Get Interest Rate Ladder for a currency
FORMAT
Each level follows: [rate, amount]
"""
endpoint = "/ir_ladders/USD"
return self.request(endpoint, params=params)
"""HTTP Authenticated API"""
def createorder(self, **params):
"""
PARAMETERS
Parameters Optional? Description
order_type limit, market or market_with_range
product_id Product ID
side buy or sell
quantity quantity to buy or sell
price price per unit of cryptocurrency
price_range true For order_type of market_with_range only, slippage of the order.
MARGIN ORDER PARAMETERS
Parameters Optional? Description
leverage_level Valid levels: 2,4,5,10,25
funding_currency Currency used to fund the trade with. Default is quoted currency (e.g a trade in BTCUSD product will use USD as the funding currency as default)
order_direction true one_direction, two_direction, netout
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/orders/"
return self.request(endpoint, "POST", params=params)
def getorder(self, id=1): # test済
"""
Parameters Optional? Description
id Order ID
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/orders/" + str(id)
return self.request(endpoint)
def getorders(self, **params): # test 済
"""
PARAMETERS:
Parameters Optional? Description
funding_currency yes filter orders based on funding currency
product_id yes filter orders based on product
status yes filter orders based on status
with_details yes return full order details (attributes between *) including executions if set to 1
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/orders"
return self.request(endpoint, params=params)
def cancelorder(self, id=1, **params):
"""
PARAMETERS:
Parameters Optional? Description
id Order ID
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/orders/" + str(id) + "/cancel"
return self.request(endpoint, method="PUT", params=params)
def editliveorder(self, id=1, **params):
"""
PARAMETERS:
Parameters Optional? Description
id Order ID
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/orders/" + str(id)
return self.request(endpoint, method="PUT", params=params)
def getorderstrade(self, id=1, **params):
"""
PARAMETERS:
Parameters Optional? Description
id Order ID
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/orders/" + str(id) + "/trades"
return self.request(endpoint)
def getmyExecution(self, **params):
"""
PARAMETERS:
Parameters Optional? Description
product_id Product ID
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = " /executions/me"
return self.request(endpoint, params=params)
def getcryptoaccount(self, **params):
"""
Get Crypto Accounts
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/crypto_accounts"
return self.request(endpoint, params=params)
def getfiataccount(self, **params):
"""
Get Fiat Accounts
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/fiat_accounts"
return self.request(endpoint, params=params)
def getallacountbalance(self, **params):
"""
Get all Account Balances
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/accounts/balance"
return self.request(endpoint, params=params)
#Assets Lendingは後で
def gettradingaccounts(self, **params):
"""
Get Trading Accounts
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/trading_accounts"
return self.request(endpoint, params=params)
def gettradingaccount(self, id=1, **params):
"""
PARAMETERS:
Parameters Optional? Description
id Trading Account ID
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/trading_accounts/" + str(id)
return self.request(endpoint, params=params)
def updateleverage(self, id=1, **params):
"""
PARAMETERS:
Parameters Optional? Description
id Trading account ID
leverage_level New leverage leve
{
"trading_account": {
"leverage_level": 25
}
}
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/trading_accounts/" + str(id)
return self.request(endpoint, method="PUT", params=params)
def gettrades(self, **params):
"""
PARAMETERS:
Parameters Optional? Description
funding_currency yes get trades of a particular funding currency
status yes open or closed
/trades?funding_currency=:funding_currency&status=:status
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/trades/"
return self.request(endpoint, params=params)
def closeatrade(self, id=1, **params):
"""
PARAMETERS:
Parameters Optional? Description
id Trade ID
closed_quantity yes The quantity you want to close
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/trades/" + str(id) + "/close"
return self.request(endpoint, method="PUT", params=params)
def closealltrade(self, **params):
"""
PARAMETERS:
Parameters Optional? Description
side yes Close all trades of this side. Close trades of both side if left blank
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/trades/close_all"
return self.request(endpoint, method="PUT", params=params)
def updateatrade(self, id=1, **params):
"""
Parameters Optional? Description
id Trade ID
stop_loss Stop Loss price
take_profit Take Profit price
{
"trade": {
"stop_loss": "300",
"take_profit": "600"
}
}
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/trades/" + str(id)
return self.request(endpoint, method="PUT", params=params)
def getatradesloan(self, id=1, **params):
"""
Parameters Optional? Description
id Trade ID
"""
if not all([self.api_key, self.api_secret]):
raise AuthException()
endpoint = "/trades/" + str(id) + "/loans"
return self.request(endpoint, params=params)
# -*- coding: utf-8 -*-
class AuthException(Exception):
def __init__(self):
msg = "Please specify your valid API Key and API Secret."
super(AuthException, self).__init__(msg)
この記事が気に入ったらサポートをしてみませんか?