以太坊 MEV 基礎設施完整工程實作指南：從搜尋者演算法到區塊構建者的技術實踐

概述

最大可提取價值（Maximal Extractable Value，MEV）是區塊鏈領域最複雜且最具爭議的技術現象之一。自 2019 年 Flashbots 首次系統性地揭示這一概念以來，MEV 已發展成一個完整的地下經濟系統，每日的 MEV 提取量從早期的數萬美元增長至 2024 年的數百萬美元。理解 MEV 基礎設施的運作原理，對於任何希望在以太坊生態系統中保持競爭力的開發者、研究者或交易者而言都至關重要。

本文從工程師的視角出發，提供 MEV 供應鏈的完整技術實作指南。我們將深入分析 MEV 搜尋者如何發現並提取機會、區塊構建者如何組織交易順序以最大化價值、提議者-構建者分離（PBS）機制如何重塑網路激勵結構，以及 Flashbots SUAVE 如何開創去中心化記憶體池的新範式。透過完整的程式碼範例與數學推導，讀者將能掌握 MEV 基礎設施的每個環節。

一、MEV 基礎理論與數學框架

1.1 MEV 的形式化定義

MEV 的正式定義為：在區塊生產過程中，由於交易排序、區塊時間操控或區塊內容選擇所產生的額外價值。這個定義涵蓋了三種主要提取方式：常規 MEV（透過改善交易排序）、時間盜竊（透過控制區塊時間戳）與區塊重組（透過分叉攻擊）。

從數學角度，我們可以將 MEV 表示為一個最佳化問題。設區塊包含一組交易 $T = \{t1, t2, ..., tn\}$，每筆交易 $ti$ 附帶一個價值函數 $v_i(S)$，表示在交易序列 $S$ 下的凈價值。區塊構建者的目標是找到最大化總價值的序列 $S^*$：

$$S^* = \arg\max{S \in \text{Perm}(T)} \sum{i=1}^{n} vi(Si)$$

其中 $Si$ 表示交易 $ti$ 在序列 $S$ 中的位置。這個問題在一般情況下是 NP 困難的，但實際上可以透過貪心演算法或啟發式方法獲得近似最優解。

1.2 常見 MEV 機會類型

套利（Arbitrage） 是最普遍的 MEV 機會之一。當同一資產在不同 DEX 上的價格存在差異時，搜尋者可以透過原子交易在不同市場間低買高賣，從價差中獲利。典型的套利利潤計算如下：

設資產 $A$ 在 Uniswap 的價格為 $P1$，在 Sushiswap 的價格為 $P2$，且 $P1 > P2$。搜尋者可以執行以下操作：

1. 在 Sushiswap 以價格 $P_2$ 買入價值 $X$ 的資產 $A$
2. 獲得數量 $\frac{X}{P_2}$ 的資產 $A$
3. 在 Uniswap 以價格 $P_1$ 賣出全部資產 $A$
4. 獲得數量 $\frac{X \cdot P1}{P2}$ 的原幣種

利潤為：$\Delta = \frac{X \cdot P1}{P2} - X = X \cdot (\frac{P1}{P2} - 1)$

扣除 Gas 費用後，若 $\Delta > G$（Gas 費用），則套利有利可圖。2024 年以太坊上每日平均發生約 5000-10000 次 DEX 套利交易，提取的總價值約 $500,000-$2,000,000 美元。

清算（Liquidation） 是 DeFi 借貸協議中的主要 MEV 來源。當借款人的抵押品價值低於清算閾值時，任何人都可以執行清算並獲得罰款性收益。假設清算獎勵比例為 $r$（通常為 5-10%），抵押品價值為 $C$，借款額為 $D$，則清算利潤為：

$$\text{Liquidator Profit} = C \cdot r - \text{Gas Cost}$$

清算人需要實時監控多個借貸協議的抵押品比率，並在觸發清算條件時立即提交交易。這要求低延遲的區塊鏈數據饋送與優化的交易廣播策略。

三明治攻擊（Sandwich Attack） 是一種針對散戶交易者的 MEV 策略。攻擊者觀察記憶體池中的待確認交易，找到大額代幣買入訂單，然後：

1. 在受害者交易前掛入買入訂單（抬價）
2. 等待受害者的交易以較高價格成交
3. 在受害者交易後掛入賣出訂單（套利）

設受害者訂單的代幣數量為 $X$，市場流動性為 $L$，則三明治利潤可以透過以下公式估算：

$$\text{Profit} \approx \frac{X^2}{L} \cdot \left( \frac{\partial P}{\partial Q} \right)$$

1.3 MEV 市場數據分析

截至 2026 年第一季度，以太坊 MEV 市場呈現以下特徵：

這些數據表明 MEV 已成為以太坊經濟學的核心組成部分，理解和參與 MEV 供應鏈對於維持網路健康至關重要。

二、搜尋者實作：從機會識別到交易構造

2.1 搜尋者架構概述

