btcde.Connection.listAddressPool() - Code Metrics - Inspection of "docs: " - peshay/btcde - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 140199...ad2fe1 )

by Andreas

created 2021-04-11 10:14 UTC

btcde.Connection.listAddressPool() A

↳ Parent: btcde

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	3
dl	0
loc	7
rs	10
c	0
b	0
f	0

#! /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
import inspect
import urllib

from urllib.parse import urlencode

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

__version__ = '4.0'

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.TRADE_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':
                caller = inspect.stack()[2][3]
                if caller in ["showMyOrders", "showMyOrderDetails"]:
                    self.error_on_invalid_value(v, self.ORDER_STATES)
                elif caller in ["showMyTrades", "showMyTradesDetails"]:
                    self.error_on_invalid_value(v, self.TRADE_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 = urlencode(sorted(self.params.items()))
            self.url = uri + '?' + self.encoded_string
        else:
            self.encoded_string = ''
            self.url = uri


    TRADING_PAIRS = ['btceur', 'bcheur', 'etheur', 'btgeur', 'bsveur', 'ltceur',
                     'iotabtc', 'dashbtc', 'gntbtc', 'ltcbtc']
    ORDER_TYPES = ['buy', 'sell']
    CURRENCIES = ['btc', 'bch', 'eth', 'btg', 'bsv', 'ltc',
                  'iota', 'dash', 'gnt']
    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']
    TRADE_STATES = [-1, 0, 1]
    ORDER_STATES = [-2, -1, 0]
    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:
        content = r.json()
        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() * 1000000)
        # Bitcoin.de API URI
        self.apihost = 'https://api.bitcoin.de'
        self.apiversion = 'v4'
        self.apibase = f'{self.apihost}/{self.apiversion}/'

    def build_hmac_sign(self, md5string, method, url):
        hmac_data = '#'.join([method, url, self.api_key, str(self.nonce), 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 = int(time.time() * 1000000)
        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(parse_float=decimal.Decimal)
            else:
                result = {}
        except requests.exceptions.RequestException as e:
            HandleRequestsException(e)
            result = {}
        return result

    def addToAddressPool(self, currency, address, **args):
        """Add address to pool"""
        uri = f'{self.apibase}{currency}/address'
        params = {'address': address}
        params.update(args)
        avail_params = ['address', 'amount_usages', 'comment']
        p = ParameterBuilder(avail_params, params, uri)
        return self.APIConnect('POST', p)

    def removeFromAddressPool(self, currency, address):
        """Remove address from pool"""
        uri = f'{self.apibase}{currency}/address/{address}'
        params = {'currency': currency, 'address': address}
        avail_params = ['currency', 'address']
        p = ParameterBuilder({}, {}, uri)
        p.verify_keys_and_values(avail_params, params)
        return self.APIConnect('DELETE', p)

    def listAddressPool(self, currency, **args):
        """List address pool"""
        uri = f'{self.apibase}{currency}/address'
        params = args
        avail_params = ['usable', 'comment', 'page']
        p = ParameterBuilder(avail_params, params, uri)
        return self.APIConnect('GET', p)

    def showOrderbook(self, order_type, trading_pair, **args):
        """Search Orderbook for offers."""
        uri = f'{self.apibase}{trading_pair}/orderbook'
        params = {'type': order_type}
        params.update(args)
        avail_params = ['type', 'trading_pair', 'amount_currency_to_trade', 'price',
                        'order_requirements_fullfilled',
                        'only_kyc_full', 'only_express_orders', 'payment_option',
                        'sepa_option', 'only_same_bankgroup', 'only_same_bic',
                        'seat_of_bank', 'page_size']
        p = ParameterBuilder(avail_params, params, uri)
        return self.APIConnect('GET', p)

    def showOrderDetails(self, trading_pair, order_id):
        """Show details for an offer."""
        uri = f'{self.apibase}{trading_pair}/orders/public/details/{order_id}'
        params = {'trading_pair': trading_pair, 'order_id': order_id}
        avail_params = ['trading_pair', 'order_id']
        p = ParameterBuilder({}, {}, uri)
        p.verify_keys_and_values(avail_params, params)
        return self.APIConnect('GET', p)

    def createOrder(self, order_type, trading_pair, max_amount_currency_to_trade, price, **args):
        """Create a new Order."""
        uri = f'{self.apibase}{trading_pair}/orders'
        # Build parameters
        params = {'type': order_type,
                  'max_amount_currency_to_trade': max_amount_currency_to_trade,
                  'price': price}
        params.update(args)
        avail_params = ['type', 'max_amount_currency_to_trade', 'price',
                        'min_amount_currency_to_trade', 'end_datetime',
                        'new_order_for_remaining_amount', 'trading_pair',
                        'min_trust_level', 'only_kyc_full', 'payment_option',
                        'sepa_option', 'seat_of_bank']
        p = ParameterBuilder(avail_params, params, uri)
        p.verify_keys_and_values(avail_params, {'trading_pair': trading_pair})
        return self.APIConnect('POST', p)

    def deleteOrder(self, order_id, trading_pair):
        """Delete an Order."""
        # Build parameters
        uri = f'{self.apibase}{trading_pair}/orders/{order_id}'
        avail_params = ['order_id', 'trading_pair']
        params = { 'order_id': order_id, 'trading_pair': trading_pair}
        p = ParameterBuilder({}, {}, uri)
        p.verify_keys_and_values(avail_params, params)
        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']
        if params.get("trading_pair"):
            uri = f'{self.apibase}{params["trading_pair"]}/orders'
            del params["trading_pair"]
        else:
            uri = f'{self.apibase}orders'
        p = ParameterBuilder(avail_params, params, uri)
        return self.APIConnect('GET', p)

    def showMyOrderDetails(self, trading_pair, order_id):
        """Details to an own Order."""
        uri = f'{self.apibase}{trading_pair}/orders/{order_id}'
        p = ParameterBuilder({}, {}, uri)
        return self.APIConnect('GET', p)

    def executeTrade(self, trading_pair, order_id, order_type, amount):
        """Buy/Sell on a specific Order."""
        uri = f'{self.apibase}{trading_pair}/trades/{order_id}'
        params = { 'type': order_type,
                   'amount_currency_to_trade': amount}
        avail_params = ['type', 'amount_currency_to_trade']
        p = ParameterBuilder(avail_params, params, uri)
        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',
                        'only_trades_with_action_for_payment_or_transfer_required',
                        'payment_method', 'date_start', 'date_end', 'page']
        if params.get("trading_pair"):
            uri = f'{self.apibase}{params["trading_pair"]}/trades'
            del params["trading_pair"]
        else:
            uri = f'{self.apibase}trades'
        p = ParameterBuilder(avail_params, params, uri)
        return self.APIConnect('GET', p)

    def showMyTradeDetails(self, trading_pair, trade_id):
        """Details to a specific Trade."""
        params = {'trading_pair': trading_pair, 'trade_id': trade_id}
        avail_params = [ 'trading_pair', 'trade_id' ]
        uri = f'{self.apibase}{trading_pair}/trades/{trade_id}'
        p = ParameterBuilder({}, {}, uri)
        p.verify_keys_and_values(avail_params, params)
        return self.APIConnect('GET', p)

    def markCoinsAsTransferred(self, trading_pair, trade_id, amount_currency_to_trade_after_fee):
        """Mark trade as transferred."""
        params = {'amount_currency_to_trade_after_fee': amount_currency_to_trade_after_fee,
                  'trading_pair': trading_pair, 'trade_id': trade_id}

        avail_params = [ 'trading_pair', 'trade_id', 'amount_currency_to_trade_after_fee' ]
        uri = f'{self.apibase}{trading_pair}/trades/{trade_id}/mark_coins_as_transferred'
        p = ParameterBuilder(avail_params,
            {'amount_currency_to_trade_after_fee': amount_currency_to_trade_after_fee}, uri)
        p.verify_keys_and_values(avail_params, params)
        return self.APIConnect('POST', p)

    def markTradeAsPaid(self, trading_pair, trade_id, volume_currency_to_pay_after_fee):
        """Mark traded as paid."""
        params = {'volume_currency_to_pay_after_fee': volume_currency_to_pay_after_fee,
                  'trading_pair': trading_pair, 'trade_id': trade_id}

        avail_params = [ 'trading_pair', 'trade_id', 'volume_currency_to_pay_after_fee' ]
        uri = f'{self.apibase}{trading_pair}/trades/{trade_id}/mark_trade_as_paid'
        p = ParameterBuilder(avail_params,
            {'volume_currency_to_pay_after_fee': volume_currency_to_pay_after_fee}, uri)
        p.verify_keys_and_values(avail_params, params)
        return self.APIConnect('POST', p)

    def markCoinsAsReceived(self, trading_pair, trade_id, amount_currency_to_trade_after_fee, rating):
        """Mark coins as received."""
        params = {'amount_currency_to_trade_after_fee': amount_currency_to_trade_after_fee,
                  'trading_pair': trading_pair, 'trade_id': trade_id, 'rating': rating}
        params_post = {'amount_currency_to_trade_after_fee': amount_currency_to_trade_after_fee,
                       'rating': rating}
        avail_params = [ 'trading_pair', 'trade_id', 'amount_currency_to_trade_after_fee', 'rating' ]
        uri = f'{self.apibase}{trading_pair}/trades/{trade_id}/mark_coins_as_received'
        p = ParameterBuilder(avail_params, params_post, uri)
        p.verify_keys_and_values(avail_params, params)
        return self.APIConnect('POST', p)

    def markTradeAsPaymentReceived(self, trading_pair, trade_id,
                                   volume_currency_to_pay_after_fee, rating,
                                   is_paid_from_correct_bank_account):
        """Mark coins as received."""
        params = {'volume_currency_to_pay_after_fee': volume_currency_to_pay_after_fee,
                  'trading_pair': trading_pair, 'trade_id': trade_id, 'rating': rating}
        params_post = {'volume_currency_to_pay_after_fee': volume_currency_to_pay_after_fee,
                       'rating': rating,
                       'is_paid_from_correct_bank_account': is_paid_from_correct_bank_account}
        avail_params = [ 'trading_pair', 'trade_id', 'volume_currency_to_pay_after_fee',
                         'rating', 'is_paid_from_correct_bank_account' ]
        uri = f'{self.apibase}{trading_pair}/trades/{trade_id}/mark_trade_as_payment_received'
        p = ParameterBuilder(avail_params, params_post, uri)
        p.verify_keys_and_values(avail_params, params)
        return self.APIConnect('POST', p)

    def addTradeRating(self, trading_pair, trade_id, rating):
        """Mark coins as received."""
        params = {'trading_pair': trading_pair, 'trade_id': trade_id, 'rating': rating}
        params_post = {'rating': rating}
        avail_params = [ 'trading_pair', 'trade_id', 'rating' ]
        uri = f'{self.apibase}{trading_pair}/trades/{trade_id}/add_trade_rating'
        p = ParameterBuilder(avail_params, params_post, uri)
        p.verify_keys_and_values(avail_params, params)
        return self.APIConnect('POST', p)

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

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

    def showPublicTradeHistory(self, trading_pair, **args):
        """All successful trades of the last 24 hours."""
        params = { 'trading_pair': trading_pair }
        params.update(args)
        avail_params = ['trading_pair', 'since_tid']
        uri = f'{self.apibase}{trading_pair}/trades/history'
        if params.get('since_tid'):
            del params["trading_pair"]
            p = ParameterBuilder(avail_params, params, uri)
        else:
            p = ParameterBuilder({}, {}, uri)
        p.verify_keys_and_values(avail_params, params)
        return self.APIConnect('GET', p)

    def showRates(self, trading_pair):
        """Query of the average rate last 3 and 12 hours."""
        uri = f'{self.apibase}{trading_pair}/rates'
        params = {'trading_pair': trading_pair}
        avail_params = ['trading_pair']
        # Build parameters
        p = ParameterBuilder({}, {}, uri)
        p.verify_keys_and_values(avail_params, params)
        return self.APIConnect('GET', p)

    def showAccountLedger(self, currency, **args):
        """Query on Account statement."""
        params = {'currency': currency}
        params.update(args)
        uri = f'{self.apibase}{currency}/account/ledger'
        avail_params = ['currency', 'type',
                        'datetime_start', 'datetime_end', 'page']
        p = ParameterBuilder(avail_params, params, uri)
        del params['currency']
        p = ParameterBuilder(avail_params, params, uri)
        return self.APIConnect('GET', p)

    def showPermissions(self):
        """Show permissions that are allowed for used API key"""
        uri = f'{self.apibase}permissions'
        p = ParameterBuilder({}, {}, uri)
        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			import inspect
13			import urllib
14
15			from urllib.parse import urlencode
16
17			logging.basicConfig()
18			log = logging.getLogger(__name__)
19			requests_log = logging.getLogger("requests.packages.urllib3")
20			requests_log.propagate = True
21
22			__version__ = '4.0'
23
24			class ParameterBuilder(object):
25			'''To verify given parameters for API.'''
26			def __init__(self, avail_params, given_params, uri):
27			self.verify_keys_and_values(avail_params, given_params)
28			self.params = given_params
29			self.create_url(uri)
30
31			def verify_keys_and_values(self, avail_params, given_params):
32			for k, v in given_params.items():
33			if k not in avail_params:
34			list_string = ', '.join(avail_params)
35			raise KeyError("{} is not any of {}".format(k, list_string))
36			if k == 'trading_pair':
37			self.error_on_invalid_value(v, self.TRADING_PAIRS)
38			elif k == 'type':
39			self.error_on_invalid_value(v, self.TRADE_TYPES)
40			elif k == 'currency':
41			self.error_on_invalid_value(v, self.CURRENCIES)
42			elif k == 'seat_of_bank':
43			self.error_on_invalid_value(v, self.BANK_SEATS)
44			elif k in ['min_trust_level', 'trust_level']:
45			self.error_on_invalid_value(v, self.TRUST_LEVELS)
46			elif k == 'payment_option':
47			self.error_on_invalid_value(v, self.PAYMENT_OPTIONS)
48			elif k == 'state':
49			caller = inspect.stack()[2][3]
50			if caller in ["showMyOrders", "showMyOrderDetails"]:
51			self.error_on_invalid_value(v, self.ORDER_STATES)
52			elif caller in ["showMyTrades", "showMyTradesDetails"]:
53			self.error_on_invalid_value(v, self.TRADE_STATES)
54
55			def error_on_invalid_value(self, value, list):
56			if value not in list:
57			list_string = ', '.join(str(x) for x in list)
58			raise ValueError("{} is not any of {}".format(value, list_string))
59
60			def create_url(self, uri):
61			if self.params:
62			self.encoded_string = urlencode(sorted(self.params.items()))
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', 'btgeur', 'bsveur', 'ltceur',
70			'iotabtc', 'dashbtc', 'gntbtc', 'ltcbtc']
71			ORDER_TYPES = ['buy', 'sell']
72			CURRENCIES = ['btc', 'bch', 'eth', 'btg', 'bsv', 'ltc',
73			'iota', 'dash', 'gnt']
74			BANK_SEATS = ['AT', 'BE', 'BG', 'CH', 'CY', 'CZ',
75			'DE', 'DK', 'EE', 'ES', 'FI', 'FR',
76			'GB', 'GR', 'HR', 'HU', 'IE', 'IS',
77			'IT', 'LI', 'LT', 'LU', 'LV', 'MT',
78			'MQ', 'NL', 'NO', 'PL', 'PT', 'RO',
79			'SE', 'SI', 'SK']
80			TRUST_LEVELS = ['bronze', 'silver', 'gold', 'platin']
81			TRADE_STATES = [-1, 0, 1]
82			ORDER_STATES = [-2, -1, 0]
83			PAYMENT_OPTIONS = [1, 2, 3]
84			TRADE_TYPES = ['all', 'buy', 'sell', 'inpayment',
85			'payout', 'affiliate', 'welcome_btc',
86			'buy_yubikey', 'buy_goldshop',
87			'buy_diamondshop', 'kickback',
88			'outgoing_fee_voluntary']
89
90			def HandleRequestsException(e):
91			"""Handle Exception from request."""
92			log.warning(e)
93
94
95			def HandleAPIErrors(r):
96			"""To handle Errors from BTCDE API."""
97			valid_status_codes = [200, 201, 204]
98			if r.status_code not in valid_status_codes:
99			content = r.json()
100			errors = content.get('errors')
101			log.warning('API Error Code: {}'.format(str(errors[0]['code'])))
102			log.warning('API Error Message: {}'.format(errors[0]['message']))
103			log.warning('API Error URL: {}'.format(r.url))
104			return False
105			else:
106			return True
107
108			class Connection(object):
109			"""To provide connection credentials to the trading API"""
110			def __init__(self, api_key, api_secret):
111			self.api_key = api_key
112			self.api_secret = api_secret
113			# set initial self.nonce
114			self.nonce = int(time.time() * 1000000)
115			# Bitcoin.de API URI
116			self.apihost = 'https://api.bitcoin.de'
117			self.apiversion = 'v4'
118			self.apibase = f'{self.apihost}/{self.apiversion}/'
119
120			def build_hmac_sign(self, md5string, method, url):
121			hmac_data = '#'.join([method, url, self.api_key, str(self.nonce), md5string])
122			hmac_signed = hmac.new(bytearray(self.api_secret.encode()), msg=hmac_data.encode(), digestmod=hashlib.sha256).hexdigest()
123			return hmac_signed
124
125			def set_header(self, url, method, encoded_string):
126			# raise self.nonce before using
127			self.nonce = int(time.time() * 1000000)
128			if method == 'POST':
129			md5_encoded_query_string = hashlib.md5(encoded_string.encode()).hexdigest()
130			else:
131			md5_encoded_query_string = hashlib.md5(b'').hexdigest()
132			hmac_signed = self.build_hmac_sign(md5_encoded_query_string,
133			method, url)
134			# set header
135			header = {'content-type':
136			'application/x-www-form-urlencoded; charset=utf-8',
137			'X-API-KEY': self.api_key,
138			'X-API-NONCE': str(self.nonce),
139			'X-API-SIGNATURE': hmac_signed }
140			return header
141
142			def send_request(self, url, method, header, encoded_string):
143			if method == 'GET':
144			r = requests.get(url, headers=(header),
145			stream=True, verify=False)
146			elif method == 'POST':
147			r = requests.post(url, headers=(header), data=encoded_string,
148			stream=True, verify=False)
149			elif method == 'DELETE':
150			r = requests.delete(url, headers=(header),
151			stream=True, verify=False)
152			return r
			0 ignored issues – show introduced 2018-11-28 19:52 UTC by Report Bug Copy Issue Report The variable `r` does not seem to be defined for all execution paths. Loading history...
