问Bitex -用于Python的加密货币交换API框架-第2轮
EN

Code Review用户

提问于 2016-12-05 02:21:11

回答 1查看 1.2K关注 0票数 11

这是对我以前的评论请求的跟进。

BitEx是我已经研究了9个多月的Python模块，作为一个附带项目。它是6个月前在GitHub上发布的，随着我接近我的1.0版本，我想借此机会在这里展示我的代码，以便纠正它。

它解决并提供给

的东西

它的目的是消除深入到REST密码交换的血淋淋的细节的需要，并为所有受支持的API提供一个统一和直观的接口。它负责身份验证过程，并为交换中所有常用的方法提供了标准化的方法集(具有相同的方法签名)(轮询订单簿和代码、下订单和取消订单等)，以及所有其他特定方法(或我有时间实现的方法)。

本质上，它是两个子包：bitex.api是负责通过requests模块设置http请求以及处理身份验证细节的后端。它可以被看作是requests的包装器，在技术上完全可以单独用于向交换发送和接收数据。

另一个是bitex.interfaces，它为所有已实现的交换提供了上述同构、标准化的方法。除了提供相同的方法签名外，它还旨在标准化方法的返回值。由于不同的exchange之间的差异很大，这些方法通过在bitex.formatters和return_json装饰器中找到的格式化程序来处理数据格式化。它依赖于bitex.api。

为什么我要把这个提交给审查

自从我开始这个项目以来，我已经重写了几次基本代码。我花了很长时间才弄清楚如何布局结构(这主要是由于我在过去一年作为软件开发学徒时的学习曲线)。

然而，在过去的两个月里，我对目前的结构变得相当喜欢和自豪，并认为它是相当体面的--因此，我可以对其进行公开审计。

我在如何从我的评论中获得最好的价值上读过元问题，最初决定对我的代码进行三轮评论：

圆：代码样式 (已完成)，PEP8，可读性，丙酮-度
圆:重构选项和当前布局的评估，特别是API类的sign()方法、return_json()装饰器和格式化程序功能的使用。
圆:缺陷，代码和逻辑的改进，bug等。

审查第2轮:重构和布局

我特别担心bitex.api子模块。sign()方法很难概括，因为输入变化很大，迫使我把所有的东西都传递给它们。我想不出一个更明智的解决办法。

在bitex.interfaces中，创建一个BaseInterface类似乎是个好主意，因为我有几种方法，它们出现在所有接口-- query_public()、query_private()和标准化方法中。然而，第二次一瞥，这个mixin类并没有使sense.This产生效果，因为在我看来，它似乎是为了拥有一个mixin类而引入一个类，因为无论如何，在大多数交换中，我都必须重写一些方法，从而产生了几乎相同的代码--只有很少的改进。

最后，使用和设计bitex.formatters来格式化通过@return_json装饰器从bitex.interfaces类返回的数据(到目前为止，这个名称可能有点不太好，因为它不再仅仅返回requests.response JSON值)。我的方法可以接受吗？它似乎是最简单的解决方案(而不是直接将每个Interface类的格式弄得乱七八糟)。

`bitex.api`

# Import Built-Ins
import logging
import requests
import time
# Import Third-Party

# Import Homebrew


log = logging.getLogger(__name__)


class RESTAPI:

    def __init__(self, uri, api_version='', key='', secret=''):
        """
        Base Class for REST API connections.
        """
        self.key = key
        self.secret = secret
        self.uri = uri
        self.apiversion = api_version
        self.req_methods = {'POST': requests.post, 'PUT': requests.put,
                            'GET': requests.get, 'DELETE': requests.delete,
                            'PATCH': requests.patch}
        log.debug("Initialized RESTAPI for URI: %s; "
                  "Will request on API version: %s" %
                  (self.uri, self.apiversion))

    def load_key(self, path):
        """
        Load key and secret from file.
        """
        with open(path, 'r') as f:
            self.key = f.readline().strip()
            self.secret = f.readline().strip()

    def nonce(self):
        return str(int(1000 * time.time()))

    def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
        """
        Dummy Signature creation method. Override this in child.
        URL is required to be returned, as some Signatures use the url for
        sig generation, and api calls made must match the address exactly.
        """
        url = self.uri

        return url, {'params': {'test_param': "authenticated_chimichanga"}}

    def query(self, method_verb, endpoint, authenticate=False,
              *args, **kwargs):
        """
        Queries exchange using given data. Defaults to unauthenticated query.
        """
        request_method = self.req_methods[method_verb]

        if self.apiversion:
            endpoint_path = '/' + self.apiversion + '/' + endpoint
        else:
            endpoint_path = '/' + endpoint

        url = self.uri + endpoint_path
        if authenticate:  # sign off kwargs and url before sending request
            url, request_kwargs = self.sign(url, endpoint, endpoint_path,
                                            method_verb, *args, **kwargs)
        else:
            request_kwargs = kwargs
        log.debug("Making request to: %s, kwargs: %s" % (url, request_kwargs))
        r = request_method(url, timeout=5, **request_kwargs)
        log.debug("Made %s request made to %s, with headers %s and body %s. "
                  "Status code %s" %
                  (r.request.method, r.request.url, r.request.headers,
                   r.request.body, r.status_code))
        return r

