set_header() - Code Metrics - Inspection of "change links for master branch" - peshay/btcde - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 907443...d1611c )

by Andreas

created 2017-10-10 09:05 UTC

set_header() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
c	1
b	0
f	0
dl	0
loc	19
rs	9.4285

#! /usr/bin/env python
"""This Module provides the functions from bitcoin.de Trading API"""

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

# these two lines enable debugging at httplib level
# (requests->urllib3->httplib)
# you will see the REQUEST, including HEADERS and DATA, and RESPONSE with
# HEADERS but without DATA.
# the only thing missing will be the response.body which is not logged.
#import http.client
#http.client.HTTPConnection.debuglevel = 1


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

__version__ = '1.4'


class Connection:
    """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

# Bitcoin.de API URI
apihost = 'https://api.bitcoin.de'
apiversion = 'v1'
orderuri = apihost + '/' + apiversion + '/' + 'orders'
tradeuri = apihost + '/' + apiversion + '/' + 'trades'
accounturi = apihost + '/' + apiversion + '/' + 'account'
# set initial nonce
nonce = int(time.time())
# disable unsecure SSL warning
requests.packages.urllib3.disable_warnings()


def HandleRequestsException(e):
    """Handle Exception from request."""
    print(e[0][0])
    print(e[0][1])


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


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

def set_header(url):
    global nonce
    # raise nonce before using
    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_data = method + '#' + \
        url + '#' + conn.api_key + \
        '#' + str(nonce) + '#' + md5_encoded_query_string
    hmac_signed = hmac.new(bytearray(conn.api_secret.encode()), msg=hmac_data.encode(), digestmod=hashlib.sha256).hexdigest()
    # set header
    header = {'content-type':
              'application/x-www-form-urlencoded; charset=utf-8',
              'X-API-KEY': conn.api_key,
                   'X-API-NONCE': str(nonce),
                   'X-API-SIGNATURE': hmac_signed }
    return header
    
def send_request(method):
    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(conn, method, params, uri):
    """Transform Parameters to URL"""
    url = params_url(params, uri)
    header = set_header(url)
    try:
        r = send_request(method)
        # Handle API Errors
        if HandleAPIErrors(r):
            # get results
            result = r.json()
        else:
            result = {}
    except requests.exceptions.RequestException as e:
        HandleRequestsException(e)
    return result


def showOrderbook(conn, OrderType, **args):
    """Search Orderbook for offers."""
    # Build parameters
    if OrderType == 'buy' or OrderType == 'sell':
        params = {'type': OrderType}
    else:
        print('problem')
    params.update(args)
    return APIConnect(conn, 'GET', params, orderuri)


def createOrder(conn, OrderType, max_amount, price, **args):
    """Create a new Order."""
    # Build parameters
    params = {'type': OrderType, 'max_amount': max_amount, 'price': price}
    params.update(args)
    return APIConnect(conn, 'POST', params, orderuri)


def deleteOrder(conn, order_id):
    """Delete an Order."""
    newuri = orderuri + "/" + order_id
    params = {'order_id': order_id}
    return APIConnect(conn, 'DELETE', params, newuri)


def showMyOrders(conn, **args):
    """Query and Filter own Orders."""
    newuri = orderuri + '/my_own'
    return APIConnect(conn, 'GET', args, newuri)


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


def executeTrade(conn, order_id, OrderType, amount):
    """Buy/Sell on a specific Order."""
    newuri = tradeuri + '/' + order_id
    params = {'order_id': order_id, 'type': OrderType, 'amount': amount}
    return APIConnect(conn, 'POST', params, newuri)


def showMyTrades(conn, **args):
    """Query and Filter on past Trades."""
    return APIConnect(conn, 'GET', args, tradeuri)


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


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


def showOrderbookCompact(conn):
    """Bids and Asks in compact format."""
    params = {}
    return APIConnect(conn, 'GET', params, orderuri + '/compact')


def showPublicTradeHistory(conn, since_tid=None):
    """All successful trades of the las 7 days."""
    if since_tid is not None:
        params = {'since_tid': since_tid}
    else:
        params = {}
    return APIConnect(conn, 'GET', params, tradeuri + '/history')


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


def showAccountLedger(conn, **args):
    """Query on Account statement."""
    return APIConnect(conn, 'GET', args, accounturi + '/ledger')


1			#! /usr/bin/env python
2			"""This Module provides the functions from 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
12			# these two lines enable debugging at httplib level
13			# (requests->urllib3->httplib)
14			# you will see the REQUEST, including HEADERS and DATA, and RESPONSE with
15			# HEADERS but without DATA.
16			# the only thing missing will be the response.body which is not logged.
17			#import http.client
18			#http.client.HTTPConnection.debuglevel = 1
19
20
21			logging.basicConfig()
22			logging.getLogger().setLevel(logging.DEBUG)
23			requests_log = logging.getLogger("requests.packages.urllib3")
24			requests_log.setLevel(logging.DEBUG)
25			requests_log.propagate = True
26
27			__version__ = '1.4'
28
29
30			class Connection:
31			"""To provide connection credentials to the trading API"""
32			def __init__(self, api_key, api_secret):
33			self.api_key = api_key
34			self.api_secret = api_secret
35
36			# Bitcoin.de API URI
37			apihost = 'https://api.bitcoin.de'
38			apiversion = 'v1'
39			orderuri = apihost + '/' + apiversion + '/' + 'orders'
40			tradeuri = apihost + '/' + apiversion + '/' + 'trades'
41			accounturi = apihost + '/' + apiversion + '/' + 'account'
42			# set initial nonce
43			nonce = int(time.time())
44			# disable unsecure SSL warning
45			requests.packages.urllib3.disable_warnings()
46
47
48			def HandleRequestsException(e):
49			"""Handle Exception from request."""
50			print(e[0][0])
51			print(e[0][1])
52
53
54			def HandleAPIErrors(r):
55			"""To handle Errors from BTCDE API."""
56			valid_status_codes = [200, 201, 204]
57			if r.status_code in valid_status_codes:
58			reader = codecs.getreader("utf-8")
59			content = json.load(reader(r.raw))
60			errors = content.get('errors')
61			print('Code: ' + str(errors[0]['code']))
62			print('Message: ' + errors[0]['message'])
63			print('With URL: ' + r.url)
64			return False
65			else:
66			return True
67
68
69			def params_url(params, uri):
70			encoded_string = ''
71			if params:
72			for key, value in sorted(params.items()):
73			encoded_string += str(key) + '=' + str(value) + '&'
74			encoded_string = encoded_string[:-1]
75			url = uri + '?' + encoded_string
76			else:
77			url = uri
78			return url
79
80			def set_header(url):
81			global nonce
82			# raise nonce before using
83			nonce += 1
84			if method == 'POST':
85			md5_encoded_query_string = hashlib.md5(encoded_string.encode()).hexdigest()
86			else:
87			md5_encoded_query_string = hashlib.md5(b'').hexdigest()
88			hmac_data = method + '#' + \
89			url + '#' + conn.api_key + \
90			'#' + str(nonce) + '#' + md5_encoded_query_string
91			hmac_signed = hmac.new(bytearray(conn.api_secret.encode()), msg=hmac_data.encode(), digestmod=hashlib.sha256).hexdigest()
92			# set header
93			header = {'content-type':
94			'application/x-www-form-urlencoded; charset=utf-8',
95			'X-API-KEY': conn.api_key,
96			'X-API-NONCE': str(nonce),
97			'X-API-SIGNATURE': hmac_signed }
98			return header
99
100			def send_request(method):
101			if method == 'GET':
102			r = requests.get(url, headers=(header),
103			stream=True, verify=False)
104			elif method == 'POST':
105			r = requests.post(url, headers=(header), data=encoded_string,
106			stream=True, verify=False)
107			elif method == 'DELETE':
108			r = requests.delete(url, headers=(header),
109			stream=True, verify=False)
110			return r
111
112			def APIConnect(conn, method, params, uri):
113			"""Transform Parameters to URL"""
114			url = params_url(params, uri)
115			header = set_header(url)
116			try:
117			r = send_request(method)
118			# Handle API Errors
119			if HandleAPIErrors(r):
120			# get results
121			result = r.json()
122			else:
123			result = {}
124			except requests.exceptions.RequestException as e:
125			HandleRequestsException(e)
126			return result
127
128
129			def showOrderbook(conn, OrderType, **args):
130			"""Search Orderbook for offers."""
131			# Build parameters
132			if OrderType == 'buy' or OrderType == 'sell':
133			params = {'type': OrderType}
134			else:
135			print('problem')
136			params.update(args)
137			return APIConnect(conn, 'GET', params, orderuri)
138
139
140			def createOrder(conn, OrderType, max_amount, price, **args):
141			"""Create a new Order."""
142			# Build parameters
143			params = {'type': OrderType, 'max_amount': max_amount, 'price': price}
144			params.update(args)
145			return APIConnect(conn, 'POST', params, orderuri)
146
147
148			def deleteOrder(conn, order_id):
149			"""Delete an Order."""
150			newuri = orderuri + "/" + order_id
151			params = {'order_id': order_id}
152			return APIConnect(conn, 'DELETE', params, newuri)
153
154
155			def showMyOrders(conn, **args):
156			"""Query and Filter own Orders."""
157			newuri = orderuri + '/my_own'
158			return APIConnect(conn, 'GET', args, newuri)
159
160
161			def showMyOrderDetails(conn, order_id):
162			"""Details to an own Order."""
163			newuri = orderuri + '/' + order_id
164			params = {'order_id': order_id}
165			return APIConnect(conn, 'GET', params, newuri)
166
167
168			def executeTrade(conn, order_id, OrderType, amount):
169			"""Buy/Sell on a specific Order."""
170			newuri = tradeuri + '/' + order_id
171			params = {'order_id': order_id, 'type': OrderType, 'amount': amount}
172			return APIConnect(conn, 'POST', params, newuri)
173
174
175			def showMyTrades(conn, **args):
176			"""Query and Filter on past Trades."""
177			return APIConnect(conn, 'GET', args, tradeuri)
178
179
180			def showMyTradeDetails(conn, trade_id):
181			"""Details to a specific Trade."""
182			newuri = tradeuri + '/' + trade_id
183			params = {'trade_id': trade_id}
184			return APIConnect(conn, 'GET', params, newuri)
185
186
187			def showAccountInfo(conn):
188			"""Query on Account Infos."""
189			params = {}
190			return APIConnect(conn, 'GET', params, accounturi)
191
192
193			def showOrderbookCompact(conn):
194			"""Bids and Asks in compact format."""
195			params = {}
196			return APIConnect(conn, 'GET', params, orderuri + '/compact')
197
198
199			def showPublicTradeHistory(conn, since_tid=None):
200			"""All successful trades of the las 7 days."""
201			if since_tid is not None:
202			params = {'since_tid': since_tid}
203			else:
204			params = {}
205			return APIConnect(conn, 'GET', params, tradeuri + '/history')
206
207
208			def showRates(conn):
209			"""Query of the average rate last 3 and 12 hours."""
210			newuri = apihost + '/' + apiversion + '/rates'
211			params = {}
212			return APIConnect(conn, 'GET', params, newuri)
213
214
215			def showAccountLedger(conn, **args):
216			"""Query on Account statement."""
217			return APIConnect(conn, 'GET', args, accounturi + '/ledger')
218

peshay / btcde

Push — master ( 907443...d1611c )

set_header() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like