一個完整的 MEV 搜尋者系統由以下元件組成：區塊鏈數據訂閱、機會識別引擎、交易構造器、Gas 估算器與交易廣播模組。以下是一個基礎搜尋者架構的 Python 實作：

from web3 import Web3
from dataclasses import dataclass
from typing import List, Dict, Optional
import asyncio
import logging

@dataclass
class MEVOpportunity:
    """MEV 機會數據結構"""
    opportunity_type: str  # 'arbitrage', 'liquidation', 'sandwich'
    expected_profit: int   # 預期利潤（wei）
    gas_cost: int          # 預估 Gas 費用（wei）
    net_profit: int         # 凈利潤（wei）
    confidence: float       # 信心度（0-1）
    deadline: int           # 截止區塊
    target_contracts: List[str]  # 目標合約地址
    encoding: Optional[bytes] = None  # 編碼後的交易資料

class MEVSearcher:
    """MEV 搜尋者核心引擎"""
    
    def __init__(self, rpc_url: str, flashbots_key: str):
        self.w3 = Web3(Web3.HTTPProvider(rpc_url))
        self.flashbots = FlashbotsSigner(flashbots_key)
        self.logger = logging.getLogger('mev_searcher')
        
        # 訂閱區塊鏈數據
        self.pending_tx_subscription = PendingTxSubscription(self.w3)
        self.block_subscription = BlockSubscription(self.w3)
        
        # 機會識別引擎
        self.opportunity_detectors = {
            'arbitrage': ArbitrageDetector(self.w3),
            'liquidation': LiquidationDetector(self.w3),
            'sandwich': SandwichDetector(self.w3),
        }
        
        # 交易池
        self.pending_opportunities: asyncio.PriorityQueue = asyncio.PriorityQueue()
        
    async def start(self):
        """啟動搜尋者主循環"""
        # 啟動數據訂閱任務
        asyncio.create_task(self.pending_tx_subscription.listen(
            callback=self.on_pending_tx
        ))
        asyncio.create_task(self.block_subscription.listen(
            callback=self.on_new_block
        ))
        
        # 啟動機會處理任務
        asyncio.create_task(self.process_opportunities())
        
        self.logger.info("MEV 搜尋者已啟動")
        
    async def on_pending_tx(self, tx: Dict):
        """處理待確認交易"""
        for name, detector in self.opportunity_detectors.items():
            opportunity = detector.detect(tx)
            if opportunity and opportunity.net_profit > 0:
                await self.pending_opportunities.put(opportunity)
                
    async def on_new_block(self, block: Dict):
        """處理新區塊"""
        block_number = block['number']
        
        # 更新 Gas 估算
        base_fee = self.estimate_base_fee(block)
        
        # 處理積壓的機會
        processed = 0
        while not self.pending_opportunities.empty() and processed < 10:
            opp = await self.pending_opportunities.get()
            
            # 重新評估機會
            if opp.deadline < block_number:
                continue  # 機會已過期
                
            # 估算 Gas
            gas_price = self.estimate_gas_price(opp, base_fee)
            opp.gas_cost = gas_price * opp.estimated_gas
            opp.net_profit = opp.expected_profit - opp.gas_cost
            
            if opp.net_profit > self.min_profit_threshold:
                await self.submit_bundle(opp)
                processed += 1

2.2 套利檢測器實作

套利檢測器需要監控所有主流 DEX 的代幣對價格，並識別任何有意義的價差。以下是一個優化的套利檢測演算法：

