Основы взаимодействия с EVM сетями

About the author

Author: https://t.me/ahillary

Resources

Channel: https://t.me/semolina_code_python

Chat: https://t.me/python_with_ahillary

YouTube: https://www.youtube.com/@semolinacode

Coding training: https://t.me/how_to_code_web3

Prop trading: https://t.me/semolina_prop

Содержание

Окружение

Версия python

Мы будем работать на python 3.11 так как последняя версия (python 3.12) нестабильно работаем со всеми функциями используемых нами библиотек

Download Python

virtualenv venv --python=python3.11

Версия web3.py

Мы будем работать с версией web3.py 6.14.0

Release Notes — web3.py 7.2.0 documentation

pip install web3==6.14.0

Остальные зависимости для урока

файл requirements.txt

web3==6.14.0
curl_cffi==0.7.1
fake-useragent==1.4.0

Дополнительные материалы

Документация web3.py

Quickstart — web3.py 7.2.0 documentation

Методички Алекса Крюгера

Данные методички позволят глубже погрузиться в web3 и понять как взаимодействовать с. web3 через код

Технический курс про DeFi для начинающих и не только.

Авторская статья “How to Code или как выносить проекты на 1000 аккаунтов”

How to Code или как выносить проекты на 1000 аккаунтов

Асинхронное подключение к блокчейну

import asyncio

from web3 import AsyncWeb3

async def main():
    rpc = '<https://rpc.ankr.com/eth/>'
    w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(rpc))
    print(await w3.is_connected())  # True

if __name__ == '__main__':
    asyncio.run(main())

RPC (Remote Procedure Call) — это протокол, который позволяет программе выполнять функции или процедуры в удаленной системе (сервере) так, как если бы они выполнялись локально в той же системе. В контексте блокчейнов и, в частности, Ethereum, RPC используется для взаимодействия с узлом (node) блокчейна, чтобы отправлять транзакции, запрашивать данные и выполнять другие действия.

Список RPC можно найти на сайте:

ChainList

Получение параметров блокчейна

import asyncio

from web3 import AsyncWeb3

async def main():
    rpc = '<https://rpc.ankr.com/eth/>'
    w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(rpc))

    # Получение gas_price
    gas_price = await w3.eth.gas_price
    print(f"gas price: {gas_price} Wei")
    print(f"gas price: {gas_price / 10 ** 9} GWei")
    print(f"gas price: {gas_price / 10 ** 18} ETH")

    # Получение max_priority_fee (для EIP-1559 транзакций. Работает не во всех сетях)
    max_priority_fee = await w3.eth.max_priority_fee
    print(f"max priority fee: {max_priority_fee}")

    # Получение id сети
    chain_id = await w3.eth.chain_id
    print(f"number of current chain is {chain_id}")

    # Получение номера текущего блока
    block_number = await w3.eth.block_number
    print(f"current block number: {block_number}")

if __name__ == '__main__':
    asyncio.run(main())

• wei — самая маленькая единица эфира (ETH), используемая для измерения и хранения значений в Ethereum (1 ETH = 10^18 wei).
• gwei — одна из единиц эфира, равная 1 миллиард wei (10^9 wei), часто используется для указания стоимости транзакций (например, цены газа).
• gas_price — стоимость, которую пользователь готов заплатить за единицу газа (операции) в Ethereum, выраженная в gwei.
• max_priority_fee — максимальная дополнительная плата, которую пользователь готов заплатить майнерам для ускорения включения транзакции в блок.
• chain_id — уникальный идентификатор конкретной сети блокчейна Ethereum (например, основная сеть, тестовая сеть).
• block_number — номер блока в цепочке блоков (блокчейне), который обозначает его позицию в последовательности всех блоков сети.

Checksum address

Checksum address (чексу́мный адрес) в Ethereum используется для проверки правильности написания и предотвращения ошибок ввода адреса.

Зачем он нужен?

1. Проверка корректности: Чексу́мный адрес включает как строчные, так и прописные буквы, что позволяет алгоритмически проверить правильность адреса. Например, если введена одна или несколько букв неправильно (например, все символы в нижнем или верхнем регистре), это будет указанием на ошибку.
2. Безопасность: Использование чексу́мных адресов снижает вероятность ошибок при копировании или ручном вводе адресов, предотвращая случайные переводы средств на неправильный адрес. Если адрес введен неверно, многие кошельки и приложения распознают это и выдадут предупреждение.

Как это работает?

Ethereum-адреса, которые изначально записаны в нижнем регистре, преобразуются в чексу́мный формат путем использования алгоритма Keccak-256. Определенные символы в адресе приводятся к верхнему регистру на основе результатов хэширования, что создает уникальный «чексу́мный» паттерн.

Как это работает в python

from web3 import AsyncWeb3

address = '0x672009c467b943eea16cdf53549c5fb94a36cfbf'
checksum_address = AsyncWeb3.to_checksum_address(address)
print(checksum_address)  # 0x672009c467B943EeA16cDf53549C5FB94a36cfBf

Получение баланса нативной монеты

import asyncio

from web3 import AsyncWeb3

async def main():
    rpc = '<https://rpc.ankr.com/eth/>'
    w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(rpc))

    address = '0x672009c467b943eea16cdf53549c5fb94a36cfbf'
    checksum_address = AsyncWeb3.to_checksum_address(address)
    balance = await w3.eth.get_balance(checksum_address)
    print(f'balance: {balance} Wei')
    print(f'balance: {balance / 10 ** 18} ETH')

if __name__ == '__main__':
    asyncio.run(main())

Перевод в разные системы измерения

from_wei

import asyncio

from web3 import AsyncWeb3

async def main():
    # баланс в Wei
    balance = 1_000_000_000_000_000_000
    print(balance)                                      # 1000000000000000000
    
    print(AsyncWeb3.from_wei(balance, 'kwei'))          # 1000000000000000
    print(AsyncWeb3.from_wei(balance, 'babbage'))       # 1000000000000000
    print(AsyncWeb3.from_wei(balance, 'femtoether'))    # 1000000000000000

    print(AsyncWeb3.from_wei(balance, 'mwei'))          # 1000000000000
    print(AsyncWeb3.from_wei(balance, 'lovelace'))      # 1000000000000
    print(AsyncWeb3.from_wei(balance, 'picoether'))     # 1000000000000

    print(AsyncWeb3.from_wei(balance, 'gwei'))          # 1000000000
    print(AsyncWeb3.from_wei(balance, 'shannon'))       # 1000000000
    print(AsyncWeb3.from_wei(balance, 'nanoether'))     # 1000000000
    print(AsyncWeb3.from_wei(balance, 'nano'))          # 1000000000

    print(AsyncWeb3.from_wei(balance, 'szabo'))         # 1000000
    print(AsyncWeb3.from_wei(balance, 'microether'))    # 1000000
    print(AsyncWeb3.from_wei(balance, 'micro'))         # 1000000

    print(AsyncWeb3.from_wei(balance, 'finney'))        # 1000
    print(AsyncWeb3.from_wei(balance, 'milliether'))    # 1000
    print(AsyncWeb3.from_wei(balance, 'milli'))         # 1000

    print(AsyncWeb3.from_wei(balance, 'ether'))         # 1

    print(AsyncWeb3.from_wei(balance, 'kether'))        # 0.001
    print(AsyncWeb3.from_wei(balance, 'grand'))         # 0.001

    print(AsyncWeb3.from_wei(balance, 'mether'))        # 0.000001

    print(AsyncWeb3.from_wei(balance, 'gether'))        # 0.000000001

    print(AsyncWeb3.from_wei(balance, 'tether'))        # 0.000000000001

if __name__ == '__main__':
    asyncio.run(main())

