> ccxt — для работы с API криптобирж

> asyncio — для асинхронных запросов

> python-dotenv — для безопасного хранения API-ключей

> requests (если нужно работать с REST API напрямую)

> Подключение к API бирж ⬇️

Используем библиотеку ccxt, чтобы подключиться к нескольким биржам (например, Binance и KuCoin)

> Загрузка всех торговых пар (token/market)

Загружаем список всех токенов и торговых пар с каждой биржи.

Это делается с помощью exchange.load_markets().

{
"BTC/USDT": ["Binance", "KuCoin"],
"ETH/USDT": ["Binance"],
}

> Сохраняем предыдущие значения цен для каждого токена 😊

> Для каждой новой цены сравниваем с предыдущей:

Если изменение цены превышает установленный порог (например, 5%), выводим уведомление:

> Используем asyncio для параллельной работы с API бирж, чтобы не блокировать выполнение.

> Каждые 30 секунд повторяем процесс мониторинга 🚬

> Подключение к API бирж 👎

> Загрузка всех токенов и определение, на каких биржах они доступны.

> Мониторинг цен для всех токенов.

> Вычисление изменений и поиск резких скачков.

> Уведомление о скачках в бот

> Пауза (несколько минут) и повторение цикла.

Multicall (MC) - смарт-контракт, принимающий массив вызовов к другим контрактам и обрабатывает их одновременно.

Он возвращает результаты всех вызовов в одном ответе снижая нагрузку на сеть и делает взаимодействие с блокчейном более эффективным.

web3.py - база

eth-brownie - фреймворк для разработки смарт-контрактов, также поддерживает multicall.

multicall.py - библиотека для работы с Multicall.

> Используем UNISWAP Factory Contract Address - контракт для управления пулами + его ABI

> Получает адреса всех пар ликвидности на Uniswap V2.

> Для каждой пары выполняем запрос к функции getReserves(), чтобы получить текущие резервы токенов.

> Используем Multicall для объединения множества вызовов в один, что снижает нагрузку на RPC и ускоряет выполнение.

> async/await

> WebSocket для подписки на данные в реальном времени

> threading или multiprocessing для параллельной обработки

captcha_request = requests.post("http://2captcha.com/in.php", data={
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": SITE_KEY,
"pageurl": URL,
"json": 1
}).json()

page.wait_for_selector('iframe[src*="hcaptcha.com"]', timeout=60000)

# Получение токена и выполнение формы
captcha_response = page.evaluate('''() => {
    return document.querySelector('[name="h-captcha-response"]').value;
}''')

scraper = cloudscraper.create_scraper()
response = scraper.get("https://binance.com")

> Автоматизации торговли (быстрые сделки, арбитраж).

> Фарминг дропов

> Минт NFT без задержек.

> Подтверждения транзакций без ручного ввода.

> Мониторинг данных (цены, ликвидность, ордербук).

> Фронт-раннинг
> Сэндвич-атаки
> DEX-арбитраж
> Ликвидация позиций

🏃 Фронт-раннинг - Бот находит выгодную транзакцию, которую кто-то собирается выполнить (например, крупную покупку токена).

Создаёт свою транзакцию с повышенной комиссией и отправляет её в блок раньше оригинальной. В результате цена токена растёт, и бот продаёт его с прибылью.

🥪 Сэндвич-атаки - Бот обнаруживает транзакцию на покупку токена, отправляет свою покупку перед этой транзакцией, поднимая цену. Затем он отправляет продажу после чужой покупки, фиксируя прибыль.

> web3.py библиотека для взаимодействия с блокчейном Ethereum.

> Infura / Alchemy для доступа к Ethereum-ноде.

> Мемпул-анализ - механизм мониторинга транзакций в реальном времени.

> Алгоритмы арбитража или фронт-раннинга для поиска прибыльных транзакций.

> Быстрое исполнение - оптимизированное создание и подпись транзакций.

pending_tx_filter = w3.eth.filter('pending')

def analyze_mempool():
pending_txs = pending_tx_filter.get_new_entries()
for tx_hash in pending_txs:
try:
tx = w3.eth.get_transaction(tx_hash)
print(f"📡 New TX: {tx['from']} -> {tx['to']} | Gas: {tx['gasPrice']}")
except Exception:
continue

while True:
analyze_mempool()

1. Бот для отслеживания крупных транзакций (Whale Tracker 2.0)

Бот мониторит блокчейн Ethereum в реальном времени, отслеживая крупные транзакции (например, больше 1000 ETH). При обнаружении отправляет уведомление в Telegram или Discord.

def filter_large_transactions(transactions):
global LAST_BLOCK
large_txs = []