class ArbitrageDetector:
    """DEX 套利檢測器"""
    
    def __init__(self, w3: Web3):
        self.w3 = w3
        self.dex_pools = self.load_dex_pools()
        self.price_cache = {}
        self.cache_ttl = 2  # 秒
        
    def load_dex_pools(self) -> Dict[str, List[Pool]]:
        """載入所有 DEX 池子資料"""
        pools = {}
        
        # Uniswap V2 工廠合約
        uniswap_v2 = UniswapV2Factory(UNISWAP_V2_FACTORY)
        for pair in uniswap_v2.get_all_pairs():
            pools.setdefault('uniswap_v2', []).append(pair)
            
        # Sushiswap
        sushi = UniswapV2Factory(SUSHISWAP_FACTORY)
        for pair in sushi.get_all_pairs():
            pools.setdefault('sushiswap', []).append(pair)
            
        # Curve
        curve_pools = CurveRegistry.get_pools()
        pools['curve'] = curve_pools
        
        return pools
        
    def detect(self, new_tx: Dict) -> Optional[MEVOpportunity]:
        """檢測新交易是否產生套利機會"""
        tx_hash = new_tx['hash']
        tx_to = new_tx.get('to', '')
        
        # 檢查是否是新池子交易
        if tx_to in self.new_pools:
            return self.check_triangular_arbitrage(tx_to, new_tx)
            
        # 檢查是否是 Swap 交易
        if self.is_swap_tx(new_tx):
            return self.check_direct_arbitrage(new_tx)
            
        return None
        
    def check_direct_arbitrage(self, swap_tx: Dict) -> Optional[MEVOpportunity]:
        """檢查直接套利機會（兩個 DEX 之間）"""
        token_in = swap_tx.get('token_in')
        token_out = swap_tx.get('token_out')
        amount_in = swap_tx.get('amount_in')
        
        # 獲取兩個市場的報價
        dex_a = self.get_quote(token_in, token_out, amount_in)
        dex_b = self.get_quote_opposite(token_in, token_out, amount_in)
        
        if not dex_a or not dex_b:
            return None
            
        # 計算套利路徑
        # 路徑 1: DEX A -> DEX B
        path1_profit = self.calculate_arbitrage_profit(
            token_in, token_out, amount_in, 'dex_a', 'dex_b'
        )
        
        # 路徑 2: DEX B -> DEX A
        path2_profit = self.calculate_arbitrage_profit(
            token_in, token_out, amount_in, 'dex_b', 'dex_a'
        )
        
        # 選擇最優路徑
        best_profit = max(path1_profit, path2_profit)
        if best_profit > self.min_profit:
            return MEVOpportunity(
                opportunity_type='arbitrage',
                expected_profit=best_profit,
                # ... 其他欄位
            )
            
        return None
        
    def calculate_arbitrage_profit(
        self, 
        token_a: str, 
        token_b: str, 
        amount: int,
        dex_buy: str,
        dex_sell: str
    ) -> int:
        """
        計算套利利潤
        
        公式推導：
        1. 在 dex_buy 以 P_buy 價格買入 token_b
           獲得: amount * P_buy / (1 - fee_buy)
        2. 在 dex_sell 以 P_sell 價格賣出 token_b
           獲得: amount * P_buy * P_sell / ((1 - fee_buy) * (1 - fee_sell))
        3. 利潤 = 最終獲得 - 初始 amount
        """
        p_buy = self.get_price(dex_buy, token_a, token_b)
        p_sell = self.get_price(dex_sell, token_b, token_a)
        
        fee_buy = 0.003  # 0.3% Uniswap/Sushi fee
        fee_sell = 0.003
        
        # 考慮費用的實際獲得量
        received = amount * p_buy * (1 - fee_buy)
        final = received * p_sell * (1 - fee_sell)
        
        profit = final - amount
        return int(profit)

2.3 清算檢測器實作

清算檢測需要實時監控借貸協議的健康因子，並在觸發清算時立即行動。以下是一個高效能的清算檢測器：

class LiquidationDetector:
    """DeFi 清算檢測器"""
    
    # 支援的借貸協議
    SUPPORTED_PROTOCOLS = {
        'aave_v3': {
            'pool': '0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2',
            'price_oracle': '0x54586bE62E3c3580375aE3723C14525306031B03',
            'health_factor_threshold': 1.0,
        },
        'compound_v3': {
            'comet': '0xc3d688B66703497DAA19211EEdff47f25384cdc3',
            'price_feed': '0x5f4eC3Df9cbd43714FE2740f5E3617235B1f0a51',
            'liquidator_points': 0.1,  # 10% of seized collateral
        },
        'euler': {
            'euler': '0x152B5mgb8BF79d36AFD0dA1fcEEdC6Fb25b4CAb',
            'markets': EulerMarkets.get_markets(),
        },
    }
    
    def __init__(self, w3: Web3):
        self.w3 = w3
        self.protocols = self.SUPPORTED_PROTOCOLS
        
        # 快取熱門用戶的健康因子
        self.health_factor_cache = {}
        
        # 事件監聽器
        self.event_listeners = {
            'borrow': self.on_borrow,
            'repay': self.on_repay,
            'supply': self.on_supply,
            'withdraw': self.on_withdraw,
            'price_update': self.on_price_update,
        }
        
    def detect(self, new_tx: Dict) -> Optional[MEVOpportunity]:
        """檢測交易是否觸發清算"""
        tx_to = new_tx.get('to', '').lower()
        
        # 檢查是否是借貸協議合約
        for protocol_name, config in self.protocols.items():
            if tx_to == config.get('pool', '').lower():
                return self.analyze_lending_protocol_tx(protocol_name, new_tx)
                
        return None
        
    def analyze_lending_protocol_tx(
        self, 
        protocol: str, 
        tx: Dict
    ) -> Optional[MEVOpportunity]:
        """分析借貸協議交易是否觸發清算"""
        
        if protocol == 'aave_v3':
            return self.detect_aave_liquidation(tx)
        elif protocol == 'compound_v3':
            return self.detect_compound_liquidation(tx)
            
        return None
        
    def detect_aave_liquidation(self, tx: Dict) -> Optional[MEVOpportunity]:
        """檢測 Aave V3 清算機會"""
        
        # 解析交易輸入資料
        func_sig = tx['input'][:10]
        
        if func_sig == '0x0c3a2c6c':  # liquidate
            # 解析清算參數
            borrower = self.w3.toChecksumAddress(tx['input'][10:74])
            
            # 檢查借款人的健康因子
            health_factor = self.get_aave_health_factor(borrower)
            
            if health_factor < 1.0:
                # 計算清算利潤
                liquidation = self.calculate_liquidation_profit(
                    protocol='aave_v3',
                    borrower=borrower,
                    debt_asset=tx.get('debt_asset'),
                    collateral_asset=tx.get('collateral_asset'),
                )
                
                if liquidation['profit'] > self.min_profit:
                    return MEVOpportunity(
                        opportunity_type='liquidation',
                        expected_profit=liquidation['profit'],
                        target_contracts=[LIQUIDATION_PROXY],
                        encoding=liquidation['calldata'],
                    )
                
    def calculate_liquidation_profit(
        self,
        protocol: str,
        borrower: str,
        debt_asset: str,
        collateral_asset: str,
    ) -> Dict:
        """
        計算清算利潤
        
        Aave V3 清算數學模型：
        - 清算閾值 (Liquidation Threshold): LT
        - 健康因子 (Health Factor): HF = Σ(Collateral * Price * LT) / Σ(Debt * Price)
        - 清算獎勵: 通常為抵押品的 5-10%
        
        清算利潤 = 抵押品價值 * 清算獎勵比例 - Gas 費用
        """
        # 獲取借款人位置
        position = self.get_borrow_position(protocol, borrower)
        
        debt_amount = position['debt_amount']
        collateral_amount = position['collateral_amount']
        debt_value = debt_amount * self.get_price(debt_asset)
        collateral_value = collateral_amount * self.get_price(collateral_asset)
        
        # Aave V3 清算參數
        liquidation_bonus = 0.10  # 10% 清算獎勵
        liquidation_threshold = 0.80  # 80% LT
        
        # 最大可清算債務量
        max_liquidation = min(
            debt_value * 0.50,  # 單次最多清算 50% 債務
            collateral_value * liquidation_threshold
        )
        
        # 清算獎勵
        liquidation_profit = max_liquidation * liquidation_bonus
        
        return {
            'profit': int(liquidation_profit),
            'debt_to_cover': int(max_liquidation / self.get_price(debt_asset)),
            'collateral_to_claim': int(max_liquidation * (1 + liquidation_bonus) / self.get_price(collateral_asset)),
            'gas_cost': self.estimate_liquidation_gas(),
        }

