Liquidation Mining（清算机器人）

日期: 2026-08-08 方向: 量化 / 统计套利 / Alpha 阶段: Phase 2 - 统计套利与Alpha Research (Day 89-102) 标签: #量化策略 #DeFi #清算 #Aave #Compound #MEV

---

今日目标

---

一、理论与模型

1.1 清算机制基础

借贷协议（Aave、Compound、MakerDAO）允许用户超额抵押借款。当抵押品价值不足以覆盖借款时，清算者可介入：

• 替借款者偿还部分借款
• 获得抵押品（含折扣，"liquidation bonus"）

1.2 Health Factor（HF）

Aave 公式： $$ HF = \frac{\sum_i (\text{Collateral}_i \cdot P_i \cdot \text{LiquidationThreshold}_i)}{\sum_j (\text{Borrow}_j \cdot P_j)} $$

• HF > 1：安全
• HF < 1：可清算

LiquidationThreshold 因资产风险不同：

• ETH: 82.5%
• WBTC: 78%
• USDC: 87%
• 长尾资产: 50-65%

1.3 清算公式（Aave V3）

清算者：

1. 选择借款人 (HF < 1)
2. 选择 debt asset 偿还（最多 50% 总 debt，称 close factor）
3. 协议给 collateral asset + bonus 给清算者

清算金额： $$ \text{Repay} \le \min(50% \cdot \text{TotalDebt}, \text{LiquidatableDebt}) $$

收益 collateral： $$ \text{Collateral Received} = \frac{\text{Repay} \cdot P_{debt}}{P_{coll}} \cdot (1 + \text{LiquidationBonus}) $$

LiquidationBonus 通常 5-15%（资产风险越高 bonus 越高）。

1.4 经济利润

利润 = Collateral 价值 - Repay 价值 - 成本

$$ \text{Profit} = \text{Collateral Received} \cdot P_{coll} - \text{Repay} \cdot P_{debt} - \text{Gas} - \text{Slippage}_{liquidate \to repay} $$

简化： $$ \text{Profit Margin} \approx \text{Bonus} - \text{Gas%} - \text{Slippage} $$

例：bonus 10%，gas $50 on $10K liq → gas 0.5%，slippage 1% → profit 8.5%

1.5 MEV 与清算

清算是公开 mempool 的，MEV bot 监控并抢跑：

• 看到 oracle 价格更新使 HF < 1 → 立刻提交清算
• 用 high gas / Flashbots 私有 mempool 优先

竞争已极激烈：Ethereum mainnet 清算 95%+ 由 < 50 个 bot 完成。

1.6 Flashloan 清算

清算者无需自有资金：

1. Flashloan 借 debt asset（同区块还）
2. 清算获得 collateral
3. DEX swap collateral → debt asset
4. 还 flashloan + 留 profit

Aave/Balancer flashloan 费 0.05-0.09%。

1.7 清算事件触发因素

---

二、直觉与陷阱

陷阱 1：竞争已极激烈

主网清算 90%+ 被专业 MEV bot 占据。新人入场胜率 < 5%。 解法：

1. 关注 L2（Arbitrum/Optimism/Base），竞争小
2. 关注小协议（Morpho、Spark、Radiant）
3. 用更激进 gas 策略
4. 自建 oracle 价格 feed（领先公开）

陷阱 2：Stale price oracle

Chainlink 1% deviation threshold + 1h heartbeat。极端行情下 oracle 滞后市场。 陷阱：清算时按 oracle 价但 swap 时按 market 价 → 严重亏损。 解法：用同样 oracle 反向计算预期 swap output。

陷阱 3：Slippage 失败

清算获得抵押品 → swap 还 flashloan。如果 collateral 是长尾资产（如 ARB、SUSHI），DEX 滑点可能 5-20%，吞噬 bonus。 解法：

1. 限制可清算的 collateral 类型
2. 用 1inch / 0x 路由聚合
3. 大额拆分成小额执行

陷阱 4：50% close factor 误用

只能清算最多 50% debt。计算清算金额需要正确：

