Graviex API

[surugh 0]

 

Исходники  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]

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]

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]

Пример бота - скупка монет на кроссах

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)