to_wei

import asyncio

from web3 import AsyncWeb3

async def main():
    balance = 1
    print(balance)                                    # 1

    print(AsyncWeb3.to_wei(balance, 'kwei'))          # 1000
    print(AsyncWeb3.to_wei(balance, 'babbage'))       # 1000
    print(AsyncWeb3.to_wei(balance, 'femtoether'))    # 1000

    print(AsyncWeb3.to_wei(balance, 'mwei'))          # 1000000
    print(AsyncWeb3.to_wei(balance, 'lovelace'))      # 1000000
    print(AsyncWeb3.to_wei(balance, 'picoether'))     # 1000000

    print(AsyncWeb3.to_wei(balance, 'gwei'))          # 1000000000
    print(AsyncWeb3.to_wei(balance, 'shannon'))       # 1000000000
    print(AsyncWeb3.to_wei(balance, 'nanoether'))     # 1000000000
    print(AsyncWeb3.to_wei(balance, 'nano'))          # 1000000000

    print(AsyncWeb3.to_wei(balance, 'szabo'))         # 1000000000000
    print(AsyncWeb3.to_wei(balance, 'microether'))    # 1000000000000
    print(AsyncWeb3.to_wei(balance, 'micro'))         # 1000000000000

    print(AsyncWeb3.to_wei(balance, 'finney'))        # 1000000000000000
    print(AsyncWeb3.to_wei(balance, 'milliether'))    # 1000000000000000
    print(AsyncWeb3.to_wei(balance, 'milli'))         # 1000000000000000

    print(AsyncWeb3.to_wei(balance, 'ether'))         # 1000000000000000000

    print(AsyncWeb3.to_wei(balance, 'kether'))        # 1000000000000000000000
    print(AsyncWeb3.to_wei(balance, 'grand'))         # 1000000000000000000000

    print(AsyncWeb3.to_wei(balance, 'mether'))        # 1000000000000000000000000

    print(AsyncWeb3.to_wei(balance, 'gether'))        # 1000000000000000000000000000

    print(AsyncWeb3.to_wei(balance, 'tether'))        # 1000000000000000000000000000000

if __name__ == '__main__':
    asyncio.run(main())

Вызов read функций смарт контракта

Сравнение read и write функций

Функции смарт-контрактов в Ethereum делятся на два типа: read (чтение) и write (запись). Они различаются по целям, затратам и последствиям взаимодействия с блокчейном.

Read-функции (чтение)

• Назначение: Используются для чтения данных из блокчейна или контракта. Эти функции не изменяют состояние блокчейна и не требуют совершения транзакции.
• Стоимость: Бесплатные, так как не требуют оплаты за газ. Поскольку состояние блокчейна не меняется, такие функции можно вызывать без затрат.
• Тип вызова: Локальный (off-chain), не требует подтверждения транзакции сетью.
• Пример: Получение баланса, проверка статуса или чтение данных, сохраненных в контракте.

Write-функции (запись)

• Назначение: Используются для изменения состояния блокчейна, таких как изменение данных контракта или отправка транзакции. Вызов таких функций влияет на глобальное состояние сети.
• Стоимость: Требуют оплату газа, поскольку вызывают изменение состояния блокчейна, и майнеры должны обработать транзакцию.
• Тип вызова: Транзакция (on-chain), требующая подтверждения в сети и записи в блокчейн.
• Пример: Отправка токенов, изменение переменных контракта, выполнение транзакций.

Таким образом, read-функции используются для доступа к данным без изменения состояния, а write-функции — для изменения данных в блокчейне, что требует затрат на газ и подтверждения транзакции.

Готовый код

import json
import asyncio

from web3 import AsyncWeb3

async def main():
    rpc = '<https://arbitrum-one.publicnode.com>'
    w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(rpc))

    my_address = AsyncWeb3.to_checksum_address('0x0bB902fC9e168343a19d622E79cE033452e64Dd8')

    arbitrum_usdc_address = AsyncWeb3.to_checksum_address('0xaf88d065e77c8cC2239327C5EDb3A432268e5831')
    arbitrum_usdc_abi = json.loads('<CONTRACT ABI HERE>')

    contract = w3.eth.contract(arbitrum_usdc_address, abi=arbitrum_usdc_abi)
    # выводит результат работы функции name()
    print('name', await contract.functions.name().call())
    # выводит результат работы функции symbol()
    print('symbol', await contract.functions.symbol().call())

    # присваивает результат работы функции decimals() в переменную decimals
    decimals = await contract.functions.decimals().call()
    print('decimals', decimals)
    # присваивает результат работы функции balanceOf(address) в переменную balance
    balance = await contract.functions.balanceOf(my_address).call()
    print('balance', balance)
    print('balance', balance / 10 ** decimals)

if __name__ == '__main__':
    asyncio.run(main())

Адрес контракта токена

arbitrum_usdc_address = '0xaf88d065e77c8cC2239327C5EDb3A432268e5831'

Здесь мы задаем адрес смарт-контракта для USDC на Arbitrum.

В данном случае адрес USDC арбитрум: 0xaf88d065e77c8cC2239327C5EDb3A432268e5831 Мы можем найти его в эксплорере:

USD Coin (USDC)

ABI контракта

arbitrum_usdc_abi = json.loads('<CONTRACT ABI HERE>')

Здесь мы задаем ABI (Application Binary Interface) для контракта USDC на Arbitrum. ABI описывает функции и события, доступные в смарт-контракте.

Для получения ABI, переходим на вкладку Contract:

На этой вкладке ищем функцию, которую хотим вызвать (ищем на вкладках Read Contract, Write Contract, Read as Proxy, Write as Proxy)

Пример 1:

Хотим вызвать функцию admin с вкладки Read Contract

В таком случае, просто переходим на вкладку Code и копируем ABI оттуда.

Пример 2:

Хотим вызвать функцию balanceOf с вкладки Read as Proxy

В таком случае нужно брать аби с Proxy контракта. Переходим на Proxy контракт:

В Proxy контракте переходим на вкладку Code и копируем ABI оттуда.

Полученное ABI вставляем в код вместо “<CONTRACT ABI HERE>«

ABI из файла

В данном случае мы ABI копируем и вставляем прямо в код:

arbitrum_usdc_abi = json.loads('<CONTRACT ABI HERE>')

Если мы работаем с большим количеством контрактов, ABI каждого из них вставлять в код неудобно так как в коде появляется большое количество символов

Хорошей практикой будет создать отдельную директорию (например abis) и все ABI контрактов хранить в ней в формате .json

Чтобы работать с этим файлом, необходимо достать оттуда ABI и преобразовать его в python объект

import json
import asyncio

from web3 import AsyncWeb3

def read_json(path: str, encoding: str | None = None) -> list | dict:
    with open(path, encoding=encoding) as f:
        return json.load(f)

async def main():
    rpc = '<https://arbitrum-one.publicnode.com>'
    w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(rpc))

    my_address = AsyncWeb3.to_checksum_address('0x0bB902fC9e168343a19d622E79cE033452e64Dd8')

    arbitrum_usdc_address = AsyncWeb3.to_checksum_address('0xaf88d065e77c8cC2239327C5EDb3A432268e5831')
    arbitrum_usdc_abi = read_json(path='<PATH TO ABI>')
    ...

if __name__ == '__main__':
    asyncio.run(main())

ABI из python объекта

Так как ABI только описывает функции и события, доступные в смарт-контракте, мы можем составить его самостоятельно и добавить только необходимые нам функции

import asyncio

from web3 import AsyncWeb3