• Health factor 评估
• 选择最优 debt-collateral 对（最大 bonus）
• 用对的 close factor

陷阱 5：reentrancy 攻击向量

清算者合约调用 flashloan + DEX swap，复杂多调用链。 陷阱：合约被 reentrancy 攻击，资金被偷。 解法：用 OZ ReentrancyGuard、严格合约审计。

陷阱 6：gas spike 时不清算

极端行情下 gas 飙升 5-10x。 陷阱：清算 P&L 由 gas 决定，gas 飙时小额清算亏损。 解法：动态 gas 估计；min profit threshold（如 $50）才执行。

陷阱 7：Failed tx 累积成本

提交清算但被抢先 → tx fail，gas 仍然要付。 解法：用 Flashbots bundle，失败不付 gas（reverted bundle）。

---

三、代码实现

3.1 完整清算 Bot 框架

# liq_bot.py
"""
DeFi Liquidation Bot Framework
- Health factor monitoring (Aave V3)
- Profitability evaluation
- Flashloan-based execution
- Multi-chain support
"""

import os
import time
import asyncio
import requests
import numpy as np
import pandas as pd
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass


# Note: 此为框架代码，链上交互需 web3.py 和实际 ABI
# 实盘 bot 需 ethers.js / forge 等结合


# ----------------------------- 数据结构 -----------------------------
@dataclass
class UserPosition:
    user: str
    collateral_assets: Dict[str, float]  # {asset: amount}
    debt_assets: Dict[str, float]
    health_factor: float
    total_collateral_usd: float
    total_debt_usd: float


@dataclass
class LiquidationOpportunity:
    user: str
    debt_asset: str
    debt_to_repay: float  # in debt asset units
    collateral_asset: str
    collateral_to_seize: float
    bonus_pct: float
    estimated_profit_usd: float
    gas_cost_usd: float
    confidence: float  # 0-1


# ----------------------------- Aave V3 监控 -----------------------------
class AaveV3Monitor:
    """监控 Aave V3 上 HF < 1.05 的用户"""

    AAVE_SUBGRAPH = "https://api.thegraph.com/subgraphs/name/aave/protocol-v3"

    LIQUIDATION_BONUSES = {
        'WETH': 0.05, 'WBTC': 0.06, 'USDC': 0.05, 'USDT': 0.05,
        'DAI': 0.05, 'AAVE': 0.075, 'LINK': 0.07,
    }

    LIQUIDATION_THRESHOLDS = {
        'WETH': 0.825, 'WBTC': 0.78, 'USDC': 0.87, 'USDT': 0.86,
        'DAI': 0.87, 'AAVE': 0.74, 'LINK': 0.65,
    }

    @staticmethod
    def fetch_at_risk_users(min_debt_usd: float = 1000) -> List[UserPosition]:
        """从 subgraph 拉取 HF < 1.05 的用户"""
        # 简化：实际 query 复杂，需用真实 schema
        query = """
        {
          users(first: 100, where: {borrowedReservesCount_gt: 0}) {
            id
            healthFactor
            totalCollateralUSD
            totalDebtUSD
            reserves {
              currentATokenBalance
              currentTotalDebt
              reserve {
                symbol
              }
            }
          }
        }
        """
        # 模拟数据（实际跑要换成 requests.post）
        mock_users = []
        for i in range(5):
            mock_users.append(UserPosition(
                user=f"0x{i:040x}",
                collateral_assets={'WETH': 10},
                debt_assets={'USDC': 25_000},
                health_factor=0.95 + i * 0.02,
                total_collateral_usd=30_000,
                total_debt_usd=25_000,
            ))
        return [u for u in mock_users if u.health_factor < 1.0]


# ----------------------------- 价格 Oracle -----------------------------
class PriceOracle:
    """获取当前价格（Chainlink / DEX TWAP）"""

    @staticmethod
    def get_price(asset: str) -> float:
        # Mock prices
        prices = {
            'WETH': 3000, 'WBTC': 60000, 'USDC': 1.0, 'USDT': 1.0,
            'DAI': 1.0, 'AAVE': 100, 'LINK': 15,
        }
        return prices.get(asset, 0)


