Розробка бота для моніторингу мемпулу

Розробка бота для моніторингу мемпула

Мемпул — це не просто черга транзакцій. Це джерело даних про намірення: що збирається зробити ринок в наступному блоці. MEV-ботів, які заробляють на цих даних, на Ethereum працює сотні. Але моніторинг мемпула потрібен не тільки для арбітражу — liquidation-боти DeFi-протоколів, системи раннього виявлення атак, захист від front-running, аналітика газових цін — усе це будується на читанні мемпула в реальному часі.

Доступ до мемпула: три рівні

Рівень 1: публічний mempool через RPC

Найпростіший спосіб — eth_subscribe("newPendingTransactions") через WebSocket:

import { ethers } from 'ethers';

const wsProvider = new ethers.WebSocketProvider(
  `wss://eth-mainnet.g.alchemy.com/v2/${process.env.ALCHEMY_KEY}`
);

wsProvider.on('pending', async (txHash: string) => {
  const tx = await wsProvider.getTransaction(txHash);
  if (!tx) return;  // транзакція могла бути dropped

  await processPendingTx(tx);
});

Проблема: публічний RPC бачить тільки частину мемпула. Нода розкриває pending-транзакцію тільки після того, як розповсюдить її в мережу. Приватні транзакції (через Flashbots, MEV Blocker, Cowswap) — не видні взагалі.

Рівень 2: власна нода з txpool API

Запустіть власний Geth/Reth з відкритим txpool namespace:

# Geth з включеним txpool API
geth --mainnet \
  --http \
  --http.api eth,net,txpool \
  --ws \
  --ws.api eth,net,txpool \
  --txpool.globalslots 8192 \
  --txpool.accountslots 64

Тепер доступен txpool_content — повний снапшот мемпула:

const txpoolContent = await provider.send('txpool_content', []);
// { pending: { [address]: { [nonce]: Transaction } }, queued: { ... } }

const txpoolStatus = await provider.send('txpool_status', []);
// { pending: '0x1234', queued: '0x56' }

Власна нода бачить усі транзакції, які p2p-мережа передає через неї. Це значно повніше публічних RPC.

Рівень 3: colocation нода + peering

Для серйозного MEV або latency-critical завдань: розміщення ноди в тому ж дата-центрі, що й крупні валідатори (Equinix NY5, Amsterdam AMS3). Настройка максимального числа peers та підключення до спеціалізованих mesh-мереж:

# Підключення до Bloxroute BDN (Blockchain Distribution Network)
geth --mainnet \
  --bootnodes "enode://bloxroute-peer@..." \
  --maxpeers 100

Це знижує latency з ~200ms до ~50ms до отримання транзакції. Критично для front-running ботів, але надлишково для більшості моніторингових завдань.

Декодування транзакцій

Raw транзакція — це набір байт. Щоб розуміти, що вона робить, потрібно декодувати calldata:

import { Interface, FunctionFragment } from 'ethers';

// Завантажуємо ABI відомих протоколів
const UNISWAP_V2_ABI = [...]; // swapExactTokensForTokens, тощо
const UNISWAP_V3_ABI = [...]; // exactInputSingle, тощо

const uniV2Iface = new Interface(UNISWAP_V2_ABI);
const uniV3Iface = new Interface(UNISWAP_V3_ABI);

function decodeUniswapTx(tx: ethers.TransactionResponse): DecodedSwap | null {
  if (!tx.data || tx.data === '0x') return null;

  const selector = tx.data.slice(0, 10);  // перші 4 байти

  // Спробуємо декодувати як Uniswap V3 exactInputSingle
  if (selector === '0x414bf389') {
    try {
      const decoded = uniV3Iface.decodeFunctionData('exactInputSingle', tx.data);
      return {
        protocol: 'UniswapV3',
        tokenIn: decoded.params.tokenIn,
        tokenOut: decoded.params.tokenOut,
        amountIn: decoded.params.amountIn,
        amountOutMinimum: decoded.params.amountOutMinimum,
        recipient: decoded.params.recipient,
      };
    } catch { return null; }
  }

  return null;
}

Для невідомих контрактів — 4byte.directory API для пошуку сигнатури за селектором:

async function resolveSelector(selector: string): Promise<string | null> {
  const res = await fetch(`https://www.4byte.directory/api/v1/signatures/?hex_signature=${selector}`);
  const data = await res.json();
  return data.results[0]?.text_signature ?? null;
}

Паттерни використання: що реально роблять з мемпулом

Liquidation bot — моніторинг позицій на Aave/Compound близьких до liquidation threshold:

