ParameterBuilder.error_on_invalid_value() - Code Metrics - Inspection of "Merge pull request #18 from peshay/refactor" - peshay/btcde - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( a48a20...2f6ef6 )

by Andreas

created 2017-12-29 18:35 UTC

ParameterBuilder.error_on_invalid_value() A

↳ Parent: ParameterBuilder

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	1

Metric	Value
cc	3
c	1
b	0
f	1
dl	0
loc	4
rs	10

#! /usr/bin/env python
"""API Wrapper for Bitcoin.de Trading API."""

import requests
import time
import json
import hmac
import hashlib
import logging
import codecs
import decimal

logging.basicConfig()
logging.getLogger().setLevel(logging.DEBUG)
log = logging.getLogger(__name__)
requests_log = logging.getLogger("requests.packages.urllib3")
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True

__version__ = '2.0'

# disable unsecure SSL warning
requests.packages.urllib3.disable_warnings()

class ParameterBuilder(object):
    '''To verify given parameters for API.'''
    def __init__(self, avail_params, given_params, uri):
        self.verify_keys_and_values(avail_params, given_params)
        self.params = given_params
        self.create_url(uri)

    def verify_keys_and_values(self, avail_params, given_params):
        for k, v in given_params.items():
            if k not in avail_params:
                list_string = ', '.join(avail_params)
                raise KeyError("{} is not any of {}".format(k, list_string))
            if k == 'trading_pair':
                self.error_on_invalid_value(v, self.TRADING_PAIRS)
            elif k == 'type':
                self.error_on_invalid_value(v, self.ORDER_TYPES)
            elif k == 'currency':
                self.error_on_invalid_value(v, self.CURRENCIES)
            elif k == 'seat_of_bank':
                self.error_on_invalid_value(v, self.BANK_SEATS)
            elif k in ['min_trust_level', 'trust_level']:
                self.error_on_invalid_value(v, self.TRUST_LEVELS)
            elif k == 'payment_option':
                self.error_on_invalid_value(v, self.PAYMENT_OPTIONS)
            elif k == 'state':
                self.error_on_invalid_value(v, self.STATES)
    
    def error_on_invalid_value(self, value, list):
        if value not in list:
            list_string = ', '.join(str(x) for x in list)
            raise ValueError("{} is not any of {}".format(value, list_string))


    def create_url(self, uri):
        if self.params:
            self.encoded_string = ''
            for key, value in sorted(self.params.items()):
                self.encoded_string += str(key) + '=' + str(value) + '&'
            self.encoded_string = self.encoded_string[:-1]
            self.url = uri + '?' + self.encoded_string
        else:
            self.encoded_string = ''
            self.url = uri

    
    TRADING_PAIRS = ['btceur', 'bcheur', 'etheur']
    ORDER_TYPES = ['buy', 'sell']
    CURRENCIES = ['btc', 'bch', 'eth']
    BANK_SEATS = ['AT', 'BE', 'BG', 'CH', 'CY', 'CZ',
                  'DE', 'DK', 'EE', 'ES', 'FI', 'FR',
                  'GB', 'GR', 'HR', 'HU', 'IE', 'IS',
                  'IT', 'LI', 'LT', 'LU', 'LV', 'MT',
                  'MQ', 'NL', 'NO', 'PL', 'PT', 'RO',
                  'SE', 'SI', 'SK']
    TRUST_LEVELS = ['bronze', 'silver', 'gold', 'platin']
    STATES = [-1, 0, 1]
    PAYMENT_OPTIONS = [1, 2, 3]
    TRADE_TYPES = ['all', 'buy', 'sell', 'inpayment',
                   'payout', 'affiliate', 'welcome_btc',
                   'buy_yubikey', 'buy_goldshop',
                   'buy_diamondshop', 'kickback',
                   'outgoing_fee_voluntary']

def HandleRequestsException(e):
    """Handle Exception from request."""
    log.warning(e)