153
154			def APIConnect(self, method, params):
155			"""Transform Parameters to URL"""
156			header = self.set_header(params.url, method,
157			params.encoded_string)
158			log.debug('Set Header: {}'.format(header))
159			try:
160			r = self.send_request(params.url, method, header,
161			params.encoded_string)
162			# Handle API Errors
163			if HandleAPIErrors(r):
164			# get results
165			result = r.json(parse_float=decimal.Decimal)
166			else:
167			result = {}
168			except requests.exceptions.RequestException as e:
169			HandleRequestsException(e)
170			result = {}
171			return result
172
173			def addToAddressPool(self, currency, address, **args):
174			"""Add address to pool"""
175			uri = f'{self.apibase}{currency}/address'
176			params = {'address': address}
177			params.update(args)
178			avail_params = ['address', 'amount_usages', 'comment']
179			p = ParameterBuilder(avail_params, params, uri)
180			return self.APIConnect('POST', p)
181
182			def removeFromAddressPool(self, currency, address):
183			"""Remove address from pool"""
184			uri = f'{self.apibase}{currency}/address/{address}'
185			params = {'currency': currency, 'address': address}
186			avail_params = ['currency', 'address']
187			p = ParameterBuilder({}, {}, uri)
188			p.verify_keys_and_values(avail_params, params)
189			return self.APIConnect('DELETE', p)
190
191			def listAddressPool(self, currency, **args):
192			"""List address pool"""
193			uri = f'{self.apibase}{currency}/address'
194			params = args
195			avail_params = ['usable', 'comment', 'page']
196			p = ParameterBuilder(avail_params, params, uri)
197			return self.APIConnect('GET', p)
198
199			def showOrderbook(self, order_type, trading_pair, **args):
200			"""Search Orderbook for offers."""
201			uri = f'{self.apibase}{trading_pair}/orderbook'
202			params = {'type': order_type}
203			params.update(args)
204			avail_params = ['type', 'trading_pair', 'amount_currency_to_trade', 'price',
205			'order_requirements_fullfilled',
206			'only_kyc_full', 'only_express_orders', 'payment_option',
207			'sepa_option', 'only_same_bankgroup', 'only_same_bic',
208			'seat_of_bank', 'page_size']
209			p = ParameterBuilder(avail_params, params, uri)
210			return self.APIConnect('GET', p)
211
212			def showOrderDetails(self, trading_pair, order_id):
213			"""Show details for an offer."""
214			uri = f'{self.apibase}{trading_pair}/orders/public/details/{order_id}'
215			params = {'trading_pair': trading_pair, 'order_id': order_id}
216			avail_params = ['trading_pair', 'order_id']
217			p = ParameterBuilder({}, {}, uri)
218			p.verify_keys_and_values(avail_params, params)
219			return self.APIConnect('GET', p)
220
221			def createOrder(self, order_type, trading_pair, max_amount_currency_to_trade, price, **args):
222			"""Create a new Order."""
223			uri = f'{self.apibase}{trading_pair}/orders'
224			# Build parameters
225			params = {'type': order_type,
226			'max_amount_currency_to_trade': max_amount_currency_to_trade,
227			'price': price}
228			params.update(args)
229			avail_params = ['type', 'max_amount_currency_to_trade', 'price',
230			'min_amount_currency_to_trade', 'end_datetime',
231			'new_order_for_remaining_amount', 'trading_pair',
232			'min_trust_level', 'only_kyc_full', 'payment_option',
233			'sepa_option', 'seat_of_bank']
234			p = ParameterBuilder(avail_params, params, uri)
235			p.verify_keys_and_values(avail_params, {'trading_pair': trading_pair})
236			return self.APIConnect('POST', p)
237
238			def deleteOrder(self, order_id, trading_pair):
239			"""Delete an Order."""
240			# Build parameters
241			uri = f'{self.apibase}{trading_pair}/orders/{order_id}'
242			avail_params = ['order_id', 'trading_pair']
243			params = { 'order_id': order_id, 'trading_pair': trading_pair}
244			p = ParameterBuilder({}, {}, uri)
245			p.verify_keys_and_values(avail_params, params)
246			return self.APIConnect('DELETE', p)
247
248			def showMyOrders(self, **args):
249			"""Query and Filter own Orders."""
250			# Build parameters
251			params = args
252			avail_params = ['type', 'trading_pair', 'state',
253			'date_start', 'date_end', 'page']
254			if params.get("trading_pair"):
255			uri = f'{self.apibase}{params["trading_pair"]}/orders'
256			del params["trading_pair"]
257			else:
258			uri = f'{self.apibase}orders'
259			p = ParameterBuilder(avail_params, params, uri)
260			return self.APIConnect('GET', p)
261
262			def showMyOrderDetails(self, trading_pair, order_id):
263			"""Details to an own Order."""
264			uri = f'{self.apibase}{trading_pair}/orders/{order_id}'
265			p = ParameterBuilder({}, {}, uri)
266			return self.APIConnect('GET', p)
267
268			def executeTrade(self, trading_pair, order_id, order_type, amount):
269			"""Buy/Sell on a specific Order."""
270			uri = f'{self.apibase}{trading_pair}/trades/{order_id}'
271			params = { 'type': order_type,
272			'amount_currency_to_trade': amount}
273			avail_params = ['type', 'amount_currency_to_trade']
274			p = ParameterBuilder(avail_params, params, uri)
275			return self.APIConnect('POST', p)
276
277			def showMyTrades(self, **args):
278			"""Query and Filter on past Trades."""
279			# Build parameters
280			params = args
281			avail_params = ['type', 'trading_pair', 'state',
282			'only_trades_with_action_for_payment_or_transfer_required',
283			'payment_method', 'date_start', 'date_end', 'page']
284			if params.get("trading_pair"):
285			uri = f'{self.apibase}{params["trading_pair"]}/trades'
286			del params["trading_pair"]
287			else:
288			uri = f'{self.apibase}trades'
289			p = ParameterBuilder(avail_params, params, uri)
290			return self.APIConnect('GET', p)
291
292			def showMyTradeDetails(self, trading_pair, trade_id):
293			"""Details to a specific Trade."""
294			params = {'trading_pair': trading_pair, 'trade_id': trade_id}
295			avail_params = [ 'trading_pair', 'trade_id' ]
296			uri = f'{self.apibase}{trading_pair}/trades/{trade_id}'
297			p = ParameterBuilder({}, {}, uri)
298			p.verify_keys_and_values(avail_params, params)
299			return self.APIConnect('GET', p)
300
301			def markCoinsAsTransferred(self, trading_pair, trade_id, amount_currency_to_trade_after_fee):
302			"""Mark trade as transferred."""
303			params = {'amount_currency_to_trade_after_fee': amount_currency_to_trade_after_fee,
304			'trading_pair': trading_pair, 'trade_id': trade_id}
305
306			avail_params = [ 'trading_pair', 'trade_id', 'amount_currency_to_trade_after_fee' ]
307			uri = f'{self.apibase}{trading_pair}/trades/{trade_id}/mark_coins_as_transferred'
308			p = ParameterBuilder(avail_params,
309			{'amount_currency_to_trade_after_fee': amount_currency_to_trade_after_fee}, uri)
310			p.verify_keys_and_values(avail_params, params)
311			return self.APIConnect('POST', p)
312
313			def markTradeAsPaid(self, trading_pair, trade_id, volume_currency_to_pay_after_fee):
314			"""Mark traded as paid."""
315			params = {'volume_currency_to_pay_after_fee': volume_currency_to_pay_after_fee,
316			'trading_pair': trading_pair, 'trade_id': trade_id}
317
318			avail_params = [ 'trading_pair', 'trade_id', 'volume_currency_to_pay_after_fee' ]
319			uri = f'{self.apibase}{trading_pair}/trades/{trade_id}/mark_trade_as_paid'
320			p = ParameterBuilder(avail_params,
321			{'volume_currency_to_pay_after_fee': volume_currency_to_pay_after_fee}, uri)
322			p.verify_keys_and_values(avail_params, params)
323			return self.APIConnect('POST', p)
324
325			def markCoinsAsReceived(self, trading_pair, trade_id, amount_currency_to_trade_after_fee, rating):
326			"""Mark coins as received."""
327			params = {'amount_currency_to_trade_after_fee': amount_currency_to_trade_after_fee,
328			'trading_pair': trading_pair, 'trade_id': trade_id, 'rating': rating}
329			params_post = {'amount_currency_to_trade_after_fee': amount_currency_to_trade_after_fee,
330			'rating': rating}
331			avail_params = [ 'trading_pair', 'trade_id', 'amount_currency_to_trade_after_fee', 'rating' ]
332			uri = f'{self.apibase}{trading_pair}/trades/{trade_id}/mark_coins_as_received'
333			p = ParameterBuilder(avail_params, params_post, uri)
334			p.verify_keys_and_values(avail_params, params)
335			return self.APIConnect('POST', p)
336
337			def markTradeAsPaymentReceived(self, trading_pair, trade_id,
338			volume_currency_to_pay_after_fee, rating,
339			is_paid_from_correct_bank_account):
340			"""Mark coins as received."""
341			params = {'volume_currency_to_pay_after_fee': volume_currency_to_pay_after_fee,
342			'trading_pair': trading_pair, 'trade_id': trade_id, 'rating': rating}
343			params_post = {'volume_currency_to_pay_after_fee': volume_currency_to_pay_after_fee,
344			'rating': rating,
345			'is_paid_from_correct_bank_account': is_paid_from_correct_bank_account}
346			avail_params = [ 'trading_pair', 'trade_id', 'volume_currency_to_pay_after_fee',
347			'rating', 'is_paid_from_correct_bank_account' ]
348			uri = f'{self.apibase}{trading_pair}/trades/{trade_id}/mark_trade_as_payment_received'
349			p = ParameterBuilder(avail_params, params_post, uri)
350			p.verify_keys_and_values(avail_params, params)
351			return self.APIConnect('POST', p)
352
353			def addTradeRating(self, trading_pair, trade_id, rating):
354			"""Mark coins as received."""
355			params = {'trading_pair': trading_pair, 'trade_id': trade_id, 'rating': rating}
356			params_post = {'rating': rating}
357			avail_params = [ 'trading_pair', 'trade_id', 'rating' ]
358			uri = f'{self.apibase}{trading_pair}/trades/{trade_id}/add_trade_rating'
359			p = ParameterBuilder(avail_params, params_post, uri)
360			p.verify_keys_and_values(avail_params, params)
361			return self.APIConnect('POST', p)
362
363			def showAccountInfo(self):
364			"""Query on Account Infos."""
365			uri = f'{self.apibase}account'
366			p = ParameterBuilder({}, {}, uri)
367			return self.APIConnect('GET', p)
368
369			def showOrderbookCompact(self, trading_pair):
370			"""Bids and Asks in compact format."""
371			params = {'trading_pair': trading_pair}
372			avail_params = ['trading_pair']
373			uri = f'{self.apibase}{trading_pair}/orderbook/compact'
374			# Build parameters
375			p = ParameterBuilder({}, {}, uri)
376			p.verify_keys_and_values(avail_params, params)
377			return self.APIConnect('GET', p)
378
379			def showPublicTradeHistory(self, trading_pair, **args):
380			"""All successful trades of the last 24 hours."""
381			params = { 'trading_pair': trading_pair }
382			params.update(args)
383			avail_params = ['trading_pair', 'since_tid']
384			uri = f'{self.apibase}{trading_pair}/trades/history'
385			if params.get('since_tid'):
386			del params["trading_pair"]
387			p = ParameterBuilder(avail_params, params, uri)
388			else:
389			p = ParameterBuilder({}, {}, uri)
390			p.verify_keys_and_values(avail_params, params)
391			return self.APIConnect('GET', p)
392
393			def showRates(self, trading_pair):
394			"""Query of the average rate last 3 and 12 hours."""
395			uri = f'{self.apibase}{trading_pair}/rates'
396			params = {'trading_pair': trading_pair}
397			avail_params = ['trading_pair']
398			# Build parameters
399			p = ParameterBuilder({}, {}, uri)
400			p.verify_keys_and_values(avail_params, params)
401			return self.APIConnect('GET', p)
402
403			def showAccountLedger(self, currency, **args):
404			"""Query on Account statement."""
405			params = {'currency': currency}
406			params.update(args)
407			uri = f'{self.apibase}{currency}/account/ledger'
408			avail_params = ['currency', 'type',
409			'datetime_start', 'datetime_end', 'page']
410			p = ParameterBuilder(avail_params, params, uri)
411			del params['currency']
412			p = ParameterBuilder(avail_params, params, uri)
413			return self.APIConnect('GET', p)
414
415			def showPermissions(self):
416			"""Show permissions that are allowed for used API key"""
417			uri = f'{self.apibase}permissions'
418			p = ParameterBuilder({}, {}, uri)
419			return self.APIConnect('GET', p)

peshay / btcde

Push — master ( 140199...ad2fe1 )

btcde.Connection.listAddressPool() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like