AI 驱动的 DEX 聚合器路由算法：最优交易路径发现与滑点预测的智能决策

AI 驱动的 DEX 聚合器路由算法：最优交易路径发现与滑点预测的智能决策

一、在流动性碎片化中寻找最优解

DeFi 生态经过数年的爆发式增长，流动性分布已经高度碎片化。同一条链上部署着数十个 DEX，每个 DEX 内部又分布着不同费率的池子。用户在完成一笔交易时，往往面临一个多跳路由问题：如何将手中资产以最低的成本兑换为目标资产？

传统 DEX 聚合器的路由算法大多基于图搜索——将代币对视为图的边，每条边的权重由流动性深度和费率决定，然后通过 Dijkstra 或变体的 Bellman-Ford 算法寻找最短路径。这套方案在简单场景下行之有效，但随着链上复杂性提升，纯图搜索暴露了两个关键缺陷：

第一，它无法处理"分裂路由"（Split Routing）的动态分配问题。一笔大额交易拆分成多路执行，每一路跨不同 DEX，最佳分配比例随市场状态实时变化，图搜索的离散枚举在大组合空间中计算成本过高。

第二，滑点预测精度不足。恒定乘积做市商（CPMM）的滑点公式在理想假设下成立，但实际上池子深度会因 MEV 攻击者抢跑、三明治攻击等因素在区块内剧烈波动，静态公式低估了实际滑点。

这两个问题恰好落在机器学习擅长解决的范畴——高维空间中的非凸优化与序列数据的动态预测。本文探讨如何用 AI 增强 DEX 聚合器的路由决策，重点覆盖路径发现与滑点预测两个核心环节。

二、从图搜索到强化学习的路由范式转变

2.1 问题建模

将 DEX 路由建模为一个马尔可夫决策过程（MDP）：

• 状态空间 S：当前所有池子的流动性分布、费率结构、Gas 基准价格、待成交金额
• 动作空间 A：选择下一跳的池子及分配金额的比例向量
• 奖励函数 R：输出代币数量 - Gas 成本 - 预估滑点损失
• 转移概率 P：状态转移由 AMM 数学公式（确定性部分）和 MEV 不确定性（随机部分）共同决定

2.2 为什么强化学习比图搜索更适合

图搜索假设边权重是静态的。但在 DeFi 中，一笔交易执行本身就是状态转移——你买入 ETH 会推高价格，影响后续跳的报价。这是典型的路由依赖问题（Route-Dependent Pricing），图搜索通过"先规划后执行"的模式难以捕捉。

强化学习的优势在于学习一种策略函数 π(a|s)，能够在高维状态空间中做在线推理。当模型训练充分后，单次推理只需一次前向传播（毫秒级），远快于枚举所有分拆组合。

2.3 模型架构选型

实践中常用的方案是 Actor-Critic 架构：

• Actor 网络：输入状态向量，输出各池子的分配权重。使用 Softmax 归一化，保证总和为 1。
• Critic 网络：评估当前状态 + 动作组合的期望回报，用于 Actor 训练时的梯度估计。

训练时使用模拟环境（Gymnasium 风格），从链上历史数据中采样状态快照，模拟交易执行。奖励使用实际链上交易结果（输出代币量 × 价格 - 输入代币价值 - Gas），避免奖励信号与真实目标不一致的问题。

2.4 滑点预测的序列模型

滑点的本质是订单对池子状态的冲击。传统公式假设池子状态在交易执行时不变，但实际上：

• 同一区块内可能有多个交易操作同一池子
• 搜索者（Searcher）可能在你的交易前后插入交易
• Flashbots 保护能减少但无法完全消除这类风险

我们对滑点的建模使用 Transformer 编码器，输入包括：

1. 过去 N 个区块中目标池子的交易序列（包括交易量、方向、Gas 价格）
2. 池子当前流动性
3. 待执行交易的金额和方向
4. 链上 MEV 活跃度指标（如 Builder 竞价数据）

输出为三档预测滑点（P50 / P90 / P99），路由引擎可根据用户的风险偏好（保守/均衡/激进）选择对应的滑点上限。

三、路由模型的工程实现