TokenABI = [
    {
        'constant': True,
        'inputs': [],
        'name': 'name',
        'outputs': [{'name': '', 'type': 'string'}],
        'payable': False,
        'stateMutability': 'view',
        'type': 'function'
    },
    {
        'constant': True,
        'inputs': [],
        'name': 'symbol',
        'outputs': [{'name': '', 'type': 'string'}],
        'payable': False,
        'stateMutability': 'view',
        'type': 'function'
    },
    {
        'constant': True,
        'inputs': [],
        'name': 'totalSupply',
        'outputs': [{'name': '', 'type': 'uint256'}],
        'payable': False,
        'stateMutability': 'view',
        'type': 'function'
    },
    {
        'constant': True,
        'inputs': [],
        'name': 'decimals',
        'outputs': [{'name': '', 'type': 'uint256'}],
        'payable': False,
        'stateMutability': 'view',
        'type': 'function'
    },
    {
        'constant': True,
        'inputs': [{'name': 'who', 'type': 'address'}],
        'name': 'balanceOf',
        'outputs': [{'name': '', 'type': 'uint256'}],
        'payable': False,
        'stateMutability': 'view',
        'type': 'function'
    },
    {
        'constant': True,
        'inputs': [{'name': '_owner', 'type': 'address'}, {'name': '_spender', 'type': 'address'}],
        'name': 'allowance',
        'outputs': [{'name': 'remaining', 'type': 'uint256'}],
        'payable': False,
        'stateMutability': 'view',
        'type': 'function'
    },
    {
        'constant': False,
        'inputs': [{'name': '_spender', 'type': 'address'}, {'name': '_value', 'type': 'uint256'}],
        'name': 'approve',
        'outputs': [],
        'payable': False,
        'stateMutability': 'nonpayable',
        'type': 'function'
    },
    {
        'constant': False,
        'inputs': [{'name': '_to', 'type': 'address'}, {'name': '_value', 'type': 'uint256'}],
        'name': 'transfer',
        'outputs': [], 'payable': False,
        'stateMutability': 'nonpayable',
        'type': 'function'
    }
]

async def main():
    rpc = '<https://arbitrum-one.publicnode.com>'
    w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(rpc))

    my_address = AsyncWeb3.to_checksum_address('0x0bB902fC9e168343a19d622E79cE033452e64Dd8')

    arbitrum_usdc_address = AsyncWeb3.to_checksum_address('0xaf88d065e77c8cC2239327C5EDb3A432268e5831')
    arbitrum_usdc_abi = TokenABI
    ...

if __name__ == '__main__':
    asyncio.run(main())

Данный метод является самым предпочтительным для работы с токенами так как 99% всех токенов реализуют интерфейс ERC-20, а это значит, что почти у всех токенов ABI будет одинаковым (одинаковые названия методов с одинаковыми параметрами)

Данную структуру python можно убрать в отдельный файл (например models.py) и импортировать оттуда

Как эта структура выглядит в json:

[
  {
    "inputs": [],
    "name": "name",
    "outputs": [
      {
        "internalType": "string",
        "name": "",
        "type": "string"
      }
    ],
    "stateMutability": "view",
    "type": "function"
  },
  {
    "inputs": [],
    "name": "symbol",
    "outputs": [
      {
        "internalType": "string",
        "name": "",
        "type": "string"
      }
    ],
    "stateMutability": "view",
    "type": "function"
  },
  {
    "inputs": [],
    "name": "totalSupply",
    "outputs": [
      {
        "internalType": "uint256",
        "name": "",
        "type": "uint256"
      }
    ],
    "stateMutability": "view",
    "type": "function"
  },
  {
    "inputs": [],
    "name": "decimals",
    "outputs": [
      {
        "internalType": "uint8",
        "name": "",
        "type": "uint8"
      }
    ],
    "stateMutability": "view",
    "type": "function"
  },
  {
    "inputs": [
      {
        "internalType": "address",
        "name": "account",
        "type": "address"
      }
    ],
    "name": "balanceOf",
    "outputs": [
      {
        "internalType": "uint256",
        "name": "",
        "type": "uint256"
      }
    ],
    "stateMutability": "view",
    "type": "function"
  },
  {
    "inputs": [
      {
        "internalType": "address",
        "name": "owner",
        "type": "address"
      },
      {
        "internalType": "address",
        "name": "spender",
        "type": "address"
      }
    ],
    "name": "allowance",
    "outputs": [
      {
        "internalType": "uint256",
        "name": "",
        "type": "uint256"
      }
    ],
    "stateMutability": "view",
    "type": "function"
  },
  {
    "inputs": [
      {
        "internalType": "address",
        "name": "spender",
        "type": "address"
      },
      {
        "internalType": "uint256",
        "name": "value",
        "type": "uint256"
      }
    ],
    "name": "approve",
    "outputs": [
      {
        "internalType": "bool",
        "name": "",
        "type": "bool"
      }
    ],
    "stateMutability": "nonpayable",
    "type": "function"
  },
  {
    "inputs": [
      {
        "internalType": "address",
        "name": "to",
        "type": "address"
      },
      {
        "internalType": "uint256",
        "name": "value",
        "type": "uint256"
      }
    ],
    "name": "transfer",
    "outputs": [
      {
        "internalType": "bool",
        "name": "",
        "type": "bool"
      }
    ],
    "stateMutability": "nonpayable",
    "type": "function"
  }
]

Умение самостоятельно составить ABI позволяет не только сократить количество символов в проекте, но и составить ABI для отправки транзакции в случаях, когда ABI недоступно

Создание объекта контракта

contract = w3.eth.contract(arbitrum_usdc_address, abi=arbitrum_usdc_abi)

Мы создаем объект контракта, используя адрес и ABI. Этот объект позволяет нам взаимодействовать с функциями контракта.

Вызов функции name

print('name', await contract.functions.name().call())

Мы вызываем функцию name контракта, которая возвращает название токена (например, «USD Coin»).

Вызов функции symbol

print('symbol', await contract.functions.symbol().call())

Мы вызываем функцию symbol контракта, которая возвращает символ токена (например, «USDC»).

Получение decimals

decimals = await contract.functions.decimals().call()
print('decimals', decimals)

Мы вызываем функцию decimals, которая возвращает количество десятичных знаков, используемых в токене (например, 6 для USDC).

Получение баланса

balance = await contract.functions.balanceOf(my_address).call()
print('balance', balance)
print('balance', balance / 10 ** decimals)

Мы вызываем функцию balanceOf, чтобы получить баланс указанного адреса. Баланс выводится сначала в «сырых» единицах (т.е. с учетом десятичных знаков), а затем в «нормализованном» виде (поделив на 10 в степени decimals).

Decimals

В контексте криптовалют и токенов, таких как USDC, функция decimals определяет, сколько десятичных знаков может иметь токен. Это важно для отображения и расчета значений токенов.

Пример

Представьте, что у вас есть 1 токен, и decimals равен 6. Это значит, что этот 1 токен можно представить как 1000000 (один с шестью десятичными знаками). Если decimals равен 18, то 1 токен будет представлен как 1000000000000000000 (один с восемнадцатью десятичными знаками).

Почему это важно?

1. Точность расчетов:
   • Чем больше десятичных знаков, тем точнее можно выразить значения. Например, в финансовых транзакциях важно иметь возможность точно указывать дробные части токенов.
2. Удобство использования:
   • Представление токенов с использованием десятичных знаков делает их более удобными для чтения и понимания. Например, 0.000001 токена с decimals 6 проще понять, чем 1 токен с decimals 6 в «сырых» единицах.

В контексте кода

# Получаем количество десятичных знаков для токена
decimals = await contract.functions.decimals().call()
print('decimals', decimals)

