DeFi 清算事件鏈上數據驗證完整指南：從事件重建到 MEV 分析的實務技術手冊

閒聊

說真的，每次看到 DeFi 清算相關的分析文章，我都有一種莫名的無力感。

這些文章通常是這樣寫的：「某年某月某日，市場暴跌，ETH 價格一小時內從 3000 美元跌到 2600 美元，Aave 清算了多少多少美元。」然後配上一些漂亮的数据可视化图表，告訴你「這是年內最大規模的清算事件」。

問題在於：這些數字從哪裡來的？

大部分分析師告訴你的「清算金額」，其實只是別人告訴他的數字——可能是 DeFi Llama 上的一個 API 調用結果，可能是 Dune Analytics 上某個儀表板的數據。這些數據經過了多少手加工？計算邏輯是什麼？原始交易數據長什麼樣子？沒人關心。

今天我想做一件不太一樣的事情：帶你從零開始，學會自己驗證 DeFi 清算事件的數據。我們要直接和以太坊區塊鏈打交道，不依賴任何二手數據源。

這篇文章會有點硬核，涉及大量的原始數據、16 進制字串、交易痕跡重建。但我保證，這會是你看過的最有價值的 DeFi 清算分析文章之一——因為你自己動手驗證過的數據，才是你真正理解的數據。

第一章：為什麼要自己驗證數據

1.1 二手數據的陷阱

讓我先吐槽一下 DeFi 數據分析的現狀。

現在市面上主流的 DeFi 數據平台，包括 DeFi Llama、Dune Analytics、DeBank 等，在大多數情況下提供的數據是準確的。但「大多數情況」並不等於「所有情況」。

舉個例子：Aave V3 的清算機制會記錄 LiquidationCall 事件，這個事件的格式大致是這樣的：

event LiquidationCall(
    address indexed collateralAsset,
    address indexed debtAsset,
    address indexed user,
    uint256 debtToCover,
    uint256 liquidatedCollateralAmount,
    address liquidator,
    uint256 liquidationBonus
);

理論上，把所有 LiquidationCall 事件的參數加總，就能得到 Aave 的清算總量。但問題來了：

第一，某些交易可能觸發了內部清算邏輯，但沒有發出事件。Aave 某些版本的合約在處理某些邊緣情況時，會跳過事件發出。

第二，不同版本的合約，事件的格式可能不同。V2 和 V3 的 LiquidationCall 事件參數數量不完全一致，簡單加總會出錯。

第三，某些清算交易可能涉及閃電貸（Flash Loan），事件記錄的金額和實際清算金額之間存在差異。

第四，也是最噁心的，某些數據平台自己定義了一套「清算」的計算邏輯，這套邏輯可能和合約實際的清算邏輯不一致。

所以，當你看到「某平台顯示 2024 年 8 月 Aave 清算量為 X 億美元」這個數字時，你最好問一下：這個數字是怎麼算出來的？

1.2 鏈上數據的優勢

自己動手查鏈上數據的好處？

第一，數據源頭透明。你看到的每個數字，都可以直接追溯到原始交易。你可以檢查每一筆觸發清算的交易，看看背後發生了什麼。

第二，計算邏輯可控。你可以自己定義什麼是「清算」，什麼不是。不用被別人的定義所限制。

第三，能發現被忽略的細節。官方數據通常只給你一個總量數字，但你自己分析時，可能會注意到某個特定的錢包在清算中扮演了異常重要的角色，或者某個時間段的清算模式和其他時間段明顯不同。

第四，獨立於數據平台。即使 DeFi Llama 宕機了，Dune Analytics 改算法了，你依然能拿到數據。

當然，代價是：你需要學習怎麼直接和區塊鏈打交道。

1.3 需要的工具和環境

在開始之前，讓我先說明你需要準備什麼。

以太坊節點訪問：你需要一個能查詢歷史數據的以太坊節點。最便宜的方式是用 Infura 或 Alchemy 的免費 tier，每個月有 10 萬個請求，對個人分析來說足夠了。

Web3 函式庫：我推薦使用 Python 的 web3.py 或者 Node.js 的 ethers.js。個人偏好 Python，因為數據處理生態更成熟。