以下给出路由优化器的核心实现骨架，涵盖状态编码、模型推理和路径输出。

import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from dataclasses import dataclass from typing import List, Tuple, Dict # --- 数据结构定义 --- @dataclass class PoolState: """单个池子的快照""" address: str token0: str token1: str reserve0: float reserve1: float fee_bps: int # 费率，基点为单位 (30 = 0.3%) dex: str # 所属 DEX 名称 tvl_usd: float @dataclass class RouteRequest: """路由请求""" token_in: str token_out: str amount_in: float max_splits: int = 5 # 最大分拆路径数 risk_profile: str = "balanced" # conservative/balanced/aggressive # --- 滑点预测模型 --- class SlippagePredictor(nn.Module): """基于 Transformer 的滑点预测器 设计决策：使用轻量级 Transformer 而非 LSTM，因为交易序列中的 关键模式（如 MEV 攻击）往往具有全局依赖，自注意力机制能更好 地捕获跨区块的价量异常模式。 """ def __init__(self, d_model=128, n_heads=4, n_layers=3): super().__init__() self.input_proj = nn.Linear(12, d_model) encoder_layer = nn.TransformerEncoderLayer( d_model=d_model, nhead=n_heads, batch_first=True ) self.transformer = nn.TransformerEncoder(encoder_layer, n_layers) # 三档输出：P50/P90/P99 滑点预测 self.output_head = nn.Linear(d_model, 3) # 使用 Softplus 保证输出为正 self.activate = nn.Softplus() def forward(self, trade_history, pool_state, trade_params): """ Args: trade_history: (B, L, 12) 历史交易序列 pool_state: (B, 4) 当前池子状态 trade_params: (B, 3) 待执行交易参数 Returns: (B, 3) P50/P90/P99 滑点 bps """ # 拼接历史与当前信息作为上下文 pool_expanded = pool_state.unsqueeze(1).expand(-1, trade_history.size(1), -1) trade_expanded = trade_params.unsqueeze(1).expand(-1, trade_history.size(1), -1) x = torch.cat([trade_history, pool_expanded, trade_expanded], dim=-1) x = self.input_proj(x) x = self.transformer(x) # 取最后一帧的输出作为全局表示 global_repr = x[:, -1, :] return self.activate(self.output_head(global_repr)) # --- 路由 Actor 网络 --- class RouteActor(nn.Module): """路由决策 Actor 设计决策：输出层使用门控机制，先预测分拆路径数 K， 再为每条路径生成分配权重。这种两阶段输出比直接固定 维度输出更灵活，避免对未使用路径分配非零权重。 """ def __init__(self, state_dim: int, max_splits: int, hidden_dim=256): super().__init__() self.max_splits = max_splits self.encoder = nn.Sequential( nn.Linear(state_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), ) # 路径数预测门 self.split_gate = nn.Linear(hidden_dim, max_splits) # 每条路径的分配权重头（基于池子特征） self.weight_head = nn.Linear(hidden_dim, max_splits) def forward(self, state: torch.Tensor): """ Args: state: (B, state_dim) 环境状态 Returns: split_weights: (B, max_splits) 归一化后的分配权重 """ encoded = self.encoder(state) gate_logits = self.split_gate(encoded) gate_probs = F.softmax(gate_logits, dim=-1) weight_logits = self.weight_head(encoded) # 累积权重做 Mask：路径 i 的掩码为前 i 个 gate 概率之和 cumulative = torch.cumsum(gate_probs, dim=-1) mask = cumulative.unsqueeze(-1).expand(-1, self.max_splits) mask = (torch.arange(self.max_splits, device=mask.device) .unsqueeze(0) < mask).float() masked_weights = weight_logits * mask # 归一化：确保总权重为 1 split_weights = F.normalize(masked_weights, p=1, dim=-1) return split_weights # --- 路由引擎 --- class AIRouteEngine: """AI 驱动的路由引擎 整合滑点预测与路由决策，输出最优执行计划。 """ def __init__( self, actor: RouteActor, slippage_model: SlippagePredictor, risk_thresholds: Dict[str, int] = None ): self.actor = actor self.slippage_model = slippage_model self.risk_thresholds = risk_thresholds or { "conservative": 0, # 使用 P99 "balanced": 1, # 使用 P90 "aggressive": 2, # 使用 P50 } def find_best_route( self, pools: List[PoolState], request: RouteRequest ) -> List[Tuple[PoolState, float]]: """ 核心路由方法：为给定请求找到最优分配方案。 流程： 1. 对每个候选池子计算滑点预测 2. 编码全局状态 3. Actor 推理得到分配权重 4. 按权重排序，过滤零权重路径 5. 返回有序执行计划 """ state_vector = self._encode_state(pools, request) state_tensor = torch.tensor(state_vector).unsqueeze(0).float() with torch.no_grad(): split_weights = self.actor(state_tensor).squeeze(0).numpy() # 过滤权重 > 1% 的路径 routes = [] for i, (pool, weight) in enumerate(zip(pools, split_weights)): if weight > 0.01: # 计算该路径的预期滑点 slippage = self._predict_slippage_for_pool( pool, request ) routes.append((pool, weight, slippage)) # 按权重降序排列，权重大的先执行以减少后续路径的价格影响 routes.sort(key=lambda x: x[1], reverse=True) return [(pool, weight) for pool, weight, _ in routes] def _encode_state(self, pools, request) -> np.ndarray: """将离散池子状态编码为固定维度向量""" features = [] for pool in pools[:request.max_splits]: # 归一化特征向量 features.extend([ pool.reserve0 / 1e9, pool.reserve1 / 1e9, pool.fee_bps / 10000.0, pool.tvl_usd / 1e9, ]) # 补齐到 max_splits padding = (request.max_splits - len(pools)) * 4 features.extend([0.0] * max(0, padding)) return np.array(features, dtype=np.float32) def _predict_slippage_for_pool(self, pool, request): """为特定池子预测滑点""" risk_idx = self.risk_thresholds[request.risk_profile] # 简化示意：实际调用 slippage_model.forward return 0.0

