-
Notifications
You must be signed in to change notification settings - Fork 65
/
Copy pathcoinigy_api_rest.py
184 lines (152 loc) · 6.72 KB
/
coinigy_api_rest.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
from collections import namedtuple
import numpy as np
import pandas as pd
import requests
credentials = namedtuple('credentials', ('api', 'secret', 'endpoint'))
connection = namedtuple('connection', ('hostname', 'port', 'secure'))
alerts = namedtuple('alerts', ('open_alerts', 'alert_history'))
class CoinigyREST:
"""
This class implements coinigy's REST api as documented in the documentation
available at
https://github.com/coinigy/api
"""
def __init__(self, acct):
self.api = acct.api
self.secret = acct.secret
self.endpoint = acct.endpoint
def request(self, method, query=None, json=False, **args):
"""
Generic interface to REST api
:param method: query name
:param query: dictionary of inputs
:param json: if True return the raw results in json format
:param args: keyword arguments added to the payload
:return:
"""
url = '{endpoint}/{method}'.format(endpoint=self.endpoint, method=method)
authAPI = {'X-API-KEY': self.api, 'X-API-SECRET': self.secret}
payload = {}
payload.update(**args)
if query is not None:
payload.update(query)
r = requests.post(url, data=payload, headers=authAPI)
if 'error' in r.json().keys():
print(r.json()['error'])
return
if json:
return r.json()
return pd.DataFrame(r.json()['data'])
def data(self, exchange, market, data_type):
"""
Common wrapper for data related queries
:param exchange:
:param market:
:param data_type: currently supported are 'history', 'bids', 'asks', 'orders'
:return:
"""
d = self.request('data', exchange_code=exchange, exchange_market=market, type=data_type, json=True)['data']
res = dict()
for key in ['history', 'bids', 'asks']:
if key in d.keys():
dat = pd.DataFrame.from_records(d[key])
if 'price' in dat.columns:
dat.price = dat.price.astype(np.float)
if 'quantity' in dat.columns:
dat.quantity = dat.quantity.astype(np.float)
if 'total' in dat.columns:
dat.total = dat.total.astype(np.float)
if 'time_local' in dat.columns:
dat.time_local = pd.to_datetime(dat.time_local, format='%Y-%m-%d %H:%M:%S')
dat.set_index('time_local', inplace=True)
if 'type' in dat.columns:
dat.type = dat.type.astype(str)
if not dat.empty:
dat['base_ccy'] = d['primary_curr_code']
dat['counter_ccy'] = d['secondary_curr_code']
res[key] = dat
return res
def accounts(self):
return self.request('accounts')
def activity(self):
return self.request('activity')
def balances(self):
return self.request('balances')
def push_notifications(self):
return self.request('pushNotifications')
def open_orders(self):
"""
FIXME: untested
"""
return self.request('orders', json=True)
def alerts(self):
all_alerts = self.request('alerts', json=True)['data']
open_alerts = pd.DataFrame(all_alerts['open_alerts'])
alert_history = pd.DataFrame(all_alerts['alert_history'])
return alerts(open_alerts=open_alerts, alert_history=alert_history)
def exchanges(self):
return self.request('exchanges')
def markets(self, exchange):
return self.request('markets',exchange_code=exchange)
def history(self, exchange, market):
return self.data(exchange=exchange, market=market, data_type='history')['history']
def asks(self, exchange, market):
return self.data(exchange=exchange, market=market, data_type='asks')['asks']
def bids(self, exchange, market):
return self.data(exchange=exchange, market=market, data_type='bids')['bids']
def orders(self, exchange, market):
return self.data(exchange=exchange, market=market, data_type='orders')
def news_feed(self):
dat = self.request('newsFeed')
dat.timestamp = pd.to_datetime(dat.timestamp)
dat.set_index('timestamp', inplace=True)
return dat
def order_types(self):
dat = self.request('orderTypes', json=True)['data']
return dict(order_types=pd.DataFrame.from_records(dat['order_types']),
price_types=pd.DataFrame.from_records(dat['price_types']))
def refresh_balance(self):
return self.request('refreshBalance', json=True)
def add_alert(self, exchange, market, price, note):
return self.request('addAlert',
exch_code=exchange,
market_name=market,
alert_price=price,
alert_note=note,
json=True)['notifications']
def delete_alert(self, alert_id):
return self.request('deleteAlert', alert_id=alert_id, json=True)['notifications']
def add_order(self, auth_id, exch_id, mkt_id, order_type_id, price_type_id, limit_price, stop_price, order_quantity):
"""
FIXME: untested
"""
return self.request('addOrder',
auth_id = auth_id,
exch_id=exch_id,
mkt_id=mkt_id,
order_type_id=order_type_id,
price_type_id=price_type_id,
limit_price=limit_price,
stop_price=stop_price,
order_quantity=order_quantity,
json=True)
def cancel_order(self, order_id):
return self.request('cancelOrder', internal_order_id=order_id, json=True)
def balance_history(self, date):
"""
NB: the timestamp columns is the time when the account was last snapshot, not the time the balances were
effectively refreshed
:param date: date str in format YYYY-MM-DD
:return: a view of the acccount balances as of the date provided
"""
bh = pd.DataFrame.from_records(self.request('balanceHistory', date=date, json=True)['data']['balance_history'])
if bh.empty:
return bh
acct = self.accounts()[['auth_id', 'exch_name']]
return pd.merge(bh, acct, on='auth_id', how='left')
if __name__ == "__main__":
credentials.api = "actual/from_file_or_env"
credentials.secret = "actual/from_file_or_env"
credentials.endpoint = "https://api.coinigy.com/api/v1/"
cr = CoinigyREST(credentials)
print(cr.balances())