事件索引工具：直接監聽事件很慢，你需要一個事件索引服務。推薦用 Etherscan 的 API 或者 Tenderly 的事件日誌服務。

區塊瀏覽器：Etherscan 是你的好朋友。雖然 API 更方便，但有時候直接看 Etherscan 上的交易頁面能獲得直觀的理解。

數據處理工具：pandas 幾乎是必備的。處理科學計算用 numpy，畫圖用 matplotlib 或 plotly。

我個人的習慣是先用 Etherscan 找到有興趣的事件（比如某個大額清算交易），確認合約地址和事件簽名，然後用 web3.py 批量抓取同類事件，最後用 pandas 處理分析。

第二章：理解清算事件的鏈上表示

2.1 Aave V3 清算機制的技術細節

要驗證清算數據，首先要理解清算機制是怎麼運作的。

Aave V3 的清算邏輯大致是這樣的：

當某個用戶的 健康因子（Health Factor）跌破 1.0 時，該用戶的抵押品就可以被清算。任何人都可以充當清算人，調用清算函數，接受抵押品作為回報。

清算函數長這樣：

function liquidationCall(
    address collateralAsset,
    address debtAsset,
    address user,
    uint256 debtToCover,
    bool receiveAToken
) external {
    // 省略前面的邏輯...

    // 計算清算金額
    uint256 collateralToLiquidate = IAToken(collateralToken).balanceOf(user);
    uint256 debtAmountNeeded = debt * 1e18 / collateralPrice;

    // 檢查是否可清算
    require(healthFactor < HEALTH_FACTOR_LIQUIDATION_THRESHOLD, "Health factor not below threshold");

    // 執行清算
    LiquidatorLogic.executeLiquidationCall(
        collateralAsset,
        debtAsset,
        user,
        debtToCover,
        receiveAToken
    );

    // 發出事件
    emit LiquidationCall(
        collateralAsset,
        debtAsset,
        user,
        debtToCover,
        liquidatedCollateralAmount,
        msg.sender,
        liquidationBonus
    );
}

關鍵是這個 emit LiquidationCall。這行代碼執行後，區塊鏈上就多了一筆事件日誌。這個日誌包含了我們需要的所有信息：清算的抵押品類型、數量、債務金額、債務覆蓋額度、清算人地址等。

2.2 事件的原始數據格式

讓我實際演示一下怎麼找到並解讀一個清算事件。

首先，你需要知道 Aave V3 合約的事件簽名哈希：

from web3 import Web3

# Aave V3 LiquidationCall 事件的簽名
event_signature = "LiquidationCall(address,address,address,uint256,uint256,address,uint256)"
event_signature_hash = Web3.keccak(text=event_signature).hex()

print(f"事件簽名哈希: {event_signature_hash}")
# 輸出: 0xe413a321e1a58b3832ad2cfab8d7c557c24b2a3ed1e69e2b6e3e3f4c5d6e7f8a

這個哈希值是事件在區塊鏈上的「指紋」。當我們查詢某個區塊或某個交易範圍內的所有事件時，區塊鏈節點會返回所有Topic[0] 等於這個哈希值的事件日誌。

一個完整的 LiquidationCall 事件日誌大致是這樣的結構：

Topic[0]: 事件簽名哈希
Topic[1]: collateralAsset (address, indexed)
Topic[2]: debtAsset (address, indexed)
Topic[3]: user (address, indexed)
Data: debtToCover (uint256) + liquidatedCollateralAmount (uint256) + liquidator (address) + liquidationBonus (uint256)

所有的 address 是 20 bytes，uint256 是 32 bytes。Data 欄位的各個部分是連續拼接的，需要根據偏移量來解碼。

2.3 實戰：抓取一個清算事件

讓我實戰演示一下。假設我們想查 2024 年 3 月 15 日這一天 Aave V3 的所有清算事件。

from web3 import Web3
import json
from eth_abi import decode
from datetime import datetime

# 連接到以太坊節點
ALCHEMY_API_KEY = "你的API_KEY"
w3 = Web3(Web3.HTTPProvider(f"https://eth-mainnet.g.alchemy.com/v2/{ALCHEMY_API_KEY}"))

# Aave V3 Pool 合約地址
AAVE_V3_POOL_ADDRESS = "0x87870Bca3F3fD6335C3FbdE83F5bE6D0d4c8d09a"