`bitex.api.rest`

# Import Built-ins
import logging
import json
import hashlib
import hmac
import base64
import time
import urllib
import urllib.parse
from requests.auth import AuthBase

# Import Third-Party

# Import Homebrew
from bitex.api.api import RESTAPI


log = logging.getLogger(__name__)


class BitfinexREST(RESTAPI):
    def __init__(self, key='', secret='', api_version='v1',
                 url='https://api.bitfinex.com'):
        super(BitfinexREST, self).__init__(url, api_version=api_version,
                                           key=key, secret=secret)

    def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
        try:
            req = kwargs['params']
        except KeyError:
            req = {}
        req['request'] = endpoint_path
        req['nonce'] = self.nonce()

        js = json.dumps(req)
        data = base64.standard_b64encode(js.encode('utf8'))

        h = hmac.new(self.secret.encode('utf8'), data, hashlib.sha384)
        signature = h.hexdigest()
        headers = {"X-BFX-APIKEY": self.key,
                   "X-BFX-SIGNATURE": signature,
                   "X-BFX-PAYLOAD": data}

        return url, {'headers': headers}


class BitstampREST(RESTAPI):
    def __init__(self, user_id='', key='', secret='', api_version='',
                 url='https://www.bitstamp.net/api'):
        self.id = user_id
        super(BitstampREST, self).__init__(url, api_version=api_version,
                                           key=key, secret=secret)

    def load_key(self, path):
        """
        Load key and secret from file.
        """
        with open(path, 'r') as f:
            self.id = f.readline().strip()
            self.key = f.readline().strip()
            self.secret = f.readline().strip()

    def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
        nonce = self.nonce()
        message = nonce + self.id + self.key

        signature = hmac.new(self.secret.encode(), message.encode(),
                             hashlib.sha256)
        signature = signature.hexdigest().upper()

        try:
            req = kwargs['params']
        except KeyError:
            req = {}
        req['key'] = self.key
        req['nonce'] = nonce
        req['signature'] = signature
        return url, {'data': req}


class BittrexREST(RESTAPI):
    def __init__(self, key='', secret='', api_version='v1.1',
                 url='https://bittrex.com/api'):
        super(BittrexREST, self).__init__(url, api_version=api_version, key=key,
                                          secret=secret)

    def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):

        try:
            params = kwargs['params']
        except KeyError:
            params = {}

        nonce = self.nonce()

        req_string = endpoint_path + '?apikey=' + self.key + "&nonce=" + nonce + '&'
        req_string += urllib.parse.urlencode(params)
        headers = {"apisign": hmac.new(self.secret.encode('utf-8'),
                                       (self.uri + req_string).encode('utf-8'),
                                       hashlib.sha512).hexdigest()}

        return self.uri + req_string, {'headers': headers, 'params': {}}


class CoincheckREST(RESTAPI):
    def __init__(self, key='', secret='', api_version='api',
                 url='https://coincheck.com'):
        super(CoincheckREST, self).__init__(url, api_version=api_version,
                                            key=key, secret=secret)

    def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):

        nonce = self.nonce()
        try:
            params = kwargs['params']
        except KeyError:
            params = {}

        params = json.dumps(params)
        # sig = nonce + url + req
        data = (nonce + endpoint_path + params).encode('utf-8')
        h = hmac.new(self.secret.encode('utf8'), data, hashlib.sha256)
        signature = h.hexdigest()
        headers = {"ACCESS-KEY": self.key,
                   "ACCESS-NONCE": nonce,
                   "ACCESS-SIGNATURE": signature}

        return url, {'headers': headers}