三、區塊構建者技術實作

3.1 區塊構建者架構

區塊構建者的核心職責是收集搜尋者提交的 Bundle，選擇能夠最大化區塊價值的交易組合，並將其組織成最終的區塊。以下是一個優化的區塊構建者架構：

import heapq
from typing import List, Tuple, Dict, Optional
from dataclasses import dataclass, field
from web3 import Web3
from web3.types import BlockNumber

@dataclass(order=True)
class BundledTransaction:
    """帶優先級的交易"""
    sort_key: float = field(compare=True)  # profit / gas
    gas_limit: int = field(compare=False)
    gas_price: int = field(compare=False)
    bundle_hash: str = field(compare=False)
    txs: List[bytes] = field(compare=False)
    refund_config: Dict = field(compare=False)
    valid_until: int = field(compare=False)
    
    def __repr__(self):
        return f"BundledTx(profit/gas={self.sort_key:.2f}, gas={self.gas_limit})"

class BlockBuilder:
    """MEV 區塊構建者"""
    
    def __init__(
        self,
        builder_key: str,
        relay_url: str,
        eth_node_url: str,
    ):
        self.signer = LocalSigner(builder_key)
        self.w3 = Web3(Web3.HTTPProvider(eth_node_url))
        self.relay = BuilderRelay(relay_url)
        
        # 交易接收器（接收搜尋者的 Bundle）
        self.bundle_receiver = BundleReceiver()
        
        # 區塊空間估算器
        self.gas_estimator = GasEstimator(self.w3)
        
        # 收益計算器
        self.profit_calculator = ProfitCalculator()
        
        # 優先級隊列（最大堆）
        self.pending_bundles: List[BundledTransaction] = []
        
    async def receive_bundle(self, bundle: Dict) -> str:
        """
        接收搜尋者提交的 Bundle
        
        Bundle 格式：
        {
            'txs': ['0x...', '0x...'],  # 交易列表
            'bundle_hash': '0x...',     # Bundle 哈希
            'max_block_number': 10,     # 有效期（區塊數）
            'gas_price': 10000000000,   # 最低 Gas 價格
            'refund_config': {          # 退款配置
                'address': '0x...',     # 退款地址
                'percent': 80,         # 退款比例
            }
        }
        """
        # 驗證 Bundle
        if not self.validate_bundle(bundle):
            raise ValueError("Invalid bundle")
            
        # 估算 Gas
        gas_limit = await self.estimate_bundle_gas(bundle['txs'])
        
        # 計算 Bundle 價值
        bundle_value = await self.profit_calculator.calculate(
            bundle['txs'],
            self.w3.eth.get_block('pending')
        )
        
        # 計算優先級分數 (profit per gas)
        gas_price = bundle.get('gas_price', 0)
        priority_score = bundle_value / max(gas_limit, 1)
        
        # 創建優先級交易物件
        bundled_tx = BundledTransaction(
            sort_key=-priority_score,  # 負數因為是最小堆
            gas_limit=gas_limit,
            gas_price=gas_price,
            bundle_hash=bundle['bundle_hash'],
            txs=bundle['txs'],
            refund_config=bundle.get('refund_config', {}),
            valid_until=bundle.get('max_block_number', 10),
        )
        
        # 加入優先級隊列
        heapq.heappush(self.pending_bundles, bundled_tx)
        
        return bundle['bundle_hash']
        
    async def build_optimized_block(
        self,
        parent_block: Dict,
        target_gas_limit: int,
    ) -> Dict:
        """
        構建優化區塊
        
        目標函數：最大化區塊價值 subject to 區塊 Gas 限制
        
        這是一個貪心近似演算法：
        1. 按 profit/gas 降序選擇 Bundle
        2. 考慮 Bundle 之間的衝突
        3. 填充剩餘空間普通交易
        """
        included_bundles = []
        remaining_gas = target_gas_limit
        
        # 從優先級隊列選擇 Bundle
        while self.pending_bundles and remaining_gas > 21000:
            bundle = heapq.heappop(self.pending_bundles)
            
            # 檢查有效性
            if bundle.valid_until < parent_block['number']:
                continue  # Bundle 已過期
                
            # 檢查 Gas 限制
            if bundle.gas_limit > remaining_gas:
                # 嘗試放入下一個區塊
                if bundle.valid_until > parent_block['number']:
                    heapq.heappush(self.pending_bundles, bundle)
                continue
                
            # 檢查衝突（Bundle 之間是否有依賴關係）
            if self.has_conflict(bundle, included_bundles):
                continue
                
            # 包含此 Bundle
            included_bundles.append(bundle)
            remaining_gas -= bundle.gas_limit
            
        # 構建區塊交易列表
        block_txs = []
        for bundle in included_bundles:
            for tx_bytes in bundle.txs:
                block_txs.append(tx_bytes)
                
        # 填充普通交易（按 Gas 價格排序）
        pending_txs = await self.get_pending_transactions()
        for tx in sorted(pending_txs, key=lambda x: x['gasPrice'], reverse=True):
            if tx['gas'] > remaining_gas:
                continue
                
            # 檢查與已包含 Bundle 的衝突
            if self.tx_conflicts_with_bundles(tx, included_bundles):
                continue
                
            block_txs.append(tx['transactionHash'])
            remaining_gas -= tx['gas']
            
            if remaining_gas < 21000:
                break
                
        return {
            'transactions': block_txs,
            'included_bundles': [b.bundle_hash for b in included_bundles],
            'total_gas_used': target_gas_limit - remaining_gas,
        }
        
    def has_conflict(
        self, 
        bundle: BundledTransaction, 
        included: List[BundledTransaction]
    ) -> bool:
        """
        檢查 Bundle 是否與已包含的 Bundle 衝突
        
        衝突檢測邏輯：
        1. 提取所有 Bundle 涉及的合約地址
        2. 檢查是否有重疊
        3. 考慮交易的 nonce 依賴
        """
        bundle_contracts = self.extract_contracts(bundle.txs)
        
        for inc_bundle in included:
            inc_contracts = self.extract_contracts(inc_bundle.txs)
            
            # 合約地址重疊檢測
            if bundle_contracts & inc_contracts:
                return True
                
            # nonce 依賴檢測
            if self.has_nonce_dependency(bundle.txs, inc_bundle.txs):
                return True
                
        return False