# ----------------------------- DEX 路由 -----------------------------
class DEXRouter:
    """模拟 DEX 路由（实际用 1inch / 0x API）"""

    @staticmethod
    def quote_swap(from_asset: str, to_asset: str, amount: float) -> Dict:
        """模拟 swap quote（含 slippage）"""
        from_price = PriceOracle.get_price(from_asset)
        to_price = PriceOracle.get_price(to_asset)
        if to_price == 0:
            return {'output': 0, 'slippage': 1.0}

        # 简化 slippage：USD 量 / 1M 的 sqrt
        usd_amount = amount * from_price
        slippage = 0.001 + 0.005 * np.sqrt(usd_amount / 1_000_000)
        slippage = min(slippage, 0.10)

        output = (amount * from_price / to_price) * (1 - slippage)
        return {'output': output, 'slippage': slippage,
                'effective_price': output * to_price / (amount * from_price)}


# ----------------------------- 利润计算 -----------------------------
def calculate_liquidation_profit(user: UserPosition,
                                   debt_asset: str,
                                   collateral_asset: str,
                                   gas_price_gwei: float = 30,
                                   gas_limit: int = 800_000,
                                   eth_price: float = 3000,
                                   flashloan_fee: float = 0.0005) -> Optional[LiquidationOpportunity]:
    if user.health_factor >= 1.0:
        return None
    if debt_asset not in user.debt_assets:
        return None
    if collateral_asset not in user.collateral_assets:
        return None

    debt_amount = user.debt_assets[debt_asset]
    coll_amount = user.collateral_assets[collateral_asset]
    debt_price = PriceOracle.get_price(debt_asset)
    coll_price = PriceOracle.get_price(collateral_asset)

    if debt_price == 0 or coll_price == 0:
        return None

    # 50% close factor
    max_repay = debt_amount * 0.5

    # Bonus
    bonus_pct = AaveV3Monitor.LIQUIDATION_BONUSES.get(collateral_asset, 0.05)

    # Collateral seized
    collateral_seized = (max_repay * debt_price / coll_price) * (1 + bonus_pct)
    if collateral_seized > coll_amount:
        # 受限于实际 collateral
        max_repay = (coll_amount * coll_price / debt_price) / (1 + bonus_pct)
        collateral_seized = coll_amount

    # Swap collateral 还 flashloan
    swap_quote = DEXRouter.quote_swap(collateral_asset, debt_asset, collateral_seized)
    debt_received_from_swap = swap_quote['output']

    # Profit (in debt asset units)
    profit_in_debt = debt_received_from_swap - max_repay * (1 + flashloan_fee)
    profit_usd = profit_in_debt * debt_price

    # Gas cost
    gas_cost_eth = gas_price_gwei * 1e-9 * gas_limit
    gas_cost_usd = gas_cost_eth * eth_price

    net_profit = profit_usd - gas_cost_usd
    confidence = 0.7 if swap_quote['slippage'] < 0.02 else 0.4

    if net_profit < 10:  # 至少 $10
        return None

    return LiquidationOpportunity(
        user=user.user,
        debt_asset=debt_asset,
        debt_to_repay=max_repay,
        collateral_asset=collateral_asset,
        collateral_to_seize=collateral_seized,
        bonus_pct=bonus_pct,
        estimated_profit_usd=net_profit,
        gas_cost_usd=gas_cost_usd,
        confidence=confidence,
    )