class GdaxAuth(AuthBase):
    def __init__(self, api_key, secret_key, passphrase):
        self.api_key = api_key.encode('utf-8')
        self.secret_key = secret_key.encode('utf-8')
        self.passphrase = passphrase.encode('utf-8')

    def __call__(self, request):
        timestamp = str(time.time())
        message = (timestamp + request.method + request.path_url +
                   (request.body or ''))
        hmac_key = base64.b64decode(self.secret_key)
        signature = hmac.new(hmac_key, message.encode('utf-8'), hashlib.sha256)
        signature_b64 = base64.b64encode(signature.digest())

        request.headers.update({
            'CB-ACCESS-SIGN': signature_b64,
            'CB-ACCESS-TIMESTAMP': timestamp,
            'CB-ACCESS-KEY': self.api_key,
            'CB-ACCESS-PASSPHRASE': self.passphrase,
            'Content-Type': 'application/json'
        })
        return request


class GDAXRest(RESTAPI):
    def __init__(self, passphrase='', key='', secret='', api_version='',
                 url='https://api.gdax.com'):
        self.passphrase = passphrase
        super(GDAXRest, self).__init__(url, api_version=api_version, key=key,
                                       secret=secret)

    def load_key(self, path):
        """
        Load key and secret from file.
        """
        with open(path, 'r') as f:
            self.passphrase = f.readline().strip()
            self.key = f.readline().strip()
            self.secret = f.readline().strip()

    def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
        auth = GdaxAuth(self.key, self.secret, self.passphrase)
        try:
            js = kwargs['params']
        except KeyError:
            js = {}

        return url, {'json': js, 'auth': auth}


class KrakenREST(RESTAPI):
    def __init__(self, key='', secret='', api_version='0',
                 url='https://api.kraken.com'):
        super(KrakenREST, self).__init__(url, api_version=api_version,
                                         key=key, secret=secret)

    def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
        try:
            req = kwargs['params']
        except KeyError:
            req = {}

        req['nonce'] = self.nonce()
        postdata = urllib.parse.urlencode(req)

        # Unicode-objects must be encoded before hashing
        encoded = (str(req['nonce']) + postdata).encode('utf-8')
        message = (endpoint_path.encode('utf-8') +
                   hashlib.sha256(encoded).digest())

        signature = hmac.new(base64.b64decode(self.secret),
                             message, hashlib.sha512)
        sigdigest = base64.b64encode(signature.digest())

        headers = {
            'API-Key': self.key,
            'API-Sign': sigdigest.decode('utf-8')
        }

        return url, {'data': req, 'headers': headers}


class ItbitREST(RESTAPI):
    def __init__(self, user_id = '', key='', secret='', api_version='v1',
                 url='https://api.itbit.com'):
        self.userId = user_id
        super(ItbitREST, self).__init__(url, api_version=api_version,
                                 key=key, secret=secret)

    def load_key(self, path):
        """
        Load user id, key and secret from file.
        """
        with open(path, 'r') as f:
            self.userId = f.readline().strip()
            self.clientKey = f.readline().strip()
            self.secret = f.readline().strip()

    def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
        try:
            params = kwargs['params']
        except KeyError:
            params = {}

        verb = method_verb

        if verb in ('PUT', 'POST'):
            body = params
        else:
            body = {}

        timestamp = self.nonce()
        nonce = self.nonce()

        message = json.dumps([verb, url, body, nonce, timestamp],
                             separators=(',', ':'))
        sha256_hash = hashlib.sha256()
        nonced_message = nonce + message
        sha256_hash.update(nonced_message.encode('utf8'))
        hash_digest = sha256_hash.digest()
        hmac_digest = hmac.new(self.secret.encode('utf-8'),
                               url.encode('utf-8') + hash_digest,
                               hashlib.sha512).digest()
        signature = base64.b64encode(hmac_digest)

        auth_headers = {
            'Authorization': self.key + ':' + signature.decode('utf8'),
            'X-Auth-Timestamp': timestamp,
            'X-Auth-Nonce': nonce,
            'Content-Type': 'application/json'
        }
        return url, {'headers': auth_headers}


class OKCoinREST(RESTAPI):
    def __init__(self, key='', secret='', api_version='v1',
                 url='https://www.okcoin.com/api'):
        super(OKCoinREST, self).__init__(url, api_version=api_version,
                                         key=key,
                                         secret=secret)

    def sign(self,url, endpoint, endpoint_path, method_verb, *args, **kwargs):
        nonce = self.nonce()

        # sig = nonce + url + req
        data = (nonce + url).encode()

        h = hmac.new(self.secret.encode('utf8'), data, hashlib.sha256)
        signature = h.hexdigest()
        headers = {"ACCESS-KEY":       self.key,
                   "ACCESS-NONCE":     nonce,
                   "ACCESS-SIGNATURE": signature}

        return url, {'headers': headers}


