balancer-assignment/utils.py at main · Mithreum/balancer-assignment · GitHub

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
import requests
import json

def encode_data(contract_address, function_signature, pool_id, kind, token_a, token_b, amount, sender, data_feed_a, data_feed_b):
    '''Converts the input params to an Ethereum API request body'''
    return {
        "jsonrpc": "2.0",
        "method": "eth_call",
        "params": [
            {
                "to": contract_address,
                "data": function_signature
                + pool_id[2:]
                + format(int(kind, 16), '064x')
                + token_a[2:].rjust(64, '0')
                + token_b[2:].rjust(64, '0')
                + format(amount, '064x')
                + sender[2:].rjust(64, '0')
                + data_feed_a[2:].rjust(64, '0')
                + data_feed_b[2:].rjust(64, '0')
            },
            "latest"],
        "id": 1
    }

def parse_signed_number(hex_string: str, num_bytes: int) -> int:
    # Convert hexadecimal string to an integer
    value = int(hex_string, 16)

    # Check if the most significant bit is set (indicating a negative number)
    if value & (1 << (num_bytes * 8 - 1)):
        # Convert the value to a negative number using two's complement
        value -= 1 << (num_bytes * 8)

    return value


def get_slippage(url, data, token_a_decimals, token_b_decimals) -> tuple:
    '''Querries the contract & parses its output'''

    try:

        # 1. Execute the request
        response = requests.post(url, json=data)

        # 2. Parse the response
        result_hex = json.loads(response.text)['result']

        # print('result_hex', result_hex)

        # 3. Parse the hexadecimal string into respective components
        step = 64 # 32 bytes * 2 chars representation
        start = 2 # 0x prefix
        stop = start + step
        slippage_percent = parse_signed_number(result_hex[start:stop], 32)
        start = stop
        stop += step
        contract_slippage = parse_signed_number(result_hex[start:stop], 32)
        start = stop
        stop += step
        token_a_value = int(result_hex[start:stop], 16)
        start = stop
        stop += step
        token_b_value = int(result_hex[start:stop], 16)
        start = stop
        stop += step
        token_a_price = int(result_hex[start:stop], 16)
        start = stop
        stop += step
        token_b_price = int(result_hex[start:stop], 16)

        return slippage_percent / 10 ** (6), contract_slippage / 10 ** (6), token_a_value / 10 ** (token_a_decimals + 8), token_b_value / 10 ** (token_b_decimals + 8), token_a_price / 1e8, token_b_price / 1e8

    except:
        print('Check the RPC URL, pool_id, contract or feed addresses')
        return 'error', 'error', 'error', 'error'

def main(tests):

    for test in tests:

        try:

            # 1. Prepare the data
            request_data = encode_data(test['contract'], test['function'], test['pool_id'], test['kind'], test['token_a'], test['token_b'], test['amount'], test['sender'], test['feed_a'], test['feed_b'])

            # print('request_data', request_data)

            # 2.
            [slippage_percent, contract_slippage, token_a_value, token_b_value,  token_a_price, token_b_price] = get_slippage(test['rpc_url'], request_data, test['a_decimals'], test['b_decimals'])

            # 3. Output the result
            print(f"{test['pair']}\n Slippage %:\t", slippage_percent, '\n', "Slippage ($):\t", contract_slippage, '\n', "Value A($):\t", token_a_value, '\n', "Value B($): \t", token_b_value, "\n", "Price A:\t", token_a_price, "\n", "Price B:\t", token_b_price)

        except Exception as e:
            print('An error occured', e)