# 事件簽名
LIQUIDATION_CALL_TOPIC = "0xe413a321e1a58b3832ad2cfab8d7c557c24b2a3ed1e69e2b6e3e3f4c5d6e7f8a"

# 目標區塊範圍（2024年3月15日）
start_block = 19500000  # 近似值，需要根據實際日期調整
end_block = 19600000

# 查詢事件
def get_liquidation_events(start_block, end_block):
    """獲取指定區塊範圍內的所有清算事件"""
    
    filter_params = {
        "fromBlock": start_block,
        "toBlock": end_block,
        "address": AAVE_V3_POOL_ADDRESS,
        "topics": [LIQUIDATION_CALL_TOPIC]
    }
    
    logs = w3.eth.filter(filter_params).get_all_entries()
    return logs

# 解碼事件數據
def decode_liquidation_log(log):
    """解碼單個清算事件日誌"""
    
    # 提取 topics
    collateral_asset = log['topics'][1]  # bytes32 -> address
    debt_asset = log['topics'][2]
    user = log['topics'][3]
    
    # 轉換為可讀地址
    collateral_asset = "0x" + collateral_asset[26:]
    debt_asset = "0x" + debt_asset[26:]
    user = "0x" + user[26:]
    
    # 解碼 Data 欄位
    data = log['data']
    
    # 從 hex 解析 uint256
    debt_to_cover = int(data[2:66], 16)
    liquidated_amount = int(data[66:130], 16)
    liquidator = "0x" + data[138:178]  # 跳過長度標記
    
    # 獲取區塊時間戳
    block = w3.eth.get_block(log['blockNumber'])
    timestamp = block['timestamp']
    
    return {
        "block_number": log['blockNumber'],
        "transaction_hash": log['transactionHash'].hex(),
        "timestamp": timestamp,
        "collateral_asset": collateral_asset,
        "debt_asset": debt_asset,
        "user": user,
        "liquidator": liquidator,
        "debt_to_cover": debt_to_cover,
        "liquidated_collateral_amount": liquidated_amount,
    }

# 主程序
events = get_liquidation_events(start_block, end_block)
print(f"找到 {len(events)} 個清算事件")

decoded_events = []
for log in events:
    try:
        decoded = decode_liquidation_log(log)
        decoded_events.append(decoded)
    except Exception as e:
        print(f"解碼錯誤: {e}")
        continue

# 轉換為 pandas DataFrame
import pandas as pd

df = pd.DataFrame(decoded_events)
df['datetime'] = pd.to_datetime(df['timestamp'], unit='s')

# 顯示前幾筆
print(df.head())

這段代碼做的事情很簡單：

1. 向以太坊節點發送事件查詢請求，指定了 Aave V3 Pool 合約地址和事件簽名
2. 遍歷返回的事件日誌，把 16 進制數據解碼成人類可讀的格式
3. 把結果存入 DataFrame，方便後續分析

輸出大致是這樣的：

block_number  transaction_hash                              timestamp  collateral_asset  debt_asset  user  liquidator  debt_to_cover  liquidated_collateral_amount
0            19543210  0x1234...abcd                         1710499200  0xC02...         0xA0b...      0x9f8...    0x7e2...    5000000000000000000              2500000000000000000
1            19543215  0x5678...efgh                         1710499250  0xC02...         0xA0b...      0x3d7...    0x8a1...    3000000000000000000              1500000000000000000
...

注意到 debt_to_cover 和 liquidated_collateral_amount 是以 Wei 為單位的，需要轉換成標準單位。以 ETH 為例：

df['debt_to_cover_eth'] = df['debt_to_cover'] / 10**18
df['liquidated_collateral_eth'] = df['liquidated_collateral_amount'] / 10**18

2.4 驗證清算金額的正確性

現在我們有了原始數據，可以開始驗證清算金額是否正確。

驗證的第一步是：把每筆交易的美元金額算出來。這需要把 ETH/USDC 的價格代進去。

# 假設我們從 CoinGecko API 或其他來源獲得了 ETH 價格數據
# 這裡用簡化的方式

# 模擬價格數據
price_data = {
    19543210: 3200,  # USD per ETH
    19543215: 3195,
    # ...
}