3.2 利潤最大化演算法

區塊構建者的核心問題是如何選擇和排序 Bundle 以最大化利潤。這是一個複雜的最佳化問題，以下是一個更精確的數學模型和實作：

class ProfitMaximizer:
    """
    區塊利潤最大化器
    
    數學模型：
    最大化: Σ(v_i * x_i) - C * G
    subject to:
        Σ(g_i * x_i) ≤ G_max  (Gas 限制)
        x_i ∈ {0, 1}  (每個 Bundle 要麼包含要麼不包含)
        
    其中：
    v_i = Bundle i 的預期利潤
    g_i = Bundle i 消耗的 Gas
    G = 區塊總 Gas 限制
    C = 每單位 Gas 的機會成本（base fee + priority fee）
    """
    
    def __init__(self, w3: Web3):
        self.w3 = w3
        self.base_fee_history = []
        
    def maximize_profit(
        self,
        bundles: List[Bundle],
        max_gas: int,
        block_base_fee: int,
    ) -> Tuple[List[Bundle], int]:
        """
        使用貪心近似演算法求解
        
        時間複雜度: O(n log n)
        近似比: (1 - 1/e) ≈ 0.63（對於 unconstrained knapsack）
        """
        
        # 計算每個 Bundle 的效益分數
        scored_bundles = []
        for bundle in bundles:
            # 估計 Gas 成本
            gas_cost = bundle.gas_limit * block_base_fee
            
            # 估計 MEV 獎勵
            mev_reward = bundle.estimated_profit
            
            # 淨效益
            net_value = mev_reward - gas_cost
            
            if net_value > 0:
                score = net_value / bundle.gas_limit
                scored_bundles.append((score, bundle, net_value))
                
        # 按分數降序排序
        scored_bundles.sort(key=lambda x: x[0], reverse=True)
        
        # 貪心選擇
        selected = []
        total_gas = 0
        total_profit = 0
        
        for score, bundle, net_value in scored_bundles:
            if total_gas + bundle.gas_limit <= max_gas:
                selected.append(bundle)
                total_gas += bundle.gas_limit
                total_profit += net_value
                
        return selected, total_profit
        
    def estimate_bundle_profit(self, bundle: Bundle) -> int:
        """
        估計 Bundle 的預期利潤
        
        利潤來源分析：
        1. 直接 MEV（套利、清算等）
        2. Gas 補貼（搜尋者願意支付的額外費用）
        3. 區塊獎勵份額
        
        利潤 = 直接 MEV + Σ(refund_i) - Gas成本
        """
        direct_mev = bundle.estimated_mev
        
        # 計算退款
        refunds = 0
        for refund_config in bundle.refund_configs:
            refund_percent = refund_config['percent']
            bundle_value = self.estimate_bundle_value(bundle)
            refunds += bundle_value * (refund_percent / 100)
            
        # Gas 成本
        gas_cost = bundle.gas_limit * self.get_current_gas_price()
        
        return int(direct_mev + refunds - gas_cost)