# ----------------------------- 主 Bot 引擎 -----------------------------
class LiquidationBot:
    def __init__(self, min_profit_usd: float = 50,
                  max_gas_gwei: float = 200):
        self.min_profit_usd = min_profit_usd
        self.max_gas_gwei = max_gas_gwei
        self.executed = []

    def scan_opportunities(self) -> List[LiquidationOpportunity]:
        users = AaveV3Monitor.fetch_at_risk_users(min_debt_usd=1000)
        opportunities = []
        for u in users:
            for debt_a in u.debt_assets:
                for coll_a in u.collateral_assets:
                    opp = calculate_liquidation_profit(
                        u, debt_a, coll_a,
                        gas_price_gwei=self._get_gas_price(),
                    )
                    if opp and opp.estimated_profit_usd >= self.min_profit_usd:
                        opportunities.append(opp)
        return sorted(opportunities, key=lambda x: x.estimated_profit_usd,
                       reverse=True)

    def _get_gas_price(self) -> float:
        try:
            r = requests.get('https://api.etherscan.io/api',
                              params={'module': 'gastracker',
                                      'action': 'gasoracle'})
            return float(r.json()['result']['ProposeGasPrice'])
        except Exception:
            return 30  # fallback

    def execute(self, opp: LiquidationOpportunity) -> Dict:
        """模拟执行（实盘要发链上 tx）"""
        gas = self._get_gas_price()
        if gas > self.max_gas_gwei:
            return {'success': False, 'reason': f'gas too high ({gas} gwei)'}

        # 模拟执行（实盘：构造 flashloan + liquidationCall）
        success = np.random.random() > 0.20  # 80% 成功（MEV 竞争失败 20%）
        if success:
            actual_profit = opp.estimated_profit_usd * np.random.uniform(0.7, 1.0)
            self.executed.append({
                'opp': opp, 'realized': actual_profit, 'time': time.time()
            })
            return {'success': True, 'profit_usd': actual_profit}
        return {'success': False, 'reason': 'lost to MEV competition'}

    def run(self, iterations: int = 10):
        for i in range(iterations):
            opps = self.scan_opportunities()
            if not opps:
                continue
            top = opps[0]
            print(f"  [{i}] Best opp: user {top.user[:8]}.. profit ${top.estimated_profit_usd:.0f}")
            result = self.execute(top)
            print(f"       {result}")
            time.sleep(1)


# ----------------------------- Flashloan 合约伪代码 -----------------------------
FLASHLOAN_LIQUIDATOR_SOLIDITY = """
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.10;

import "@aave/protocol-v3/contracts/flashloan/base/FlashLoanReceiverBase.sol";
import "@uniswap/v3-periphery/contracts/interfaces/ISwapRouter.sol";

contract FlashLiquidator is FlashLoanReceiverBase {
    ISwapRouter public swapRouter;
    address public owner;

    constructor(address provider, address router) FlashLoanReceiverBase(provider) {
        swapRouter = ISwapRouter(router);
        owner = msg.sender;
    }

    /// @notice 触发清算
    function liquidate(
        address user,
        address debtAsset,
        address collateralAsset,
        uint256 debtToRepay
    ) external onlyOwner {
        // Step 1: Flashloan debtAsset
        address[] memory assets = new address[](1);
        assets[0] = debtAsset;
        uint256[] memory amounts = new uint256[](1);
        amounts[0] = debtToRepay;
        uint256[] memory modes = new uint256[](1);
        modes[0] = 0;

        bytes memory params = abi.encode(user, collateralAsset);
        POOL.flashLoan(address(this), assets, amounts, modes, address(this), params, 0);
    }

    /// @notice Aave 调用，flashloan 到账后执行
    function executeOperation(
        address[] calldata assets,
        uint256[] calldata amounts,
        uint256[] calldata premiums,
        address initiator,
        bytes calldata params
    ) external override returns (bool) {
        require(msg.sender == address(POOL), "only pool");
        (address user, address collateralAsset) = abi.decode(params, (address, address));

        // Step 2: Approve and call liquidationCall
        IERC20(assets[0]).approve(address(POOL), amounts[0]);
        POOL.liquidationCall(collateralAsset, assets[0], user, amounts[0], false);

        // Step 3: Swap collateral back to debtAsset
        uint256 collBalance = IERC20(collateralAsset).balanceOf(address(this));
        IERC20(collateralAsset).approve(address(swapRouter), collBalance);

        ISwapRouter.ExactInputSingleParams memory swapParams = ISwapRouter.ExactInputSingleParams({
            tokenIn: collateralAsset,
            tokenOut: assets[0],
            fee: 3000,
            recipient: address(this),
            deadline: block.timestamp,
            amountIn: collBalance,
            amountOutMinimum: amounts[0] + premiums[0],
            sqrtPriceLimitX96: 0
        });
        swapRouter.exactInputSingle(swapParams);

        // Step 4: Approve repay
        IERC20(assets[0]).approve(address(POOL), amounts[0] + premiums[0]);

        // Step 5: Profit transfer to owner
        uint256 leftover = IERC20(assets[0]).balanceOf(address(this)) - (amounts[0] + premiums[0]);
        if (leftover > 0) {
            IERC20(assets[0]).transfer(owner, leftover);
        }

        return true;
    }

    modifier onlyOwner() {
        require(msg.sender == owner, "only owner");
        _;
    }
}
"""