def HandleAPIErrors(r):
    """To handle Errors from BTCDE API."""
    valid_status_codes = [200, 201, 204]
    if r.status_code not in valid_status_codes:
        reader = codecs.getreader("utf-8")
        content = json.load(reader(r.raw))
        errors = content.get('errors')
        log.warning('API Error Code: {}'.format(str(errors[0]['code'])))
        log.warning('API Error Message: {}'.format(errors[0]['message']))
        log.warning('API Error URL: {}'.format(r.url))
        return False
    else:
        return True


class Connection(object):
    """To provide connection credentials to the trading API"""
    def __init__(self, api_key, api_secret):
        self.api_key = api_key
        self.api_secret = api_secret
        # set initial self.nonce
        self.nonce = int(time.time())
        # Bitcoin.de API URI
        self.apihost = 'https://api.bitcoin.de'
        self.apiversion = 'v2'
        self.orderuri = self.apihost + '/' + self.apiversion + '/' + 'orders'
        self.tradeuri = self.apihost + '/' + self.apiversion + '/' + 'trades'
        self.accounturi = self.apihost + '/' + self.apiversion + '/' + 'account'

    def build_hmac_sign(self, md5string, method, url):
        hmac_data = '{method}#{url}#{key}#{nonce}#{md5}'\
                    .format(method=method, url=url,
                            key=self.api_key, nonce=str(self.nonce),
                            md5=md5string)
        hmac_signed = hmac.new(bytearray(self.api_secret.encode()), msg=hmac_data.encode(), digestmod=hashlib.sha256).hexdigest()
        return hmac_signed
        
    def set_header(self, url, method, encoded_string):
        # raise self.nonce before using
        self.nonce += 1
        if method == 'POST':
            md5_encoded_query_string = hashlib.md5(encoded_string.encode()).hexdigest()
        else:
            md5_encoded_query_string = hashlib.md5(b'').hexdigest()
        hmac_signed = self.build_hmac_sign(md5_encoded_query_string,
                                           method, url)
        # set header
        header = {'content-type':
                  'application/x-www-form-urlencoded; charset=utf-8',
                  'X-API-KEY': self.api_key,
                  'X-API-NONCE': str(self.nonce),
                  'X-API-SIGNATURE': hmac_signed }
        return header

    def send_request(self, url, method, header, encoded_string):
        if method == 'GET':
            r = requests.get(url, headers=(header),
                             stream=True, verify=False)
        elif method == 'POST':
            r = requests.post(url, headers=(header), data=encoded_string,
                              stream=True, verify=False)
        elif method == 'DELETE':
            r = requests.delete(url, headers=(header),
                                stream=True, verify=False)
        return r

    def APIConnect(self, method, params):
        """Transform Parameters to URL"""
        header = self.set_header(params.url, method,
                                 params.encoded_string)
        log.debug('Set Header: {}'.format(header))
        try:
            r = self.send_request(params.url, method, header,
                                  params.encoded_string)
            # Handle API Errors
            if HandleAPIErrors(r):
                # get results
                result = r.json()
            else:
                result = {}
        except requests.exceptions.RequestException as e:
            HandleRequestsException(e)
            result = {}
        return result

    def showOrderbook(self, order_type, trading_pair, **args):
        """Search Orderbook for offers."""
        params = {'type': order_type,
                  'trading_pair': trading_pair}
        params.update(args)
        avail_params = ['type', 'trading_pair', 'amount', 'price',
                        'order_requirements_fullfilled',
                        'only_kyc_full', 'only_express_orders',
                        'only_same_bankgroup', 'only_same_bic',
                        'seat_of_bank']
        p = ParameterBuilder(avail_params, params, self.orderuri)
        return self.APIConnect('GET', p)


    def createOrder(self, order_type, trading_pair, max_amount, price, **args):
        """Create a new Order."""
        # Build parameters
        params = {'type': order_type,
                  'max_amount': max_amount,
                  'price': price,
                  'trading_pair': trading_pair}
        params.update(args)
        avail_params = ['type', 'trading_pair', 'max_amount', 'price',
                        'min_amount', 'new_order_for_remaining_amount',
                        'min_trust_level', 'only_kyc_full', 'payment_option',
                        'seat_of_bank']
        p = ParameterBuilder(avail_params, params, self.orderuri)
        return self.APIConnect('POST', p)


    def deleteOrder(self, order_id, trading_pair):
        """Delete an Order."""
        # Build parameters
        params = {'order_id': order_id,
                  'trading_pair': trading_pair}
        avail_params = ['order_id', 'trading_pair']
        newuri = self.orderuri + "/" + order_id + "/" + trading_pair
        p = ParameterBuilder(avail_params, params, newuri)
        p.encoded_string = ''
        p.url = newuri
        return self.APIConnect('DELETE', p)


    def showMyOrders(self, **args):
        """Query and Filter own Orders."""
        # Build parameters
        params = args
        avail_params = ['type', 'trading_pair', 'state',
                        'date_start', 'date_end', 'page']
        newuri = self.orderuri + '/my_own'
        p = ParameterBuilder(avail_params, params, newuri)
        return self.APIConnect('GET', p)


    def showMyOrderDetails(self, order_id):
        """Details to an own Order."""
        newuri = self.orderuri + '/' + order_id
        p = ParameterBuilder({}, {}, newuri)
        return self.APIConnect('GET', p)


    def executeTrade(self, order_id, order_type, trading_pair, amount):
        """Buy/Sell on a specific Order."""
        newuri = self.tradeuri + '/' + order_id
        params = {'order_id': order_id,
                  'type': order_type,
                  'trading_pair': trading_pair,
                  'amount': amount}
        avail_params = ['order_id', 'type', 'trading_pair',
                        'amount']
        p = ParameterBuilder(avail_params, params, newuri)
        return self.APIConnect('POST', p)


    def showMyTrades(self, **args):
        """Query and Filter on past Trades."""
        # Build parameters
        params = args
        avail_params = ['type', 'trading_pair', 'state',
                        'date_start', 'date_end', 'page']
        p = ParameterBuilder(avail_params, params, self.tradeuri)
        return self.APIConnect('GET', p)


    def showMyTradeDetails(self, trade_id):
        """Details to a specific Trade."""
        newuri = self.tradeuri + '/' + trade_id
        params = {}
        p = ParameterBuilder({}, {}, newuri)
        return self.APIConnect('GET', p)


    def showAccountInfo(self):
        """Query on Account Infos."""
        p = ParameterBuilder({}, {}, self.accounturi)
        return self.APIConnect('GET', p)


    def showOrderbookCompact(self, trading_pair):
        """Bids and Asks in compact format."""
        params = {'trading_pair': trading_pair}
        # Build parameters
        avail_params = ['trading_pair']
        p = ParameterBuilder(avail_params, params,
                             self.orderuri + '/compact')
        return self.APIConnect('GET', p)


    def showPublicTradeHistory(self, trading_pair, **args):
        """All successful trades of the las 7 days."""
        params = {'trading_pair': trading_pair}
        params.update(args)
        avail_params = ['trading_pair', 'since_tid']
        p = ParameterBuilder(avail_params, params,
                             self.tradeuri + '/history')
        return self.APIConnect('GET', p)


    def showRates(self, trading_pair):
        """Query of the average rate last 3 and 12 hours."""
        newuri = self.apihost + '/' + self.apiversion + '/rates'
        params = {'trading_pair': trading_pair}
        avail_params = ['trading_pair']
        p = ParameterBuilder(avail_params, params, newuri)
        return self.APIConnect('GET', p)


    def showAccountLedger(self, currency, **args):
        """Query on Account statement."""
        params = {'currency': currency}
        params.update(args)
        avail_params = ['currency', 'type',
                        'date_start', 'date_end', 'page']
        p = ParameterBuilder(avail_params, params,
                             self.accounturi + '/ledger')
        return self.APIConnect('GET', p)