四、提議者-構建者分離（PBS）機制

4.1 PBS 運作原理

提議者-構建者分離是以太坊解決 MEV 中心化問題的核心機制。在 PBS 之前，驗證者（提議者）同時負責區塊提議和區塊構建，這導致大型質押池可以透過 MEV 獲得不成比例的收益，造成網路激勵的失衡。

PBS 將區塊構建權責從驗證者分離出來，由專業的區塊構建者負責。提議者只需選擇價值最高的區塊提案，而無需關心區塊內容。這種設計的優點包括：

1. 降低驗證者複雜度：驗證者無需運行 MEV 搜尋基礎設施
2. 減少 MEV 提取的中心化：小型質押者透過 MEV-Boost 獲得 MEV 收益
3. 提高抗審查能力：構建者多樣性減少單點審查風險

PBS 的完整流程如下：

1. 區塊構建者：
   - 接收搜尋者的 Bundle
   - 選擇並排序交易
   - 計算區塊 value（支付給提議者的費用）
   - 發送區塊投標到中繼網路

2. 中繼網路（Relay）：
   - 驗證構建者的投標
   - 計算區塊頭
   - 提供區塊頭給提議者

3. 驗證者（提議者）：
   - 從多個中繼接收區塊投標
   - 選擇價值最高的區塊頭
   - 使用自己的私鑰簽署區塊頭
   - 發送簽名的區塊頭回中繼

4. 中繼完成：
   - 驗證提議者簽名
   - 向構建者發送完整的區塊
   - 將區塊提交到網路

4.2 MEV-Boost 實作

Flashbots MEV-Boost 是目前最廣泛使用的 PBS 實現。以下是一個簡化的 MEV-Boost 用戶端實作：

import httpx
from eth_typing import BlockNumber, ChecksumAddress
from web3 import Web3
from web3.types import SignedBeaconBlock, ExecutionPayload

class MEVBoostClient:
    """MEV-Boost 用戶端（驗證者端）"""
    
    def __init__(
        self,
        consensus_node_url: str,
        execution_node_url: str,
        relay_urls: List[str],
    ):
        self.consensus = ConsensusClient(consensus_node_url)
        self.execution = ExecutionClient(execution_node_url)
        self.relays = [RelayClient(url) for url in relay_urls]
        
        # 驗證者私鑰（僅用於簽署區塊頭）
        self.validator_sk = None
        
    async def get_best_bid(
        self, 
        slot: int
    ) -> Optional[Dict]:
        """
        從所有中繼獲取最佳投標
        
        返回價值最高的區塊投標
        """
        tasks = [
            relay.get_header(slot) 
            for relay in self.relays
        ]
        
        bids = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 過濾無效投標
        valid_bids = [
            bid for bid in bids 
            if isinstance(bid, dict) and bid.get('value', 0) > 0
        ]
        
        if not valid_bids:
            return None
            
        # 選擇價值最高的投標
        best_bid = max(valid_bids, key=lambda x: int(x['value'], 16))
        
        return best_bid
        
    async def propose_block(
        self,
        slot: int,
        best_bid: Dict,
    ) -> bool:
        """
        提議區塊
        
        步驟：
        1. 驗證投標有效性
        2. 簽署區塊頭
        3. 發送給中繼
        4. 等待確認
        """
        relay = self._find_relay_for_bid(best_bid)
        
        # 獲取完整區塊內容
        payload = await relay.get_payload(best_bid['blockHash'])
        
        # 構造信標區塊
        signed_block = await self.construct_beacon_block(
            slot=slot,
            execution_payload=payload,
        )
        
        # 簽署區塊頭
        domain = self.consensus.get_domain(
            fork_version=self.consensus.get_fork_version(slot),
            domain_type=b'\x01',  # DOMAIN_BEACON_PROPOSER
        )
        
        signing_root = self.consensus.compute_signing_root(
            block_root=best_bid['blockHash'],
            domain=domain,
        )
        
        signature = self.validator_sk.sign(signing_root)
        
        # 發送已簽署的區塊
        response = await relay.submit_block(
            signed_block=signed_block,
            signature=signature,
        )
        
        return response.get('status') == 'VALID'
        
    def _find_relay_for_bid(self, bid: Dict) -> RelayClient:
        """根據投標找到對應的中繼"""
        pubkey = bid.get('pubkey')
        for relay in self.relays:
            if relay.pubkey == pubkey:
                return relay
        raise ValueError("Unknown relay for bid")

