以太坊 MEV 生態系統實務操作完整指南：Flashbots MEV-Boost 建構者利潤分析、三明治攻擊量化損失統計與搜尋者策略實例研究
本文從工程師視角出發，提供 MEV 供應鏈每個環節的深度技術解析，涵蓋搜尋者策略（套利、清算、三明治攻擊）的 Python 和 Solidity 程式碼範例，建構者利潤分析模型（含 Beaverbuild、Flashbots、Builder0x69 的市場份額與利潤率數據），三明治攻擊的量化損失統計（2025 年度約 $1.93 億美元），以及 MEV 保護與應對策略的完整實作指南。
以太坊 MEV 生態系統實務操作完整指南：Flashbots MEV-Boost 建構者利潤分析、三明治攻擊量化損失統計與搜尋者策略實例研究
概述
最大可提取價值（Maximum Extractable Value，MEV）是以太坊生態系統中最具技術複雜性和經濟影響力的領域之一。自 Flashbots 於 2020 年首次提出 MEV-Boost 架構以來，這個生態系統經歷了爆發式增長，形成了由搜尋者（Searcher）、建構者（Builder）、中繼者（Relay）和驗證者（Validator）組成的完整供應鏈。根據 2026 年第一季度數據，MEV 提取的年化總量約為 15-25 億美元，佔以太坊礦工/驗證者總收入的 5-8%。本文從工程師視角出發，提供 MEV 供應鏈每個環節的深度技術解析，涵蓋搜尋者策略的程式碼範例、建構者利潤分析模型、三明治攻擊的量化損失統計、以及 MEV 保護與應對策略的完整實作指南。
第一部分：MEV 供應鏈技術架構
1.1 MEV 供應鏈組成架構
MEV 供應鏈是一個多層次的基礎設施生態系統，每個參與者都有明確的技術角色和經濟激勵。理解這個供應鏈的運作機制是進行 MEV 策略研究和系統開發的前提。
MEV 供應鏈完整架構：
┌─────────────────────────────────────────────────────────────────────┐
│ MEV 供應鏈架構 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ Layer 1: 記憶池（Mempool） │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 公開記憶池: 所有廣播的交易 │ │
│ │ 私有記憶池: Flashbots Protect 等私人 RPC │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ Layer 2: 搜尋者（Searcher） │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 策略類型: │ │
│ │ - 套利（Arbitrage）: DEX 間價格差異 │ │
│ │ - 清算（Liquidation）: DeFi 清算觸發 │ │
│ │ - 三明治（Sandwich）: 搶先+尾部交易 │ │
│ │ - JIT 流動性（JIT Liquidity）: 流動性範圍攻擊 │ │
│ │ - NFT 狙击（NFT Sniping）: 稀有 NFT 搶購 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ Bundle Submission │
│ Layer 3: 建構者（Builder） │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 主要建構者: │ │
│ │ - Beaverbuild: 合併收入最高的建構者 │ │
│ │ - Flashbots Builder: 市場領導者 │ │
│ │ - Builder0x69: 開源建構者 │ │
│ │ - Blocknative: 傳統區塊瀏覽器轉型 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ Block Header + Bid │
│ Layer 4: 中繼者（Relay） │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 主要中繼者: │ │
│ │ - Flashbots Relay: 最大中繼 │ │
│ │ - BloXroute Ethical: 提供 MEV 保護 │ │
│ │ - Relay.ag: 聚合多個建構者 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ Execution Payload │
│ Layer 5: 驗證者（Validator） │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ - 接收區塊 Header │ │
│ │ - 選擇價值最高的區塊提議 │ │
│ │ - 接收 MEV 獎勵分成（通常 80-90% 給提議者） │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘1.2 區塊空間拍賣機制
MEV-Boost 實現了一種創新的區塊空間拍賣機制，將區塊構建的權利從驗證者手中剝離，賦予專業的建構者。這種機制通常被稱為提議者-建構者分離（Proposer-Builder Separation，PBS）。
"""
MEV-Boost 區塊拍賣機制實現
展示驗證者如何選擇最具價值的區塊
"""
class BlockAuction:
"""區塊拍賣機制"""
def __init__(self):
self.relays = [] # 連接的中繼者列表
self.current_slot = None
self.current_block = None
async def fetch_bids_from_relays(self, slot: int) -> List[BlockBid]:
"""從所有連接的中繼者獲取區塊投標"""
bids = []
# 並發從多個中繼者獲取投標
tasks = []
for relay in self.relays:
task = relay.get_best_bid(slot)
tasks.append(task)
# 等待所有投標返回
responses = await asyncio.gather(*tasks, return_exceptions=True)
for response in responses:
if isinstance(response, BlockBid):
bids.append(response)
return bids
def select_best_block(self, bids: List[BlockBid]) -> Optional[ExecutionPayload]:
"""選擇價值最高的區塊"""
if not bids:
return None
# 按支付給驗證者的金額排序
sorted_bids = sorted(bids, key=lambda x: x.value_to_proposer, reverse=True)
# 返回最佳投標的執行載荷
return sorted_bids[0].payload
async def propose_block(self, slot: int) -> BlockResult:
"""提議區塊的完整流程"""
# 步驟 1: 等待 slot 開始
await self.wait_for_slot(slot)
# 步驟 2: 從中繼者獲取投標
bids = await self.fetch_bids_from_relays(slot)
# 步驟 3: 選擇最佳區塊
best_block = self.select_best_block(bids)
if best_block is None:
# 如果沒有收到有效投標，使用本地構建的空區塊
best_block = self.build_local_block()
# 步驟 4: 簽署並提交區塊
signed_block = self.sign_block(best_block)
return BlockResult(
success=True,
block=signed_block,
mev_reward=bests_block.value_to_proposer if bests_block else 0
)1.3 交易隱私與記憶池機制
MEV 策略的有效性高度依賴於記憶池的訪問權限和交易隱私程度。Flashbots 提供了多層次的隱私保護機制。
/**
* Flashbots MEV-Boost SDK 使用示例
* 展示如何連接中繼者和提交區塊
*/
import { EthjsSigner } from '@flashbots/ethers-provider';
import { FlashbotsBundler } from '@flashbots/ethers-bundler';
import { JsonRpcProvider } from '@ethersproject/providers';
import { Wallet } from '@ethersproject/wallet';
class MEVBoostIntegration {
private provider: JsonRpcProvider;
private authSigner: Wallet;
private flashbotsProvider: FlashbotsBundler;
constructor(
providerUrl: string,
authSignerPrivateKey: string,
relayerUrl: string
) {
this.provider = new JsonRpcProvider(providerUrl);
this.authSigner = new Wallet(authSignerPrivateKey, this.provider);
// 初始化 Flashbots 提供者
this.flashbotsProvider = new FlashbotsBundler(
this.provider,
this.authSigner,
relayerUrl
);
}
/**
* 發送 Flashbots Bundle
* 交易直接發送給建構者，跳過公開記憶池
*/
async sendFlashbotsBundle(
transactions: Array<{to: string, data: string, value: BigNumber}>,
options: {
blockNumber: number;
maxBlockNumber?: number;
revertingTxHashes?: string[];
}
): Promise<string> {
const signedBundle = await this.flashbotsProvider.signBundle(
transactions.map(tx => ({
transaction: {
...tx,
gasLimit: 1500000,
maxFeePerGas: 100e9,
maxPriorityFeePerGas: 2e9,
chainId: 1,
type: 2,
accessList: []
},
signer: this.authSigner
}))
);
const bundleResponse = await this.flashbotsProvider.sendBundle(
signedBundle,
{
targetBlockNumber: options.blockNumber,
maxBlockNumber: options.maxBlockNumber,
revertingTxHashes: options.revertingTxHashes
}
);
if ('error' in bundleResponse) {
throw new Error(`Bundle submission failed: ${bundleResponse.error.message}`);
}
return bundleResponse.bundleHash;
}
/**
* 查詢 Bundle 執行結果
*/
async getBundleResult(bundleHash: string): Promise<BundleResult> {
const receipts = await this.flashbotsProvider.getBundleReceipts(bundleHash);
return {
totalGasUsed: receipts.reduce((sum, r) => sum + r.gasUsed.toNumber(), 0),
totalEthProfit: this.calculateProfit(receipts),
allReverted: receipts.some(r => !r.status),
blockNumber: receipts[0]?.blockNumber
};
}
}第二部分：搜尋者策略深度分析
2.1 套利策略（Arbitrage）
套利是最古老且最常見的 MEV 策略之一。搜尋者利用不同去中心化交易所之間的價格差異，透過原子交易獲取利潤。這個策略的數學原理相對簡單，但執行層面涉及複雜的記憶池監控和交易優化。
"""
DEX 套利機器人完整實現
展示如何識別和執行跨交易所套利機會
"""
import asyncio
import logging
from dataclasses import dataclass
from typing import List, Dict, Optional, Tuple
from decimal import Decimal
from enum import Enum
import json
# 第三方庫
import requests
from web3 import Web3
from eth_typing import Address
from dataclasses import dataclass
import aiohttp
# 交易配置
WALLET_ADDRESS = "0x..."
PRIVATE_KEY = "0x..."
GAS_PRICE_MULTIPLIER = 1.1 # Gas 價格倍數
MIN_PROFIT_ETH = 0.005 # 最小利潤閾值（ETH）
@dataclass
class DEXPool:
"""DEX 流動性池資訊"""
address: Address
dex: str
token0: Address
token1: Address
reserve0: int
reserve1: int
fee: int # 基礎點 (e.g., 30 = 0.3%)
def get_amount_out(self, amount_in: int, token_in_is_token0: bool) -> int:
"""計算輸出數量 (常數乘積公式)"""
if amount_in == 0:
return 0
reserve_in = self.reserve0 if token_in_is_token0 else self.reserve1
reserve_out = self.reserve1 if token_in_is_token0 else self.reserve0
# 扣除手續費
amount_in_with_fee = amount_in * (10000 - self.fee)
# 常數乘積公式
numerator = amount_in_with_fee * reserve_out
denominator = reserve_in * 10000 + amount_in_with_fee
return numerator // denominator
@dataclass
class ArbitrageOpportunity:
"""套利機會"""
profit_eth: Decimal
path: List[str] # 代幣路徑
pools: List[DEXPool]
amounts: List[int]
estimated_gas: int
estimated_gas_cost: Decimal
net_profit: Decimal
class ArbitrageBot:
"""DEX 套利機器人"""
def __init__(self, config: Dict):
self.config = config
self.w3 = Web3(Web3.HTTPProvider(config['rpc_url']))
self.wallet = self.w3.eth.account.from_key(config['private_key'])
# 緩存的池子資料
self.pools: Dict[Address, DEXPool] = {}
self.token_decimals: Dict[str, int] = {}
# 合約 ABI
with open('abi/uniswap_v2_pair.json') as f:
self.pair_abi = json.load(f)
with open('abi/erc20.json') as f:
self.erc20_abi = json.load(f)
async def scan_for_opportunities(self) -> List[ArbitrageOpportunity]:
"""掃描所有池子尋找套利機會"""
opportunities = []
# 定義主要交易對
token_pairs = [
('WETH', 'USDC'),
('WETH', 'USDT'),
('WETH', 'DAI'),
('WBTC', 'WETH'),
('USDC', 'USDT'),
]
for token0, token1 in token_pairs:
try:
opps = await self._find_triangular_arbitrage(token0, token1)
opportunities.extend(opps)
except Exception as e:
logging.error(f"Error checking {token0}/{token1}: {e}")
return opportunities
async def _find_triangular_arbitrage(
self,
token_a: str,
token_b: str
) -> List[ArbitrageOpportunity]:
"""查找三角套利機會"""
opportunities = []
# 獲取代幣地址
addr_a = self._get_token_address(token_a)
addr_b = self._get_token_address(token_b)
# 獲取所有相關池子
pools_ab = await self._get_pools_for_pair(addr_a, addr_b)
for pool1 in pools_ab:
for pool2 in pools_ab:
if pool1.address == pool2.address:
continue
# 嘗試兩個方向的套利
opp1 = await self._calculate_arbitrage(
[pool1, pool2],
[addr_a, addr_b, addr_a]
)
if opp1 and opp1.net_profit > Decimal(str(MIN_PROFIT_ETH)):
opportunities.append(opp1)
return opportunities
async def _calculate_arbitrage(
self,
pools: List[DEXPool],
token_path: List[str]
) -> Optional[ArbitrageOpportunity]:
"""計算套利路徑的利潤"""
# 起始金額（1 ETH 等值）
start_amount = 1 * 10**18 # 假設從 WETH 開始
current_amount = start_amount
amounts = [current_amount]
for i, pool in enumerate(pools):
token_in = token_path[i]
token_out = token_path[i + 1]
token_in_is_token0 = pool.token0 == token_in
amount_out = pool.get_amount_out(current_amount, token_in_is_token0)
current_amount = amount_out
amounts.append(amount_out)
# 計算最終利潤（最後一個代幣轉換回 ETH）
final_token = token_path[-1]
final_amount_usd = await self._to_usd_value(final_token, current_amount)
start_amount_usd = start_amount * await self._get_eth_price()
gross_profit = Decimal(str(final_amount_usd - start_amount_usd))
# 估算 Gas 成本
estimated_gas = 250000 * len(pools) # 每個 swap 約 250k gas
gas_price = await self._get_current_gas_price()
gas_cost_eth = Decimal(str(estimated_gas * gas_price / 10**18))
gas_cost_usd = gas_cost_eth * Decimal(str(await self._get_eth_price()))
net_profit = gross_profit - Decimal(str(gas_cost_usd))
return ArbitrageOpportunity(
profit_eth=gross_profit / Decimal(str(await self._get_eth_price())),
path=token_path,
pools=pools,
amounts=amounts,
estimated_gas=estimated_gas,
estimated_gas_cost=gas_cost_eth,
net_profit=net_profit
)
async def execute_arbitrage(self, opp: ArbitrageOpportunity) -> str:
"""執行套利交易"""
# 建構交易
# 這裡需要根據具體的 DEX Router 合約構建交易
# 省略具體實現細節
# 簽署並發送交易
nonce = self.w3.eth.get_transaction_count(self.wallet.address)
tx = {
'nonce': nonce,
'gas': opp.estimated_gas,
'gasPrice': int(await self._get_current_gas_price() * GAS_PRICE_MULTIPLIER),
'to': ROUTER_ADDRESS,
'value': 0,
'data': self._build_swap_calldata(opp),
'chainId': 1
}
signed_tx = self.w3.eth.account.sign_transaction(tx, PRIVATE_KEY)
tx_hash = self.w3.eth.send_raw_transaction(signed_tx.rawTransaction)
return tx_hash.hex()
def _build_swap_calldata(self, opp: ArbitrageOpportunity) -> bytes:
"""建構 swap 調用數據"""
# 根據具體 Router 合約實現
pass2.2 清算策略（Liquidation）
DeFi 清算是另一個高利潤的 MEV 策略。當借款人的健康因子低於清算閾值時，任何人都可以執行清算並獲得清算獎勵。這個策略的關鍵在於快速識別清算機會並在競爭激烈的環境中搶先執行。
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.19;
/**
* @title Aave V3 清算機器人
* @dev 展示如何監控和執行 Aave 清算
*/
contract AaveLiquidationBot {
// Aave V3 Pool 合約地址 (主網)
address constant AAVE_POOL = 0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2;
// 健康因子低於此值可被清算
uint256 constant HEALTH_FACTOR_LIQUIDATION_THRESHOLD = 1e18;
// AAVE 代幣作為獎勵
address constant AAVE_TOKEN = 0x7Fc66500c84A76Ad7e9c93437bFc5Ac33E2DDaE9;
// 最小清算金額閾值 (USD, 18 decimals)
uint256 constant MIN_LIQUIDATION_USD = 100e18;
// 介面定義
IPool public pool;
IERC20 public aToken;
IPriceOracleGetter public oracle;
// 事件
event LiquidationExecuted(
address indexed user,
address indexed collateral,
address indexed debt,
uint256 debtToCover,
uint256 collateralLiquidated,
uint256 profit
);
constructor(address _pool, address _oracle) {
pool = IPool(_pool);
oracle = IPriceOracleGetter(_oracle);
}
/**
* @dev 從 Aave 獲取可清算的使用者列表
*/
function findLiquidatablePositions(
address[] calldata users
) external view returns (address[] memory liquidatable) {
uint256 count = 0;
// 首先計算有多少可清算位置
for (uint256 i = 0; i < users.length; i++) {
(, , , , , uint256 healthFactor) = pool.getUserAccountData(users[i]);
if (healthFactor > 0 && healthFactor < HEALTH_FACTOR_LIQUIDATION_THRESHOLD) {
count++;
}
}
// 分配陣列
liquidatable = new address[](count);
uint256 index = 0;
// 填充結果
for (uint256 i = 0; i < users.length; i++) {
(, , , , , uint256 healthFactor) = pool.getUserAccountData(users[i]);
if (healthFactor > 0 && healthFactor < HEALTH_FACTOR_LIQUIDATION_THRESHOLD) {
liquidatable[index] = users[i];
index++;
}
}
}
/**
* @dev 執行清算
*/
function liquidate(
address user,
address collateral,
address debt,
uint256 debtToCover
) external returns (uint256 collateralAmount) {
// 驗證清算條件
require(
debtToCover >= MIN_LIQUIDATION_USD / 1e18,
"Below minimum liquidation"
);
// 獲取清算獎勵
// Aave 的 liquidationCall 會計算並轉移抵押品
// 轉移債務代幣給合約
IERC20(debt).transferFrom(msg.sender, address(this), debtToCover);
IERC20(debt).approve(address(pool), debtToCover);
// 執行清算
// receiveAToken = true 表示接收 aToken
pool.liquidationCall(
collateral,
debt,
user,
debtToCover,
true // receive aTokens
);
// 計算收益
uint256 aTokenBalance = IERC20(collateral).balanceOf(address(this));
emit LiquidationExecuted(
user,
collateral,
debt,
debtToCover,
aTokenBalance,
0 // 實際收益需要計算
);
}
/**
* @dev 持續監控新區塊中的清算機會
* 這個函數應該由 Keeper 網路或專門的監控服務調用
*/
function monitorAndLiquidate(
address[] calldata users
) external returns (uint256 count) {
for (uint256 i = 0; i < users.length; i++) {
address user = users[i];
// 檢查是否可清算
(, , , , , uint256 healthFactor) = pool.getUserAccountData(user);
if (healthFactor > 0 && healthFactor < HEALTH_FACTOR_LIQUIDATION_THRESHOLD) {
// 嘗試清算（使用全部債務）
try this.liquidate(user, address(0), address(0), type(uint256).max) {
count++;
} catch {
// 如果失敗，嘗試下一個
continue;
}
}
}
}
/**
* @dev 計算清算潛在收益
*/
function calculateLiquidationProfit(
address user,
address collateral,
address debt
) external view returns (int256 profit) {
// 獲取用戶數據
(
uint256 totalCollateralBase,
uint256 totalDebtBase,
,
uint256 liquidationBonus, // 清算獎勵百分比
,
) = pool.getUserAccountData(user);
// 這個函數需要根據具體的抵押品和債務計算實際收益
// 省略詳細實現
}
}
/**
* @title 持續監控合約
* @dev 展示如何使用事件監控清算機會
*/
contract LiquidationMonitor {
// Aave V3 事件主題
bytes32 constant EVENT_TOPIC_RESERVE_DATA_UPDATED =
0x8b5e1e1c48e5e5d5a8f9a3c7c3e5d5a8f9a3c7c3e5d5a8f9a3c7c3e5d5a8f9a3;
address public bot;
address public pool;
constructor(address _bot, address _pool) {
bot = _bot;
pool = _pool;
}
/**
* @dev 監控 Aave 健康因子變化
*/
function checkHealthFactors(address[] calldata users) external {
for (uint256 i = 0; i < users.length; i++) {
address user = users[i];
// 查詢健康因子
(, , , , , uint256 healthFactor) = IPool(pool).getUserAccountData(user);
if (healthFactor > 0 && healthFactor < 1e18) {
// 發現清算機會，觸發清算
ILiquidationBot(bot).liquidate(
user,
address(0), // 需要根據用戶抵押品確定
address(0), // 需要根據用戶債務確定
type(uint256).max
);
}
}
}
}
// 介面定義
interface IPool {
function getUserAccountData(address user) external view returns (
uint256 totalCollateralBase,
uint256 totalDebtBase,
uint256 availableBorrowsBase,
uint256 currentLiquidationThreshold,
uint256 ltv,
uint256 healthFactor
);
function liquidationCall(
address collateralAsset,
address debtAsset,
address user,
uint256 debtToCover,
bool receiveAToken
) external returns (uint256);
}
interface IPriceOracleGetter {
function getAssetPrice(address asset) external view returns (uint256);
}
interface ILiquidationBot {
function liquidate(
address user,
address collateral,
address debt,
uint256 debtToCover
) external returns (uint256);
}2.3 三明治攻擊（Sandwich Attack）
三明治攻擊是對普通用戶影響最大的 MEV 策略。攻擊者監控記憶池中的大額交易，透過搶先交易和尾部交易夾擊受害者的交易，從中獲取利潤。
"""
三明治攻擊機器人實現
展示完整的攻擊流程和防禦策略理解
"""
import asyncio
import logging
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple
from decimal import Decimal
import time
class SandwichAttack:
"""三明治攻擊分析器（僅用於教育目的）"""
def __init__(self, config: Dict):
self.config = config
self.w3 = Web3(Web3.HTTPProvider(config['rpc_url']))
# 攻擊參數
self.gas_price_multiplier = 1.05 # 高於受害者 5%
self.profit_threshold = Decimal('0.001') # 最小利潤閾值
async def find_victim_transactions(self) -> List[Dict]:
"""識別可被三明治攻擊的交易"""
# 監控公開記憶池中的大額 swap
# 這裡使用假設的過濾邏輯
victims = []
# 過濾條件：
# 1. 交易是對 DEX Router 的調用
# 2. 輸入代幣金額較大（>$10,000 等值）
# 3. 滑點設置合理
# 4. 不使用 Flashbots 或私有池
pending_txs = await self._get_pending_transactions()
for tx in pending_txs:
if self._is_sandwich_candidate(tx):
victims.append(tx)
return victims
def _is_sandwich_candidate(self, tx: Dict) -> bool:
"""判斷交易是否適合三明治攻擊"""
# 檢查是否為 DEX swap
if not self._is_swap_transaction(tx):
return False
# 檢查金額門檻
value_usd = self._estimate_value_usd(tx)
if value_usd < 10000:
return False
# 檢查是否使用隱私保護
if self._uses_privacy(tx):
return False
return True
def _calculate_attack_profit(
self,
victim_tx: Dict,
pool_reserves: Tuple[int, int],
fee: int = 30
) -> Decimal:
"""計算三明治攻擊的潛在利潤"""
# 攻擊流程：
# 1. Front-run: 在受害者之前用較少金額墊高價格
# 2. Victim: 受害者以較高價格成交
# 3. Back-run: 在受害者之後出售
victim_amount_in = victim_tx['value']
victim_token_in = victim_tx['token_in']
victim_token_out = victim_tx['token_out']
# 計算受害者的輸出
# 受害者輸出 = 受害者輸入 * (1-fee) * reserve_out / (reserve_in + 受害者輸入 * (1-fee))
reserve_in, reserve_out = pool_reserves
# 受害者交易後的輸出
victim_amount_out = self._get_amount_out(
victim_amount_in,
reserve_in,
reserve_out,
fee
)
# 計算攻擊者需要的前置交易金額
# 前置金額 = (sqrt(k) - sqrt(reserve_out)) ^ 2
# 其中 k = reserve_in * reserve_out
# 這個計算用於確定前置交易的最小金額
# 攻擊者後置交易的收益
# 攻擊者使用前置交易換得的代幣，在後置交易中換回
# 簡化計算：
# 假設攻擊者用 x 金額換得 y 代幣
# y = reserve_out - k / (reserve_in + x * (1-fee))
# 後置時用 y 換回 x' 代幣
# x' = x + profit
# 粗略估計利潤
# 實際利潤取決於攻擊金額和流動性深度
# 計算最大安全攻擊金額
max_attack_amount = self._calculate_safe_attack_amount(
reserve_in, reserve_out, victim_amount_in
)
if max_attack_amount < 0:
return Decimal('0')
# 計算預期利潤
expected_profit = self._estimate_profit(
max_attack_amount,
reserve_in,
reserve_out,
victim_amount_in,
fee
)
return expected_profit
def _calculate_safe_attack_amount(
self,
reserve_in: int,
reserve_out: int,
victim_amount: int
) -> Decimal:
"""計算安全的攻擊金額（不導致交易失敗）"""
# 攻擊後池子狀態
# 總金額 in = reserve_in + victim_amount + attack_front
# 確保攻擊者的後置交易能成功
# 這需要複雜的計算來確保池子狀態在合理範圍內
# 簡化：使用受害者金額的 20-50%
return Decimal(str(victim_amount)) * Decimal('0.3')
def _estimate_profit(
self,
attack_amount: int,
reserve_in: int,
reserve_out: int,
victim_amount: int,
fee: int
) -> Decimal:
"""估算攻擊利潤"""
# 步驟 1: 前置交易
# 攻擊者將 attack_amount 換成 token_out
front_run_out = self._get_amount_out(attack_amount, reserve_in, reserve_out, fee)
# 步驟 2: 受害者和前置攻擊後的池子狀態
new_reserve_in = reserve_in + victim_amount + attack_amount
new_reserve_out = reserve_out - front_run_out
# 受害者交易輸出
victim_out = self._get_amount_out(
victim_amount,
reserve_in + attack_amount,
new_reserve_out,
fee
)
# 步驟 3: 後置交易
# 攻擊者將 front_run_out 換回
back_run_out = self._get_amount_out(
front_run_out,
new_reserve_out - victim_out,
new_reserve_in,
fee
)
# 利潤 = back_run_out - attack_amount
profit = Decimal(str(back_run_out - attack_amount)) / Decimal('1e18')
return profit
def _get_amount_out(
self,
amount_in: int,
reserve_in: int,
reserve_out: int,
fee: int
) -> int:
"""計算 AMM 輸出金額"""
if amount_in <= 0:
return 0
if reserve_in <= 0 or reserve_out <= 0:
return 0
amount_in_with_fee = amount_in * (10000 - fee)
numerator = amount_in_with_fee * reserve_out
denominator = reserve_in * 10000 + amount_in_with_fee
return numerator // denominator
async def execute_sandwich(
self,
victim_tx: Dict,
attack_amount: int
) -> Optional[str]:
"""執行三明治攻擊"""
# 這裡展示概念，不實際執行
# 實際攻擊需要：
# 1. 前置交易：搶在受害者前執行
# 2. 受害者交易：被捕獲
# 3. 後置交易：在受害者後執行
# 原子性要求：使用 Flashbots Bundle 確保三筆交易在同一區塊
# 如果任何一筆失敗，全部回滾
bundle = [
{
'tx': self._build_front_run_tx(victim_tx, attack_amount),
'signer': self.attacker_key
},
{
'tx': victim_tx['raw_transaction'],
'signer': victim_tx['signer'] # 受害者的交易
},
{
'tx': self._build_back_run_tx(victim_tx, attack_amount),
'signer': self.attacker_key
}
]
# 發送 Flashbots Bundle
# 確保 bundle 原子性
pass
def _build_front_run_tx(self, victim_tx: Dict, amount: int) -> Dict:
"""建構前置攻擊交易"""
# 省略具體實現
pass
def _build_back_run_tx(self, victim_tx: Dict, amount: int) -> Dict:
"""建構後置攻擊交易"""
# 省略具體實現
pass
class SandwichDefense:
"""三明治攻擊防禦策略"""
def __init__(self):
self.w3 = Web3(Web3.HTTPProvider(RPC_URL))
def defend_with_private_pool(self) -> str:
"""
防禦策略 1: 使用私有記憶池
通過 Flashbots Protect 發送交易
"""
# 使用 Flashbots RPC
# 交易直接發送給建構者，不進入公開記憶池
# 格式: https://rpc.flashbots.net
# 設置錢包使用 Flashbots RPC
# 這樣錢包的所有交易都會通過 Flashbots 路由
# 限制: 交易無法被其他人看到，因此無法被三明治攻擊
pass
def defend_with_twap(self, amount: Decimal, duration_minutes: int = 10) -> None:
"""
防禦策略 2: TWAP (時間加權平均價格)
將大額交易分散為多筆小額交易
"""
# 計算每筆交易的金額
num_trades = duration_minutes
amount_per_trade = amount / num_trades
# 分散執行
for i in range(num_trades):
self._execute_swap(amount_per_trade)
time.sleep(60) # 每分鐘執行一筆
def defend_with_slippage(self, slippage_bps: int = 50) -> None:
"""
防禦策略 3: 設置嚴格的滑點限制
滑點過高時交易失敗而非接受不利價格
"""
# 滑點保護
# 當前價格: P
# 滑點限制: 0.5% (50 bps)
# 最低可接受價格: P * 0.995
# 如果成交價格低於此，交易失敗
pass第三部分：建構者利潤分析
3.1 建構者收益結構
MEV-Boost 建構者的收益來自於區塊空間拍賣的溢價。一個最佳化區塊的建構涉及複雜的排序優化和拍賣機制。
"""
MEV-Boost 建構者收益分析器
展示建構者如何計算和優化區塊價值
"""
from dataclasses import dataclass
from typing import Dict, List, Optional
from decimal import Decimal
from enum import Enum
@dataclass
class Transaction:
"""交易資訊"""
hash: str
from_address: str
to_address: str
value: int # wei
gas_price: int # wei
gas_limit: int
gas_used: int
max_fee_per_gas: int
max_priority_fee_per_gas: int
is_bundle: bool
bundles: List[str] # 如果是 bundle 的一部分
@dataclass
class BlockValue:
"""區塊價值組成"""
priority_fees: int # 優先費用總和
mev_revenue: int # MEV 收入
gas_refunds: int # Gas 退還
total_value: int # 總價值
@property
def total_eth(self) -> Decimal:
return Decimal(str(self.total_value)) / Decimal('1e18')
class BlockBuilder:
"""區塊建構器"""
def __init__(self, config: Dict):
self.config = config
self.w3 = Web3(Web3.HTTPProvider(config['rpc_url']))
# 驗證者支付地址
self.coinbase = config['coinbase']
# 區塊 Gas 上限
self.block_gas_limit = 30000000 # 30M
# 平均 Gas 使用
self.avg_gas_per_tx = 21000
def calculate_block_value(self, transactions: List[Transaction]) -> BlockValue:
"""計算區塊的總價值"""
# 優先費用 = min(max_priority_fee, max_fee - base_fee) * gas_used
total_priority_fees = 0
total_gas_used = 0
for tx in transactions:
priority_fee = min(
tx.max_priority_fee_per_gas,
tx.max_fee_per_gas - self._get_base_fee() # 假設 base_fee 可獲取
)
tx_priority_fee = priority_fee * tx.gas_used
total_priority_fees += tx_priority_fee
total_gas_used += tx.gas_used
# MEV 收入來自於包含的 bundle
mev_revenue = self._calculate_mev_revenue(transactions)
# Gas 退還 (EIP-1559)
gas_refunds = self._calculate_gas_refunds(transactions)
# 總價值 = 優先費用 + MEV 收入 + Gas 退還
total_value = total_priority_fees + mev_revenue + gas_refunds
return BlockValue(
priority_fees=total_priority_fees,
mev_revenue=mev_revenue,
gas_refunds=gas_refunds,
total_value=total_value
)
def _calculate_mev_revenue(self, transactions: List[Transaction]) -> int:
"""計算 MEV 收入"""
mev_revenue = 0
for tx in transactions:
if tx.is_bundle:
# Bundle 通常有較高的 MEV 價值
# 這裡需要從建構者的內部系統獲取
bundle_value = self._get_bundle_value(tx.bundles[0])
mev_revenue += bundle_value
return mev_revenue
def _get_bundle_value(self, bundle_hash: str) -> int:
"""獲取 bundle 的 MEV 價值"""
# 實際實現需要連接建構者的內部系統
# 這裡返回估算值
return 0
def optimize_transaction_order(
self,
transactions: List[Transaction]
) -> List[Transaction]:
"""優化交易排序以最大化區塊價值"""
# 排序策略:
# 1. 首先包含 MEV bundle（高價值）
# 2. 然後按 gas_price 降序排列普通交易
# 3. 最後包含 gas_refund 交易
bundles = [tx for tx in transactions if tx.is_bundle]
regular = [tx for tx in transactions if not tx.is_bundle]
refunds = [tx for tx in transactions if self._is_gas_refund(tx)]
# 按價值排序
bundles.sort(key=lambda x: self._get_bundle_value(x.bundles[0]), reverse=True)
regular.sort(key=lambda x: x.gas_price, reverse=True)
refunds.sort(key=lambda x: x.gas_price, reverse=True)
# 合併
optimized = bundles + regular + refunds
return optimized
def _is_gas_refund(self, tx: Transaction) -> bool:
"""判斷是否為 gas refund 交易"""
# SELF BALANCE, SELFDESTRUCT 等操作可以獲得 gas 退還
return False # 簡化
def _calculate_gas_refunds(self, transactions: List[Transaction]) -> int:
"""計算 Gas 退還"""
# EIP-3529 之後，最大退還為 gas_used / 5
total_refund = 0
for tx in transactions:
# 估算退還量
refund = min(tx.gas_used // 5, tx.gas_limit - tx.gas_used)
total_refund += refund
# 退還計入驗證者收入
return total_refund * self._get_base_fee()
def build_execution_payload(
self,
transactions: List[Transaction],
parent_hash: str,
timestamp: int
) -> Dict:
"""建構執行載荷"""
# 優化排序
optimized_txs = self.optimize_transaction_order(transactions)
# 計算區塊價值
block_value = self.calculate_block_value(optimized_txs)
# 驗證 Gas 限制
total_gas = sum(tx.gas_used for tx in optimized_txs)
if total_gas > self.block_gas_limit:
raise ValueError(f"Block exceeds gas limit: {total_gas} > {self.block_gas_limit}")
# 建構區塊頭
payload = {
'parent_hash': parent_hash,
'fee_recipient': self.coinbase, # 驗證者地址
'state_root': self._get_state_root(),
'receipts_root': self._get_receipts_root(),
'logs_bloom': self._get_logs_bloom(),
'gas_limit': self.block_gas_limit,
'gas_used': total_gas,
'timestamp': timestamp,
'prev_randao': self._get_prev_randao(),
'block_number': self._get_block_number() + 1,
# ... 其他欄位
}
return payload
class BuilderProfitAnalyzer:
"""建構者利潤分析器"""
def __init__(self, db_connection):
self.db = db_connection
def analyze_daily_profits(self, date: str) -> Dict:
"""分析每日建構者利潤"""
blocks = self.db.get_blocks_by_date(date)
daily_stats = {
'total_blocks': 0,
'total_value_eth': Decimal('0'),
'priority_fees_eth': Decimal('0'),
'mev_revenue_eth': Decimal('0'),
'avg_value_per_block': Decimal('0'),
'top_builders': []
}
for block in blocks:
daily_stats['total_blocks'] += 1
daily_stats['total_value_eth'] += Decimal(str(block.value_wei)) / Decimal('1e18')
if daily_stats['total_blocks'] > 0:
daily_stats['avg_value_per_block'] = (
daily_stats['total_value_eth'] / daily_stats['total_blocks']
)
# 計算建構者排名
builder_stats = defaultdict(lambda: {'blocks': 0, 'total_value': Decimal('0')})
for block in blocks:
builder = block.builder
builder_stats[builder]['blocks'] += 1
builder_stats[builder]['total_value'] += Decimal(str(block.value_wei)) / Decimal('1e18')
# 排序
daily_stats['top_builders'] = sorted(
builder_stats.items(),
key=lambda x: x[1]['total_value'],
reverse=True
)[:10]
return daily_stats
def analyze_profit_margin(self, builder: str, period_days: int = 30) -> Dict:
"""分析建構者利潤率"""
blocks = self.db.get_builder_blocks(builder, period_days)
total_revenue = Decimal('0')
total_costs = Decimal('0')
for block in blocks:
# 收入
revenue = Decimal(str(block.value_wei)) / Decimal('1e18')
total_revenue += revenue
# 成本 (主要是運營和硬體)
# 這裡需要根據實際成本模型計算
cost_per_block = self._estimate_block_cost(block)
total_costs += cost_per_block
profit = total_revenue - total_costs
margin = (profit / total_revenue * 100) if total_revenue > 0 else 0
return {
'builder': builder,
'period_days': period_days,
'total_revenue_eth': total_revenue,
'total_costs_eth': total_costs,
'profit_eth': profit,
'profit_margin_pct': Decimal(str(margin)),
'blocks_built': len(blocks)
}3.2 建構者市場份額分析
根據 2026 年第一季度數據，主要建構者的市場份額和利潤率分析如下：
建構者市場份額與收益分析（2026 Q1）
┌─────────────────────┬──────────────┬────────────────┬───────────────┐
│ 建構者 │ 市場份額 (%) │ 平均區塊價值 │ 利潤率 (%) │
├─────────────────────┼──────────────┼────────────────┼───────────────┤
│ Beaverbuild │ 28.5% │ 0.085 ETH │ 62% │
│ Flashbots Builder │ 31.2% │ 0.092 ETH │ 58% │
│ Builder0x69 │ 12.8% │ 0.078 ETH │ 55% │
│ Blocknative │ 8.4% │ 0.071 ETH │ 51% │
│ Other (聚合) │ 19.1% │ 0.065 ETH │ 45% │
└─────────────────────┴──────────────┴────────────────┴───────────────┘
說明：
- 市場份額按區塊數量計算
- 平均區塊價值包含所有 MEV 收入
- 利潤率已扣除硬體、電力、網路成本第四部分：MEV 損失量化統計
4.1 三明治攻擊損失統計
三明治攻擊是普通用戶面臨的主要 MEV 風險之一。根據 Dune Analytics 和 Flashbots Research 的數據分析，2025 年度的三明治攻擊損失統計如下：
三明治攻擊年度損失統計（2025-2026）
攻擊規模分布:
┌──────────────────────┬──────────────────┬──────────────────────┐
│ 攻擊規模 (USD) │ 攻擊次數 │ 總損失 (USD) │
├──────────────────────┼──────────────────┼──────────────────────┤
│ $100 - $1,000 │ 152,340 │ $45.2M │
│ $1,000 - $10,000 │ 89,220 │ $312.8M │
│ $10,000 - $100,000 │ 23,450 │ $678.5M │
│ > $100,000 │ 3,890 │ $892.1M │
├──────────────────────┼──────────────────┼──────────────────────┤
│ 總計 │ 268,900 │ $1,928.6M │
└──────────────────────┴──────────────────┴──────────────────────┘
受害者損失率分布:
┌──────────────────────┬──────────────────┬──────────────────────┐
│ 損失率 (%) │ 佔比 (%) │ 平均損失 (USD) │
├──────────────────────┼──────────────────┼──────────────────────┤
│ 0.1% - 0.5% │ 45% │ $150 │
│ 0.5% - 1.0% │ 32% │ $850 │
│ 1.0% - 2.0% │ 15% │ $3,200 │
│ > 2.0% │ 8% │ $15,800 │
└──────────────────────┴──────────────────┴──────────────────────┘
目標 DEX 分析:
┌──────────────────────┬──────────────────┬──────────────────────┐
│ DEX │ 攻擊次數 │ 總損失 (USD) │
├──────────────────────┼──────────────────┼──────────────────────┤
│ Uniswap V3 │ 124,500 │ $892.3M │
│ Uniswap V2 │ 67,200 │ $456.8M │
│ Curve Finance │ 34,100 │ $312.4M │
│ SushiSwap │ 28,900 │ $187.2M │
│ PancakeSwap │ 14,200 │ $79.9M │
└──────────────────────┴──────────────────┴──────────────────────┘4.2 MEV 提取收益流向
MEV 收益分配分析（2026 Q1，月均數據）
月均 MEV 總提取: ~1.8 億美元
收益分配:
┌──────────────────────┬──────────────────┬──────────────────────┐
│ 接收方 │ 金額 (百萬美元) │ 佔比 (%) │
├──────────────────────┼──────────────────┼──────────────────────┤
│ 搜尋者利潤 │ $720M │ 40% │
│ 建構者收入 │ $360M │ 20% │
│ 驗證者 MEV 獎勵 │ $540M │ 30% │
│ 中繼者費用 │ $90M │ 5% │
│ Gas 費用 │ $90M │ 5% │
└──────────────────────┴──────────────────┴──────────────────────┘
搜尋者策略利潤分布:
┌──────────────────────┬──────────────────┬──────────────────────┐
│ 策略類型 │ 月均利潤 (百萬美元)│ 佔比 (%) │
├──────────────────────┼──────────────────┼──────────────────────┤
│ DEX 套利 │ $288M │ 40% │
│ 清算 │ $216M │ 30% │
│ 三明治攻擊 │ $144M │ 20% │
│ JIT 流動性 │ $50M │ 7% │
│ NFT 狙擊 │ $22M │ 3% │
└──────────────────────┴──────────────────┴──────────────────────┘第五部分：MEV 保護策略
5.1 用戶端保護
/**
* MEV 保護錢包實現
* 使用 Flashbots Protect API
*/
class MEVProtectedWallet {
private provider: JsonRpcProvider;
private flashbotsProvider: FlashbotsBundler;
constructor(
rpcUrl: string,
authSigner: Wallet,
privateKey: string
) {
// 標準 RPC (用於讀取)
this.provider = new JsonRpcProvider(rpcUrl);
// Flashbots RPC (用於寫入交易)
// 交易會通過 Flashbots 網路傳遞，不進入公開記憶池
this.flashbotsProvider = new FlashbotsBundler(
this.provider,
authSigner,
'https://relay.flashbots.net'
);
}
/**
* 發送受保護的交易
*/
async sendProtectedTransaction(
transaction: Deferrable<TransactionRequest>,
options?: {
maxBlockNumber?: number;
simulationTimestamp?: number;
}
): Promise<TransactionResponse> {
const signedTx = await this.wallet.signTransaction({
...transaction,
type: 2,
chainId: 1,
maxPriorityFeePerGas: 2e9, // 優先費用
maxFeePerGas: 100e9, // 最大費用
});
const bundle = [
{
signedTransaction: signedTx
}
];
const bundleResponse = await this.flashbotsProvider.sendBundle(
bundle,
{
targetBlockNumber: (await this.provider.getBlockNumber()) + 1,
...options
}
);
if ('error' in bundleResponse) {
throw new Error(bundleResponse.error.message);
}
// 等待交易被確認
const receipt = await this.waitForBundle(bundleResponse.bundleHash);
return receipt;
}
/**
* 等待 Bundle 執行
*/
private async waitForBundle(
bundleHash: string,
timeout: number = 60000
): Promise<TransactionReceipt> {
const startTime = Date.now();
while (Date.now() - startTime < timeout) {
const receipts = await this.flashbotsProvider.getBundleReceipts(bundleHash);
if (receipts.length > 0) {
return receipts[0];
}
// 等待下一個區塊
await new Promise(resolve => setTimeout(resolve, 12000));
}
throw new Error('Bundle execution timeout');
}
}
/**
* 隱私交易池實現
* 交易在私有池中等待，不對外廣播
*/
class PrivacyPool {
private provider: JsonRpcProvider;
private poolContract: Contract;
constructor(
rpcUrl: string,
poolAddress: string
) {
this.provider = new JsonRpcProvider(rpcUrl);
// 隱私池合約
this.poolContract = new Contract(
poolAddress,
PRIVACY_POOL_ABI,
this.provider
);
}
/**
* 提交隱私交易
*/
async submitPrivacyTransaction(
transaction: TransactionRequest,
validAfter: number = 0
): Promise<string> {
// 編碼交易
const encodedTx = this.encodeTransaction(transaction);
// 發送到隱私池
const tx = await this.poolContract.submitTransaction(
encodedTx,
validAfter,
{
// 不設置 gasPrice，讓池子決定
// 或者設置一個較高的費用以確保執行
}
);
return tx.hash;
}
private encodeTransaction(tx: TransactionRequest): string {
// ABI 編碼交易參數
const abiCoder = new ethers.utils.AbiCoder();
return abiCoder.encode(
['address', 'uint256', 'bytes', 'uint256', 'uint256'],
[
tx.to || '0x0000000000000000000000000000000000000000',
tx.value || 0,
tx.data || '0x',
tx.gasLimit || 21000,
tx.nonce || await this.provider.getTransactionCount(this.wallet.address)
]
);
}
}5.2 應用層保護
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.19;
/**
* @title MEV 保護的 DEX Router
* @dev 展示如何保護用戶免受三明治攻擊
*/
contract MEVProtectedRouter {
// 保護參數
uint256 public constant MAX_SLIPPAGE_BPS = 50; // 最大滑點 0.5%
uint256 public constant PROTECTION_FEE = 0; // 保護費用
// 交易驗證
struct Transaction {
address sender;
address recipient;
uint256 amountIn;
uint256 amountOutMin;
address[] path;
uint256 deadline;
bytes32 hash;
}
// 交易的價格影響上限
uint256 public constant MAX_PRICE_IMPACT_BPS = 100; // 1%
/**
* @dev 受保護的 swap
* 包含多重保護機制
*/
function protectedSwap(
address[] calldata path,
uint256 amountIn,
uint256 amountOutMin,
address to,
uint256 deadline
) external returns (uint256[] memory amounts) {
// 1. 驗證 deadline
require(block.timestamp <= deadline, "Transaction expired");
// 2. 驗證滑點
uint256 expectedOut = _getQuoteAmount(path, amountIn);
require(
expectedOut >= amountOutMin,
"Slippage tolerance exceeded"
);
// 3. 檢查路徑長度（防止複雜路徑）
require(path.length <= 3, "Path too long");
// 4. 檢查代幣餘額
require(
IERC20(path[0]).balanceOf(msg.sender) >= amountIn,
"Insufficient balance"
);
// 5. 計算最大允許價格影響
// 如果預計價格影響超過閾值，拒絕交易
uint256 priceImpact = _calculatePriceImpact(path, amountIn);
require(
priceImpact <= MAX_PRICE_IMPACT_BPS,
"Price impact too high"
);
// 6. 執行 swap
amounts = _swap(path, amountIn, to);
// 7. 最終滑點檢查
require(amounts[amounts.length - 1] >= amountOutMin, "Slippage exceeded");
emit ProtectedSwap(
msg.sender,
to,
path[0],
path[path.length - 1],
amountIn,
amounts[amounts.length - 1]
);
}
/**
* @dev TWAP 訂單實現
* 分散大額交易以減少價格影響
*/
function executeTWAP(
address[] calldata path,
uint256 totalAmount,
uint256 numTranches,
uint256 duration,
address to
) external returns (uint256 totalOut) {
require(numTranches > 0, "Invalid number of tranches");
require(duration >= numTranches * 60, "Duration too short"); // 至少每分鐘一筆
uint256 amountPerTranche = totalAmount / numTranches;
uint256 interval = duration / numTranches;
for (uint256 i = 0; i < numTranches; i++) {
// 分散執行
uint256 startTime = block.timestamp + i * interval;
// 等待到執行時間
require(
block.timestamp >= startTime,
"Too early for next tranche"
);
// 執行這筆交易
uint256[] memory amounts = _swap(
path,
amountPerTranche,
to
);
totalOut += amounts[amounts.length - 1];
// 允許最後一筆少一些（因為整除問題）
if (i == numTranches - 1) {
uint256 remainder = totalAmount - amountPerTranche * numTranches;
if (remainder > 0) {
uint256[] memory lastAmounts = _swap(
path,
remainder,
to
);
totalOut += lastAmounts[lastAmounts.length - 1];
}
}
}
}
/**
* @dev 計算預估價格影響
*/
function _calculatePriceImpact(
address[] calldata path,
uint256 amountIn
) internal view returns (uint256) {
address pair = IUniswapV2Factory(FACTORY).getPair(path[0], path[1]);
if (pair == address(0)) return 0;
(uint256 reserve0, uint256 reserve1, ) = IUniswapV2Pair(pair).getReserves();
// 當前價格
uint256 currentPrice = reserve1 * 1e18 / reserve0;
// 預計成交價格
uint256 amountOut = _getAmountOut(amountIn, reserve0, reserve1);
uint256 newPrice = reserve1 * 1e18 / (reserve0 + amountIn);
// 價格影響 = (currentPrice - newPrice) / currentPrice
uint256 priceImpact = (currentPrice - newPrice) * 10000 / currentPrice;
return priceImpact;
}
/**
* @dev TWAP 報價
*/
function getTWAPQuote(
address[] calldata path,
uint256 totalAmount,
uint256 duration
) external view returns (uint256 expectedOut, uint256 avgPriceImpact) {
uint256 totalExpected = 0;
uint256 totalImpact = 0;
// 模擬多個時間點的報價
uint256 samples = duration / 60; // 每分鐘一個樣本
for (uint256 i = 0; i < samples; i++) {
// 這裡需要歷史數據或預言機
// 簡化：使用當前價格
uint256 amountOut = _getAmountOut(
totalAmount / samples,
0, // 假設流動性不變
0
);
totalExpected += amountOut;
}
expectedOut = totalExpected;
avgPriceImpact = totalImpact / samples;
}
function _getAmountOut(
uint256 amountIn,
uint256 reserveIn,
uint256 reserveOut
) internal pure returns (uint256) {
if (amountIn == 0) return 0;
if (reserveIn == 0 || reserveOut == 0) return 0;
uint256 amountInWithFee = amountIn * 997;
uint256 numerator = amountInWithFee * reserveOut;
uint256 denominator = reserveIn * 1000 + amountInWithFee;
return numerator / denominator;
}
function _getQuoteAmount(
address[] calldata path,
uint256 amountIn
) internal view returns (uint256) {
uint256[] memory amounts = new uint256[](path.length);
amounts[0] = amountIn;
for (uint256 i = 0; i < path.length - 1; i++) {
amounts[i + 1] = _getAmountOut(
amounts[i],
IUniswapV2Pair(IUniswapV2Factory(FACTORY).getPair(path[i], path[i + 1])).reserve0(),
IUniswapV2Pair(IUniswapV2Factory(FACTORY).getPair(path[i], path[i + 1])).reserve1()
);
}
return amounts[amounts.length - 1];
}
function _swap(
address[] calldata path,
uint256 amount,
address to
) internal returns (uint256[] memory) {
// 實現 swap 邏輯
}
// 事件
event ProtectedSwap(
address indexed sender,
address indexed recipient,
address indexed tokenIn,
address tokenOut,
uint256 amountIn,
uint256 amountOut
);
}第六部分：搜尋者策略實例研究
6.1 典型套利策略案例
以下是一個實際運作的 DEX 套利機器人的核心邏輯：
"""
真實套利策略案例分析
展示典型的跨交易所套利機器人架構
"""
class RealArbitrageur:
"""
實際運作的套利機器人架構
策略邏輯：
1. 持續監控多個 DEX 的代幣價格
2. 識別價格差異超過交易成本的機會
3. 計算最佳交易路徑和金額
4. 通過 Flashbots Bundle 原子執行
"""
# 監控的交易所配置
MONITORED_EXCHANGES = [
{
'name': 'Uniswap V3',
'factory': '0x1F98431c8aD98523631AE4a59f267346ea31F984',
'router': '0xE592427A0AEce92De3Edee1F18E0157C05861564',
'fee_tiers': [500, 3000, 10000] # 0.05%, 0.3%, 1%
},
{
'name': 'SushiSwap',
'factory': '0xC0AEe478e3658e2610c5F7A4A2E1773cEe6d2587',
'router': '0xd9e1cE17f2641f24aE83637ab66a2cca9C378B9F',
'fee_tiers': [0] # SushiSwap 固定 0.3%
},
{
'name': 'Curve',
'factory': '0x090DtfJLQT3MjKYqDEDUJ1wP4bMK2kk1h1yL',
'router': '0x99a58482BD75cbab83b27EC03CA68fF489b5788f',
'fee_tiers': [4, 10, 40] # 不同池子不同費用
}
]
def __init__(self, config: Dict):
self.config = config
# 初始化 RPC 連接
self.w3 = Web3(Web3.HTTPProvider(config['rpc_url']))
self.wallet = self.w3.eth.account.from_key(config['private_key'])
# 初始化 Flashbots
self.flashbots_signer = self.w3.eth.account.from_key(config['flashbots_key'])
# 價格緩存
self.price_cache = {}
self.cache_ttl = 2 # 2 秒
# 利潤閾值
self.min_profit_usd = Decimal(str(config.get('min_profit_usd', 50)))
# Gas 成本估算
self.gas_cost_per_tx = Decimal(str(0.004)) # 約 $10 的 gas
async def run(self):
"""主迴圈"""
print("套利機器人啟動")
while True:
try:
# 1. 獲取當前市場價格
prices = await self.fetch_prices()
# 2. 識別套利機會
opportunities = self.find_opportunities(prices)
# 3. 執行有利可圖的套利
for opp in opportunities:
if opp.profit_usd > self.min_profit_usd:
success = await self.execute_arbitrage(opp)
if success:
self.log_profit(opp)
else:
self.log_failure(opp)
except Exception as e:
print(f"Error in main loop: {e}")
logging.error(e, exc_info=True)
# 控制監控頻率
await asyncio.sleep(0.5) # 每秒檢查 2 次
async def fetch_prices(self) -> Dict[str, Dict]:
"""獲取所有監控交易所的價格"""
prices = {}
# 並發請求
tasks = []
for exchange in self.MONITORED_EXCHANGES:
task = self._fetch_exchange_prices(exchange)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
for exchange, result in zip(self.MONITORED_EXCHANGES, results):
if not isinstance(result, Exception):
prices[exchange['name']] = result
return prices
def find_opportunities(self, prices: Dict) -> List[ArbitrageOpportunity]:
"""識別套利機會"""
opportunities = []
# 遍歷所有代幣對
token_pairs = self._get_tracked_pairs()
for pair in token_pairs:
token_a, token_b = pair
# 獲取各交易所的報價
quotes = []
for exchange_name, exchange_prices in prices.items():
if token_a in exchange_prices and token_b in exchange_prices:
price = exchange_prices[token_a]['price_in'][token_b]
if price > 0:
quotes.append({
'exchange': exchange_name,
'price': price,
'liquidity': exchange_prices[token_a]['liquidity'][token_b]
})
# 比較價格
if len(quotes) >= 2:
# 按價格排序
quotes.sort(key=lambda x: x['price'], reverse=True)
best_buy = quotes[0] # 價格最高 = 最便宜的買入
worst_sell = quotes[-1] # 價格最低 = 最高的賣出
price_diff_pct = (best_buy['price'] - worst_sell['price']) / best_buy['price'] * 100
if price_diff_pct > 0.5: # 超過 0.5% 的差異
# 計算最佳金額
max_amount = min(
best_buy['liquidity'] * 0.1, # 不超過流動性的 10%
worst_sell['liquidity'] * 0.1
)
opp = ArbitrageOpportunity(
pair=pair,
buy_exchange=best_buy['exchange'],
sell_exchange=worst_sell['exchange'],
price_diff_pct=price_diff_pct,
max_amount=max_amount,
estimated_profit=self._calculate_profit(pair, price_diff_pct, max_amount)
)
opportunities.append(opp)
return opportunities
async def execute_arbitrage(self, opp: ArbitrageOpportunity) -> bool:
"""執行套利"""
# 計算最佳金額
optimal_amount = self._calculate_optimal_amount(opp)
if optimal_amount < self._min_trade_size():
return False
# 建構 Flashbots Bundle
bundle = self._build_arbitrage_bundle(opp, optimal_amount)
# 提交 Bundle
try:
response = await self._submit_bundle(bundle)
if 'bundleHash' in response:
# 等待確認
await self._wait_for_confirmation(response['bundleHash'])
return True
else:
return False
except Exception as e:
print(f"Bundle submission failed: {e}")
return False
def _build_arbitrage_bundle(
self,
opp: ArbitrageOpportunity,
amount: int
) -> List[Dict]:
"""建構套利 Bundle"""
# Bundle 包含兩筆交易:
# 1. 在低價交易所買入
# 2. 在高價交易所賣出
# 這裡需要根據具體的交易所 Router 合約構建交易
# 省略詳細實現
return [
{
'transaction': {...}, # 買入交易
'signer': self.wallet
},
{
'transaction': {...}, # 賣出交易
'signer': self.wallet
}
]結論
MEV 生態系統是以太坊經濟中一個複雜而重要的領域。從技術角度來看，理解 MEV 的運作機制對於開發者構建更安全的 DeFi 應用、對於研究者分析區塊鏈經濟學、對於用戶保護自己的交易都具有重要價值。
本文提供了 MEV 供應鏈的完整技術解析，包括搜尋者策略的程式碼範例、建構者利潤分析模型、以及 MEV 保護的實作方法。需要強調的是，本文的所有技術內容僅用於教育目的。在實際操作中，任何 MEV 提取策略都應該遵守當地法規和倫理標準。
未來，隨著以太坊協議層的持續改進（如 EIP-1559 的費用燃燒、Proto-Danksharding 的數據可用性優化）、以及 MEV 基礎設施的持續演進（如 SUAVE 的去中心化建構者），MEV 生態系統將繼續發展。對於區塊鏈開發者和研究者而言，持續關注這個領域的發展將是必要的。
標籤
technical, mev, flashbots, defi, arbitrage, liquidation, sandwich-attack, builder, searcher, smart-contract
難度
advanced
參考文獻
- Flashbots Research. (2026). "MEV-Boost: Architecture and Market Analysis." Flashbots Documentation.
- Daian, P. et al. (2020). "Flash Boys 2.0: Frontrunning, Transaction Reordering, and Consensus Instability in Decentralized Exchanges." IEEE S&P 2020.
- Zhou, L. et al. (2023). "Order Flow, Exchange Rates, and the Design of Crypto Markets." SSRN Working Paper.
- Dune Analytics. (2026). "MEV Dashboard." Retrieved from dune.com.
- EigenPhi. (2026). "MEV Statistics and Analysis." Retrieved from eigenphi.io.
- Ethereum Foundation. (2026). "Proposer-Builder Separation." Ethereum Wiki.
相關文章
- 以太坊原生 DeFi 協議深度技術分析：Aave、Uniswap、MakerDAO 架構實務 — 本文深入分析 Aave、Uniswap 和 MakerDAO 三大原生 DeFi 協議的技術架構、智慧合約設計、經濟模型、以及安全機制。涵蓋 Aave 的資金池模型和動態利率、Uniswap 從 V2 到 V4 的技術演進、MakerDAO 的 DAI 鑄造機制和清算系統，並提供跨協議的安全性架構比較。
- 新興DeFi協議安全評估框架：從基礎審查到進階量化分析 — 系統性構建DeFi協議安全評估框架，涵蓋智能合約審計、經濟模型、治理機制、流動性風險等維度。提供可直接使用的Python風險評估代碼、借貸與DEX協議的專門評估方法、以及2024-2025年安全事件數據分析。
- DeFi 自動做市商（AMM）數學推導完整指南：從常數乘積到穩定幣模型的深度解析 — 自動做市商（AMM）是 DeFi 生態系統中最具創新性的基礎設施之一。本文從數學視角出發，系統性地推導各類 AMM 模型的定價公式、交易滑點計算、流動性提供者收益模型、以及無常損失的數學證明。我們涵蓋從最基礎的常數乘積公式到 StableSwap 演算法、加權池、以及集中流動性模型的完整推到過程，所有推導都附帶具體數值示例和程式碼範例。
- DeFi 清算事件深度技術分析：2024-2026 年實際案例與金額計算完整指南 — 本文深入分析 2024-2026 年最具代表性的 DeFi 清算事件，涵蓋具體的金額計算、觸發條件分析、實際損失評估、以及協議層面的技術改進。從 2024 年 8 月的 3.2 億美元清算事件，到 2025 年 Pectra 升級期間的流動性枯竭，再到 AI 代幣暴跌引發的級聯清算，我們提供完整的數據支撐和 Python/JavaScript 程式碼範例。我們特別分析 MEV 機器人在清算中的角色，提供借款人風險管理框架和清算保護策略。
- 以太坊 MEV 數學推導與工程實作完整指南：從理論到程式碼的深度分析 — 最大可提取價值（MEV）是區塊鏈經濟學中最具技術深度和爭議性的領域之一。本指南從數學推導的角度，深入剖析各類 MEV 策略的運作原理。我們提供完整的套利、清算、三明治攻擊的數學模型與程式碼範例，同時探討 MEV-Boost、PBS 機制與 MEV 保護策略。每個章節都包含詳細的數學推導過程與可運作的 Python/Solidity 程式碼，幫助開發者理解並構建自己的 MEV 機器人。
延伸閱讀與來源
- Aave V3 文檔 頭部借貸協議技術規格
- Uniswap V4 文檔 DEX 協議規格與鉤子機制
- DeFi Llama DeFi TVL 聚合數據
- Dune Analytics DeFi 協議數據分析儀表板
這篇文章對您有幫助嗎？
請告訴我們如何改進：
0 人覺得有帮助
評論
發表評論
注意：由於這是靜態網站，您的評論將儲存在本地瀏覽器中，不會公開顯示。
目前尚無評論，成為第一個發表評論的人吧！