class BTCERest(RESTAPI):
    def __init__(self, key='', secret='', api_version='3',
                 url='https://btc-e.com/api'):
        super(BTCERest, self).__init__(url, api_version=api_version, key=key,
                                         secret=secret)

    def sign(self, url, endpoint, endpoint_path, method_verb, *args, **kwargs):
        nonce = self.nonce()
        try:
            params = kwargs['params']
        except KeyError:
            params = {}
        post_params = params
        post_params.update({'nonce': nonce, 'method': endpoint.split('/', 1)[1]})
        post_params = urllib.parse.urlencode(post_params)

        signature = hmac.new(self.secret.encode('utf-8'),
                             post_params.encode('utf-8'), hashlib.sha512)
        headers = {'Key': self.key, 'Sign': signature.hexdigest(),
                   "Content-type": "application/x-www-form-urlencoded"}

        # split by tapi str to gain clean url;
        url = url.split('/tapi', 1)[0] + '/tapi'

        return url, {'headers': headers, 'params': params}


class CCEXRest(RESTAPI):
    def __init__(self, key='', secret='', api_version='',
                 url='https://c-cex.com/t'):
        super(CCEXRest, self).__init__(url, api_version=api_version, key=key,
                                         secret=secret)

    def sign(self, uri, endpoint, endpoint_path, method_verb, *args, **kwargs):
        nonce = self.nonce()
        try:
            params = kwargs['params']
        except KeyError:
            params = {}

        params['apikey'] = self.key
        params['nonce'] = nonce
        post_params = params
        post_params.update({'nonce': nonce, 'method': endpoint})
        post_params = urllib.parse.urlencode(post_params)

        url = uri + post_params

        sig = hmac.new(url, self.secret, hashlib.sha512)
        headers = {'apisign': sig}

        return url, {'headers': headers}


class CryptopiaREST(RESTAPI):
    def __init__(self, key='', secret='', api_version='',
                 url='https://www.cryptopia.co.nz/api'):
        super(CryptopiaREST, self).__init__(url, api_version=api_version, key=key,
                                         secret=secret)

    def sign(self, uri, endpoint, endpoint_path, method_verb, *args, **kwargs):
        nonce = self.nonce()
        try:
            params = kwargs['params']
        except KeyError:
            params = {}


        post_data = json.dumps(params)
        md5 = base64.b64encode(hashlib.md5().updated(post_data).digest())

        sig = self.key + 'POST' + urllib.parse.quote_plus(uri).lower() + nonce + md5
        hmac_sig = base64.b64encode(hmac.new(base64.b64decode(self.secret),
                                              sig, hashlib.sha256).digest())
        header_data = 'amx' + self.key + ':' + hmac_sig + ':' + nonce
        headers = {'Authorization': header_data,
                   'Content-Type': 'application/json; charset=utf-8'}

        return uri, {'headers': headers, 'data': post_data}


class GeminiREST(RESTAPI):
    def __init__(self, key='', secret='', api_version='v1',
                 url='https://api.gemini.com'):
        super(GeminiREST, self).__init__(url, api_version=api_version, key=key,
                                         secret=secret)

    def sign(self, uri, endpoint, endpoint_path, method_verb, *args, **kwargs):
        nonce = self.nonce()
        try:
            params = kwargs['params']
        except KeyError:
            params = {}
        payload = params
        payload['nonce'] = nonce
        payload['request'] = endpoint_path
        payload = base64.b64encode(json.dumps(payload))
        sig = hmac.new(self.secret, payload, hashlib.sha384).hexdigest()
        headers = {'X-GEMINI-APIKEY': self.key,
                   'X-GEMINI-PAYLOAD': payload,
                   'X-GEMINI-SIGNATURE': sig}
        return uri, {'headers': headers}


class YunbiREST(RESTAPI):
    def __init__(self, key='', secret='', api_version='v2',
                 url='https://yunbi.com/api'):
        super(YunbiREST, self).__init__(url, api_version=api_version, key=key,
                                         secret=secret)

    def sign(self, uri, endpoint, endpoint_path, method_verb, *args, **kwargs):
        nonce = self.nonce()
        try:
            params = kwargs['params']
        except KeyError:
            params = {}
        params['tonce'] = nonce
        params['access_key'] = self.key
        post_params = urllib.parse.urlencode(params)
        msg = '%s|%s|%s' % (method_verb, endpoint_path, post_params)

        sig = hmac.new(self.secret, msg, hashlib.sha256).hexdigest()
        uri += post_params + '&signature=' + sig

        return uri, {}