Этот код вызывает функцию decimals у смарт-контракта, чтобы узнать, сколько десятичных знаков используется для данного токена. После этого, когда мы получаем баланс токенов, мы можем разделить «сырое» значение баланса на 10 в степени decimals, чтобы получить удобочитаемое значение.

# Получаем баланс адреса
balance = await contract.functions.balanceOf(my_address).call()
# Выводим баланс в "сырых" единицах
print('balance', balance)
# Преобразуем баланс в удобочитаемое значение
print('balance', balance / 10 ** decimals)

Здесь balance сначала выводится в «сырых» единицах, а затем делится на 10 в степени decimals, чтобы получить значение в удобочитаемом формате. Например, если баланс равен 1000000 и decimals равен 6, то это будет отображено как 1.0 токена.

Создание account

account из модуля web3.eth.account в библиотеке web3.py используется для создания и управления Ethereum-аккаунтами, а также для выполнения операций, таких как подпись транзакций и сообщений. Этот модуль помогает разработчикам безопасно взаимодействовать с аккаунтами в локальном контексте, не полагаясь на внешние кошельки или приложения.

Создание account из приватного ключа

import asyncio

from eth_account.signers.local import LocalAccount
from web3 import AsyncWeb3

async def main():
    rpc = '<https://rpc.ankr.com/eth/>'
    w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(rpc))

    private_key = '<YOUR PRIVATE KEY HERE>'
    account: LocalAccount = w3.eth.account.from_key(private_key)
    address = account.address
    print(address)

if __name__ == '__main__':
    asyncio.run(main())

В данном случае аннотацию типа account: LocalAccount четко прописываем так как в некоторых случаях редактор кода не может точно определить тип и не будет показывать доступные поля и методы

Создание account из мнемонической фразы

import asyncio

from eth_account.signers.local import LocalAccount
from web3 import AsyncWeb3

async def main():
    rpc = '<https://rpc.ankr.com/eth/>'
    w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(rpc))

    w3.eth.account.enable_unaudited_hdwallet_features()    

    mnemonic = '<YOUR MNEMONIC PHRASE HERE>'
    account: LocalAccount = w3.eth.account.from_mnemonic(mnemonic)
    address = account.address
    print(address)

if __name__ == '__main__':
    asyncio.run(main())

По-умолчанию account нельзя создать просто из мнемонической фразы. Вылетает ошибка (Использование мнемонических функций Account отключено по умолчанию до стабилизации API)

AttributeError: The use of the Mnemonic features of Account is disabled by 
default until its API stabilizes. 
**To use these features, please enable them by running 
Account.enable_unaudited_hdwallet_features()** and try again.

Из ошибки понятно, что нужно включить непроверенные функции hdwallet перед использованием w3.eth.account.from_mnemonic(mnemonic):

w3.eth.account.enable_unaudited_hdwallet_features()

Вызов write функций смарт контракта

Класс Client

Для отправки write транзакций нам потребуется объект AsyncWeb3 и account поэтому для более удобной работы создадим класс Client:

from web3 import AsyncWeb3
from eth_account.signers.local import LocalAccount

class Client:
    private_key: str
    rpc: str
    w3: AsyncWeb3
    account: LocalAccount

    def __init__(self, private_key: str, rpc: str):
        self.private_key = private_key
        self.rpc = rpc
        self.w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(rpc))
        self.account = self.w3.eth.account.from_key(private_key)

Метод send_transaction

from web3 import AsyncWeb3
from eth_typing import ChecksumAddress, HexStr
from eth_account.signers.local import LocalAccount

class Client:
    ...

    async def send_transaction(
            self,
            to: str | ChecksumAddress,
            data: HexStr | None = None,
            from_: str | ChecksumAddress | None = None,
            increase_gas: float = 1,
            value: int | None = None,
    ) -> HexBytes | None:
        if not from_:
            from_ = self.account.address

        tx_params = {
            'chainId': await self.w3.eth.chain_id,
            'nonce': await self.w3.eth.get_transaction_count(self.account.address),
            'from': AsyncWeb3.to_checksum_address(from_),
            'to': AsyncWeb3.to_checksum_address(to),
            'gasPrice': await self.w3.eth.gas_price
        }
        if data:
            tx_params['data'] = data
        if value:
            tx_params['value'] = value

        gas = await self.w3.eth.estimate_gas(tx_params)
        tx_params['gas'] = int(gas * increase_gas)

        sign = self.w3.eth.account.sign_transaction(tx_params, self.private_key)
        return await self.w3.eth.send_raw_transaction(sign.rawTransaction)

Добавили в класс метод send_transaction, который позволит отправить транзакцию

Основные понятия

• Транзакция — это действие в сети Ethereum, например, отправка ETH другому пользователю или взаимодействие с контрактом.
• Подпись транзакции — это процесс, в котором вы подтверждаете, что транзакция исходит от вас, с помощью вашего приватного ключа.
• Отправка транзакции — процесс передачи подписанной транзакции в сеть, где майнеры ее обрабатывают.

Аргументы метода send_transaction

1. to: str | ChecksumAddress — это адрес получателя, то есть тот, кому вы отправляете средства или адрес смарт контракта, с которым взаимодействуете. Адрес может быть строкой или форматом Ethereum Checksum Address (адрес с проверкой на правильность).
2. data: HexStr | None — это дополнительные данные для контракта, если вы хотите взаимодействовать со смарт контрактами. По умолчанию None, то есть для простой транзакции не нужно передавать данные.
3. from_: str | ChecksumAddress | None — это адрес отправителя. Если не указано, по умолчанию используется ваш аккаунт (self.account.address).
4. increase_gas: float = 1 — это коэффициент увеличения количества газа, который вы готовы заплатить. По умолчанию установлен в 1 (то есть расчетная стоимость газа используется как есть), но можно увеличить, если хотите ускорить транзакцию.
5. value: int | None — это сумма средств (в Wei), которую вы хотите отправить. Если не указываете, то транзакция не включает перевод ETH.

Внутри метода send_transaction

1. Адрес отправителя: if not from_: from_ = self.account.address Если вы не указали адрес отправителя, используется ваш аккаунт, который был создан при инициализации класса.
2. Формирование параметров транзакции (tx_params): tx_params = { 'chainId': await self.w3.eth.chain_id, 'nonce': await self.w3.eth.get_transaction_count(self.account.address), 'from': AsyncWeb3.to_checksum_address(from_), 'to': AsyncWeb3.to_checksum_address(to), 'gasPrice': await self.w3.eth.gas_price } Здесь собираются данные для транзакции:
   • chainId — это ID сети Ethereum (например, основная сеть или тестовая сеть).
   • nonce — это число транзакций, отправленных вашим аккаунтом, которое обеспечивает уникальность каждой транзакции.
   • from и to — адреса отправителя и получателя.
   • gasPrice — цена за единицу газа (комиссии) в сети.
3. Добавление данных и value (если указаны): if data: tx_params['data'] = data if value: tx_params['value'] = value Если вы передали дополнительные данные или value для отправки, они добавляются в параметры транзакции.
4. Оценка газа: gas = await self.w3.eth.estimate_gas(tx_params) tx_params['gas'] = int(gas * increase_gas) Тут метод оценивает, сколько газа потребуется для выполнения транзакции. Этот шаг помогает избежать ошибок, связанных с нехваткой газа. Если вы хотите ускорить транзакцию, можно увеличить значение газа через параметр increase_gas.
5. Подпись транзакции: sign = self.w3.eth.account.sign_transaction(tx_params, self.private_key) После того как параметры собраны и газ рассчитан, транзакция подписывается с помощью вашего приватного ключа. Это подтверждает, что транзакция отправлена именно от вашего имени.
6. Отправка транзакции в сеть: return await self.w3.eth.send_raw_transaction(sign.rawTransaction) Подписанная транзакция отправляется в сеть Ethereum. Вы получите уникальный хеш транзакции, который можно использовать для отслеживания ее статуса.