for tx in transactions:
    block_number = int(tx["blockNumber"])
    if LAST_BLOCK and block_number <= LAST_BLOCK:
        continue  # Пропускаем уже обработанные транзакции

    eth_value = int(tx["value"]) / 10**18  # Переводим в ETH
    if eth_value >= ETH_THRESHOLD:
        large_txs.append(tx)

if transactions:
    LAST_BLOCK = int(transactions[0]["blockNumber"])  # Обновляем последний блок
return large_txs

🗒 Приложение анализирует историю транзакций кошелька и рассчитывает прибыль/убыток на основе покупки/продажи токенов и стоимости газа.

Основной анализатор прибыли и убытков

def analyze_pnl():
transactions = get_wallet_transactions()
total_income = 0
total_expense = 0
gas_fees = 0

for tx in transactions:
    eth_value = int(tx["value"]) / 10**18  # ETH перевод из Wei
    gas_cost = (int(tx["gasPrice"]) * int(tx["gasUsed"])) / 10**18  # Стоимость газа в ETH
    timestamp = int(tx["timeStamp"])
    date_str = datetime.utcfromtimestamp(timestamp).strftime('%d-%m-%Y')

    # Получаем цену ETH на дату транзакции
    eth_price = get_eth_price(date_str)
    if not eth_price:
        continue  # Пропускаем, если цену не удалось получить

    usd_value = eth_value * eth_price
    gas_usd = gas_cost * eth_price

    if tx["to"].lower() == ETH_ADDRESS.lower():
        total_income += usd_value  # Приход ETH
    else:
        total_expense += usd_value  # Отправка ETH
        gas_fees += gas_usd  # Комиссия за газ

    time.sleep(60)  # Чтоб не перегружать API

# 🔥 Итоговая прибыль (PnL)
pnl = total_income - (total_expense + gas_fees)
print(f"💰 Приход: ${total_income:.2f}\\\\n📉 Расходы: ${total_expense:.2f}\\\\n⛽️ Комиссия за газ: ${gas_fees:.2f}\\\\n📊 Итоговый PnL: ${pnl:.2f}")
analyze_pnl()

3. Дашборд активности смарт-контракта

Веб-приложение, отображающее информацию о вызовах функций конкретного смарт-контракта, количестве пользователей и обороте токенов.

блокчейн - изолированная среда, он не может просто так получать данные с интернета - цены активов, мировые событие etc и ChainLink решает эту проблему

Оракулы для взаимодействия с реальным миром
Если нужно подтягивать данные с бирж, API погоды или спортивных событий, можем написать Python-скрипт, который отправляет данные в смарт-контракт через Chainlink.

