surugh 0 Опубликовано: 12 декабря 2020 Share Опубликовано: 12 декабря 2020 (изменено) Исходники GitHub Рецепты API v3 Graviex python 3.9 GET: import time import hmac import hashlib import requests from pprint import pprint ''' All private API requires 3 authentication parameters and zero or more API specific parameters. access_key Your access key. tonce tonce is a timestamp in integer, stands for milliseconds elapsed since Unix epoch. Tonce must be within 30 seconds of server's current time. Each tonce can only be used once. signature Signature of the API request, generated by you, use your secret key. if it useful you can make a donation GIO: GQksGKKU6PotrByC8hQzBRwMAN7FuMNugF ETH: 0x290fE67efA690AE7924DB416d4239daC9c309b97 ''' DEBUG = True # your API keys access_key = '' secret_key = b'' tonce = int(time.time()) * 1000 if DEBUG: print(f'tonce: {tonce}') # the request query sorted in alphabetic order, including access_key and tonce, # e.g. access_key=xxx&foo=bar&tonce=123456789 request = f'access_key={access_key}&tonce={tonce}' # HTTP verb like GET/POST in uppercase verb = 'GET' # request path like /webapi/v3/markets uri = '/webapi/v3/deposits' # The combined string looks like: GET|/webapi/v3/markets|access_key=xxx&foo=bar&tonce=123456789 message = f'{verb}|{uri}|{request}' if DEBUG: print(f'request: {request}\nmessage: {message}') signature = hmac.new( secret_key, message.encode(), hashlib.sha256 ).hexdigest() if DEBUG: print(f'signature: {signature}') query = f'https://graviex.net{uri}?{request}&signature={signature}' response = requests.get(query) if DEBUG: print(f'query: {query}\nresponse: {response}') content = response.json() pprint(f'content: {content}') POST: import time import hmac import hashlib import requests from pprint import pprint ''' All private API requires 3 authentication parameters and zero or more API specific parameters. access_key Your access key. tonce tonce is a timestamp in integer, stands for milliseconds elapsed since Unix epoch. Tonce must be within 30 seconds of server's current time. Each tonce can only be used once. signature Signature of the API request, generated by you, use your secret key. if it useful you can make a donation GIO: GQksGKKU6PotrByC8hQzBRwMAN7FuMNugF DOGE: D8Hy4stikB4Pk8rnSfhQxgdZ59GqmasdPj ETH: 0x290fE67efA690AE7924DB416d4239daC9c309b97 ''' DEBUG = True # your API keys access_key = '' secret_key = b'' tonce = str(int(time.time()) * 1000) if DEBUG: print(f'tonce: {tonce}') # the request query sorted in alphabetic order, including access_key and tonce, # e.g. access_key=xxx&foo=bar&tonce=123456789 request = f'access_key={access_key}&market=gioeth&price=0.00003&side=buy&tonce={tonce}&volume=2.0' # HTTP verb like GET/POST in uppercase verb = 'POST' # request path like /webapi/v3/markets uri = '/webapi/v3/orders' # The combined string looks like: GET|/webapi/v3/markets|access_key=xxx&foo=bar&tonce=123456789 message = f'{verb}|{uri}|{request}' if DEBUG: print(f'request: {request}\nmessage: {message}') signature = hmac.new( secret_key, message.encode(), hashlib.sha256 ).hexdigest() if DEBUG: print(f'signature: {signature}') query = f'https://graviex.net{uri}?{request}&signature={signature}' response = requests.post(query) if DEBUG: print(f'query: {query}\nresponse: {response}') content = response.json() pprint(f'content: {content}') Звонилка: Реализация: Все публичные методы вызываются через приватный апи, ключи обязательны! Переключатель GET/POST get=True/False import time import hmac import hashlib import requests """ WebSite https://graviex.net/desktop API v3 https://graviex.net/documents/api_v3 API Usage Policy https://graviex.net/documents/user-agreement if it useful you can make a donation GIO: GQksGKKU6PotrByC8hQzBRwMAN7FuMNugF ETH: 0x290fE67efA690AE7924DB416d4239daC9c309b97 """ DEBUG = True class GraviexAPI: def __init__(self, access_key=None, secret_key=None): self.access = access_key self.secret = secret_key self.uri_main = "/webapi/v3/" self.base_url = "https://graviex.net/" # or use # self.base_url = "https://graviex.net:443/" def call_api(self, uri_path, param_path=None, params=None, get=True): # create tonce & sorting parameters to make sure the request is correct time.sleep(2) # edit as needed tonce = int(time.time()) * 1000 if not params: params = {} params.update(dict(tonce=tonce, access_key=self.access)) params_keys = list(params.keys()) params_keys.sort() if DEBUG: print(f'params: {params}') # create {canonical_query} & {canonical_uri} request = '' for key in params_keys: value = params[key] request += f'{key}={value}&' if DEBUG: print(f'request: {request}') if param_path: uri = f'{self.uri_main}{uri_path}/{param_path}' else: uri = f'{self.uri_main}{uri_path}' if DEBUG: print(f'uri: {uri}') # create payload & signature (hash of the request) # with crop the last character "&" in request, otherwise the signature will be incorrect if get: verb = 'GET' else: verb = 'POST' message = f'{verb}|{uri}|{request[:-1]}' if DEBUG: print(f'message: {message}') if self.secret: signature = hmac.new( # if api Secret Key not in bytes replace self.secret to # bytes(self.secret, encoding='utf-8'), self.secret, message.encode(), hashlib.sha256 ).hexdigest() # now we have a signed request which can be used like url query = f'{self.base_url}{uri}?{request}signature={signature}' else: query = f'{self.base_url}{uri}' print(f'query: {query}') # send requests and get responses if get: response = requests.get(query) else: response = requests.post(query) return response.json() if __name__ == '__main__': # 6000 requests/keypair/5 minutes, you can contact GRAVIEX if you need more. gex_private = GraviexAPI( access_key='', secret_key=b'' ) market = 'giobtc' currency = 'gio' timestamp = int(time.time() - 1000000000000) * 1000 public_apis = [ {'path': 'markets', 'sub_path': None, 'params': {}}, {'path': 'markets', 'sub_path': market, 'params': {'market': market}}, {'path': 'tickers', 'sub_path': None, 'params': {}}, {'path': 'markets', 'sub_path': market, 'params': {'market': market}}, {'path': 'depth', 'sub_path': None, 'params': {'market': market, 'limit': 1, 'order': None}}, {'path': 'trades', 'sub_path': None, 'params': {'market': market, 'limit': 1, 'order_by': 'desc', 'timestamp': timestamp, 'from': timestamp, 'to': timestamp}}, {'path': 'trades_simple', 'sub_path': None, 'params': {'market': market}}, {'path': 'k', 'sub_path': None, 'params': {'market': market, 'limit': 1, 'period': 1, 'timestamp': timestamp}}, {'path': 'k_with_pending_trades', 'sub_path': None, 'params': {'market': market, 'trade_id': 1, 'limit': 1, 'period': 1, 'timestamp': timestamp}}, {'path': 'timestamp', 'sub_path': None, 'params': {}}, {'path': 'currency', 'sub_path': 'info', 'params': {'currency': currency}}, ] for api in public_apis: r = gex_private.call_api(api['path'], param_path=api['sub_path'], params=api['params']) print(f'api: {api}:\n{r}\n') Изменено 12 декабря 2020 пользователем surugh Цитата Ссылка на сообщение Поделиться на других сайтах
surugh 0 Опубликовано: 13 декабря 2020 Автор Share Опубликовано: 13 декабря 2020 public api python wrapper import requests ''' All private API requires 3 authentication parameters and zero or more API specific parameters. access_key Your access key. tonce tonce is a timestamp in integer, stands for milliseconds elapsed since Unix epoch. Tonce must be within 30 seconds of server's current time. Each tonce can only be used once. signature Signature of the API request, generated by you, use your secret key. if it useful you can make a donation GIO: GQksGKKU6PotrByC8hQzBRwMAN7FuMNugF DOGE: D8Hy4stikB4Pk8rnSfhQxgdZ59GqmasdPj ETH: 0x290fE67efA690AE7924DB416d4239daC9c309b97 ''' class GraviexPublicAPI: def __init__(self): self.base_url = "https://graviex.net/" self.uri_main = "/webapi/v3/" def call_api(self, path, sub_path=None, params=None): if sub_path: uri = f'{self.uri_main}{path}/{sub_path}' else: uri = f'{self.uri_main}{path}' if params: params_keys = list(params.keys()) params_keys.sort() params_sorted = '' for key in params_keys: value = params[key] if value: params_sorted += f'{key}={value}&' url = f'{self.base_url}{uri}?{params_sorted}' else: url = f'{self.base_url}{uri}' r = requests.get(url) return r.json() def markets(self, market=None): return self.call_api('markets', sub_path=market) def tickers(self, market=None): return self.call_api('tickers', sub_path=market) def depth(self, market, limit=None, order=None): return self.call_api('depth', params=dict(market=market, limit=limit, order=order)) def trades(self, market, limit=None, timestamp=None, from_id=None, to_id=None, order_by=None): return self.call_api('depth', params=dict(market=market, limit=limit, timestamp=timestamp, from_id=from_id, to_id=to_id, order_by=order_by)) def trades_simple(self, market): return self.call_api('trades_simple', params=dict(market=market)) def k(self, market, limit=None, period=None, timestamp=None): return self.call_api('k', params=dict(market=market, limit=limit, period=period, timestamp=timestamp)) def k_with_pending_trades(self, market, trade_id, limit=None, timestamp=None): return self.call_api('k_with_pending_trades', params=dict(market=market, trade_id=trade_id, limit=limit, timestamp=timestamp)) def timestamp(self): return self.call_api('timestamp') def currency(self, unit): return self.call_api('currency', sub_path='info', params=dict(currency=unit)) if __name__ == '__main__': # подключаемся gex_public = GraviexPublicAPI() # получаем все пары markets = gex_public.markets() print(markets) # пара market_id = 'giobtc' # информация по паре market_info = gex_public.markets(market_id) print(market_info) # получаем все тикеры tickers = gex_public.tickers() print(tickers) # тикер по паре ticker_pair = gex_public.tickers(market_id) print(ticker_pair) # глубина по паре depth = gex_public.depth(market_id) print(depth) # сделки по паре trade = gex_public.trades(market_id) print(trade) # сделки с минимальными свойствами trade_simple = gex_public.trades_simple(market_id) print(trade_simple) # OHLC (K line) по паре k_line = gex_public.k(market_id) print(k_line) # Получите K данных с отложенными сделками, которые еще не включены в K данных, # потому что существует задержка между сделкой, сгенерированной и обработанной генератором данных K k_with_pending_trades = gex_public.k_with_pending_trades(market_id, 1) print(k_with_pending_trades) # timestamp сервера timestamp = gex_public.timestamp() print(timestamp) # информация по валюте currency = 'gio' currency_info = gex_public.currency(currency) print(currency_info) Цитата Ссылка на сообщение Поделиться на других сайтах
surugh 0 Опубликовано: 6 января Автор Share Опубликовано: 6 января (изменено) private_api_python_wrapper import time import hmac import hashlib import requests ''' All private API requires 3 authentication parameters and zero or more API specific parameters. access_key Your access key. tonce tonce is a timestamp in integer, stands for milliseconds elapsed since Unix epoch. Tonce must be within 30 seconds of server's current time. Each tonce can only be used once. signature Signature of the API request, generated by you, use your secret key. if it useful you can make a donation GIO: GQksGKKU6PotrByC8hQzBRwMAN7FuMNugF DOGE: D8Hy4stikB4Pk8rnSfhQxgdZ59GqmasdPj ETH: 0x290fE67efA690AE7924DB416d4239daC9c309b97 ''' class GraviexPrivateAPI: def __init__(self, access_key=None, secret_key=None): self.base_url = "https://graviex.net/" self.uri_main = "/webapi/v3/" self.access = access_key self.secret = secret_key def call_api(self, path, sub_path=None, params=None, get=True): # Если много запросов летит сразу тут можно еще поколдовать # Tonce must be within 30 seconds of server's current time. tonce = int(time.time()) * 1000 if not params: params = {} # Чистим параметры от None for key in list(params): if params[key] is None: del params[key] # Костыль для функций использующих параметр from bad_key = 'frome' if bad_key in params: params['from'] = params.pop(bad_key) # Добавляем обязательные параметры и сортируем ключи params.update(dict(tonce=tonce, access_key=self.access)) params_keys = list(params.keys()) params_keys.sort() # Формируем запросы для корректной подписи request = '' for key in params_keys: value = params[key] request += f'{key}={value}&' if sub_path: uri = f'{self.uri_main}{path}/{sub_path}' else: uri = f'{self.uri_main}{path}' if get: verb = 'GET' else: verb = 'POST' message = f'{verb}|{uri}|{request[:-1]}' # Наколдовали теперь подпишем signature = hmac.new( # if api Secret Key not in bytes replace self.secret to # bytes(self.secret, encoding='utf-8'), self.secret, message.encode(), hashlib.sha256 ).hexdigest() # Слепим из приготовленнго URL и скормим зверюге requests query = f'{self.base_url}{uri}?{request}signature={signature}' if get: response = requests.get(query) else: response = requests.post(query) return response.json() def get_members_me(self): return self.call_api('members', sub_path='me') def get_account_history(self, currency=None, limit=None, tx_type=None, from_date=None, to_date=None, page=None, sorted_by=None): return self.call_api('account', sub_path='history', params=dict(currency=currency, limit=limit, type=tx_type, frome=from_date, to=to_date, page=page, order_by=sorted_by)) def get_deposits(self, currency=None, limit=None, state=None): return self.call_api('deposits', params=dict(currency=currency, limit=limit, state=state)) def get_deposit(self, tx_id): return self.call_api('deposit', params=dict(tx_id=tx_id)) def get_deposit_address(self, currency): return self.call_api('deposit_address', params=dict(currency=currency)) def get_gen_deposit_address(self, currency): return self.call_api('gen_deposit_address', params=dict(currency=currency)) def get_orders(self, market=None, state=None, limit=None, page=None, order_by=None): return self.call_api('orders', params=dict(market=market, state=state, limit=limit, page=page, order_by=order_by)) def get_orders_history(self, market=None, state=None, limit=None, from_date=None, to_date=None, page=None, order_by=None): return self.call_api('orders', sub_path='history', params=dict(market=market, state=state, limit=limit, frome=from_date, to=to_date, page=page, order_by=order_by)) def get_order(self, order_id): return self.call_api('order', params=dict(id=order_id)) def get_order_book(self, market, asks_limit=None, bids_limit=None): return self.call_api('order_book', params=dict(market=market, asks_limit=asks_limit, bids_limit=bids_limit)) def get_my_trades(self, limit=None, timestamp=None, from_id=None, to_id=None, order_by=None): return self.call_api('trades', sub_path='my', params=dict(limit=limit, timestamp=timestamp, frome=from_id, to=to_id, order_by=order_by)) def get_trades_history(self, market=None, limit=None, from_date=None, to_date=None, page=None, order_by=None): return self.call_api('trades', sub_path='history', params=dict(market=market, limit=limit, frome=from_date, to=to_date, page=page, order_by=order_by)) def get_settings(self): return self.call_api('settings') def get_withdraws(self, limit=None, state=None): return self.call_api('withdraws', params=dict(limit=limit, state=state)) def get_fund_sources(self, currency): return self.call_api('currency', params=dict(currency=currency)) def get_strategies_list(self): return self.call_api('strategies', sub_path='list') def get_my_strategies(self): return self.call_api('strategies', sub_path='my') def post_register_device(self, device_id): return self.call_api('members', sub_path='me/register_device', params=dict(device=device_id), get=False) def post_update_preferences(self, locale, notify_balances, notify_trades, notify_withdraws, notify_deposits): return self.call_api('members', sub_path='me/update_preferences', params=dict(locale=locale, notify_balances=notify_balances, notify_trades=notify_trades, notify_withdraws=notify_withdraws, notify_deposits=notify_deposits), get=False) def post_orders(self, market, side, volume, price=None, ord_type=None): return self.call_api('orders', params=dict(market=market, side=side, volume=volume, price=price, ord_type=ord_type), get=False) def post_orders_clear(self, side): return self.call_api('orders', sub_path='clear', params=dict(side=side), get=False) def post_order_delete(self, order_id): return self.call_api('order', sub_path='delete', params=dict(id=order_id), get=False) def post_settings_store(self, data): return self.call_api('settings', sub_path='store', params=dict(data=data), get=False) def post_create_withdraw(self, currency, address, amount, provider=None, speed_up=None): return self.call_api('create_withdraw', params=dict(currency=currency, fund_uid=address, sum=amount, provider=provider, speed_up=speed_up), get=False) def post_create_fund_source(self, currency, address, label, provider=None): return self.call_api('create_fund_source', params=dict(currency=currency, uid=address, extra=label, fund_uid=provider), get=False) def post_remove_fund_source(self, source_id): return self.call_api('remove_fund_source', params=dict(id=source_id), get=False) def post_strategy_cancel(self, strategy_id): return self.call_api('strategy', sub_path='cancel', params=dict(id=strategy_id), get=False) def post_strategy_create(self, market=None, strategy=None, config=None): return self.call_api('strategy', sub_path='create', params=dict(market=market, strategy=strategy, config=config), get=False) def post_strategy_update(self, strategy_id, config=None): return self.call_api('strategy', sub_path='update', params=dict(id=strategy_id, config=config), get=False) def post_strategy_activate(self, strategy_id): return self.call_api('strategy', sub_path='activate', params=dict(id=strategy_id), get=False) def post_strategy_deactivate(self, strategy_id): return self.call_api('strategy', sub_path='deactivate', params=dict(id=strategy_id), get=False) if __name__ == '__main__': # 6000 requests/keypair/5 minutes, you can contact GRAVIEX if you need more. gex_private = GraviexPrivateAPI( access_key='', secret_key=b'' ) # Get your profile and accounts info # profile = gex_private.get_members_me() # print(profile) # acc_history = gex_private.get_account_history() # print(acc_history) # deposits = gex_private.get_deposits() # print(deposits) Изменено 7 января пользователем surugh Цитата Ссылка на сообщение Поделиться на других сайтах
surugh 0 Опубликовано: 9 января Автор Share Опубликовано: 9 января Пример бота - скупка монет на кроссах import time from graviex_publc_api import GraviexPublicAPI from graviex_private_api import GraviexPrivateAPI gex_public = GraviexPublicAPI() gex_private = GraviexPrivateAPI( access_key='', secret_key=b'' ) PAIR = 'giodoge' # Начинаем скупать монету по самой низкой доступной цене до нужного колличества MIN_BASE = 10000.0 # Покупать будем частями пока не выкупим необходимое кол-во ORDER_SIZE = 100 # Получаем информацию по паре market_info = gex_public.markets(PAIR) # print(market_info) # Выделяем базовую и котируемую валюты pair_info = market_info['attributes'] base_curr = pair_info['base_unit'] quote_curr = pair_info['quote_unit'] pair_curr = base_curr, quote_curr while True: # Узнаем края стаканов ticker = gex_public.tickers(PAIR) # print(ticker) ask = float(ticker['sell']) bid = float(ticker['buy']) print(f'Pair: {PAIR}\nBID: {bid}\tASK: {ask}') # Узнаем максимальную цену покупки base_btc = gex_public.tickers(base_curr + 'btc') quote_btc = gex_public.tickers(quote_curr + 'btc') base_sellers = float(base_btc['sell']) quote_sellers = float(quote_btc['sell']) max_buy_price = round(base_sellers / quote_sellers, 8) print(f'Max buy price:\n{max_buy_price}') # Нужно проверить наличие открытых ордеров order = gex_private.get_orders(market=PAIR, limit=1) # print(f'Last order:\n{orders}') time.sleep(1) if order: # Если ордер выше/равен максимальной цены закупки if max_buy_price >= float(order[0]['price']): # Если ордер по краю стакана, курим if bid == float(order[0]['price']): print('Wait...') # Снимаем все ордера на покупку для коррекции else: gex_private.post_orders_clear(side='buy') print('Clear order') time.sleep(1) # Нужно выставить ордер else: # Получаем баланс по валютам из пары profile = gex_private.get_members_me() time.sleep(1) # print(profile) balance = profile['accounts_filtered'] # print(balance) pair_balance = [] for unit in balance: for curr in pair_curr: if unit['currency'] == curr: pair_balance.append(dict(currency=unit['currency'], balance=unit['balance'])) print(f'Pair balance:\n{pair_balance}') # Узнаем сколько у нас базовй валюты base_balance = [] for info in pair_balance: if info['currency'] == base_curr: base_balance.append(info['balance']) print(f'Base balance:\n{base_balance}') # Если базовой валюты на балансе меньше нужного кол-ва if float(base_balance[0]) <= MIN_BASE: # Выкупаем ордер по выгодной цене if ask <= max_buy_price: gex_private.post_orders(PAIR, 'buy', str(MIN_BASE), str(ask)) print(f'Выкупаем ордер по ASK: {ask}') # Ставим ордер на покупку по BID ниже либо равную максимальной закупочной цены else: if bid <= max_buy_price: gex_private.post_orders(PAIR, 'buy', str(ORDER_SIZE), str(bid)) print(f'Новый ордер по BID: {bid} {PAIR}') # Снимаем все ордера и выходим (при желании прикурчиваем отправку сообщения в телеграм) else: gex_private.post_orders_clear(side='buy') print('Done') quit() print('-'*15) time.sleep(60) Цитата Ссылка на сообщение Поделиться на других сайтах
Рекомендованные сообщения
Join the conversation
You can post now and register later. If you have an account, sign in now to post with your account.