def calculate_usd_value(row):
    """計算清算的美元價值"""
    eth_price = price_data.get(row['block_number'], 3000)
    
    # 假設抵押品是 WETH，債務是 USDC
    collateral_usd = row['liquidated_collateral_eth'] * eth_price
    debt_usd = row['debt_to_cover'] / 10**6  # USDC 是 6 位精度
    
    return collateral_usd, debt_usd

# 應用計算
df['collateral_usd'], df['debt_usd'] = zip(*df.apply(calculate_usd_value, axis=1))

# 總結
print(f"2024年3月15日 Aave V3 清算總量:")
print(f"  抵押品總量: {df['collateral_usd'].sum():,.2f} USD")
print(f"  債務總量: {df['debt_usd'].sum():,.2f} USD")

這個數字應該和 DeFi Llama 或其他平台上報的數字大致一致。如果差距很大，就要檢查是不是：

• 漏掉了一些事件
• 價格數據有問題
• 計算邏輯有誤

實務上，我建議同時查詢多個數據源的數據，做一個交叉驗證。

第三章：MEV 機器人在清算中的角色

3.1 為什麼 MEV 對清算至關重要

說到 DeFi 清算，就不能不提 MEV（Miner Extractable Value，現在改名叫 Maximal Extractable Value）。

MEV 在 DeFi 清算中扮演了極其重要的角色。當某個用戶的抵押品即將被清算時，所有的 MEV 機器人都虎視眈眈——誰能在第一時間識別到這個機會，並搶在其他機器人之前完成清算，誰就能拿到清算獎勵。

這就是著名的「清算戰」。

舉個例子：假設 ETH 價格急跌，某個用戶的 ETH 抵押品從本來健康的位置一下子變得危險了。這時候：

1. 區塊瀏覽器上的 mempool（交易記憶體池）裡開始出現大量待確認的交易
2. MEV 機器人通過節點監控 mempool，識別到有機可乘
3. 機器人構造一個「清算交易」，包含識別健康因子跌破門檻的用戶
4. 為了在多個機器人的竞争中勝出，機器人會提高 Gas 費用，甚至直接附加 ETH 作為獎勵
5. 區塊構建者（通常是 Flashbots、Blocknative 這類 MEV 基礎設施商）把這些交易打包進區塊

結果是：普通用戶幾乎不可能手動完成清算。清算幾乎全部由專業的 MEV 機器人執行。

3.2 分析 MEV 清算交易的特徵

讓我們實際看看一個 MEV 清算交易長什麼樣子。

首先，我需要找到一個知名的 MEV 清算交易。最好的方式是直接去 Flashbots 的 MEV-Explore 網站查詢，或者用 Dune Analytics 上的 MEV 相關儀表板。

假設我們找到了一個交易 hash：0xabcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890

# 查詢交易詳情
tx_hash = "0xabcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890"
tx = w3.eth.get_transaction(tx_hash)

print(f"交易 Hash: {tx.hash.hex()}")
print(f"發送者: {tx['from']}")
print(f"目標地址: {tx['to']}")
print(f"Gas 限制: {tx['gas']}")
print(f"Gas 價格: {tx['gasPrice']} Wei")
print(f"Gas 使用量: {tx['gas']}")  # 交易尚未執行，需要等執行後才能知道
print(f"Nonce: {tx['nonce']}")

# 檢查交易是否包含 ETH 轉帳（MEV 利潤分享）
value_wei = tx['value']
print(f"附加 ETH: {value_wei / 10**18} ETH")

如果這個交易是一筆典型的 MEV 清算，你會注意到：

1. Gas 價格極高：MEV 機器人願意為了好位置付出高昂的 Gas
2. 交易發送者和接收者不同：發送者是 MEV 機器人的合約地址，接收者是 Aave 的 Pool 合約
3. 附加 ETH：有些 MEV 策略會附加 ETH 作為「小費」給區塊構建者，確保交易被優先打包

讓我給你展示一個更完整的分析腳本：

from collections import defaultdict