contract OracleConsumer {
int256 public btcPrice;
address public oracle;

event PriceUpdated(int256 newPrice);

modifier onlyOracle() {
    require(msg.sender == oracle, "Not authorized");
    _;
}

constructor(address _oracle) {
    oracle = _oracle; // Адрес оракула
}

function updatePrice(int256 _price) external onlyOracle {
    btcPrice = _price;
    emit PriceUpdated(_price);
}

# Получение цены BTC/USD
def get_btc_price():
url = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usd"
response = requests.get(url)
data = response.json()
return int(data["bitcoin"]["usd"] * 10**8)  # Конвертируем в формат int256

# Обновление цены в смарт-контракте
def update_smart_contract():
btc_price = get_btc_price()
print(f"Полученная цена BTC/USD: {btc_price / 10**8} USD")

# Формирование транзакции
tx = contract.functions.updatePrice(btc_price).build_transaction({
    'from': ORACLE_ADDRESS,
    'nonce': w3.eth.get_transaction_count(ORACLE_ADDRESS),
    'gas': 100000,
    'gasPrice': w3.to_wei('10', 'gwei')
})

Python-скрипт запрашивает цену BTC/USD с CoinGecko 💰

> Формирует транзакцию и подписывает её.

> Отправляет данные в смарт-контракт.

> Контракт обновляет цену, и любое приложение в блокчейне может её использовать.

Доступ к блокчейн-данным через Python 🤩

скрипт показывает, как получить:

✅ Баланс адреса в ETH

✅ Историю транзакций

✅ Актуальную цену ETH/USD через Chainlink

# Получение баланса ETH
def get_eth_balance(address):
balance_wei = w3.eth.get_balance(address)
balance_eth = w3.from_wei(balance_wei, 'ether')
return balance_eth

# Получение актуальной цены ETH/USD через Chainlink

def get_eth_price():
contract = w3.eth.contract(address=CHAINLINK_FEED, abi=FEED_ABI)
latest_data = contract.functions.latestRoundData().call()
price = latest_data[1] / 10**8  # Chainlink возвращает цену с 8 знаками после запятой
return price


# Получение последних 10 транзакций с Etherscan API

def get_recent_transactions(address):
ETHERSCAN_API_KEY = "YOUR_ETHERSCAN_API_KEY"
url = f"https://api.etherscan.io/api?module=account&action=txlist&address={address}&startblock=0&endblock=99999999&sort=desc&apikey={ETHERSCAN_API_KEY}"
response = requests.get(url).json()
transactions = response["result"][:10]  # Берем последние 10 транзакций
return transactions

Подключение к Solana (Devnet/Testnet/Mainnet) ⬇️

> Генерация нового ключа и кошелька.

> Получение баланса кошелька.

> Отправка SOL с одного адреса на другой.

from solana.rpc.api import Client

# Подключаемся к Solana Devnet (для тестирования)

client = Client("https://api.devnet.solana")

# работает ли соединение

print(client.is_connected())

keypair = Keypair.generate()
public_key = keypair.public_key
private_key = base58.b58encode(keypair.secret()).decode()  # Для сохранения

# Получаем баланс

def get_balance(pubkey):
balance = client.get_balance(pubkey)
return balance["result"]["value"]

# Запрашиваем airdrop (только в Devnet)

print("🚀 Запрашиваем airdrop 1 SOL...")
client.request_airdrop(public_key, 1_000_000_000)
time.sleep(20)  # Нужно подождать подтверждения транзакции

# Отправка 0.01 SOL другому кошельку

receiver = PublicKey("8K4wR1vYyqM3Z9oeGypvVX5vKYJefT3gT9sAeZKqPmWv") 
amount = 10_000_000  # 0.01 SOL в лампортах

txn = Transaction().add(
transfer({"from_pubkey": public_key, "to_pubkey": receiver, "lamports": amount})
)

print("📤 Отправляем 0.01 SOL...")

try:
txn_signature = client.send_transaction(txn, keypair, opts=TxOpts(skip_confirmation=False))
print(f"✅ Транзакция отправлена! Signature: {txn_signature['result']}")
except Exception as e:
print(f"❌ Ошибка: {e}")

Работа со смарт-контрактами Solana

> Создание и управление токенами SPL

> Подключение и взаимодействие с Raydium и другими Solana Dex

> Работа с WebSockets

IDL (Interface Definition Language) JSON-файл, описывающий, какие методы есть в программе, какие аргументы они принимают, и какие аккаунты участвуют в транзакциях.

🤩 На основе IDL создается объект программы, который предоставляет методы для взаимодействия с программой на Solana.

from anchorpy import Idl, Program
from solana.rpc.api import Client
from solana.publickey import PublicKey
from solana.account import Account
import asyncio

# 1. Подключение к Solana RPC

client = Client("[https://api.devnet.solana.com](https://api.devnet.solana.com/)")

#2. Загрузка IDL

with open("idl.json", "r") as f:
idl = Idl.from_json(f.read())

#3. Создание объекта программы

program_id = PublicKey("ПрограммаПубличныйКлюч")
program = Program(idl, program_id, client)

#4. Подготовка данных

from_account = PublicKey("ПубличныйКлючОтправителя")
to_account = PublicKey("ПубличныйКлючПолучателя")
amount = 1_000_000  # 0.001 SOL

#5. Приватный ключ отправителя

private_key = bytes([...])  # Замените на ваш приватный ключ
signer = Account(private_key)

#6. Вызов метода программы

async def call_transfer():
tx = await program.rpc["transfer"](
amount,
accounts={
"from": from_account,
"to": to_account,
"system_program": PublicKey("11111111111111111111111111111111")
},
signers=[signer]
)
print("Транзакция отправлена:", tx)

#Запуск асинхронной функции

asyncio.run(call_transfer())

1. Формирование инструкции:

> Клиентский код на основе IDL формирует инструкцию для вызова метода программы

2. Создание транзакции:

> Инструкция добавляется в транзакцию.

3. Подпись транзакции:

> Транзакция подписывается отправителем.

4. Отправка транзакции:

> Транзакция отправляется в сеть Solana через RPC.

⌨️ Хотя Anchor автоматически генерирует IDL и клиентский код, нам всё равно нужно:

1. Настроить подключение к Solana:

> Указать, к какому RPC-узлу подключаться (например, Devnet или Mainnet)

2. Указать параметры вызова

> Передать аргументы метода (сумму перевода, etc).

> Указать аккаунты, которые участвуют в транзакции (отправитель и получатель).

> Подписать транзакцию Указать, кто подписывает транзакцию (отправитель).

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

sender = Keypair()  # Ваш кошелек
receiver = PublicKey("RECEIVER_PUBLIC_KEY")  # Публичный ключ получателя
amount = 1000000  # Сумма в lamports (1 SOL = 1000000000 lamports)

# Создаем транзакцию с повышенной комиссией
txn = Transaction().add(transfer(TransferParams(
    from_pubkey=sender.public_key,
    to_pubkey=receiver,
    lamports=amount
)))

# Указываем приоритет (добавляем комиссию)
txn.recent_blockhash = client.get_recent_blockhash()["result"]["value"]["blockhash"]
txn.fee_payer = sender.public_key
txn.sign(sender)

client = Client("тут_ссылка_на_rpc")

import asyncio
from solana.rpc.async_api import AsyncClient

async def send_transaction_async():
    client = AsyncClient("https://api.mainnet-beta.solana.com")
    # Ваш код для создания и отправки транзакции
    await client.send_transaction(txn)
    await client.close()

asyncio.run(send_transaction_async())

В Solana есть механизм Durable Nonce, который позволяет избежать проблем с устаревшими nonce и ускорить обработку транзакций.

😱 nonce - уникальное число, которое добавляется к данным для обеспечения их уникальности или для выполнения определенных условий

nonce_account = Keypair()
txn = Transaction().add(create_nonce_account(
    from_pubkey=sender.public_key,
    nonce_pubkey=nonce_account.public_key,
    authority=sender.public_key,
    lamports=1000000
))

import requests

# Эндпоинт для получения маршрутов

url = "https://api.jup.ag/swap/v1/quote"

# Параметры запроса

params = {
"inputMint": "So11111111111111111111111111111111111111112",  # SOL
"outputMint": "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v",  # USDC
"amount": 1000000000,  # 1 SOL (в lamports)
"slippageBps": 50,  # Допустимое проскальзывание (0.5%)
}

# Отправляем запрос

response = requests.get(url, params=params)

# Проверяем ответ

if response.status_code == 200:
routes = response.json()
print("Available routes:", routes)
else:
print("Error:", response.status_code, response.text)

from solana.rpc.api import Client
from solana.transaction import Transaction
from solana.keypair import Keypair
import base64

# Подключаемся к devnet

client = Client("https://api.devnet.solana.com")

# твой кошель

wallet = Keypair.from_secret_key(твой_приватник)  

# Эндпоинт для выполнения свопа

swap_url = "https://api.jup.ag/swap/v1/swap"

# Данные для свопа

swap_data = {
"quoteResponse": routes,  # Используем маршрут из предыдущего шага
"userPublicKey": str(wallet.public_key),
"wrapAndUnwrapSol": True,  # Автоматически конвертировать SOL в wSOL и обратно
}

# Получаем транзакцию для свопа

swap_response = requests.post(swap_url, json=swap_data)

if swap_response.status_code == 200:
swap_transaction = swap_response.json()
print("Swap transaction:", swap_transaction)

# Декодируем транзакцию
raw_transaction = base64.b64decode(swap_transaction["swapTransaction"])

# Создаем объект транзакции
transaction = Transaction.deserialize(raw_transaction)

# Подписываем транзакцию
transaction.sign(wallet)

# Отправляем транзакцию
txid = client.send_transaction(transaction, wallet)
print("Transaction sent:", txid)

else:
print("Error:", swap_response.status_code, swap_response.text)

tokens_url = "https://api.jup.ag/tokens/v1/mints/all"

tokens_response = requests.get(tokens_url)

if tokens_response.status_code == 200:
tokens = tokens_response.json()
print("Available tokens:", tokens)
else:
print("Error:", tokens_response.status_code, tokens_response.text)

pip install solana bip32utils mnemonic (либы для соланы и мнем-фразы)

from solana.keypair import Keypair

# создаем новый KeyPair

keypair = Keypair()

# публичный и приватный ключи

print("Public Key:", keypair.public_key)
print("Private Key:", keypair.secret_key)

from mnemonic import Mnemonic

# генерация мнемонической фразы

mnemo = Mnemonic("english")
mnemonic_phrase = mnemo.generate(strength=128)  # 12 слов

print("Mnemonic Phrase:", mnemonic_phrase)

import hashlib
import hmac
from bip32utils import BIP32Key

# преобразуем мнемоническую фразу в seed

seed = mnemo.to_seed(mnemonic_phrase)

# деривация приватного ключа по пути BIP-44

# путь для Solana: m/44'/501'/0'/0'

bip32_key = BIP32Key.fromEntropy(seed)
derived_key = bip32_key.ChildKey(44 + BIP32Key.HARDEN).ChildKey(501 + BIP32Key.HARDEN).ChildKey(0 + BIP32Key.HARDEN).ChildKey(0).ChildKey(0)

# приватный ключ (32 байта)

private_key = derived_key.PrivateKey()
print("Derived Private Key:", private_key.hex())

from solana.keypair import Keypair

# Преобразуем приватный ключ в KeyPair

keypair = Keypair.from_secret_key(private_key)

# публичный и приватный ключи

print("Public Key from Derived Private Key:", keypair.public_key)
print("Private Key from Derived Private Key:", keypair.secret_key.hex())

😭 Генерируется 12 слов, которые можно использовать для восстановления кошелька.

Мнемоническая фраза преобразуется в seed с помощью алгоритма PBKDF2.