1		#! /usr/bin/env python
2		"""API Wrapper for Bitcoin.de Trading API."""
3
4		import requests
5		import time
6		import json
7		import hmac
8		import hashlib
9		import logging
10		import codecs
11		import decimal
12
13		logging.basicConfig()
14		logging.getLogger().setLevel(logging.DEBUG)
15		log = logging.getLogger(__name__)
16		requests_log = logging.getLogger("requests.packages.urllib3")
17		requests_log.setLevel(logging.DEBUG)
18		requests_log.propagate = True
19
20		__version__ = '2.0'
21
22		# disable unsecure SSL warning
23		requests.packages.urllib3.disable_warnings()
24
25		class ParameterBuilder(object):
26		'''To verify given parameters for API.'''
27		def __init__(self, avail_params, given_params, uri):
28		self.verify_keys_and_values(avail_params, given_params)
29		self.params = given_params
30		self.create_url(uri)
31
32		def verify_keys_and_values(self, avail_params, given_params):
33		for k, v in given_params.items():
34		if k not in avail_params:
35		list_string = ', '.join(avail_params)
36		raise KeyError("{} is not any of {}".format(k, list_string))
37		if k == 'trading_pair':
38		self.error_on_invalid_value(v, self.TRADING_PAIRS)
39		elif k == 'type':
40		self.error_on_invalid_value(v, self.ORDER_TYPES)
41		elif k == 'currency':
42		self.error_on_invalid_value(v, self.CURRENCIES)
43		elif k == 'seat_of_bank':
44		self.error_on_invalid_value(v, self.BANK_SEATS)
45		elif k in ['min_trust_level', 'trust_level']:
46		self.error_on_invalid_value(v, self.TRUST_LEVELS)
47		elif k == 'payment_option':
48		self.error_on_invalid_value(v, self.PAYMENT_OPTIONS)
49		elif k == 'state':
50		self.error_on_invalid_value(v, self.STATES)
51
52		def error_on_invalid_value(self, value, list):
53		if value not in list:
54		list_string = ', '.join(str(x) for x in list)
55		raise ValueError("{} is not any of {}".format(value, list_string))
56	View Code Duplication
		0 ignored issues – show Duplication introduced 2017-12-29 18:25 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