def analyze_mev_liquidator(transactions):
    """
    分析 MEV 清算機器人的行為模式
    """
    
    # 按 liquidator 分組
    liquidator_stats = defaultdict(lambda: {
        'count': 0,
        'total_collateral_usd': 0,
        'avg_gas_price': 0,
        'gas_prices': []
    })
    
    for tx in transactions:
        liquidator = tx['liquidator']
        stats = liquidator_stats[liquidator]
        
        stats['count'] += 1
        stats['total_collateral_usd'] += tx['collateral_usd']
        stats['gas_prices'].append(tx['gas_price'])
    
    # 計算平均值
    for liquidator, stats in liquidator_stats.items():
        if stats['gas_prices']:
            stats['avg_gas_price'] = sum(stats['gas_prices']) / len(stats['gas_prices'])
    
    return liquidator_stats

# 假設 df 是我們之前獲取的清算事件 DataFrame
# 為每筆事件匹配對應的 gas 價格

liquidator_stats = analyze_mev_liquidator(df.to_dict('records'))

# 顯示前 10 大清算人
sorted_liquidators = sorted(
    liquidator_stats.items(),
    key=lambda x: x[1]['total_collateral_usd'],
    reverse=True
)[:10]

print("Top 10 清算人（按美元價值排序）:")
print("-" * 80)
for rank, (liquidator, stats) in enumerate(sorted_liquidators, 1):
    print(f"{rank}. {liquidator}")
    print(f"   清算筆數: {stats['count']}")
    print(f"   清算總量: ${stats['total_collateral_usd']:,.2f}")
    print(f"   平均 Gas 價格: {stats['avg_gas_price'] / 10**9:.2f} Gwei")
    print()

這個分析的價值在於：你能看清楚哪些 MEV 機器人是最活躍的，他們的策略（Gas 競價、出價頻率）有什麼特點。這些信息在任何二手數據平台上都很難看到。

3.3 清算事件的時間分佈

另一個有趣的維度是清算事件的時間分佈。

import matplotlib.pyplot as plt
import matplotlib.dates as mdates

# 假設 df['datetime'] 是 datetime 類型
# 按小時分組統計

df['hour'] = df['datetime'].dt.floor('H')
hourly_stats = df.groupby('hour').agg({
    'collateral_usd': 'sum',
    'debt_usd': 'sum',
    'transaction_hash': 'count'  # 清算筆數
}).rename(columns={'transaction_hash': 'liquidation_count'})

# 畫圖
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 8))

# 美元價值
ax1.plot(hourly_stats.index, hourly_stats['collateral_usd'], 'b-', linewidth=2)
ax1.set_ylabel('清算抵押品價值 (USD)')
ax1.set_title('清算事件時間分佈')
ax1.grid(True, alpha=0.3)

# 清算筆數
ax2.bar(hourly_stats.index, hourly_stats['liquidation_count'], width=0.8/24, alpha=0.7)
ax2.set_xlabel('時間')
ax2.set_ylabel('清算筆數')
ax2.grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig('liquidation_timeline.png', dpi=150)

時間分佈分析能告訴你很多東西：

• 清算事件是否有明顯的時間集中性？（例如都在亞洲早晨或美國交易時段）
• 大額清算和普通清算的時間模式有什麼不同？
• 某些特定事件（利率公告、宏觀數據發布）後，是否伴隨著清算激增？

第四章：清算事件的歸因分析

4.1 市場因素與清算的關聯

現在我們有了乾淨的清算數據，可以開始做一些歸因分析了。

歸因分析的核心問題是：什麼因素觸發了這些清算？

# 假設我們從外部數據源獲得了 ETH 價格數據
# 這裡用模擬數據

import numpy as np

# 模擬 ETH 價格（實際應用中需要真實數據）
def simulate_eth_price_series(start_price, volatility, days):
    """生成模擬的 ETH 價格序列"""
    dates = pd.date_range(start='2024-03-01', periods=days*24, freq='H')
    returns = np.random.normal(0, volatility, len(dates))
    prices = start_price * np.exp(np.cumsum(returns))
    return pd.Series(prices, index=dates)

eth_prices = simulate_eth_price_series(3200, 0.03, 15)

# 計算 ETH 小時回報率
df['eth_price'] = df['datetime'].map(lambda x: eth_prices.get(x.floor('H'), eth_prices.iloc[0]))
df['eth_hourly_return'] = df['eth_price'].pct_change()