Метод verif_tx

Метод verif_tx используется для проверки статуса транзакции в блокчейне Ethereum. Он ждет, пока транзакция завершится (попадет в блокчейн), и затем проверяет, успешно ли она была выполнена. Если транзакция прошла успешно, метод возвращает хеш транзакции. Если транзакция не удалась, он генерирует ошибку.

from hexbytes import HexBytes

from web3 import AsyncWeb3
from web3.exceptions import Web3Exception
from eth_account.signers.local import LocalAccount

class Client:
    ...

    async def verif_tx(self, tx_hash: HexBytes, timeout: int = 200) -> str:
        data = await self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=timeout)
        if data.get('status') == 1:
            return tx_hash.hex()
        raise Web3Exception(f'transaction failed {data["transactionHash"].hex()}')

    ...

Аргументы метода

1. tx_hash: HexBytes — это хеш транзакции, которую нужно проверить. Хеш — это уникальный идентификатор транзакции в сети Ethereum.
2. timeout: int = 200 — максимальное время ожидания завершения транзакции, в секундах. Если за это время транзакция не будет завершена, метод вернет ошибку.

Как работает метод

1. Ожидание завершения транзакции: data = await self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=timeout) Этот код использует метод wait_for_transaction_receipt, который «ждет», пока транзакция будет включена в блокчейн. Это функция блокирует выполнение программы до тех пор, пока транзакция не попадет в блок и не будет завершена, либо пока не истечет время ожидания (timeout).
2. Проверка статуса транзакции: if data.get('status') == 1: Когда транзакция завершится, метод проверяет ее статус:
   • Статус 1 означает, что транзакция прошла успешно.
   • Если статус другой (например, 0), это означает, что транзакция не удалась (например, не хватило газа или произошла ошибка в контракте).
3. Возвращение хеша транзакции при успехе: return tx_hash.hex() Если транзакция была успешной, метод возвращает хеш транзакции в виде строки.
4. Генерация ошибки при неудаче: raise Web3Exception(f'transaction failed {data["transactionHash"].hex()}') Если транзакция не удалась, метод генерирует исключение (ошибку) с сообщением, указывающим на хеш транзакции. Это полезно для отладки и уведомления о том, что транзакция не прошла.

Зачем нужен метод

• Этот метод полезен, когда вам нужно убедиться, что транзакция успешно завершена, прежде чем продолжить выполнение программы.
• В Ethereum после отправки транзакции она не моментально добавляется в блокчейн — требуется время, чтобы майнеры обработали ее. Этот метод ждет завершения этого процесса.
• Если транзакция завершится с ошибкой, программа сможет узнать об этом и обработать исключение (например, повторить транзакцию или вывести сообщение об ошибке).

Пример использования send_transaction на примере approve

Разбор транзакции в експлорере

Пример транзакции approve в сканере:

В данном случае мы даём апрув протоколу uniswap на использование наших токенов USDT.

Из транзакции видно, что она уходит на адрес контракта USDT:

Если посмотреть data транзакции, то мы увидим какому контракту мы даём апрув (параметр spender). В данном случае это контракт uniswap

Также в data мы видим количество токенов в Wei, которые мы апруваем для uniswap:

Параметр value тоже можно посмотреть в експлорере. В данном случае он равняется 0:

Отправляем транзакцию

import asyncio

class Client:
    ...

async def main():
    # Задача: сделать апрув всего баланса USDT для uniswap

    token_address = AsyncWeb3.to_checksum_address('0x55d398326f99059fF775485246999027B3197955')
    spender = AsyncWeb3.to_checksum_address('0x000000000022D473030F116dDEE9F6B43aC78BA3')

    client = Client(
        private_key='<ТУТ ВАШ ПРИВАТНЫЙ КЛЮЧ>',
        rpc='<https://1rpc.io/bnb>'
    )

    contract = client.w3.eth.contract(
        address=token_address,
        abi=TokenABI
    )

    decimals = await contract.functions.decimals().call()
    token_balance = await contract.functions.balanceOf(client.account.address).call()
    approved_amount = await contract.functions.allowance(client.account.address, spender).call()

    print('decimals:', decimals)
    print('token_balance:', token_balance / 10 ** decimals)
    print('approved_amount:', approved_amount / 10 ** decimals)

    if approved_amount < token_balance:
        tx_hash = await client.send_transaction(
            to=token_address,
            data=contract.encodeABI('approve',
                                    args=(
                                        spender,
                                        token_balance
                                    ))
        )
        if tx_hash:
            try:
                await client.verif_tx(tx_hash=tx_hash)
                print(f'Transaction success!! tx_hash: {tx_hash.hex()}')
            except Exception as err:
                print(f'Transaction error!! tx_hash: {tx_hash.hex()}; error: {err}')
        else:
            print(f'Transaction error!!')
    else:
        print('Already approved')

if __name__ == '__main__':
    asyncio.run(main())

Основные шаги

1. Определение важных адресов:
   • token_address — это адрес смарт-контракта токена USDT на сети BNB Chain.
   • spender — это адрес смарт-контракта Uniswap (или другого контракта), который будет использовать одобренные токены.
2. Создание клиента: client = Client( private_key='<ТУТ ВАШ ПРИВАТНЫЙ КЛЮЧ>', rpc='<https://1rpc.io/bnb>' ) Здесь создается объект клиента для работы с блокчейном. В качестве параметров передаются приватный ключ (который дает доступ к вашему аккаунту) и адрес RPC-сервера для BNB Chain (это публичный сервер, через который происходит подключение к сети).
3. Получение контракта токена: contract = client.w3.eth.contract( address=token_address, abi=TokenABI ) Мы получаем доступ к контракту USDT, указав его адрес и ABI (Application Binary Interface) — это описание функций, которые доступны в смарт-контракте.
4. Получение важной информации: decimals = await contract.functions.decimals().call() token_balance = await contract.functions.balanceOf(client.account.address).call() approved_amount = await contract.functions.allowance(client.account.address, spender).call()
   • decimals — это количество знаков после запятой у токена USDT. У USDT обычно 6 знаков (это означает, что 1 USDT = 1 * 10^6 в «сырых» данных).
   • token_balance — это текущий баланс USDT на вашем аккаунте.
   • approved_amount — это количество токенов, которые вы уже одобрили для использования spender (Uniswap).
5. Вывод информации: print('decimals:', decimals) print('token_balance:', token_balance / 10 ** decimals) print('approved_amount:', approved_amount / 10 ** decimals) Здесь выводится информация о балансе и уже одобренных токенах в удобном формате, чтобы увидеть значения в USDT, а не в «сырых» единицах.
6. Проверка, нужно ли делать approve: if approved_amount < token_balance: Если уже одобренная сумма меньше, чем текущий баланс, нужно обновить разрешение на использование токенов. Если баланс меньше, одобрение не требуется.
7. Создание и отправка транзакции approve: tx_hash = await client.send_transaction( to=token_address, data=contract.encodeABI('approve', args=( spender, token_balance )) ) Здесь создается и отправляется транзакция для вызова функции approve в контракте токена USDT. Эта функция разрешает Uniswap (spender) использовать ваш баланс токенов. Функция использует encodeABI, чтобы подготовить правильный вызов функции approve с аргументами: spender и сумма токенов для одобрения.
8. Проверка результата транзакции:
   • Если транзакция успешно отправлена, метод пытается дождаться ее подтверждения через verif_tx.Если транзакция прошла успешно, выводится сообщение с хешем транзакции:Если транзакция завершилась с ошибкой, выводится сообщение с ошибкой: await client.verif_tx(tx_hash=tx_hash) print(f'Transaction success!! tx_hash: {tx_hash.hex()}') print(f'Transaction error!! tx_hash: {tx_hash.hex()}; error: {err}')
   • Если транзакция не была создана, выводится сообщение об ошибке: print(f'Transaction error!!')