四、边界与挑战

4.1 训练数据偏差

现有历史数据中成功的交易占多数，失败的或被抢跑的交易数据不完整。训练模型时容易过度乐观，低估极端市场下的滑点。解决方案包括：对训练数据进行对抗增强，引入人工构造的 MEV 攻击场景；实用离线策略评估（OPE）验证模型在下行风险下的表现。

4.2 推理延迟与链下推理的矛盾

RL 模型推理在 CPU 上约 10-50ms，对于高频场景可接受，但对于需要严格原子性保证的路由执行，链下推理-链上执行的间隙是核心脆弱点。一个可行的折中是：推理得到路由方案后，通过智能合约设置最小输出金额（minAmountOut）和区块号上限（deadline），将延迟风险转化为可量化的容忍参数。

4.3 跨链路由的外推能力

当前训练数据通常来自单链。跨链桥的引入会使状态空间爆炸，且桥的延迟不确定性很大。此处更务实的做法是先做跨链资产的标准价格修正，然后将跨链部分作为路由图中的特殊边处理。

4.4 对抗性攻击面

攻击者可能反过来利用 AI 模型的已知行为发起对抗性攻击。例如，通过制造特定交易模式诱导路由模型选择高滑点路径。防御策略包括模型集成（Ensemble）——同时运行 3-5 个独立训练的路由策略，取中位数方案，增加攻击者操纵的成本。

五、总结

DEX 聚合器的路由问题是 DeFi 中少有的"最适合 AI"的命题：高维组合空间、实时动态变化、非凸目标函数——这些恰好是传统图搜索的软肋，也是深度学习的强项。

但需要冷静看待的是，当前阶段 AI 路由更现实的定位是"增强"而非"替代"确定性算法。图搜索作为保底路径提供可审计性，AI 模型负责在复杂场景（大额交易、多跳路由、跨 DEX 分拆）中提供优化建议。两者组合使用的"混合路由"方案，是当前工程上最稳健的选择。

在接下来的迭代中，链上推理能力的成熟（如 zkML、TEE 推理）将是关键拐点。当路由模型能直接在链上执行推理时，前述的推理-执行延迟问题将被根本性地解决。