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

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

Мемпул — это не просто очередь транзакций. Это источник данных о намерениях: что собирается сделать рынок в следующем блоке. MEV-ботов, которые зарабатывают на этих данных, на Ethereum работают сотни. Но мониторинг мемпула нужен не только для арбитража — liquidation-боты DeFi-протоколов, системы раннего обнаружения атак, front-running protection, аналитика gas prices — всё это строится на чтении мемпула в реальном времени.

Доступ к мемпулу: три уровня

Уровень 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;  // транзакция могла быть дропнута

  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: node 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, etc.
const UNISWAP_V3_ABI = [...]; // exactInputSingle, etc.

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;

    // Ищем уже отправленные транзакции на тот же пул с более высоким gas
    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.