9. Уже одобрено: Если уже одобренный лимит (approved_amount) больше или равен текущему балансу, выводится сообщение: print('Already approved')

Подробнее про contract.encodeABI

Метод encodeABI кодирует вызов функции в виде строки, содержащей эти байты, которые можно включить в данные транзакции (data), передаваемой контракту. Он используется для взаимодействия с контрактами в сети.

Блокчейн работает с байтами и он ждёт информацию в таком виде:

Тут мы можем увидеть байты, с которыми работает блокчейн.

Кроме этого, мы тут можем подглять название функции, которая вызывается (approve).

Сигнатура функции ontract.encodeABI() выглядит следующим образом:

contract.encodeABI(fn_name='function_name', args=[arg1, arg2, ...])

То есть первый параметр — это название функции (в нашем случае approve), а второй параметр — агрументы функции, которые мы можем посмотреть в експлорере после нажатия кнопки Decode Input Data:

В нашем случае функция будет выглядеть следующим образом:

data=contract.encodeABI('approve',
                        args=(
                            spender,
                            token_balance
                        ))

До этого мы создавали контракт

contract = client.w3.eth.contract(
    address=token_address,
    abi=TokenABI
)

И указали в нём abi=TokenABI

Таким образом, функция encodeABI(…) по названию функции находит в TokenABI нужную функции и кодирует наши параметры в байт код (то есть нам не обязательно знать полный ABI контракта. Мы можем самостоятельно написать ABI и использовать его)

EIP-1559 транзакции

EIP-1559 — это улучшение сети Ethereum, предложенное и внедренное с целью изменить структуру комиссии за транзакции в блокчейне. Оно было активировано с обновлением London Hard Fork в августе 2021 года. EIP-1559 делает комиссии за транзакции более предсказуемыми и снижает вероятность переплаты за газ.

Ключевые изменения в EIP-1559

1. Base Fee (базовая комиссия) — минимальная комиссия, которая автоматически рассчитывается на уровне протокола и зависит от загруженности сети. Эта комиссия сжигается (удаляется из оборота), что делает систему более предсказуемой и может положительно влиять на цену ETH за счет дефляционного механизма.
2. Max Fee Per Gas — это максимальная сумма, которую пользователь готов заплатить за единицу газа для выполнения транзакции.
3. Max Priority Fee Per Gas — это дополнительные чаевые майнерам, чтобы стимулировать их обрабатывать транзакцию быстрее.

Как EIP-1559 транзакции отличаются от «обычных» транзакций (legacy)

До EIP-1559 транзакции работали по более простому принципу:

• Указывалась только gas price (цена за единицу газа), которую пользователь готов заплатить.
• Все комиссии платились майнерам.

В EIP-1559 вместо единого gas price теперь используются:

• Base Fee (базовая комиссия, динамически регулируется протоколом).
• Max Fee Per Gas (максимальная сумма за единицу газа).
• Max Priority Fee (приоритетная комиссия для майнеров).

Отличия в Python