57		def create_url(self, uri):
58		if self.params:
59		self.encoded_string = ''
60		for key, value in sorted(self.params.items()):
61		self.encoded_string += str(key) + '=' + str(value) + '&'
62		self.encoded_string = self.encoded_string[:-1]
63		self.url = uri + '?' + self.encoded_string
64		else:
65		self.encoded_string = ''
66		self.url = uri
67
68
69		TRADING_PAIRS = ['btceur', 'bcheur', 'etheur']
70		ORDER_TYPES = ['buy', 'sell']
71		CURRENCIES = ['btc', 'bch', 'eth']
72		BANK_SEATS = ['AT', 'BE', 'BG', 'CH', 'CY', 'CZ',
73		'DE', 'DK', 'EE', 'ES', 'FI', 'FR',
74		'GB', 'GR', 'HR', 'HU', 'IE', 'IS',
75		'IT', 'LI', 'LT', 'LU', 'LV', 'MT',
76		'MQ', 'NL', 'NO', 'PL', 'PT', 'RO',
77		'SE', 'SI', 'SK']
78		TRUST_LEVELS = ['bronze', 'silver', 'gold', 'platin']
79		STATES = [-1, 0, 1]
80		PAYMENT_OPTIONS = [1, 2, 3]
81		TRADE_TYPES = ['all', 'buy', 'sell', 'inpayment',
82		'payout', 'affiliate', 'welcome_btc',
83		'buy_yubikey', 'buy_goldshop',
84		'buy_diamondshop', 'kickback',
85		'outgoing_fee_voluntary']
86
87		def HandleRequestsException(e):
88		"""Handle Exception from request."""
89		log.warning(e)
90
91
92		def HandleAPIErrors(r):
93		"""To handle Errors from BTCDE API."""
94		valid_status_codes = [200, 201, 204]
95		if r.status_code not in valid_status_codes:
96		reader = codecs.getreader("utf-8")
97		content = json.load(reader(r.raw))
98		errors = content.get('errors')
99		log.warning('API Error Code: {}'.format(str(errors[0]['code'])))
100		log.warning('API Error Message: {}'.format(errors[0]['message']))
101		log.warning('API Error URL: {}'.format(r.url))
102		return False
103		else:
104		return True
105
106
107		class Connection(object):
108		"""To provide connection credentials to the trading API"""
109		def __init__(self, api_key, api_secret):
110		self.api_key = api_key
111		self.api_secret = api_secret
112		# set initial self.nonce
113		self.nonce = int(time.time())
114		# Bitcoin.de API URI
115		self.apihost = 'https://api.bitcoin.de'
116		self.apiversion = 'v2'
117		self.orderuri = self.apihost + '/' + self.apiversion + '/' + 'orders'
118		self.tradeuri = self.apihost + '/' + self.apiversion + '/' + 'trades'
119		self.accounturi = self.apihost + '/' + self.apiversion + '/' + 'account'
120
121		def build_hmac_sign(self, md5string, method, url):
122		hmac_data = '{method}#{url}#{key}#{nonce}#{md5}'\
123		.format(method=method, url=url,
124		key=self.api_key, nonce=str(self.nonce),
125		md5=md5string)
126		hmac_signed = hmac.new(bytearray(self.api_secret.encode()), msg=hmac_data.encode(), digestmod=hashlib.sha256).hexdigest()
127		return hmac_signed
128
129		def set_header(self, url, method, encoded_string):
130		# raise self.nonce before using
131		self.nonce += 1
132		if method == 'POST':
133		md5_encoded_query_string = hashlib.md5(encoded_string.encode()).hexdigest()
134		else:
135		md5_encoded_query_string = hashlib.md5(b'').hexdigest()
136		hmac_signed = self.build_hmac_sign(md5_encoded_query_string,
137		method, url)
138		# set header
139		header = {'content-type':
140		'application/x-www-form-urlencoded; charset=utf-8',
141		'X-API-KEY': self.api_key,
142		'X-API-NONCE': str(self.nonce),
143		'X-API-SIGNATURE': hmac_signed }
144		return header
145
146		def send_request(self, url, method, header, encoded_string):
147		if method == 'GET':
148		r = requests.get(url, headers=(header),
149		stream=True, verify=False)
150		elif method == 'POST':
151		r = requests.post(url, headers=(header), data=encoded_string,
152		stream=True, verify=False)
153		elif method == 'DELETE':
154		r = requests.delete(url, headers=(header),
155		stream=True, verify=False)
156		return r
157
158		def APIConnect(self, method, params):
159		"""Transform Parameters to URL"""
160		header = self.set_header(params.url, method,
161		params.encoded_string)
162		log.debug('Set Header: {}'.format(header))
163		try:
164		r = self.send_request(params.url, method, header,
165		params.encoded_string)
166		# Handle API Errors
167		if HandleAPIErrors(r):
168		# get results
169		result = r.json()
170		else:
171		result = {}
172		except requests.exceptions.RequestException as e:
173		HandleRequestsException(e)
174		result = {}
175		return result
176
177		def showOrderbook(self, order_type, trading_pair, **args):
178		"""Search Orderbook for offers."""
179		params = {'type': order_type,
180		'trading_pair': trading_pair}
181		params.update(args)
182		avail_params = ['type', 'trading_pair', 'amount', 'price',
183		'order_requirements_fullfilled',
184		'only_kyc_full', 'only_express_orders',
185		'only_same_bankgroup', 'only_same_bic',
186		'seat_of_bank']
187		p = ParameterBuilder(avail_params, params, self.orderuri)
188		return self.APIConnect('GET', p)
189
190
191		def createOrder(self, order_type, trading_pair, max_amount, price, **args):
192		"""Create a new Order."""
193		# Build parameters
194		params = {'type': order_type,
195		'max_amount': max_amount,
196		'price': price,
197		'trading_pair': trading_pair}
198		params.update(args)
199		avail_params = ['type', 'trading_pair', 'max_amount', 'price',
200		'min_amount', 'new_order_for_remaining_amount',
201		'min_trust_level', 'only_kyc_full', 'payment_option',
202		'seat_of_bank']
203		p = ParameterBuilder(avail_params, params, self.orderuri)
204		return self.APIConnect('POST', p)
205
206
207		def deleteOrder(self, order_id, trading_pair):
208		"""Delete an Order."""
209		# Build parameters
210		params = {'order_id': order_id,
211		'trading_pair': trading_pair}
212		avail_params = ['order_id', 'trading_pair']
213		newuri = self.orderuri + "/" + order_id + "/" + trading_pair
214		p = ParameterBuilder(avail_params, params, newuri)
215		p.encoded_string = ''
216		p.url = newuri
217		return self.APIConnect('DELETE', p)
218
219
220		def showMyOrders(self, **args):
221		"""Query and Filter own Orders."""
222		# Build parameters
223		params = args
224		avail_params = ['type', 'trading_pair', 'state',
225		'date_start', 'date_end', 'page']
226		newuri = self.orderuri + '/my_own'
227		p = ParameterBuilder(avail_params, params, newuri)
228		return self.APIConnect('GET', p)
229
230
231		def showMyOrderDetails(self, order_id):
232		"""Details to an own Order."""
233		newuri = self.orderuri + '/' + order_id
234		p = ParameterBuilder({}, {}, newuri)
235		return self.APIConnect('GET', p)
236
237
238		def executeTrade(self, order_id, order_type, trading_pair, amount):
239		"""Buy/Sell on a specific Order."""
240		newuri = self.tradeuri + '/' + order_id
241		params = {'order_id': order_id,
242		'type': order_type,
243		'trading_pair': trading_pair,
244		'amount': amount}
245		avail_params = ['order_id', 'type', 'trading_pair',
246		'amount']
247		p = ParameterBuilder(avail_params, params, newuri)
248		return self.APIConnect('POST', p)
249
250
251		def showMyTrades(self, **args):
252		"""Query and Filter on past Trades."""
253		# Build parameters
254		params = args
255		avail_params = ['type', 'trading_pair', 'state',
256		'date_start', 'date_end', 'page']
257		p = ParameterBuilder(avail_params, params, self.tradeuri)
258		return self.APIConnect('GET', p)
259
260
261		def showMyTradeDetails(self, trade_id):
262		"""Details to a specific Trade."""
263		newuri = self.tradeuri + '/' + trade_id
264		params = {}
265		p = ParameterBuilder({}, {}, newuri)
266		return self.APIConnect('GET', p)
267
268
269		def showAccountInfo(self):
270		"""Query on Account Infos."""
271		p = ParameterBuilder({}, {}, self.accounturi)
272		return self.APIConnect('GET', p)
273
274
275		def showOrderbookCompact(self, trading_pair):
276		"""Bids and Asks in compact format."""
277		params = {'trading_pair': trading_pair}
278		# Build parameters
279		avail_params = ['trading_pair']
280		p = ParameterBuilder(avail_params, params,
281		self.orderuri + '/compact')
282		return self.APIConnect('GET', p)
283
284
285		def showPublicTradeHistory(self, trading_pair, **args):
286		"""All successful trades of the las 7 days."""
287		params = {'trading_pair': trading_pair}
288		params.update(args)
289		avail_params = ['trading_pair', 'since_tid']
290		p = ParameterBuilder(avail_params, params,
291		self.tradeuri + '/history')
292		return self.APIConnect('GET', p)
293
294
295		def showRates(self, trading_pair):
296		"""Query of the average rate last 3 and 12 hours."""
297		newuri = self.apihost + '/' + self.apiversion + '/rates'
298		params = {'trading_pair': trading_pair}
299		avail_params = ['trading_pair']
300		p = ParameterBuilder(avail_params, params, newuri)
301		return self.APIConnect('GET', p)
302
303
304		def showAccountLedger(self, currency, **args):
305		"""Query on Account statement."""
306		params = {'currency': currency}
307		params.update(args)
308		avail_params = ['currency', 'type',
309		'date_start', 'date_end', 'page']
310		p = ParameterBuilder(avail_params, params,
311		self.accounturi + '/ledger')
312		return self.APIConnect('GET', p)
313

peshay / btcde

Push — master ( a48a20...2f6ef6 )

ParameterBuilder.error_on_invalid_value() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like