かぴぱら（@kapipara180）です。

界隈でにわかに注目を浴びているQuoineのAPIをpythonから使えるようにラッピングするライブラリを作ったので公開しますね。

開発はpybitflyerをベースにさせていただきました。この場を借りて製作者様にお礼申し上げます！

取り急ぎテキストで置いておきます。gitに挙げろよという批判はあると思いますが、いったんこれで勘弁してください。笑

2020/1/17：　スナフキンさんがさらに改良してくれたコードがGitにあるのでそちらをご利用ください。

https://github.com/Snufkin0866/pyliquid

あまり使わないライブラリなので「jwt」をインストールする人が多いと思いますが、pip install jwtはNGです。pip intall PyJWTが正解です。詳しくは以下のページをご覧ください。

pyquoine.py

# -*- 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)

exception.py

# -*- coding: utf-8 -*-

class AuthException(Exception):

    def __init__(self):
        msg = "Please specify your valid API Key and API Secret."
        super(AuthException, self).__init__(msg)