1. Legacy-транзакции (до EIP-1559): При работе с legacy-транзакциями до EIP-1559 вам нужно было указывать только gasPrice — цену, которую вы готовы заплатить за каждый «газ». Пример транзакции: tx_params = { 'chainId': await self.w3.eth.chain_id, 'nonce': await self.w3.eth.get_transaction_count(self.account.address), 'from': AsyncWeb3.to_checksum_address(from_), 'to': AsyncWeb3.to_checksum_address(to), 'gasPrice': await self.w3.eth.gas_price, 'data': data, # если есть 'value': value, # если есть } Здесь единственная комиссия — это gasPrice, которая выплачивается полностью майнерам.
2. EIP-1559 транзакции: С EIP-1559 транзакции немного усложнились, так как теперь вам нужно указывать больше параметров:
   • maxFeePerGas: максимальная комиссия за единицу газа (включает базовую комиссию и чаевые).
   • maxPriorityFeePerGas: приоритетная комиссия (чаевые майнеру).
   Пример транзакции: max_priority_fee_per_gas = await self.w3.eth.max_priority_fee base_fee = (await self.w3.eth.get_block('latest'))['baseFeePerGas'] max_fee_per_gas = base_fee + max_priority_fee_per_gas tx_params = { 'chainId': await self.w3.eth.chain_id, 'nonce': await self.w3.eth.get_transaction_count(self.account.address), 'from': AsyncWeb3.to_checksum_address(from_), 'to': AsyncWeb3.to_checksum_address(to), 'maxFeePerGas': max_fee_per_gas, # максимальная общая комиссия 'maxPriorityFeePerGas': max_priority_fee_per_gas, # приоритетная комиссия майнеру 'data': data, # если есть 'value': value, # если есть } Расчет приоритетной комиссии (max_priority_fee_per_gas) max_priority_fee_per_gas = await self.w3.eth.max_priority_fee
   • max_priority_fee_per_gas — это приоритетная комиссия (чаевые), которую вы платите майнерам, чтобы стимулировать их обработать вашу транзакцию быстрее.
   • self.w3.eth.max_priority_fee — это асинхронный вызов, который запрашивает у сети Ethereum текущую среднюю приоритетную комиссию. Она может варьироваться в зависимости от того, как загружена сеть.
   Получение Base Fee (базовой комиссии) base_fee = (await self.w3.eth.get_block('latest'))['baseFeePerGas']
   • base_fee — это базовая комиссия за газ, которая автоматически рассчитывается сетью Ethereum для каждого блока. Она изменяется в зависимости от загруженности сети и устанавливается на уровне протокола.
   • self.w3.eth.get_block('latest') — асинхронный запрос последнего блока. В нем содержится параметр baseFeePerGas, который и есть текущая базовая комиссия.
   Расчет Max Fee Per Gas max_fee_per_gas = base_fee + max_priority_fee_per_gas
   • max_fee_per_gas — это максимальная комиссия, которую вы готовы заплатить за каждую единицу газа. Она состоит из двух компонентов:
     1. Base Fee (базовая комиссия) — сжигается сетью.
     2. Priority Fee (приоритетная комиссия) — выплачивается майнерам как чаевые.
   Здесь максимальная комиссия складывается из суммы базовой комиссии и приоритетной комиссии.

Измененный код send_transaction с учетом eip-1559

class Client:
    ...

    async def send_transaction(
            self,
            to: str | ChecksumAddress,
            data: HexStr | None = None,
            from_: str | ChecksumAddress | None = None,
            increase_gas: float = 1,
            value: int | None = None,
            eip1559: bool = True,
            max_priority_fee_per_gas: int | None = None
    ) -> HexBytes | None:
        if not from_:
            from_ = self.account.address

        tx_params = {
            'chainId': await self.w3.eth.chain_id,
            'nonce': await self.w3.eth.get_transaction_count(self.account.address),
            'from': AsyncWeb3.to_checksum_address(from_),
            'to': AsyncWeb3.to_checksum_address(to),
        }

        if eip1559:
            if max_priority_fee_per_gas is None:
                max_priority_fee_per_gas = await self.w3.eth.max_priority_fee
            base_fee = (await self.w3.eth.get_block('latest'))['baseFeePerGas']
            max_fee_per_gas = base_fee + max_priority_fee_per_gas
            tx_params['maxFeePerGas'] = max_fee_per_gas  # максимальная общая комиссия
            tx_params['maxPriorityFeePerGas'] = max_priority_fee_per_gas  # приоритетная комиссия майнеру
        else:
            tx_params['gasPrice'] = await self.w3.eth.gas_price

        if data:
            tx_params['data'] = data
        if value:
            tx_params['value'] = value

        gas = await self.w3.eth.estimate_gas(tx_params)
        tx_params['gas'] = int(gas * increase_gas)

        sign = self.w3.eth.account.sign_transaction(tx_params, self.private_key)
        return await self.w3.eth.send_raw_transaction(sign.rawTransaction)

    ...

В данном коде мы добавили новый параметр eip1559 в функцию send_transaction Если этот параметр True, в словарь tx_params добавятся ключи maxFeePerGas и maxPriorityFeePerGas. В ином случае в словарь будет добавлен ключ gasPrice.

Также добавили параметр max_priority_fee_per_gas, чтобы была возможность задать это значение вручную

Добавление proxy и headers в Client

from fake_useragent import UserAgent
from web3 import AsyncWeb3
from eth_account.signers.local import LocalAccount

class Client:
    private_key: str
    rpc: str
    proxy: str | None
    w3: AsyncWeb3
    account: LocalAccount

    def __init__(self, private_key: str, rpc: str, proxy: str | None = None):
        self.private_key = private_key
        self.rpc = rpc
        self.proxy = proxy
        
        if self.proxy:
            if '://' not in self.proxy:
                self.proxy = f'http://{self.proxy}'

        self.headers = {
            'accept': '*/*',
            'accept-language': 'en-US,en;q=0.9',
            'content-type': 'application/json',
            'user-agent': UserAgent().chrome
        }

        self.w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(
            endpoint_uri=rpc,
            request_kwargs={'proxy': self.proxy, 'headers': self.headers}
        ))
        self.account = self.w3.eth.account.from_key(private_key)

API для получения цены

Код функции get_token_price()

from curl_cffi.requests import AsyncSession

async def get_token_price(token_symbol='ETH') -> float | None:
    token_symbol = token_symbol.upper()

    if token_symbol in ('USDC', 'USDT', 'DAI', 'CEBUSD', 'BUSD', 'USDC.E'):
        return 1
    if token_symbol == 'WETH':
        token_symbol = 'ETH'
    if token_symbol == 'WBTC':
        token_symbol = 'BTC'

    for _ in range(5):
        try:
            async with AsyncSession() as session:
                response = await session.get(f'<https://api.binance.com/api/v3/depth?limit=1&symbol={token_symbol}USDT>')
                result_dict = response.json()
                if 'asks' not in result_dict:
                    return
                return float(result_dict['asks'][0][0])
        except Exception:
            await asyncio.sleep(5)
    raise ValueError(f'Can not get {token_symbol} price from Binance API')

Очень часто нам необходимо узнать актуальную цену токена (например для того, чтобы высчитать slippage)

Для решения этой задачи можно использовать Binance API

Для оптимизации работы функции, в неё внедрена дополнительная логика в первом ветвлении

В данном примере мы асинхронно обращаемся к Binance API и получаем стакан в отношении к USDT

Библиотеки для выполнения асинхронных запросов

Для реализации асинхронных запросов используется библиотека curl_cffi

curl-cffi

Но также может подойти любая другая библиотека для асинхронных запросов:

aiohttp

httpx

Как выполнить асинхронный GET запрос?

from curl_cffi.requests import AsyncSession

async with AsyncSession() as session:
    response = await session.get(
        f'<https://api.binance.com/api/v3/depth?limit=1&symbol={token_symbol}USDT>'
    )
    result_dict = response.json()
    if 'asks' not in result_dict:
        return
    return float(result_dict['asks'][0][0])

Как это работает:

1. Асинхронная сессия:
   • AsyncSession() — это асинхронный объект сессии, который можно использовать для выполнения HTTP-запросов.
   • Сессия создается в контексте async with, что автоматически управляет созданием и закрытием соединений, упрощая работу с ресурсами.
2. Асинхронный запрос GET:
   • await session.get(...) — асинхронный вызов HTTP-запроса метода GET к указанному URL. В данном примере запрашивается информация о торговом символе с использованием API Binance.
   • await используется, потому что запрос выполняется асинхронно. Это позволяет другим частям программы продолжать выполняться, пока идет ожидание ответа.
3. Получение и разбор ответа:
   • response.json() — возвращает ответ в формате JSON, преобразованный в словарь Python. Эта функция автоматически разбирает текст ответа как JSON.
4. Проверка содержимого:
   • Если в полученном словаре нет ключа 'asks', функция возвращает None.
   • Если ключ существует, то берется первое значение цены из массива 'asks', и оно преобразуется в число типа float.

Добавление метода get_token_price() в Client

Данный метод можно добавить к нам в клиента как статический метод

import asyncio

from curl_cffi.requests import AsyncSession

class Client:
    ...

    @staticmethod
    async def get_token_price(token_symbol='ETH') -> float | None:
        token_symbol = token_symbol.upper()

        if token_symbol in ('USDC', 'USDT', 'DAI', 'CEBUSD', 'BUSD', 'USDC.E'):
            return 1
        if token_symbol == 'WETH':
            token_symbol = 'ETH'
        if token_symbol == 'WBTC':
            token_symbol = 'BTC'

        for _ in range(5):
            try:
                async with AsyncSession() as session:
                    response = await session.get(
                        f'<https://api.binance.com/api/v3/depth?limit=1&symbol={token_symbol}USDT>')
                    result_dict = response.json()
                    if 'asks' not in result_dict:
                        return
                    return float(result_dict['asks'][0][0])
            except Exception:
                await asyncio.sleep(5)
        raise ValueError(f'Can not get {token_symbol} price from Binance API')

    ...

Полный код (в один файл)

import asyncio
from hexbytes import HexBytes

from curl_cffi.requests import AsyncSession
from fake_useragent import UserAgent

from web3 import AsyncWeb3
from web3.exceptions import Web3Exception
from eth_typing import ChecksumAddress, HexStr
from eth_account.signers.local import LocalAccount

TokenABI = [
    {
        'constant': True,
        'inputs': [],
        'name': 'name',
        'outputs': [{'name': '', 'type': 'string'}],
        'payable': False,
        'stateMutability': 'view',
        'type': 'function'
    },
    {
        'constant': True,
        'inputs': [],
        'name': 'symbol',
        'outputs': [{'name': '', 'type': 'string'}],
        'payable': False,
        'stateMutability': 'view',
        'type': 'function'
    },
    {
        'constant': True,
        'inputs': [],
        'name': 'totalSupply',
        'outputs': [{'name': '', 'type': 'uint256'}],
        'payable': False,
        'stateMutability': 'view',
        'type': 'function'
    },
    {
        'constant': True,
        'inputs': [],
        'name': 'decimals',
        'outputs': [{'name': '', 'type': 'uint256'}],
        'payable': False,
        'stateMutability': 'view',
        'type': 'function'
    },
    {
        'constant': True,
        'inputs': [{'name': 'who', 'type': 'address'}],
        'name': 'balanceOf',
        'outputs': [{'name': '', 'type': 'uint256'}],
        'payable': False,
        'stateMutability': 'view',
        'type': 'function'
    },
    {
        'constant': True,
        'inputs': [{'name': '_owner', 'type': 'address'}, {'name': '_spender', 'type': 'address'}],
        'name': 'allowance',
        'outputs': [{'name': 'remaining', 'type': 'uint256'}],
        'payable': False,
        'stateMutability': 'view',
        'type': 'function'
    },
    {
        'constant': False,
        'inputs': [{'name': '_spender', 'type': 'address'}, {'name': '_value', 'type': 'uint256'}],
        'name': 'approve',
        'outputs': [],
        'payable': False,
        'stateMutability': 'nonpayable',
        'type': 'function'
    },
    {
        'constant': False,
        'inputs': [{'name': '_to', 'type': 'address'}, {'name': '_value', 'type': 'uint256'}],
        'name': 'transfer',
        'outputs': [], 'payable': False,
        'stateMutability': 'nonpayable',
        'type': 'function'
    }
]

class Client:
    private_key: str
    rpc: str
    proxy: str | None
    w3: AsyncWeb3
    account: LocalAccount

    def __init__(self, private_key: str, rpc: str, proxy: str | None = None):
        self.private_key = private_key
        self.rpc = rpc
        self.proxy = proxy

        self.headers = {
            'accept': '*/*',
            'accept-language': 'en-US,en;q=0.9',
            'content-type': 'application/json',
            'user-agent': UserAgent().chrome
        }

        self.w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(
            endpoint_uri=rpc,
            request_kwargs={'proxy': self.proxy, 'headers': self.headers}
        ))
        self.account = self.w3.eth.account.from_key(private_key)

        if self.proxy:
            if 'http' not in self.proxy:
                self.proxy = f'http://{self.proxy}'

    async def send_transaction(
            self,
            to: str | ChecksumAddress,
            data: HexStr | None = None,
            from_: str | ChecksumAddress | None = None,
            increase_gas: float = 1,
            value: int | None = None,
            eip1559: bool = True,
            max_priority_fee_per_gas: int | None = None
    ) -> HexBytes | None:
        if not from_:
            from_ = self.account.address

        tx_params = {
            'chainId': await self.w3.eth.chain_id,
            'nonce': await self.w3.eth.get_transaction_count(self.account.address),
            'from': AsyncWeb3.to_checksum_address(from_),
            'to': AsyncWeb3.to_checksum_address(to),
        }

        if eip1559:
            if max_priority_fee_per_gas is None:
                max_priority_fee_per_gas = await self.w3.eth.max_priority_fee
            base_fee = (await self.w3.eth.get_block('latest'))['baseFeePerGas']
            max_fee_per_gas = base_fee + max_priority_fee_per_gas
            tx_params['maxFeePerGas'] = max_fee_per_gas  # максимальная общая комиссия
            tx_params['maxPriorityFeePerGas'] = max_priority_fee_per_gas  # приоритетная комиссия майнеру
        else:
            tx_params['gasPrice'] = await self.w3.eth.gas_price

        if data:
            tx_params['data'] = data
        if value:
            tx_params['value'] = value

        gas = await self.w3.eth.estimate_gas(tx_params)
        tx_params['gas'] = int(gas * increase_gas)

        sign = self.w3.eth.account.sign_transaction(tx_params, self.private_key)
        return await self.w3.eth.send_raw_transaction(sign.rawTransaction)

    async def verif_tx(self, tx_hash: HexBytes, timeout: int = 200) -> str:
        data = await self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=timeout)
        if data.get('status') == 1:
            return tx_hash.hex()
        raise Web3Exception(f'transaction failed {data["transactionHash"].hex()}')

    @staticmethod
    async def get_token_price(token_symbol='ETH') -> float | None:
        token_symbol = token_symbol.upper()

        if token_symbol in ('USDC', 'USDT', 'DAI', 'CEBUSD', 'BUSD', 'USDC.E'):
            return 1
        if token_symbol == 'WETH':
            token_symbol = 'ETH'
        if token_symbol == 'WBTC':
            token_symbol = 'BTC'

        for _ in range(5):
            try:
                async with AsyncSession() as session:
                    response = await session.get(
                        f'<https://api.binance.com/api/v3/depth?limit=1&symbol={token_symbol}USDT>')
                    result_dict = response.json()
                    if 'asks' not in result_dict:
                        return
                    return float(result_dict['asks'][0][0])
            except Exception:
                await asyncio.sleep(5)
        raise ValueError(f'Can not get {token_symbol} price from Binance API')