class RockTradingREST(RESTAPI):
    def __init__(self, key='', secret='', api_version='v1',
                 url='https://api.therocktrading.com'):
        super(RockTradingREST, self).__init__(url, api_version=api_version,
                                        key=key,
                                        secret=secret)

    def sign(self, uri, endpoint, endpoint_path, method_verb, *args, **kwargs):
        nonce = self.nonce()
        try:
            params = kwargs['params']
        except KeyError:
            params = {}
        payload = params
        payload['nonce'] = int(nonce)
        payload['request'] = endpoint_path

        msg = nonce + uri
        sig = hmac.new(self.secret.encode(), msg.encode(), hashlib.sha384).hexdigest()
        headers = {'X-TRT-APIKEY': self.key,
                   'X-TRT-Nonce': nonce,
                   'X-TRT-SIGNATURE': sig, 'Content-Type': 'application/json'}
        return uri, {'headers': headers}


class PoloniexREST(RESTAPI):
    def __init__(self, key='', secret='', api_version='',
                 url='https://poloniex.com'):
        super(PoloniexREST, self).__init__(url, api_version=api_version,
                                           key=key, secret=secret)

    def sign(self, uri, endpoint, endpoint_path, method_verb, *args, **kwargs):
        try:
            params = kwargs['params']
        except KeyError:
            params = {}
        params['nonce'] = self.nonce()
        payload = params

        msg = urllib.parse.urlencode(payload).encode('utf-8')
        sig = hmac.new(self.secret.encode('utf-8'), msg, hashlib.sha512).hexdigest()
        headers = {'Key': self.key, 'Sign': sig}
        return uri, {'headers': headers, 'data': params}

`bitex.interfaces`

`bitex.interfaces.kraken`

"""
https:/kraken.com/help/api
"""

# Import Built-Ins
import logging

# Import Third-Party

# Import Homebrew
from bitex.api.rest import KrakenREST
from bitex.utils import return_json
from bitex.formatters.kraken import cancel, trade, order_book

# Init Logging Facilities
log = logging.getLogger(__name__)


class Kraken(KrakenREST):
    def __init__(self, key='', secret='', key_file=''):
        super(Kraken, self).__init__(key, secret)
        if key_file:
            self.load_key(key_file)

    def make_params(self, *pairs, **kwargs):
        q = {'pair': ','.join(pairs)}
        q.update(kwargs)
        return q

    def public_query(self, endpoint, **kwargs):
        path = 'public/' + endpoint
        return self.query('GET', path, **kwargs)

    def private_query(self, endpoint, **kwargs):
        path = 'private/' + endpoint
        return self.query('POST', path, authenticate=True, **kwargs)

    """
    BitEx Standardized Methods
    """

    @return_json(None)
    def ticker(self, *pairs):
        q = self.make_params(*pairs)
        return self.public_query('Ticker', params=q)

    @return_json(order_book)
    def order_book(self, pair, **kwargs):
        q = self.make_params(pair, **kwargs)
        return self.public_query('Depth', params=q)

    @return_json(None)
    def trades(self, pair, **kwargs):
        q = self.make_params(pair, **kwargs)
        return self.public_query('Trades', params=q)

    def _add_order(self, pair, side, price, amount, **kwargs):
        q = {'pair': pair, 'type': side, 'price': price,
             'ordertype': 'limit', 'volume': amount,
             'trading_agreement': 'agree'}
        q.update(kwargs)
        return self.private_query('AddOrder', params=q)

    @return_json(trade)
    def bid(self, pair, price, amount, **kwargs):
        return self._add_order(pair, 'buy', price, amount, **kwargs)

    @return_json(trade)
    def ask(self, pair, price, amount, **kwargs):
        return self._add_order(pair, 'sell', price, amount, **kwargs)

    @return_json(cancel)
    def cancel_order(self, order_id, **kwargs):
        q = {'txid': order_id}
        q.update(kwargs)
        return self.private_query('CancelOrder', params=q)

    @return_json(None)
    def order_info(self, *txids, **kwargs):
        if len(txids) > 1:
            q = {'txid': txids}
        elif txids:
            txid, *_ = txids
            q = {'txid': txid}
        else:
            q = {}
        q.update(kwargs)
        return self.private_query('QueryOrders', params=q)

    @return_json(None)
    def balance(self, **kwargs):
        return self.private_query('Balance')

    @return_json(None)
    def withdraw(self, _type, source_wallet, amount, tar_addr, **kwargs):
        raise NotImplementedError()

    @return_json(None)
    def deposit_address(self, **kwargs):
        raise NotImplementedError()

    """
    Exchange Specific Methods
    """

    @return_json(None)
    def time(self):
        return self.public_query('Time')

    @return_json(None)
    def assets(self, **kwargs):
        return self.public_query('Assets', params=kwargs)

    @return_json(None)
    def pairs(self, **kwargs):
        return self.public_query('AssetPairs', params=kwargs)

    @return_json(None)
    def ohlc(self, pair, **kwargs):
        q = self.make_params(pair, **kwargs)
        return self.public_query('OHLC', params=q)

    @return_json(None)
    def spread(self, pair, **kwargs):
        q = self.make_params(pair, **kwargs)
        return self.public_query('Spread', params=q)

    @return_json(None)
    def orders(self, **kwargs):
        q = kwargs
        return self.private_query('OpenOrders', params=q)

    @return_json(None)
    def closed_orders(self, **kwargs):
        q = kwargs
        return self.private_query('ClosedOrders', params=q)

    @return_json(None)
    def trade_history(self, **kwargs):
        q = kwargs
        return self.private_query('TradesHistory', params=q)

    @return_json(None)
    def fees(self, pair=None):
        q = {'fee-info': True}

        if pair:
            q['pair'] = pair
        return self.private_query('TradeVolume', params=q)