4.3 PBS 經濟學分析

PBS 機制重塑了以太坊驗證者的激勵結構。讓我們分析 PBS 的經濟學影響：

驗證者收益變化：

中繼市場結構：

截至 2026 年第一季度，主要中繼提供商包括：

• Flashbots（~65% 市場份額）
• Blocknative（~10%）
• Blocksmith（~8%）
• BloXroute（~7%）
• 其他（~10%）

這種市場集中度引發了對 PBS 抗審查性的擔憂。若單一中繼控制過大市場份額，可能成為網路審查的潛在攻擊面。

五、Flashbots SUAVE：去中心化記憶體池

5.1 SUAVE 設計理念

Flashbots SUAVE（Settler Unauthorized VALue Extraction）是 Flashbots 提出的下一代 MEV 基礎設施願景，旨在建立一個去中心化的記憶體池和區塊構建網路。SUAVE 的核心設計目標是：

1. 去中心化：消除單點故障和審查風險
2. 隱私：保護交易免受搶先交易和三明治攻擊
3. 效率：最大化 MEV 價值的公平分配
4. 可組合性：允許跨鏈 MEV 策略

SUAVE 採用分層架構：

┌─────────────────────────────────────────┐
│           User Intent Layer             │
│  (意圖表達：我想用 X 換 Y)              │
└─────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────┐
│          Preference Pool                │
│  (用戶偏好：隱私、速度、費用)            │
└─────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────┐
│        Solver Network (求解器網路)       │
│  (竞標滿足用戶意圖的最佳方案)            │
└─────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────┐
│        Universal Mempool                │
│  (共享記憶體池，跨區塊鏈)                │
└─────────────────────────────────────────┘

5.2 SUAVE 合約實作

SUAVE 的核心是一組部署在 Solidity 的智慧合約。以下是簡化的關鍵合約介面：

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.19;

/**
 * @title ISuavePool
 * @notice SUAVE 記憶體池介面
 */
interface ISuavePool {
    /**
     * @notice 提交交易到記憶體池
     * @param  tx 交易資料
     * @param  confidentialInputs 機密輸入（用於隱私計算）
     */
    function submitTransaction(
        bytes calldata tx,
        bytes calldata confidentialInputs
    ) external payable returns (bytes32 txHash);
    
    /**
     * @notice 批量提交交易
     */
    function submitBundle(
        Transaction[] calldata txs,
        bytes[] calldata confidentialInputs
    ) external payable returns (bytes32 bundleHash);
    
    /**
     * @notice 查詢交易狀態
     */
    function getTransactionStatus(
        bytes32 txHash
    ) external view returns (TransactionStatus);
}

/**
 * @title ISolver
 * @notice 求解器介面
 */
interface ISolver {
    /**
     * @notice 提交求解方案
     * @param  bundleHash 捆綁包哈希
     * @param  solution 解題方案（包含交易序列和支付金額）
     * @param  bond 質押保證金
     */
    function submitSolution(
        bytes32 bundleHash,
        Solution calldata solution,
        uint256 bond
    ) external;
    
    /**
     * @notice 領取獎勵
     */
    function claimReward(bytes32 bundleHash) external;
}

/**
 * @title SuavePool
 * @notice SUAVE 記憶體池主合約
 */