async def main():
    # Задача: сделать апрув всего баланса USDT для uniswap

    token_address = AsyncWeb3.to_checksum_address('0x55d398326f99059fF775485246999027B3197955')
    spender = AsyncWeb3.to_checksum_address('0x000000000022D473030F116dDEE9F6B43aC78BA3')

    client = Client(
        private_key='<YOUR PRIVATE KEY HERE>',
        rpc='<https://1rpc.io/bnb>',
        proxy='<YOUR PROXY HERE>'
    )

    contract = client.w3.eth.contract(
        address=token_address,
        abi=TokenABI
    )

    decimals = await contract.functions.decimals().call()
    token_balance = await contract.functions.balanceOf(client.account.address).call()
    approved_amount = await contract.functions.allowance(client.account.address, spender).call()

    print('decimals:', decimals)
    print('token_balance:', token_balance / 10 ** decimals)
    print('approved_amount:', approved_amount / 10 ** decimals)

    if approved_amount < token_balance:
        tx_hash = await client.send_transaction(
            to=token_address,
            data=contract.encodeABI('approve',
                                    args=(
                                        spender,
                                        token_balance
                                    ))
        )
        if tx_hash:
            try:
                await client.verif_tx(tx_hash=tx_hash)
                print(f'Transaction success!! tx_hash: {tx_hash.hex()}')
            except Exception as err:
                print(f'Transaction error!! tx_hash: {tx_hash.hex()}; error: {err}')
        else:
            print(f'Transaction error!!')
    else:
        print('Already approved')

if __name__ == '__main__':
    asyncio.run(main())

Распределение по файлам

Писать код в один файл – неудобно так как код в любом случае у нас будет расти и в конечном итоге мы получим код на 34872364826348 строк, которые будет невозможно прочитать и поддерживать. Чтобы такого с нами не случилось, мы сделаем небольшой рефакторинг кода.

Класс TokenABI вынесем в файл data.models

Класс Client вынесем в файл client

Директорию abis перенесем в директорию data

Таким образом, файл main.py будет выглядеть следующим образом:

import asyncio

from web3 import AsyncWeb3

from data.models import TokenABI
from client import Client

async def main():
    # Задача: сделать апрув всего баланса USDT для uniswap

    token_address = AsyncWeb3.to_checksum_address('0x55d398326f99059fF775485246999027B3197955')
    spender = AsyncWeb3.to_checksum_address('0x000000000022D473030F116dDEE9F6B43aC78BA3')

    client = Client(
        private_key='<YOUR PRIVATE KEY HERE>',
        rpc='<https://1rpc.io/bnb>',
        proxy='<YOUR PROXY HERE>'
    )

    contract = client.w3.eth.contract(
        address=token_address,
        abi=TokenABI
    )

    decimals = await contract.functions.decimals().call()
    token_balance = await contract.functions.balanceOf(client.account.address).call()
    approved_amount = await contract.functions.allowance(client.account.address, spender).call()

    print('decimals:', decimals)
    print('token_balance:', token_balance / 10 ** decimals)
    print('approved_amount:', approved_amount / 10 ** decimals)

    if approved_amount < token_balance:
        tx_hash = await client.send_transaction(
            to=token_address,
            data=contract.encodeABI('approve',
                                    args=(
                                        spender,
                                        token_balance
                                    ))
        )
        if tx_hash:
            try:
                await client.verif_tx(tx_hash=tx_hash)
                print(f'Transaction success!! tx_hash: {tx_hash.hex()}')
            except Exception as err:
                print(f'Transaction error!! tx_hash: {tx_hash.hex()}; error: {err}')
        else:
            print(f'Transaction error!!')
    else:
        print('Already approved')

if __name__ == '__main__':
    asyncio.run(main())

Финальный код на GitHub

https://github.com/fidry/python_web3_course_3/tree/main/lesson_02_send_transaction