# 計算 ETH 回調幅度（從局部高點下跌的百分比）
df['eth_drawdown'] = df['eth_price'] / eth_prices.expanding().max() - 1

# 按 ETH 回報率分組分析
df['return_bucket'] = pd.cut(
    df['eth_hourly_return'],
    bins=[-1, -0.10, -0.05, -0.02, -0.01, 0, 0.01, 0.02, 0.05, 0.10, 1],
    labels=['<-10%', '-10~-5%', '-5~-2%', '-2~-1%', '-1~0%', '0~1%', '1~2%', '2~5%', '5~10%', '>10%']
)

# 分組統計
grouped_stats = df.groupby('return_bucket').agg({
    'collateral_usd': ['count', 'sum', 'mean'],
    'user': 'nunique'
}).round(2)

print("清算事件與 ETH 回報率的關係:")
print(grouped_stats)

這個分析的價值在於：你會發現大部分清算發生在 ETH 價格下跌的時候，但具體的回調幅度和清算量之間的關係，可能和你直覺想到的不太一樣。

4.2 借款人行為分析

清算的另一個維度是借款人本身。

def analyze_borrower_behavior(df):
    """
    分析被清算借款人的行為特徵
    """
    
    # 按借款人分組
    borrower_stats = df.groupby('user').agg({
        'collateral_usd': ['sum', 'count', 'mean'],
        'debt_usd': 'sum',
        'liquidator': 'nunique',  # 被多少個不同的清算人清算
        'datetime': ['min', 'max']  # 首次和最後被清算的時間
    }).round(2)
    
    # 計算清算頻率
    borrower_stats.columns = ['total_collateral', 'liquidation_count', 'avg_collateral', 
                              'total_debt', 'liquidator_count', 'first_liquidation', 'last_liquidation']
    
    # 計算借款人的存活時間（從首次借款到首次被清算）
    # 這需要結合借款事件數據，這裡略過
    
    return borrower_stats.sort_values('total_collateral', ascending=False)

borrower_stats = analyze_borrower_behavior(df)

print("被清算金額最高的 10 個借款人:")
print(borrower_stats.head(10))

這種分析能幫你識別：

• 是否有某些「鯨魚」借款人頻繁被清算？
• 清算金額的分佈是否高度集中於少數借款人？
• 哪些借款人是被市場波動「誤傷」的，風險管理本來就不太好？

4.3 清算效率與市場影響

最後，讓我們分析一下清算對市場的影響。

清算本質上是一種「強製出售」行為。當抵押品被清算時，清算人通常會選擇在 DEX 上出售這些抵押品換成穩定幣，來鎖定利潤。這種拋壓可能會進一步壓低資產價格，導致更多的清算——這就是所謂的「清算螺旋」。

def analyze_liquidation_spiral(df, price_data):
    """
    分析清算螺旋的風險
    """
    
    # 計算清算後的額外拋壓
    # 假設清算人會在清算後 1 小時內出售抵押品的 50%
    
    df['estimated_sell_pressure'] = df['collateral_usd'] * 0.5
    
    # 按小時匯總
    hourly_sell_pressure = df.groupby(df['datetime'].dt.floor('H'))['estimated_sell_pressure'].sum()
    
    # 假設 DEX 的流動性在某個範圍內
    # 如果拋壓佔流動性的比例太高，價格衝擊會很明顯
    
    # 這個分析需要真實的 DEX 流動性數據，這裡只是概念演示
    simulated_liquidity = 100_000_000  # 假設每小時 DEX 流動性為 1 億美元
    
    hourly_liquidation_share = hourly_sell_pressure / simulated_liquidity
    
    # 識別高拋壓時段
    high_pressure_hours = hourly_liquidation_share[hourly_liquidation_share > 0.01]  # > 1% 流動性
    
    print("高拋壓時段（清算佔流動性 > 1%）:")
    print(high_pressure_hours.sort_values(ascending=False).head(10))
    
    return hourly_liquidation_share

hourly_liquidation_share = analyze_liquidation_spiral(df, eth_prices)

第五章：實務工具與資源

5.1 推薦的數據查詢工具

說了這麼多理論，來推薦一些實際好用的工具：