`bitex.interfaces.bitfinex`

"""
http://docs.bitfinex.com/
"""

# Import Built-Ins
import logging

# Import Third-Party

# Import Homebrew
from bitex.api.rest import BitfinexREST
from bitex.utils import return_json
from bitex.formatters.bitfinex import trade, cancel, order_status
# Init Logging Facilities
log = logging.getLogger(__name__)


class Bitfinex(BitfinexREST):
    def __init__(self, key='', secret='', key_file=''):
        super(Bitfinex, self).__init__(key, secret)
        if key_file:
            self.load_key(key_file)

    def public_query(self, endpoint, **kwargs):
        return self.query('GET', endpoint, **kwargs)

    def private_query(self, endpoint, **kwargs):
        return self.query('POST', endpoint, authenticate=True, **kwargs)

    """
    BitEx Standardized Methods
    """
    @return_json(None)
    def order_book(self, pair, **kwargs):
        return self.public_query('book/%s' % pair, params=kwargs)

    @return_json(None)
    def ticker(self, pair, **kwargs):
        return self.public_query('pubticker/%s' % pair, params=kwargs)

    @return_json(None)
    def trades(self, pair, **kwargs):
        return self.public_query('trades/%s' % pair, params=kwargs)

    def _place_order(self, pair, amount, price, side, replace, **kwargs):
        q = {'symbol': pair, 'amount': amount, 'price': price, 'side': side,
             'type': 'exchange limit'}
        q.update(kwargs)
        if replace:
            return self.private_query('order/cancel/replace', params=q)
        else:
            return self.private_query('order/new', params=q)

    @return_json(trade)
    def bid(self, pair, price, amount, replace=False, **kwargs):
        return self._place_order(pair, amount, price, 'buy', replace=replace,
                                 **kwargs)

    @return_json(trade)
    def ask(self, pair, price, amount, replace=False, **kwargs):
        return self._place_order(pair, str(amount), str(price), 'sell',
                                 replace=replace, **kwargs)

    @return_json(cancel)
    def cancel_order(self, order_id, all=False, **kwargs):

        q = {'order_id': int(order_id)}
        q.update(kwargs)
        if not all:
            return self.private_query('order/cancel', params=q)
        else:
            endpoint = 'order/cancel/all'
            return self.private_query(endpoint)

    @return_json(order_status)
    def order(self, order_id, **kwargs):
        q = {'order_id': order_id}
        q.update(kwargs)
        return self.private_query('order/status', params=q)

    @return_json(None)
    def balance(self, **kwargs):
        return self.private_query('balances', params=kwargs)

    @return_json(None)
    def withdraw(self, _type, source_wallet, amount, tar_addr, **kwargs):
        q = {'withdraw_type': _type, 'walletselected': source_wallet,
             'amount': amount, 'address': tar_addr}
        q.update(kwargs)
        return self.private_query('withdraw', params=q)

    @return_json(None)
    def deposit_address(self, **kwargs):
        q = {'method': currency, 'wallet_name': target_wallet}
        q.update(kwargs)
        return self.private_query('deposit/new', params=kwargs)

    """
    Exchange Specific Methods
    """

    @return_json(None)
    def statistics(self, pair):
        return self.public_query('stats/%s' % pair)

    @return_json(None)
    def funding_book(self, currency, **kwargs):
        return self.public_query('lendbook/%s' % currency, params=kwargs)

    @return_json(None)
    def lends(self, currency, **kwargs):
        return self.public_query('lends/%s' % currency, params=kwargs)

    @return_json(None)
    def pairs(self, details=False):
        if details:
            return self.public_query('symbols_details')
        else:
            return self.public_query('symbols')

    @return_json(None)
    def fees(self):
        return self.private_query('account_infos')

    @return_json(None)
    def orders(self):
        return self.private_query('orders')

    @return_json(None)
    def balance_history(self, currency, **kwargs):
        q = {'currency': currency}
        q.update(kwargs)
        return self.private_query('history/movements', params=q)

    @return_json(None)
    def trade_history(self, pair, since, **kwargs):
        q = {'symbol': pair, 'timestamp': since}
        q.update(kwargs)
    return self.private_query('mytrades', params=q)