// Слухаємо ціни в мемпулі — транзакції Chainlink oracle updater
async function watchChainlinkUpdates(tx: ethers.TransactionResponse) {
  if (tx.to !== CHAINLINK_AGGREGATOR) return;

  // Декодуємо нову ціну з calldata
  const newPrice = decodeOracleUpdate(tx.data);

  // Симулюємо liquidation з новою ціною
  const liquidatablePositions = await simulateLiquidations(newPrice);

  if (liquidatablePositions.length > 0) {
    // Відправляємо liquidation транзакцію з вищим gas price
    await sendLiquidationTx(liquidatablePositions[0], {
      maxFeePerGas: tx.maxFeePerGas! * 2n,  // обгоняємо оракул
    });
  }
}

Sandwich attack detection — для захисту вашого протоколу:

interface SandwichCandidate {
  victimTx: string;
  frontRunTx: string;
  estimatedProfit: bigint;
}

async function detectPotentialSandwich(
  pendingTxs: Map<string, ethers.TransactionResponse>
): Promise<SandwichCandidate[]> {
  const candidates: SandwichCandidate[] = [];

  for (const [hash, tx] of pendingTxs) {
    const swap = decodeUniswapTx(tx);
    if (!swap) continue;

    // Шукаємо вже відправлені транзакції на той же пул з вищим газом
    const frontRuns = [...pendingTxs.values()].filter(other =>
      other.hash !== hash &&
      decodeUniswapTx(other)?.tokenIn === swap.tokenIn &&
      other.maxFeePerGas! > tx.maxFeePerGas!
    );

    if (frontRuns.length > 0) {
      candidates.push({ victimTx: hash, frontRunTx: frontRuns[0].hash, estimatedProfit: 0n });
    }
  }

  return candidates;
}

Gas price forecasting — аналіз distribution газу в мемпулі для передбачення наступного base fee:

async function analyzeGasDistribution() {
  const txpoolContent = await provider.send('txpool_content', []);
  const gasPrices: bigint[] = [];

  for (const addrTxs of Object.values(txpoolContent.pending)) {
    for (const tx of Object.values(addrTxs as any)) {
      gasPrices.push(BigInt((tx as any).maxFeePerGas || (tx as any).gasPrice));
    }
  }

  gasPrices.sort((a, b) => (a < b ? -1 : 1));

  return {
    p25: gasPrices[Math.floor(gasPrices.length * 0.25)],
    p50: gasPrices[Math.floor(gasPrices.length * 0.50)],  // медіана
    p75: gasPrices[Math.floor(gasPrices.length * 0.75)],
    p95: gasPrices[Math.floor(gasPrices.length * 0.95)],
    count: gasPrices.length,
  };
}

Архітектура високонагруженого бота

На mainnet Ethereum мемпул містить 50–200k транзакцій. Обробка кожної синхронно неможлива:

WebSocket listener
      │ raw tx hash
      ▼
  Queue (Redis Streams)
      │
  Workers (x8 процесів)
      │ декодування + фільтрація
      ▼
  Signal Bus (Redis Pub/Sub)
      │ тільки релевантні подій
      ▼
  Strategy Handlers
      │ liquidation / arb / alert
      ▼
  Execution Engine (з rate limiting)

Воркери декодують транзакції паралельно, в чергу сигналів попадає тільки те, що пройшло фільтр. Execution engine управляє відправкою транзакцій, щоб не flood-ити мережу при масових подіях.

Flashbots та приватні транзакції

Частина транзакцій ніколи не попадає в публічний мемпул — відправляються напрямки валідаторам через Flashbots MEV-Boost або Flashbots Protect. Для моніторингу цього трафіку потрібен окремий підхід — підписка на Flashbots event stream або партнерство з block builder'ами.

Для більшості завдань (ліквідацій, моніторингу протоколу) публічного мемпула достатньо. Для конкурентного MEV арбітражу — без Flashbots інтеграції важко працювати: боти з приватними транзакціями мають структурну перевагу.

Latency: вимірювання та оптимізація

class LatencyTracker {
  private samples: number[] = [];

  recordTxSeen(txHash: string, seenAt: number) {
    // seenAt — timestamp отримання з WebSocket (performance.now())
    this.txSeenMap.set(txHash, seenAt);
  }

  recordTxMined(txHash: string, blockTimestamp: number) {
    const seenAt = this.txSeenMap.get(txHash);
    if (!seenAt) return;
    this.samples.push(Date.now() - seenAt);
  }

  getP50LatencyMs(): number {
    const sorted = [...this.samples].sort((a, b) => a - b);
    return sorted[Math.floor(sorted.length / 2)];
  }
}

Типові цифри: публічний Alchemy WebSocket — 150–400ms від broadcast до отримання. Власна нода з 50+ peers — 50–150ms. Node colocation в Equinix — 10–50ms.