# ----------------------------- 主程序 -----------------------------
if __name__ == '__main__':
    print("=" * 60)
    print("Liquidation Bot Demo (using mock data)")
    bot = LiquidationBot(min_profit_usd=50, max_gas_gwei=100)
    bot.run(iterations=5)

    if bot.executed:
        print("\nExecuted Liquidations:")
        for r in bot.executed:
            print(f"  User {r['opp'].user[:10]}.. profit ${r['realized']:.2f}")
        total = sum(r['realized'] for r in bot.executed)
        print(f"Total profit: ${total:.2f}")

---

四、真实数据/案例

案例 1：2020-03-12 黑色星期四的 MakerDAO 灾难

BTC 一日 -50%，ETH gas 飙升 100x：

• MakerDAO 价格 oracle 滞后
• $0 出价的清算获得 $4M+ ETH（无人竞争）
• 协议产生 $4M 坏账，被迫 mint MKR 拍卖

教训：

• 极端事件下 keeper 网络可能失败
• gas spike 让小型 keeper 退出
• 需要 fallback 机制（Aave 后续加 reserve factor）

案例 2：Aave 历史最大单笔清算

2022-11-22：用户 0x57e0...（外号 "AvaxOG"）借了 $5M USDC，抵押 $7M AVAX。 AVAX 单日 -20%，HF 跌破 1。 清算 bot 群体 30 笔交易完成全部清算，bonus 5% × $5M = $250K bonus。 重点：50% close factor 限制使得需要多笔交易。

案例 3：MEV bot 的清算业务

Wintermute / Flashbots 报告（2023）：

• Top 3 bot 占 70% 主网清算量
• 月利润 $0.5-3M（净 gas 后）
• 平均单笔毛利 $200-500，净利 $50-200
• 95% 失败 tx（被对手抢）
• 只 5% 成功，但累计盈利

案例 4：L2 清算机会（2024）

Arbitrum / Base 上 Radiant、Aave 清算：

• 竞争比主网少 10x
• 单笔利润相似但成功率更高（50%+）
• 但流动性较小，单笔规模有限

案例 5：Compound V3 清算改革

V3 改用单 collateral 模式（每市场只支持 1 borrow asset），简化清算：

• Liquidation 无 close factor 限制
• Bonus 固定（如 USDC market 的 6%）
• 但 collateral 仍多种（WETH/WBTC/COMP/UNI）

---

五、CEX vs DEX 策略差异

清算机会几乎全部是 DeFi 原生（CEX 自己强平，无第三方机会）。

CEX 的"等价"是：CEX 用户被强平时，强平订单进 order book，但这是 CEX 自身风控操作，外部参与者只能买强平产生的"低价"订单（短暂、需快速反应）。

---

六、风险管理

6.1 清算 bot 风险

6.2 执行优先级

def priority_score(opp: LiquidationOpportunity, gas_price: float) -> float:
    """打分排序，先做高分"""
    profit_per_gas = opp.estimated_profit_usd / (gas_price * 800_000 * 1e-9 * 3000)
    return profit_per_gas * opp.confidence

