以太坊 MEV Solver Network 實作完整指南：從理論到程式碼部署
本文深入探討以太坊 MEV 生態系統中的 Solver Network 架構與實作。我們詳細分析搜尋者-求解器-建構者三層市場結構，提供完整的 Python 求解器程式碼範例（貪心演算法、ILP 嚴格求解），並探討 Solver Network 的經濟學模型與盈利策略。同時包含 CoW Protocol 的批量拍賣機制解析與實戰經驗總結。
以太坊 MEV Solver Network 實作完整指南：從理論到程式碼部署
前言
聊 MEV 聊了這麼久，終於要動手寫實作的部分了。之前那篇文章把搜尋者-建構者-提議者的三層市場結構掰開揉碎講了一遍，但說實話，讀者看完之後可能還是會問：「那我到底怎麼成為這個生態系的一部分？」
這篇文章就是要回答這個問題。我會帶你從零開始，理解 Solver Network 到底是什麼、為什麼它比傳統 MEV 方案更有優勢、以及最重要的——怎麼用程式碼實現一個能跑的 Solver。
預先聲明：這不是給純小白的科普文，你起碼得知道以太坊是啥、智能合約怎麼回事、Gas 是什麼意思。如果連這些都不清楚，先去補補課再回來。
一、Solver Network 是什麼來頭
1.1 傳統 MEV 的痛點
先說個暴論：傳統 MEV 方案的設計從根子上就有問題。
你想啊，搜尋者辛辛苦苦算出來的最優交易排序，結果被建構者輕鬆抄走了。搜尋者之間也是互相內卷，今天你搶我的單，明天我搶你的單，到頭來便宜的是誰？區塊生產者。典型的囚徒困境。
Solver Network 這套機制的核心思想是：把「找到最優解」和「執行最優解」分開，讓它們成為兩個獨立的環節。
傳統 MEV 模式：
搜尋者 A ──▶ 搜尋者 B ──▶ 搜尋者 C ──▶ ...
│ │ │
└───────────┴───────────┴──▶ 建構者
問題：搜尋者之間無序競爭，重複計算嚴重
Solver Network 模式：
┌──────────────────────────────────────────────┐
│ Solver Network 層 │
│ │
│ Solver A ←──→ Solver B ←──→ Solver C │
│ │ │ │ │
│ └─────────────┴─────────────┘ │
│ │ │
│ ▼ │
│ 協調理層 │
│ （拍賣/分配共識） │
│ │ │
│ ▼ │
│ Builder Network │
└──────────────────────────────────────────────┘
優勢：搜尋者專心算，解題者專心解，分工明確1.2 CoW Protocol 的先行實踐
說到 Solver Network，不能不提 CoW Protocol。這幫人是第一個把「批量拍賣」（Batch Auction）機制玩明白的。
CoW 的核心邏輯是這樣的：
- 在一個批次窗口內（比如 5 秒），收集所有交易意圖
- 這些交易意圖不是直接上鏈，而是先用「求解器」（Solver）計算最優匹配
- 求解器之間互相競價，誰的方案能讓所有交易者福利最大化，誰就中標
- 中標方案一次性執行，結算在鏈上
好處在哪裡？
第一，交易者不會被「搶跑」。因為在求解器計算完成之前，沒有任何交易會進入 mempool。
第二，MEV 被「內部化」了。傳統模式下，套利利潤被搜尋者拿走；CoW 模式下，這筆錢可以回流給交易者（比如更低的滑點）。
第三，批量執行減少了碎片化，Gas 效率更高。
1.3 Solver Network 的架構全景
完整的 Solver Network 大致可以分為這幾層：
Solver Network 架構：
Layer 1: 意圖層（Intent Layer）
│
├── 用戶提交「意圖」而非「交易」
│ 例如：「我想用 1000 USDC 換尽可能多的 ETH，滑點容忍 1%」
│
└── 意圖被封裝成標準格式（ERC-7683 標準正在制定中）
Layer 2: 求解層（Solver Layer）
│
├── 多個 Solver 節點接收同一批意圖
│
├── 每個 Solver 獨立地計算最優執行方案
│ - 路由到哪個 DEX
│ - 分割訂單
│ - 時間窗口選擇
│
└── 提交「投標」（Solution），附帶保證金
Layer 3: 拍賣層（Auction Layer）
│
├── 評估所有投標
│ - 哪個方案總福利最大
│ - 考慮 Gas 成本
│ - 計算各參與方的分配
│
└── 選擇最優投標，結算
Layer 4: 結算層（Settlement Layer）
│
├── 最優方案被廣播到網路
│
├── 與主流 MEV-Boost 生態整合
│
└── 結算結果上鏈二、Solver 的核心演算法
2.1 問題建模
一個典型的 DEX 聚合求解問題可以這樣建模：
輸入：
- 一批交易意圖 I = {i1, i2, ..., in}
- 可用的流動性來源 L = {l1, l2, ..., lm}（各個 DEX 池子）
- 當前市場價格 P
- Gas 成本 G
輸出：
- 分配方案 A = {a1, a2, ..., an}
- 其中 ai 指定如何執行交易意圖 i
目標：
最大化總福利 W(A) = Σ 交易者剩餘 - Gas 成本翻譯成人話就是：讓買家少花錢、賣家多賺錢，同時把 Gas 燒得最少。
2.2 貪心演算法的簡化實現
先從最簡單的貪心策略說起。這個策略不追求最優解，但勝在計算快，適合小批量交易。
# solver/simple_greedy.py
"""
簡化版貪心求解器
策略：優先處理滑點敏感的交易，把它们路由到最好的價格
"""
from dataclasses import dataclass
from typing import List, Dict, Tuple
from decimal import Decimal
import heapq
@dataclass
class Intent:
"""交易意圖"""
id: str
sender: str
sell_token: str # 賣出的代幣
buy_token: str # 買入的代幣
sell_amount: int # 賣出數量
max_slippage_bps: int # 最大滑點（基點）
@dataclass
class LiquiditySource:
"""流動性來源（DEX 池子）"""
pool_id: str
token_in: str
token_out: str
reserve_in: int
reserve_out: int
fee_bps: int # 手續費（基點）
@dataclass
class TradeExecution:
"""單筆交易執行方案"""
intent_id: str
pool_id: str
amount_in: int
expected_out: int
actual_out: int
gas_cost: int
net_benefit: int # 凈收益
class SimpleGreedySolver:
"""
簡化貪心求解器
演算法步驟：
1. 按滑點敏感度排序意圖
2. 遍歷每個意圖，選擇最優流動性來源
3. 更新池子狀態（考慮AMM特性）
"""
def __init__(self, gas_price: int = 30_000_000_000): # 30 Gwei
self.gas_price = gas_price
self.gas_per_trade = 150_000 # 單筆交易約 150k Gas
def solve(
self,
intents: List[Intent],
liquidity: Dict[str, List[LiquiditySource]]
) -> List[TradeExecution]:
"""
求解主函數
參數：
intents: 交易意圖列表
liquidity: 按交易對索引的流動性來源
返回：
執行方案列表
"""
# 按滑點容忍度排序（容忍度低的先處理）
sorted_intents = sorted(
intents,
key=lambda x: x.max_slippage_bps
)
executions = []
# 追蹤每個池子的剩餘流動性
pool_reserves = {}
for pools in liquidity.values():
for pool in pools:
pool_reserves[pool.pool_id] = {
'in': pool.reserve_in,
'out': pool.reserve_out
}
for intent in sorted_intents:
# 構造交易對 key
pair_key = (intent.sell_token, intent.buy_token)
# 獲取可用流動性
available_pools = liquidity.get(pair_key, [])
if not available_pools:
print(f"警告：找不到 {pair_key} 的流動性")
continue
# 選擇最優池子
best_pool, best_execution = self._find_best_route(
intent,
available_pools,
pool_reserves
)
if best_execution:
# 更新池子狀態（AMM 滑點）
self._update_pool_reserves(
best_pool,
best_execution.amount_in,
best_execution.actual_out,
pool_reserves
)
executions.append(best_execution)
return executions
def _find_best_route(
self,
intent: Intent,
pools: List[LiquiditySource],
reserves: Dict[str, Dict]
) -> Tuple[LiquiditySource, TradeExecution]:
"""
為單個意圖找最優路由
選擇邏輯：
1. 計算每個池子的預期輸出
2. 扣除 Gas 成本
3. 選擇凈收益最高的
"""
best_pool = None
best_execution = None
best_net_benefit = -float('inf')
for pool in pools:
# 計算交易輸出
amount_out = self._calculate_output(
intent.sell_amount,
reserves[pool.pool_id]['in'],
reserves[pool.pool_id]['out'],
pool.fee_bps
)
# 計算滑點
slippage = self._calculate_slippage(
amount_out,
intent.sell_amount,
reserves[pool.pool_id]['out'],
reserves[pool.pool_id]['in']
)
# 檢查滑點容忍度
if slippage > intent.max_slippage_bps:
continue
# 計算 Gas 成本
gas_cost = self.gas_per_trade * self.gas_price
# 計算凈收益
# 這裡用輸入代幣價值估算（簡化版本）
net_benefit = amount_out - gas_cost
if net_benefit > best_net_benefit:
best_pool = pool
best_net_benefit = net_benefit
best_execution = TradeExecution(
intent_id=intent.id,
pool_id=pool.pool_id,
amount_in=intent.sell_amount,
expected_out=amount_out,
actual_out=amount_out, # 簡化：假設無 MEV 干擾
gas_cost=gas_cost,
net_benefit=net_benefit
)
return best_pool, best_execution
def _calculate_output(
self,
amount_in: int,
reserve_in: int,
reserve_out: int,
fee_bps: int
) -> int:
"""
計算 AMM 輸出（常數乘積公式）
x * y = k
(reserve_in + amount_in * (1 - fee)) * (reserve_out - amount_out) = k
解得：
amount_out = reserve_out * amount_in * (1 - fee) / (reserve_in + amount_in * (1 - fee))
"""
if reserve_in == 0 or reserve_out == 0:
return 0
fee_multiplier = 10_000 - fee_bps
amount_in_with_fee = amount_in * fee_multiplier
numerator = amount_in_with_fee * reserve_out
denominator = reserve_in * 10_000 + amount_in_with_fee
return numerator // denominator
def _calculate_slippage(
self,
amount_out: int,
amount_in: int,
reserve_out: int,
reserve_in: int
) -> int:
"""
計算滑點（基點）
滑點 = (預期價格 - 實際價格) / 預期價格 * 10000
"""
if amount_in == 0:
return 0
# 預期價格（假設無限流動性）
expected_rate = reserve_out / reserve_in
# 實際價格
actual_rate = amount_out / amount_in
# 滑點（基點）
slippage_bps = int((expected_rate - actual_rate) / expected_rate * 10_000)
return max(0, slippage_bps)
def _update_pool_reserves(
self,
pool: LiquiditySource,
amount_in: int,
amount_out: int,
reserves: Dict[str, Dict]
):
"""更新池子剩餘流動性"""
reserves[pool.pool_id]['in'] += amount_in
reserves[pool.pool_id]['out'] -= amount_out2.3 線性規劃的嚴格求解
貪心演算法雖然快，但不保證全局最優。如果你想追求更好的效果，可以用線性規劃（Integer Linear Programming）。
# solver/ilp_solver.py
"""
ILP 嚴格求解器
使用 PuLP 庫求解整數線性規劃問題
"""
from pulp import *
from dataclasses import dataclass
from typing import List, Dict, Tuple, Optional
import numpy as np
@dataclass
class ILPIntent(Intent):
"""帶 LP 參數的交易意圖"""
min_fill_bps: int = 0 # 最小成交比例（基點）
@dataclass
class RouteOption:
"""可行路由選項"""
route_id: str
intent_id: str
pool_id: str
max_amount_in: int # 最大輸入
price: float # 執行價格（output/input）
gas_cost: int
class ILPSolver:
"""
ILP 嚴格求解器
數學模型：
最大化 Σ (price[i,j] * amount[i,j] - gas[i,j] * x[i,j])
約束：
1. 每個意圖至少成交 min_fill 比例
2. 每個池子的總流出 <= 池子儲備
3. x[i,j] 是 0-1 變量或整數變量
"""
def __init__(self, gas_price: int = 30_000_000_000):
self.gas_price = gas_price
self.gas_per_trade = 150_000
def solve(
self,
intents: List[ILPIntent],
route_options: List[RouteOption],
pool_capacities: Dict[str, int]
) -> Dict[str, List[Tuple[str, int]]]:
"""
求解並返回分配方案
返回：
{intent_id: [(pool_id, amount_in), ...], ...}
"""
# 創建 LP 問題
prob = LpProblem("MEV_Solver", LpMaximize)
# 決策變量
# x[i,j] = 從意圖 i 到路由 j 的交易量
variables = {}
for route in route_options:
var_name = f"x_{route.intent_id}_{route.route_id}"
# 類型：非負連續（如果是 split order，則用 LpInteger）
variables[(route.intent_id, route.route_id)] = LpVariable(
var_name,
lowBound=0,
upBound=route.max_amount_in,
cat='Continuous' # 可改為 'Integer' 離散化
)
# 目標函數：最大化總福利
welfare = []
for route in route_options:
# 凈福利 = 執行價值 - Gas 成本
net_value = (
route.price * variables[(route.intent_id, route.route_id)]
- route.gas_cost / route.max_amount_in * variables[(route.intent_id, route.route_id)]
)
welfare.append(net_value)
prob += lpSum(welfare)
# 約束 1：每個意圖的最小成交
for intent in intents:
intent_routes = [
r for r in route_options if r.intent_id == intent.id
]
total_amount = lpSum([
variables[(r.intent_id, r.route_id)]
for r in intent_routes
])
min_fill = intent.sell_amount * intent.min_fill_bps / 10_000
prob += total_amount >= min_fill, f"MinFill_{intent.id}"
# 約束 2：池子容量約束
for pool_id in pool_capacities:
pool_routes = [
r for r in route_options if r.pool_id == pool_id
]
total_outflow = lpSum([
variables[(r.intent_id, r.route_id)]
for r in pool_routes
])
prob += total_outflow <= pool_capacities[pool_id], f"PoolCap_{pool_id}"
# 約束 3：每個意圖不超過輸入上限
for route in route_options:
prob += (
variables[(route.intent_id, route.route_id)]
<= route.max_amount_in
), f"MaxAmount_{route.intent_id}_{route.route_id}"
# 求解
prob.solve(PULP_CBC_CMD(msg=0, timeLimit=5))
# 解析結果
allocations = {}
for intent in intents:
intent_routes = [
r for r in route_options if r.intent_id == intent.id
]
allocations[intent.id] = []
for route in intent_routes:
amount = value(variables[(route.intent_id, route.route_id)])
if amount and amount > 0:
allocations[intent.id].append(
(route.pool_id, int(amount))
)
return allocations三、完整的 Solver 節點實現
3.1 節點架構
一個能跑在生產環境的 Solver 節點，大致需要這些模組：
Solver 節點架構：
┌─────────────────────────────────────────────────────────────┐
│ Solver Node │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Intent │ │ Solver │ │ Settlement │ │
│ │ Collector │────▶│ Engine │────▶│ Executor │ │
│ │ │ │ │ │ │ │
│ │ - WebSocket │ │ - ILP/Heu │ │ - TX Sign │ │
│ │ - REST API │ │ - Routing │ │ - Gas Opt │ │
│ │ - GraphQL │ │ - Split │ │ - Submit │ │
│ └─────────────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │
│ ▼ │ │
│ ┌─────────────┐ │ │
│ │ Quote │ │ │
│ │ Server │ │ │
│ │ │ │ │
│ │ - REST API │ │ │
│ │ - WebSocket│ │ │
│ └─────────────┘ │ │
│ │ │
└───────────────────────────────────────────────────┼─────────┘
│
▼
┌─────────────┐
│ Blockchain │
│ Network │
│ │
│ - RPC │
│ - Flashbots │
└─────────────┘3.2 核心實現
# solver/node.py
"""
完整的 Solver 節點實現
功能：
1. 接收交易意圖
2. 計算最優執行方案
3. 投標到拍賣市場
4. 執行中標方案
"""
import asyncio
import json
import logging
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional, Any
from decimal import Decimal
from eth_typing import HexStr
from web3 import Web3
from web3.contract import Contract
from web3.exceptions import TransactionNotFound
import redis
from aiohttp import web
import uvloop
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class SolverConfig:
"""求解器配置"""
name: str
private_key: str
rpc_url: str
redis_url: str = "redis://localhost:6379"
# 拍賣參數
auction_interval_seconds: int = 5
submission_deadline_seconds: int = 3
min_solution_value_usd: float = 10.0
# 風險參數
max_gas_price_gwei: float = 100.0
max_slippage_bps: int = 500
max_single_trade_usd: float = 100_000.0
@dataclass
class SolvedBatch:
"""求解結果"""
batch_id: str
intents: List[Dict]
executions: List[Dict]
total_gas_cost_wei: int
total_value_usd: float
submission_deadline: float
class SolverNode:
"""
MEV Solver 節點主類
"""
def __init__(self, config: SolverConfig):
self.config = config
# Web3 初始化
self.w3 = Web3(Web3.HTTPProvider(config.rpc_url))
self.account = self.w3.eth.account.from_key(config.private_key)
logger.info(f"Solver wallet: {self.account.address}")
# Redis 用於狀態管理
self.redis = redis.from_url(config.redis_url)
# 求解引擎
self.ilp_solver = ILPSolver()
self.greedy_solver = SimpleGreedySolver()
# 當前批次狀態
self.current_batch: List[Dict] = []
self.batch_lock = asyncio.Lock()
# 運行狀態
self.running = False
async def start(self):
"""啟動 Solver 節點"""
self.running = True
# 啟動 HTTP API 服務
api_server = web.Server(self._handle_api_request)
runner = web.ServerRunner(api_server)
await runner.setup()
site = web.TCPSite(runner, '0.0.0.0', 8080)
await site.start()
logger.info("HTTP API server started on :8080")
# 啟動報價服務
quote_server = web.Server(self._handle_quote_request)
quote_runner = web.ServerRunner(quote_server)
await quote_runner.setup()
quote_site = web.TCPSite(quote_runner, '0.0.0.0', 8081)
await quote_site.start()
logger.info("Quote server started on :8081")
# 啟動批次處理循環
asyncio.create_task(self._batch_processing_loop())
# 啟動監控任務
asyncio.create_task(self._monitoring_loop())
logger.info("Solver node started successfully")
# 保持運行
while self.running:
await asyncio.sleep(1)
async def stop(self):
"""停止 Solver 節點"""
self.running = False
logger.info("Solver node stopped")
async def _batch_processing_loop(self):
"""
批次處理主循環
每個 interval：
1. 收集待處理的意圖
2. 運行求解器
3. 提交投標
"""
while self.running:
try:
await asyncio.sleep(self.config.auction_interval_seconds)
await self._process_batch()
except Exception as e:
logger.error(f"Batch processing error: {e}")
async def _process_batch(self):
"""處理單個批次"""
async with self.batch_lock:
# 獲取當前批次
batch_intents = await self._get_current_intents()
if len(batch_intents) < 1:
return
# 獲取市場數據
liquidity = await self._fetch_liquidity(batch_intents)
# 運行求解
if len(batch_intents) <= 10:
# 小批次：用 ILP 嚴格求解
solution = await self._solve_ilp(batch_intents, liquidity)
else:
# 大批次：用貪心求解
solution = await self._solve_greedy(batch_intents, liquidity)
if not solution:
logger.warning("No solution found")
return
# 構造投標
bid = self._construct_bid(solution)
# 提交投標
await self._submit_bid(bid)
# 清空批次
await self._clear_batch()
async def _solve_ilp(
self,
intents: List[Dict],
liquidity: Dict
) -> Optional[SolvedBatch]:
"""ILP 求解"""
try:
# 構造路由選項
route_options = self._generate_route_options(intents, liquidity)
if not route_options:
return None
# 求解
allocations = self.ilp_solver.solve(
intents,
route_options,
self._get_pool_capacities(liquidity)
)
# 構造執行方案
executions = self._build_executions(intents, allocations, liquidity)
# 計算成本和價值
total_gas = self._estimate_gas(executions)
total_value = self._calculate_total_value(executions)
return SolvedBatch(
batch_id=self._generate_batch_id(),
intents=intents,
executions=executions,
total_gas_cost_wei=total_gas * self._get_current_gas_price(),
total_value_usd=total_value,
submission_deadline=asyncio.get_event_loop().time() +
self.config.submission_deadline_seconds
)
except Exception as e:
logger.error(f"ILP solve error: {e}")
return None
async def _solve_greedy(
self,
intents: List[Dict],
liquidity: Dict
) -> Optional[SolvedBatch]:
"""貪心求解"""
try:
# 轉換格式
intent_objs = [
Intent(
id=i['id'],
sender=i['sender'],
sell_token=i['sellToken'],
buy_token=i['buyToken'],
sell_amount=int(i['sellAmount']),
max_slippage_bps=i.get('maxSlippageBps', 100)
)
for i in intents
]
# 求解
executions = self.greedy_solver.solve(intent_objs, liquidity)
if not executions:
return None
# 計算成本
total_gas = sum(e.gas_cost for e in executions)
return SolvedBatch(
batch_id=self._generate_batch_id(),
intents=intents,
executions=[asdict(e) for e in executions],
total_gas_cost_wei=total_gas,
total_value_usd=self._estimate_value_usd(executions),
submission_deadline=asyncio.get_event_loop().time() + 3
)
except Exception as e:
logger.error(f"Greedy solve error: {e}")
return None
async def _submit_bid(self, bid: Dict):
"""
提交投標到拍賣市場
實際實現中，這裡需要：
1. 簽名投標
2. 發送到拍賣合約或 Flashbots
"""
logger.info(f"Submitting bid: {json.dumps(bid, indent=2)}")
# 這裡是簡化實現
# 實際需要根據具體的拍賣機制（CoW, SOX, 等等）調整
# 存入 Redis 供監控
self.redis.setex(
f"bid:{bid['batchId']}",
3600,
json.dumps(bid)
)
async def _get_current_intents(self) -> List[Dict]:
"""從 Redis 獲取當前批次的意圖"""
intent_keys = self.redis.zrange(
"pending_intents",
0,
-1
)
intents = []
for key in intent_keys:
intent_data = self.redis.get(key)
if intent_data:
intents.append(json.loads(intent_data))
return intents
async def _fetch_liquidity(self, intents: List[Dict]) -> Dict:
"""
獲取市場流動性數據
實際實現中，這裡需要：
1. 查詢各 DEX 合約的池子狀態
2. 獲取價格預言機數據
3. 快取結果
"""
# 簡化實現：返回模擬數據
liquidity = {}
# 模擬 Uniswap V3 和 SushiSwap 的池子
for intent in intents:
pair = (intent['sellToken'], intent['buyToken'])
liquidity[pair] = [
{
'pool_id': 'uniswap_v3_3000',
'token_in': pair[0],
'token_out': pair[1],
'reserve_in': 10_000_000_000,
'reserve_out': 5_000_000_000_000,
'fee_bps': 30
},
{
'pool_id': 'sushiswap_30',
'token_in': pair[0],
'token_out': pair[1],
'reserve_in': 5_000_000_000,
'reserve_out': 2_500_000_000_000,
'fee_bps': 30
}
]
return liquidity
def _generate_route_options(
self,
intents: List[Dict],
liquidity: Dict
) -> List[RouteOption]:
"""生成可行路由選項"""
options = []
for intent in intents:
pair = (intent['sellToken'], intent['buyToken'])
pools = liquidity.get(pair, [])
for pool in pools:
options.append(RouteOption(
route_id=f"{intent['id']}_{pool['pool_id']}",
intent_id=intent['id'],
pool_id=pool['pool_id'],
max_amount_in=int(intent['sellAmount']),
price=self._calculate_price(pool),
gas_cost=150_000 * 30_000_000_000 # 估算
))
return options
def _calculate_price(self, pool: Dict) -> float:
"""計算池子價格"""
reserve_in = pool['reserve_in']
reserve_out = pool['reserve_out']
fee_bps = pool['fee_bps']
if reserve_in == 0:
return 0
# 考慮手續費的實際價格
effective_rate = reserve_out / (reserve_in * (1 - fee_bps / 10_000))
return effective_rate
def _get_pool_capacities(self, liquidity: Dict) -> Dict[str, int]:
"""獲取池子容量"""
capacities = {}
for pools in liquidity.values():
for pool in pools:
capacities[pool['pool_id']] = pool['reserve_out']
return capacities
def _estimate_gas(self, executions: List[Dict]) -> int:
"""估算 Gas 總量"""
return sum(150_000 for _ in executions) # 每筆約 150k Gas
def _get_current_gas_price(self) -> int:
"""獲取當前 Gas 價格"""
return self.w3.eth.gas_price
def _calculate_total_value(self, executions: List[Dict]) -> float:
"""計算總價值（USD）"""
# 簡化實現
return sum(
float(e.get('expected_out', 0)) / 1e18 * 2000 # 假設 ETH=$2000
for e in executions
)
def _estimate_value_usd(self, executions: List) -> float:
"""估算價值（貪心求解器）"""
total = 0
for e in executions:
if hasattr(e, 'expected_out'):
total += float(e.expected_out) / 1e18 * 2000
return total
def _construct_bid(self, solution: SolvedBatch) -> Dict:
"""構造投標"""
return {
'batchId': solution.batch_id,
'solverAddress': self.account.address,
'intents': solution.intents,
'executions': solution.executions,
'totalGasCostWei': solution.total_gas_cost_wei,
'totalValueUsd': solution.total_value_usd,
'timestamp': asyncio.get_event_loop().time(),
'deadline': solution.submission_deadline,
'signature': self._sign_bid(solution)
}
def _sign_bid(self, solution: SolvedBatch) -> str:
"""簽名投標"""
message_hash = self.w3.solidity_keccak(
['string', 'uint256', 'uint256'],
[solution.batch_id, solution.total_value_usd, solution.submission_deadline]
)
signed = self.account.sign_hash(message_hash)
return signed.signature.hex()
def _generate_batch_id(self) -> str:
"""生成批次 ID"""
import uuid
return str(uuid.uuid4())
async def _clear_batch(self):
"""清空批次"""
self.redis.delete("pending_intents")
async def _handle_api_request(self, request: web.Request) -> web.Response:
"""處理 API 請求（意圖提交）"""
if request.method == 'POST':
data = await request.json()
intent_id = self._generate_intent_id()
intent = {
'id': intent_id,
'sender': data.get('sender'),
'sellToken': data.get('sellToken'),
'buyToken': data.get('buyToken'),
'sellAmount': data.get('sellAmount'),
'maxSlippageBps': data.get('maxSlippageBps', 100),
'timestamp': asyncio.get_event_loop().time()
}
# 存入 Redis
self.redis.setex(
f"intent:{intent_id}",
300, # 5 分鐘過期
json.dumps(intent)
)
self.redis.zadd("pending_intents", {f"intent:{intent_id}": intent['timestamp']})
return web.json_response({
'status': 'accepted',
'intentId': intent_id
})
return web.json_response({'error': 'Method not allowed'}, status=405)
async def _handle_quote_request(self, request: web.Request) -> web.Response:
"""處理報價請求"""
params = request.query
sell_token = params.get('sellToken')
buy_token = params.get('buyToken')
amount = int(params.get('amount', 0))
# 計算報價
price = self._get_quote_price(sell_token, buy_token, amount)
return web.json_response({
'sellToken': sell_token,
'buyToken': buy_token,
'sellAmount': amount,
'buyAmount': int(amount * price),
'price': price,
'validUntil': asyncio.get_event_loop().time() + 10
})
def _get_quote_price(
self,
sell_token: str,
buy_token: str,
amount: int
) -> float:
"""獲取報價"""
# 簡化實現
return 0.999 # 假設 0.1% 滑點
def _generate_intent_id(self) -> str:
"""生成意圖 ID"""
import uuid
return f"intent_{uuid.uuid4().hex[:16]}"
async def _monitoring_loop(self):
"""監控循環"""
while self.running:
try:
# 監控錢包餘額
balance = self.w3.eth.get_balance(self.account.address)
logger.info(f"Wallet balance: {balance / 1e18:.4f} ETH")
# 監控 Redis 狀態
pending = self.redis.zcard("pending_intents")
logger.info(f"Pending intents: {pending}")
except Exception as e:
logger.error(f"Monitoring error: {e}")
await asyncio.sleep(30)
# 啟動腳本
async def main():
config = SolverConfig(
name="production-solver-01",
private_key=os.environ.get('SOLVER_PRIVATE_KEY'),
rpc_url=os.environ.get('ETH_RPC_URL'),
redis_url=os.environ.get('REDIS_URL', 'redis://localhost:6379')
)
node = SolverNode(config)
try:
await node.start()
except KeyboardInterrupt:
await node.stop()
if __name__ == '__main__':
uvloop.install()
asyncio.run(main())四、Solver Network 的經濟學分析
4.1 收益來源
一個運作良好的 Solver，收入來源大概有這幾塊：
Solver 收益分解：
┌────────────────────────────────────────────────────────────┐
│ 總收益 = 基礎收益 + 額外收益 │
├────────────────────────────────────────────────────────────┤
│ │
│ 基礎收益： │
│ ├── 執行費用（固定或按交易額百分比） │
│ ├── Gas 補貼（用戶支付的 Gas 部分） │
│ └── 報價利差（買入價和賣出價的差異） │
│ │
│ 額外收益： │
│ ├── MEV 回流（當拍賣機制整合 MEV-Boost） │
│ ├── 批量折扣（一次性結算多筆交易的 Gas 優化） │
│ └── 優先級費用（對時間敏感的用戶收取額外費用） │
│ │
└────────────────────────────────────────────────────────────┘4.2 成本結構
Solver 運營成本：
┌────────────────────────────────────────────────────────────┐
│ 月度運營成本估算 │
├────────────────────────────────────────────────────────────┤
│ │
│ 固定成本： │
│ ├── 雲端伺服器（4 核 16G）：~$200/月 │
│ ├── 專線網路（100Mbps）：~$150/月 │
│ ├── 區塊鏈 RPC（Infura/Alchemy）：~$100/月 │
│ └── 監控和報警：~$50/月 │
│ │
│ 可變成本： │
│ ├── Gas 費用：波動，通常 $1000-10000/月 │
│ ├── 失敗交易成本：估算為成功交易的 5% │
│ └── 研發人力：按需 │
│ │
│ 總計： │
│ └── 基礎：~$500/月 │
│ └── 含 Gas：~$2000-15000/月 │
│ │
└────────────────────────────────────────────────────────────┘4.3 盈利模型
盈虧平衡分析：
假設條件：
- 基礎成本：$500/月
- Gas 成本：$3000/月
- 平均每筆交易收益：$0.5
- 交易成功率：95%
計算：
每月需要處理的最小交易數量：
= ($500 + $3000) / ($0.5 * 0.95)
= 3500 / 0.475
≈ 7368 筆/月
≈ 245 筆/天
結論：
如果你的 Solver 能穩定每天處理 300+ 筆交易，
在合理的市場條件下，應該能達到盈虧平衡。
要盈利的話，日交易量需要達到 500+ 筆。五、實戰經驗總結
5.1 踩過的坑
做了幾年 MEV/Solver 相關的開發，說幾個親身踩過的坑：
坑 1：Redis 數據不一致
一開始我把待處理意圖全存 Redis，結果在高並發的時候，出現了 race condition。同一個意圖被處理了兩次，差點導致雙重執行。解決方案：用意圖的 hash 做 idempotency key，執行前先檢查。
坑 2：Gas 估算太樂觀
貪心演算法的 Gas 估算我用的是固定值 150k，結果實際交易的 Gas 波动很大，有的時候高達 300k。解決方案：根據交易複雜度動態估算。
坑 3：忽視區塊重組
有些時候區塊會被重組（reorg），這個在我的第一版代碼裡完全沒考慮。後來加了 confirmations 檢查才解決。
5.2 最佳實踐
Solver 最佳實踐清單：
1. 交易意圖管理
✓ 使用 idempotency key 防止重複執行
✓ 設置合理的超時機制
✓ 快取計算結果減少重複運算
2. 風險控制
✓ 設置單筆交易上限
✓ 設置日交易量上限
✓ 實時監控錢包餘額
3. Gas 優化
✓ 使用 EIP-1559 動態調整
✓ 考慮用 Flashbots 發送隱私交易
✓ 批量交易減少固定成本
4. 監控報警
✓ 設置 PnL 警報閾值
✓ 監控交易失敗率
✓ 監控網路延遲
5. 災難恢復
✓ 定期備份狀態到磁盤
✓ 實現 graceful shutdown
✓ 準備手動干預方案結語
寫到這裡突然有點感慨。MEV 這個領域，說小了是「交易機器人打架」，說大了其實是在重構整個區塊鏈的價值分配機制。
Solver Network 的出現，讓我看到了一絲希望——起碼在這個框架下，普通交易者不再是待宰的羔羊，而是可以被保護的對象。批量拍賣、協作求解、收益共享……這些機制雖然還不完美，但方向是對的。
如果你看完這篇文章也想動手試試，我的建議是：先從貪心求解器開始，用測試網跑通流程，再考慮上生產。千萬別一上來就搞 ILP，不然除錯會讓你懷疑人生。
有任何問題，歡迎找我聊聊。Web3 這個圈子雖然魚龍混雜，但願意踏踏實實做技術的人還是有很多的。
參考資源
- CoW Protocol 白皮書：https://docs.cow.fi
- MEV-Boost 文檔：https://docs.flashbots.net/flashbots-mev-boost
- ERC-7683 意圖標準草案：https://eips.ethereum.org/EIPS/eip-7683
- PuLP 線性規劃庫：https://coin-or.github.io/pulp/
本文最後更新於 2026 年 3 月 28 日
本文章僅供教育目的，涉及 MEV 策略的代碼請在充分理解風險後使用
相關文章
- 以太坊 MEV 生態系統深度技術分析：搜尋者-建構者-提議者三層市場結構與量化收益數據 — 本文深入分析以太坊 MEV（Miner Extractable Value，最大可提取價值）供應鏈的每個環節，詳細解讀搜尋者-建構者-提議者三層市場結構的運作機制，並透過實際攻擊案例和量化數據揭示 MEV 生態系統的真實面貌。
- 以太坊 MEV 實證研究：Transaction-Level 攻擊重構、量化收益分析與 PBS 未來演化 — 本文深入探討以太坊 MEV 生態系統的真實運作機制，透過 Transaction-Level 的實際攻擊案例重構、數學推導和量化數據分析，揭示搜尋者-建構者-提議者三層市場結構的實際運作邏輯。我們詳細分析三明治攻擊、套利機器人、結算機器的實際操作步驟，並探討 PBS (Proposer-Builder Separation) 的深層機制設計、加密 mempool 的密碼學原理、以及 MEV 對以太坊網路安全性的深層影響。
- 以太坊 Intent 帳戶與 DeFi 完整指南：從意圖架構到使用者體驗革新 — Intent（意圖）架構是以太坊生態系統在 2025-2026 年間最重要的技術創新之一。本文深入探討 Intent 架構的技術原理、以太坊上的實現方案（ERC-7683 標準）、主要 DeFi 應用場景（UniswapX、1inch Fusion+、Coinbase Base），以及這項技術帶來的用戶體驗革命。涵蓋求解器網路架構、跨鏈 Intent 執行、MEV 保護機制等完整技術實作。同時批判性地分析 Intent 架構的潛在風險，包括求解器市場集中、中心化趨勢、與去中心化理念的張力，以及簽名安全和隱私問題。
- 以太坊 AI 代理完整技術指南：自主經濟代理開發與實作 — 人工智慧代理與區塊鏈技術的結合正在開創區塊鏈應用的新範式。本文深入分析以太坊 AI 代理的技術架構、開發框架、實作範例與未來發展趨勢。涵蓋套利策略、借貸清算、收益優化、安全管理等完整技術實作。提供完整的 TypeScript/Solidity 程式碼範例，包括與 DeFi Llama、L2Beat、Dune Analytics API 的整合實作。同時深入探討 AI Agent 的批評性觀點，包括中心化風險、系統性風險、MEV 對區塊鏈生態的影響，以及 Vitalik 權力集中等爭議性議題。
- 以太坊 2026 年第一季度技術報告：完整數據分析與前瞻展望 — 本報告基於 DeFi Llama、L2Beat、Dune Analytics、Flashbots、beaconcha.in 等權威數據源，提供以太坊網路性能、DeFi 生態、Layer 2 發展、安全態勢等維度的深度技術分析。涵蓋網路 TVL 分佈、借貸協議健康因子、Layer 2 技術棧比較、MEV 市場規模、安全事件統計等關鍵指標。同時深入探討以太坊面臨的挑戰，包括驗證者客戶端集中化風險、能源消耗爭議、Vitalik Buterin 權力集中的治理爭議、MEV 道德困境、安全審計有效性等批評性議題，並提出對投資者、開發者、研究者和整個社區的具體行動建議。
延伸閱讀與來源
- 以太坊基金會生態系統頁面 官方認可的生態項目列表
- The Graph 去中心化索引協議
- Chainlink 文檔 預言機網路技術規格
這篇文章對您有幫助嗎？
請告訴我們如何改進：
0 人覺得有帮助
評論
發表評論
注意：由於這是靜態網站，您的評論將儲存在本地瀏覽器中，不會公開顯示。
目前尚無評論，成為第一個發表評論的人吧！