QgGHva4z

1############################### CODE ###############################
2
3import ccxt
4
5
6DEFAULT_RATE_LIMIT = 1000
7
8
9def get_ccxt_object(ccxt_id, credentials, headers=None, testnet=False):
10    """
11    Creates instance of ccxt.
12    :param ccxt_id: id of exchange in ccxt
13    :param credentials: all required credentials for exchange api
14    :param headers: additional headers for exchange pi
15    :param testnet: tries to use testnet url, if possible
16    :return: instance of ccxt, or raise exception in case of any error
17    """
18    if ccxt_id not in ccxt.exchanges:
19        raise ccxt.BaseError(f'Exchange {ccxt_id} is not supported.')
20
21    ccxt_credentials = create_ccxt_credentials(credentials)
22
23    # prepare headers
24    _headers = {
25        'enableRateLimit': True,
26        'rateLimit': DEFAULT_RATE_LIMIT
27    }
28    _headers.update(ccxt_credentials)
29    if headers:
30        _headers.update(headers)
31
32    ccxt_object = getattr(ccxt, ccxt_id)(_headers)
33
34    # change to testnet url
35    if testnet:
36        if not ccxt_object.urls.get('test'):
37            raise ccxt.BaseError(f"Exchange {ccxt_id} hasn't testnet support.")
38
39        ccxt_object.urls['api'] = ccxt_object.urls['test']
40
41    return ccxt_object
42
43
44def create_ccxt_credentials(credentials):
45    """
46    Converts model's credentials to ccxt credentials format.
47    :param credentials: from model/request
48    :return: ccxt credentials
49    """
50    ccxt_credentials = credentials.copy()
51
52    if credentials.get('api_key'):
53        ccxt_credentials['apiKey'] = ccxt_credentials.pop('api_key')
54    if credentials.get('secret_key'):
55        ccxt_credentials['secret'] = ccxt_credentials.pop('secret_key')
56
57    return ccxt_credentials
58
59############################### TEST ###############################
60
61import ccxt
62import pytest
63
64from src.exchanges_fetcher.core.ccxt_utils import (
65    get_ccxt_object, create_ccxt_credentials
66)
67
68
69def test_create_ccxt_credentials():
70    credentials = {
71        'api_key': 'some_api_key',
72        'secret_key': 'some_secret_key',
73        'uid': 'some_uid'
74    }
75    ccxt_credentials = create_ccxt_credentials(credentials)
76    assert len(ccxt_credentials.keys()) == 3
77    assert credentials['api_key'] == ccxt_credentials['apiKey']
78    assert credentials['secret_key'] == ccxt_credentials['secret']
79
80
81def test_get_valid_ccxt_object():
82    ccxt_id = ccxt.exchanges[0]
83    valid_headers = {'options': {'adjustForTimeDifference': True}}
84    assert get_ccxt_object(ccxt_id, credentials={}, headers=valid_headers)
85
86
87def test_get_invalid_ccxt_object():
88    with pytest.raises(ccxt.BaseError):
89        _ = get_ccxt_object('ccxt_id', credentials={})
90
91
92def test_get_ccxt_object_for_existing_testnet(mocker):
93    # for first ccxt class set allowance for testnet
94    ccxt_id = ccxt.exchanges[0]
95    ccxt_class = mocker.MagicMock()
96    ccxt_class.urls = {'test': 'test_url'}
97    mocker.patch.object(ccxt, ccxt_id, ccxt_class)
98
99    assert get_ccxt_object(ccxt_id, credentials={}, testnet=True)
100
101
102def test_get_ccxt_object_for_not_existing_testnet(mocker):
103    # for first ccxt class set allowance for testnet
104    ccxt_id = ccxt.exchanges[0]
105    ccxt_class = mocker.Mock()
106    ccxt_class.urls = {'api': 'url'}
107    mocker.patch.object(ccxt, ccxt_id, ccxt_class)
108
109    with pytest.raises(ccxt.BaseError):
110        _ = get_ccxt_object('ccxt_id', credentials={}, testnet=True)