6.3 多链分散

30% 主网（高竞争但高量）
30% Arbitrum
20% Base
10% Optimism
10% Polygon / BSC

---

七、关键速查

关键参数

Bot 推荐配置

config = {
    'min_profit_usd': 50,
    'max_gas_gwei': 100,
    'min_confidence': 0.5,
    'use_flashbots': True,
    'allowed_collaterals': ['WETH', 'WBTC', 'USDC', 'USDT', 'DAI'],
    'min_swap_liquidity': 100_000,  # USD in DEX
}

Subgraph 查询

Aave V3 用户 HF 监控：

{
  users(where: { borrowedReservesCount_gt: 0 }, first: 1000) {
    id
    healthFactor
    totalCollateralUSD
    totalDebtUSD
  }
}

---

八、面试题

Q1：清算 bot 普通人还能做吗？

答：主网很难（< 50 个专业 bot 占 95%），但仍有机会：

1. L2 / 新链：竞争 < 主网 10x
2. 新协议：Spark/Euler V2 等推出后 6 月内
3. 小币种 / 长尾：bonus 高，bot 不愿处理（slippage 大）
4. 跨清算：发现协议 + 抢跑机会
5. 链下监控 + 链上执行优化：自建 oracle 数据流抢先 1-2s 普通人入门门槛：~$10K（gas + dev cost），月收入潜力 $500-2000（L2 + 新协议）。

Q2：解释 Aave 清算公式中的 50% close factor 设计意图？

答：

• 保护借款人：避免一次清算 100% 仓位（即使 HF 略低于 1）
• 避免级联强平：防止单次大额清算造成的滑点放大其他用户清算
• 保留可复活机会：用户可补充抵押或还款一半 debt 后避免 100% 损失
• 流动性管理：close factor 50% 让小笔清算更频繁，单笔小额避免 DEX 冲击
• trade-off：bot 需要分多笔才能完全清算大仓位，效率降

Q3：怎么集成 Flashbots 避免 MEV 损失？

答：

1. Bundle 提交：把 liquidationCall + swap 打包成 atomic bundle
2. 私有 mempool：tx 不进公开 mempool，对手看不到
3. Failure protection：bundle 失败时不上链（不付 gas）
4. Bidding：通过支付高额给 builder 优先包含
5. MEV-Share：分享部分 profit 与 user
6. 缺点：仅 ETH mainnet 支持完善，L2 替代方案有限（Optimism MEV-Auction、Arbitrum Sequencer 公平排序）

Q4：清算最大的不可控风险？

答：

1. 协议漏洞：你的 bot 调用协议时遇到 bug → 资金损失（虽然你只 flashloan，但 DEX swap 可能损失）
2. Oracle 操纵：低流动性资产被攻击者操纵价格触发"伪清算"，你的 bot 被骗执行
3. Reorg：成功的清算被 reorg 撤销
4. 外部依赖：DEX 路由失败、桥下线、subgraph 慢
5. Smart contract bug：你的 bot 合约被 reentrancy / 闪电贷攻击

• 缓解：小额起步、多次 audit、紧急 pause、保险

Q5：解释清算如何贡献协议安全？

答：

• 健康度维持：清算激励第三方监控并消除坏账风险
• 市场效率：bonus 是清算者的劳动报酬，市场化定价
• 去中心化：任何人可清算，无信任要求
• vs CeFi 中心化强平：CEX 强平由所内部执行，外部不可监督；DeFi 清算公开透明
• 激励对齐：bonus 大小匹配资产风险，长尾资产 bonus 高吸引清算者
• 价格发现：清算引发的 DEX swap 帮助市场快速 mark-down 长尾资产真实价值
• 风险：bonus 太低 → 没人清算，协议出现坏账（如 Compound 几次 OUSD/cUSD 事件）

---

明日预告

Day 100: DeFi 原生收益策略 — Looping、Delta-neutral、Pendle PT/YT 深度。今天的清算是 short-burst alpha，明天的收益策略是 sustained yield，两者互补。