`bitex.interfaces.gdax`

"""
https://docs.gdax.com/
"""

# Import Built-Ins
import logging

# Import Third-Party

# Import Homebrew
from bitex.api.rest import GDAXRest
from bitex.utils import return_json

# Init Logging Facilities
log = logging.getLogger(__name__)


class GDAX(GDAXRest):
    def __init__(self, key='', secret='', key_file=''):
        super(GDAX, self).__init__(key, secret)
        if key_file:
            self.load_key(key_file)

    def public_query(self, endpoint, **kwargs):
        return self.query('GET', endpoint, **kwargs)

    def private_query(self, endpoint, method_verb='POST', **kwargs):
        return self.query(method_verb, endpoint, authenticate=True, **kwargs)

    """
    BitEx Standardized Methods
    """

    @return_json(None)
    def ticker(self, pair, **kwargs):
        return self.public_query('products/%s/ticker' % pair, params=kwargs)

    @return_json(None)
    def order_book(self, pair, **kwargs):
        return self.public_query('products/%s/book' % pair, params=kwargs)

    @return_json(None)
    def trades(self, pair, **kwargs):
        return self.public_query('products/%s/trades' % pair, params=kwargs)

    @return_json(None)
    def bid(self, pair, price, size, **kwargs):
        q = {'side': 'buy', 'type': 'market', 'product_id': pair,
             'price': price, 'size': size}
        q.update(kwargs)
        return self.private_query('orders', params=q)

    @return_json(None)
    def ask(self, pair, price, amount, **kwargs):
        q = {'side': 'sell', 'type': 'market', 'product_id': pair,
             'price': price, 'size': size}
        q.update(kwargs)
        return self.private_query('orders', params=q)

    @return_json(None)
    def cancel_order(self, order_id, all=False, **kwargs):

        if not all:
            return self.private_query('orders/%s' % order_id,
                                      method_verb='DELETE', params=kwargs)
        else:
            return self.private_query('orders', method_verb='DELETE',
                                      params=kwargs)

    @return_json(None)
    def order(self, order_id, **kwargs):
        return self.private_query('orders/%s' % order_id, method_verb='GET',
                                  params=kwargs)

    @return_json(None)
    def balance(self, **kwargs):
        return self.private_query('accounts', method_verb='GET', params=kwargs)

    @return_json(None)
    def withdraw(self, _type, source_wallet, amount, tar_addr, **kwargs):
        raise NotImplementedError()

    @return_json(None)
    def deposit_address(self, **kwargs):
        raise NotImplementedError()

    """
    Exchange Specific Methods
    """

    @return_json
    def time(self):
        return self.public_query('time')

    @return_json(None)
    def currencies(self):
        return self.public_query('currencies')

    @return_json(None)
    def pairs(self):
        return self.public_query('products')

    @return_json(None)
    def ohlc(self, pair, **kwargs):
        return self.public_query('products/%s/candles' % pair, params=kwargs)

    @return_json(None)
    def stats(self, pair, **kwargs):
        return self.public_query('products/%s/stats' % pair, params=kwargs)

`bitex.utils`

# Import Built-Ins
import logging
import json
import requests
# Import Third-Party

# Import Homebrew

# Init Logging Facilities
log = logging.getLogger(__name__)