contract SuavePool is ISuavePool {
    // 儲存交易
    mapping(bytes32 => Transaction) public transactions;
    
    // 儲存捆綁包
    mapping(bytes32 => Bundle) public bundles;
    
    // 求解器註冊
    mapping(address => Solver) public solvers;
    
    // 全域設定
    uint256 public constant MIN_BOND = 0.1 ether;
    uint256 public constant SOLUTION_TIMEOUT = 10 minutes;
    uint256 public constant FEE_PERCENTAGE = 1; // 1% 協議費
    
    // 事件
    event TransactionSubmitted(
        bytes32 indexed txHash,
        address indexed submitter,
        uint256 gasLimit
    );
    
    event BundleCreated(
        bytes32 indexed bundleHash,
        Transaction[] txs,
        uint256 bundleValue
    );
    
    event SolutionSubmitted(
        bytes32 indexed bundleHash,
        address indexed solver,
        uint256 payment
    );
    
    function submitTransaction(
        bytes calldata tx,
        bytes calldata confidentialInputs
    ) external payable override returns (bytes32 txHash) {
        // 解碼交易
        (address to, uint256 value, bytes memory data) = abi.decode(
            tx,
            (address, uint256, bytes)
        );
        
        // 計算交易哈希
        txHash = keccak256(abi.encodePacked(tx, confidentialInputs));
        
        // 儲存交易
        transactions[txHash] = Transaction({
            to: to,
            value: value,
            data: data,
            sender: msg.sender,
            confidentialInputs: keccak256(confidentialInputs),
            gasLimit: tx gasLimit,
            status: TransactionStatus.Pending,
            blockNumber: 0
        });
        
        emit TransactionSubmitted(txHash, msg.sender, tx.gasLimit);
    }
    
    function submitBundle(
        Transaction[] calldata txs,
        bytes[] calldata confidentialInputs
    ) external payable override returns (bytes32 bundleHash) {
        require(txs.length > 0, "Empty bundle");
        require(txs.length == confidentialInputs.length, "Length mismatch");
        
        // 驗證所有交易
        uint256 totalValue = 0;
        for (uint i = 0; i < txs.length; i++) {
            bytes32 txHash = keccak256(abi.encodePacked(
                txs[i],
                confidentialInputs[i]
            ));
            
            // 驗證交易存在且狀態正確
            require(
                transactions[txHash].status == TransactionStatus.Pending,
                "Invalid transaction"
            );
            
            totalValue += txs[i].value;
        }
        
        // 創建捆綁包
        bundleHash = keccak256(abi.encodePacked(
            block.timestamp,
            msg.sender,
            keccak256(abi.encode(txs))
        ));
        
        bundles[bundleHash] = Bundle({
            transactions: txs,
            confidentialInputs: confidentialInputs,
            submitter: msg.sender,
            status: BundleStatus.Open,
            deadline: block.timestamp + SOLUTION_TIMEOUT,
            totalValue: totalValue,
            bestPayment: 0,
            bestSolver: address(0)
        });
        
        emit BundleCreated(bundleHash, txs, totalValue);
    }
    
    function submitSolution(
        bytes32 bundleHash,
        Solution calldata solution,
        uint256 bond
    ) external override {
        require(bond >= MIN_BOND, "Bond too low");
        require(
            bundles[bundleHash].status == BundleStatus.Open,
            "Bundle not open"
        );
        require(
            block.timestamp <= bundles[bundleHash].deadline,
            "Deadline passed"
        );
        
        // 轉移保證金
        solvers[msg.sender].bond += bond;
        
        // 驗證解決方案
        uint256 payment = _calculatePayment(bundleHash, solution);
        require(
            payment <= bundles[bundleHash].totalValue,
            "Payment exceeds value"
        );
        
        // 檢查是否是最優方案
        if (payment > bundles[bundleHash].bestPayment) {
            bundles[bundleHash].bestPayment = payment;
            bundles[bundleHash].bestSolver = msg.sender;
        }
        
        emit SolutionSubmitted(bundleHash, msg.sender, payment);
    }
}

六、結論與展望

MEV 基礎設施是以太坊生態系統中最具技術挑戰性和經濟價值的領域之一。從本文的分析可以看出，MEV 供應鏈的每個環節——搜尋者、區塊構建者、提議者和中繼——都有其獨特的技術複雜性和經濟激勵結構。

關鍵要點回顧：

1. MEV 是結構性現象：只要區塊空間存在稀缺性、交易排序有價值、記憶體池不完美透明，MEV 就會持續存在。

1. PBS 是必要的演進：透過分離區塊構建與提議職責，PBS 降低了 MEV 對網路去中心化的負面影響，同時為小型驗證者提供了參與 MEV 收益的途徑。

1. SUAVE 代表未來方向：去中心化記憶體池和求解器網路有望創造一個更公平、更高效的 MEV 生態系統。

1. 安全性仍是挑戰：MEV 基礎設施涉及大量資金流動，智慧合約漏洞可能導致災難性損失。形式化驗證和嚴格的審計是不可或缺的。

展望未來，隨著以太坊持續演進（The Surge、The Scourge 等升級）以及跨鏈互操作性的提升，MEV 的形態將繼續演化。開發者應持續關注這個快速變化的領域，並在構建 MEV 相關系統時始終將安全性和抗審查性放在首位。

參考資料

以太坊官方文檔

EIP-1559 提案

Flashbots MEV-Boost

MEV-Explore 數據看板

Etherscan