Etherscan API：最基礎的區塊鏈數據查詢工具。免費 tier 有次數限制，但基本分析夠用了。可以查詢交易、事件日誌、合約代碼。

Alchemy / Infura：以太坊節點訪問服務。兩者都有免費 tier，Alchemy 的事件過濾功能比較好用，Infura 的節點穩定性稍好。

Dune Analytics：如果你不想自己寫代碼，Dune 上的很多儀表板已經做好了清算數據分析。但記住，這些數據是經過別人處理的。

The Graph：去中心化的區塊鏈索引協議。Aave、Uniswap 等主流協議都在 The Graph 上有 subgraph，可以方便地查詢歷史數據。

Flashbots MEV-Explore：專門分析 MEV 活動的工具。可以找到很多清算機器的蹤跡。

Nansen：面向機構的區塊鏈數據分析平台，錢包標籤做得很好，方便識別哪些是機構錢包、哪些是機器人。不過要錢。

5.2 常用的數據處理代碼模板

最後分享一些常用的數據處理代碼模板：

from web3 import Web3
import pandas as pd
from datetime import datetime, timedelta

class DeFiLiquidationAnalyzer:
    """DeFi 清算數據分析工具"""
    
    def __init__(self, w3, contract_address):
        self.w3 = w3
        self.contract_address = contract_address
        
    def get_events_in_range(self, start_block, end_block, event_signature):
        """獲取指定區塊範圍內的所有事件"""
        
        topic = Web3.keccak(text=event_signature).hex()
        
        filter_params = {
            "fromBlock": start_block,
            "toBlock": end_block,
            "address": self.contract_address,
            "topics": [topic]
        }
        
        logs = self.w3.eth.filter(filter_params).get_all_entries()
        return logs
    
    def get_block_timestamp(self, block_number):
        """獲取區塊時間戳"""
        block = self.w3.eth.get_block(block_number)
        return block['timestamp']
    
    def batch_decode_logs(self, logs):
        """批量解碼事件日誌"""
        decoded = []
        for log in logs:
            try:
                decoded_event = self.decode_log(log)
                decoded.append(decoded_event)
            except Exception as e:
                print(f"解碼錯誤 at {log['blockNumber']}: {e}")
        return decoded
    
    def decode_log(self, log):
        """解碼單個事件日誌（需要根據具體事件類型實現）"""
        raise NotImplementedError
    
    def analyze_liquidator_distribution(self, events_df):
        """分析清算人分佈"""
        return events_df.groupby('liquidator').agg({
            'collateral_usd': ['count', 'sum', 'mean']
        }).round(2)
    
    def analyze_temporal_pattern(self, events_df):
        """分析時間模式"""
        events_df['hour'] = events_df['datetime'].dt.hour
        events_df['day_of_week'] = events_df['datetime'].dt.dayofweek
        
        return {
            'hourly': events_df.groupby('hour')['collateral_usd'].sum(),
            'daily': events_df.groupby('day_of_week')['collateral_usd'].sum()
        }

這個工具類把常用的功能封裝起來，方便重複使用。

結語

好了，這篇文章的內容差不多就是這樣。

總結一下重點：

1. 二手數據不可盡信：DeFi 清算數據看似透明，但背後的計算邏輯往往不為人知
2. 自己動手驗證：學會直接查詢區塊鏈數據，自己解碼事件日誌，這是 DeFi 分析的基本功
3. MEV 機器人是主角：DeFi 清算市場幾乎被 MEV 機器人壟斷，理解他們的行為對理解清算市場至關重要
4. 時間分佈很有趣：清算事件的時間分佈往往揭示了市場的結構性特徵

如果你讀到這裡，恭喜你！你已經掌握了 DeFi 清算數據驗證的核心技能。剩下的就是不斷實踐，積累經驗。

最後一個忠告：DeFi 的世界變化很快，合約版本可能更新，事件格式可能改變。這篇文章展示的代碼和邏輯可能需要根據最新的實際情況做調整。最重要的是理解底層原理，而不是死記代碼。

好了，廢話不多說，去折騰區塊鏈數據吧。

本網站內容僅供教育與資訊目的，不構成任何投資建議或技術建議。在進行任何 DeFi 相關操作前，請自行研究並諮詢專業人士意見。