def return_json(formatter=None):
    def decorator(func):
        def wrapper(*args, **kwargs):
            try:
                r = func(*args, **kwargs)
            except Exception as e:
                log.error("return_json(): Error during call to "
                          "%s(%s, %s) %s" % (func.__name__, args, kwargs, e))
                raise

            try:
                r.raise_for_status()
            except requests.HTTPError as e:
                log.error("return_json: HTTPError for url %s: "
                          "%s" % (r.request.url, e))
                return None, r

            try:
                data = r.json()
            except json.JSONDecodeError:
                log.error('return_json: Error while parsing json. '
                          'Request url was: %s, result is: '
                          '%s' % (r.request.url, r.text))
                return None, r
            except Exception as e:
                log.error("return_json(): Unexpected error while parsing json "
                          "from %s: %s" % (r.request.url, e))
                raise

            # Apply formatter and return
            if formatter is not None:
                return formatter(data, *args, **kwargs), r
            else:
                return data, r
        return wrapper
return decorator

您还可以在其GitHub存储库中找到代码。我省略了一些接口类--我提供的三个类与它们所提供的类一样多样。

GitHub仓库 (dev分支)

framework

cryptocurrency

python

python-3.x

rest

有奖征集｜云上CPU玩转AIGC挑战赛

参加活动赢取洛斐键盘、小米电纸书、智能音响等鹅厂周边！

回答 1

Code Review用户

回答已采纳

发布于 2017-01-12 14:27:43

这里有很多代码，所以我只想回顾一下bitex.api。你会发现这里有很多可以供一次审查的。也许在“代码评审”中的其他评审人员会检查您的其他代码。

没有模块文档字符串。这个模块的目的是什么？里面有什么？
import requests位于“内置”部分，但据我所知，这个模块不是内置到Python中的，因此它应该位于“第三方”部分。
RESTAPI的名称可以改进。这个类的实例是代表API客户端还是API服务器？名字没有说清楚。
RESTAPI类没有docstring。这个类的实例代表什么？你能举几个例子说明一下它的用法吗？在对post的介绍中，有一些文本可以用作docstring的起点。
当您记录消息时，不要使用%操作符来格式化消息，而是分别传递格式字符串和格式参数如文件所述。(这为记录器提供了更大的灵活性，例如，如果日志记录被禁止，那么消息可能永远不需要格式化。)
__init__方法的docstring需要记录该方法，而不是整个类。所以应该开始“创建RESTAPI对象.”然后继续记录论点的意义。
这很小，但是为什么参数用下划线拼写api_version，而属性拼写为apiversion却没有？
使用不安全的key和secret的默认参数是有风险的。有一种风险是，由于疏忽，load_key方法可能不会被调用(特别是因为没有文档来解释何时需要调用它)。这将使API变得不安全。如果代码在默认情况下是安全的，那么情况要好得多。因此，我宁愿使用无效的默认参数。
self.req_methods每次都是相同的，所以它应该是全局常量或类属性，而不是实例属性。
可以避免self.req_methods结构中的重复，例如：#支持的HTTP谓词。REQ_METHODS ={谓词:getattr(请求，verb.lower())，用于动词}，但实际上，我认为这些数据结构是不必要的。而不是: request_method = self.req_methods方法_动词 #。R= request_method(url，timeout=5，**request_kwargs)为什么不使用requests.request？R= requests.request(method_verb，url，timeout=5，**request_kwargs)
load_key的docstring应该解释path参数的含义。(这是一个文本文件的路径，其前两行分别是密钥和秘密。)
open的第二个参数默认为'r'，因此可以省略这一点。
方法nonce没有docstring。
仅以time.time的输出为基础是有风险的，因为这可能会倒退：>>>导入时间>>> time.get_clock_info(' time ').monotonic False，这意味着nonces可能会重复。
sign的docstring并不解释方法应该做什么。
默认情况下，sign方法是不安全的。很容易忘记推翻它，这样就不会有什么问题，但是所有的签名都是假的。在默认情况下最好是安全的。一个很好的方法是使用来自abc (抽象基类)模块的工具:从abc导入ABCMeta，抽象方法类RESTAPI(metaclass=ABCMeta)：#.@抽象方法def签名(self，.)：#.现在，在不重写RESTAPI方法的情况下从sign继承的任何尝试都会引发异常。
query方法的docstring并不解释参数的含义或返回的内容。
timeout=5关键字参数是硬编码的。如果有人需要查询比此更长的超时时间的API，该怎么办？该值应该是query函数的关键字参数(或者可能是类或对象的属性)。