DeFi 清算機制實時模擬器教學：風險參數回測與歷史情境重現完整指南

摘要

去中心化金融（DeFi）借貸協議的清算機制是維持系統健康的關鍵機制。2021 年 5 月 19 日、2022 年 11 月 FTX 崩潰等歷史事件揭示了極端市場條件下清算機制的重要性與潛在風險。本文提供完整的 DeFi 清算機制實時模擬器教學，包含 Python 程式碼範例、風險參數回測框架、以及歷史情境重現系統。讀者可透過本指南建立自己的量化模型，深入理解清算觸發條件、健康因子計算、以及不同市場參數下的清算閾值優化策略。

1. DeFi 清算機制基礎

1.1 清算的核心概念

在 DeFi 借貸協議中，清算（Liquidation）是一種自動執行機制，用於確保借款人的抵押品價值始終足以覆蓋其借款金額。當抵押品價值下跌或借款金額上升（因利息累積）導致健康因子低於清算閾值時，任何人都可以作為「清算人」執行清算。

關鍵術語定義：

清算觸發條件：Health Factor < Liquidation Threshold
健康因子計算：HF = (抵押品價值 × 清算閾值) / 借款總價值

清算 penalty：借款人被處罰的金額，通常為借款金額的 5-10%
清算獎勵：清算人可獲得的獎勵，通常為抵押品的 5-10%

1.2 主流借貸協議清算參數對比

2. 清算模擬器架構設計

2.1 系統架構

┌─────────────────────────────────────────────────────────────┐
│                   DeFi 清算模擬器架構                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐ │
│  │  數據源層     │───▶│  引擎層       │───▶│  策略層      │ │
│  │              │    │              │    │              │ │
│  │ - 鏈上數據    │    │ - 健康因子    │    │ - 清算策略   │ │
│  │ - 價格餽給   │    │ - 清算計算   │    │ - 風險管理   │ │
│  │ - 歷史事件   │    │ - 模擬引擎   │    │ - 回測框架   │ │
│  └──────────────┘    └──────────────┘    └──────────────┘ │
│         │                   │                   │           │
│         ▼                   ▼                   ▼           │
│  ┌──────────────────────────────────────────────────────┐ │
│  │                      輸出層                           │ │
│  │  - 即時儀表板  - 回測報告  - 風險警示  - 歷史重現     │ │
│  └──────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

2.2 核心資料結構

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from decimal import Decimal
from enum import Enum
from datetime import datetime
import numpy as np
import pandas as pd

@dataclass
class PriceData:
    """價格數據結構"""
    timestamp: datetime
    symbol: str
    price: Decimal
    source: str  # 'binance', 'coinbase', 'uniswap', 'chainlink'
    
    def to_float(self) -> float:
        return float(self.price)


@dataclass
class TokenConfig:
    """代幣配置"""
    symbol: str
    address: str
    decimals: int
    oracle_address: str
    volatility_30d: float = 0.0  # 30 天波動率
    liquidity_score: float = 0.0  # 流動性評分 (0-1)
    is_collateral: bool = True
    liquidation_threshold: float = 0.80  # 清算閾值
    collateral_factor: float = 0.80  # 抵押因子
    interest_rate_model: str = "Aave V3"
    
    def __post_init__(self):
        """驗證配置參數"""
        if not 0 < self.liquidation_threshold <= 1:
            raise ValueError(f"Liquidation threshold must be in (0, 1], got {self.liquidation_threshold}")
        if not 0 < self.collateral_factor <= 1:
            raise ValueError(f"Collateral factor must be in (0, 1], got {self.collateral_factor}")


@dataclass
class Position:
    """借款頭寸"""
    owner: str
    collateral_assets: Dict[str, Decimal] = field(default_factory=dict)  # 抵押資產
    borrow_assets: Dict[str, Decimal] = field(default_factory=dict)  # 借款資產
    creation_timestamp: datetime = field(default_factory=datetime.now)
    last_update_timestamp: datetime = field(default_factory=datetime.now)
    
    def get_collateral_value(
        self, 
        prices: Dict[str, Decimal],
        token_configs: Dict[str, TokenConfig]
    ) -> Decimal:
        """計算抵押品總價值（USD）"""
        total = Decimal('0')
        for symbol, amount in self.collateral_assets.items():
            if symbol in prices and symbol in token_configs:
                price = prices[symbol]
                config = token_configs[symbol]
                total += amount * price * config.collateral_factor
        return total
    
    def get_borrow_value(
        self,
        prices: Dict[str, Decimal],
        interest_accrued: Dict[str, Decimal] = None
    ) -> Decimal:
        """計算借款總價值（USD，含利息）"""
        total = Decimal('0')
        for symbol, amount in self.borrow_assets.items():
            if symbol in prices:
                principal = amount
                interest = interest_accrued.get(symbol, Decimal('0')) if interest_accrued else Decimal('0')
                total += (principal + interest) * prices[symbol]
        return total
    
    def calculate_health_factor(
        self,
        prices: Dict[str, Decimal],
        token_configs: Dict[str, TokenConfig],
        interest_accrued: Dict[str, Decimal] = None
    ) -> Optional[Decimal]:
        """
        計算健康因子
        
        HF = (抵押品價值 × 清算閾值) / 借款總價值
        
        Returns:
            Health Factor，若借款為 0 返回 None（視為無限大）
        """
        collateral_value = self.get_collateral_value(prices, token_configs)
        borrow_value = self.get_borrow_value(prices, interest_accrued)
        
        if borrow_value == 0:
            return None  # 無借款，視為無限大
        
        return (collateral_value * Decimal('1')) / borrow_value  # 清算閾值已內含


@dataclass
class LiquidationEvent:
    """清算事件"""
    timestamp: datetime
    position_owner: str
    liquidated_collateral: Dict[str, Decimal]
    repaid_debt: Dict[str, Decimal]
    collateral_received: Dict[str, Decimal]  # 清算人收到的抵押品
    protocol_fee: Dict[str, Decimal]  # 協議收取的費用
    health_factor_at_liquidation: Decimal
    price_data: Dict[str, PriceData]
    
    @property
    def total_collateral_usd(self) -> Decimal:
        """總抵押品價值（USD）"""
        return sum(self.collateral_received.values())
    
    @property
    def total_debt_repaid_usd(self) -> Decimal:
        """總償還債務（USD）"""
        return sum(self.repaid_debt.values())
    
    @property
    def liquidation_bonus(self) -> Decimal:
        """清算獎勵"""
        return self.total_collateral_usd - self.total_debt_repaid_usd

3. 健康因子模擬引擎

3.1 即時健康因子計算

class HealthFactorEngine:
    """
    健康因子計算引擎
    
    提供即時的健康因子計算、預測和監控功能
    """
    
    def __init__(self, token_configs: Dict[str, TokenConfig]):
        self.token_configs = token_configs
        self._build_price_cache()
    
    def _build_price_cache(self):
        """初始化價格緩存"""
        self._price_cache: Dict[str, List[PriceData]] = {
            symbol: [] for symbol in self.token_configs.keys()
        }
    
    def update_price(self, price_data: PriceData):
        """更新價格數據"""
        if price_data.symbol in self._price_cache:
            self._price_cache[price_data.symbol].append(price_data)
            # 保持最近 1000 筆價格數據
            if len(self._price_cache[price_data.symbol]) > 1000:
                self._price_cache[price_data.symbol] = \
                    self._price_cache[price_data.symbol][-1000:]
    
    def get_current_prices(self) -> Dict[str, Decimal]:
        """獲取最新價格"""
        prices = {}
        for symbol, prices_list in self._price_cache.items():
            if prices_list:
                prices[symbol] = prices_list[-1].price
        return prices
    
    def calculate_health_factor(
        self,
        position: Position,
        interest_accrued: Dict[str, Decimal] = None
    ) -> Optional[Decimal]:
        """計算當前健康因子"""
        prices = self.get_current_prices()
        return position.calculate_health_factor(
            prices=prices,
            token_configs=self.token_configs,
            interest_accrued=interest_accrued
        )
    
    def calculate_liquidation_price(
        self,
        position: Position,
        collateral_symbol: str,
        borrow_asset: str = "ETH",
        interest_accrued: Dict[str, Decimal] = None
    ) -> Optional[Decimal]:
        """
        計算指定抵押資產的清算觸發價格
        
        當該抵押資產的價格低於此值時，頭寸將被清算
        
        Args:
            position: 借款頭寸
            collateral_symbol: 要計算的抵押資產符號
            borrow_asset: 借款資產符號
            interest_accrued: 已累積利息
            
        Returns:
            清算觸發價格，若無法計算返回 None
        """
        if collateral_symbol not in position.collateral_assets:
            return None
        
        borrow_value = position.get_borrow_value(
            prices=self.get_current_prices(),
            interest_accrued=interest_accrued
        )
        
        # 需要的抵押品價值 = 借款價值 / 清算閾值
        required_collateral_value = borrow_value / Decimal(
            str(self.token_configs[collateral_symbol].liquidation_threshold)
        )
        
        # 單位的抵押品價值
        collateral_amount = position.collateral_assets[collateral_symbol]
        if collateral_amount == 0:
            return None
        
        price_per_unit = required_collateral_value / collateral_amount
        
        return price_per_unit
    
    def predict_health_factor_trajectory(
        self,
        position: Position,
        price_scenarios: Dict[str, List[float]],
        time_horizon_hours: int = 24,
        interest_rate_estimate: Dict[str, float] = None
    ) -> pd.DataFrame:
        """
        預測健康因子軌跡
        
        基於不同的價格情景，模擬未來的健康因子變化
        
        Args:
            position: 借款頭寸
            price_scenarios: 價格情景，格式為 {symbol: [price_t0, price_t1, ...]}
            time_horizon_hours: 預測時間範圍（小時）
            interest_rate_estimate: 借款資產的利率估計（年化，小時利率）
            
        Returns:
            DataFrame，包含每個時間點的健康因子預測
        """
        current_prices = self.get_current_prices()
        
        # 計算當前狀態
        current_hf = self.calculate_health_factor(position)
        current_collateral_value = position.get_collateral_value(
            current_prices, self.token_configs
        )
        current_borrow_value = position.get_borrow_value(
            current_prices, None
        )
        
        # 生成時間序列
        hours = np.arange(0, time_horizon_hours + 1, 1)  # 每小時一個點
        results = []
        
        for scenario_name, prices in price_scenarios.items():
            hf_values = []
            
            for i, hour in enumerate(hours):
                if i >= len(prices):
                    break
                    
                price = Decimal(str(prices[i]))
                
                # 計算利息累積
                interest_multiplier = Decimal('1')
                if interest_rate_estimate:
                    for borrow_symbol, borrow_amount in position.borrow_assets.items():
                        hourly_rate = interest_rate_estimate.get(borrow_symbol, 0) / (365 * 24)
                        accumulated_interest = borrow_amount * Decimal(str(hourly_rate)) * Decimal(str(hour))
                        interest_multiplier += accumulated_interest / position.get_borrow_value(
                            {borrow_symbol: Decimal('1')}, None
                        )
                
                # 重新計算抵押品價值
                new_collateral_value = Decimal('0')
                for symbol, amount in position.collateral_assets.items():
                    if symbol == collateral_symbol_for_update:
                        new_collateral_value += amount * price * self.token_configs[symbol].collateral_factor
                    else:
                        new_collateral_value += amount * current_prices.get(symbol, Decimal('0')) * \
                            self.token_configs[symbol].collateral_factor
                
                # 計算新的借款價值
                new_borrow_value = current_borrow_value * interest_multiplier
                
                # 計算健康因子
                if new_borrow_value > 0:
                    hf = new_collateral_value / new_borrow_value
                else:
                    hf = None
                
                hf_values.append(float(hf) if hf else float('inf'))
            
            for hour, hf in zip(hours[:len(hf_values)], hf_values):
                results.append({
                    'hour': hour,
                    'scenario': scenario_name,
                    'health_factor': hf
                })
        
        return pd.DataFrame(results)
    
    def find_liquidation_threshold_sensitivity(
        self,
        position: Position,
        threshold_range: Tuple[float, float] = (0.5, 0.95),
        step: float = 0.01
    ) -> pd.DataFrame:
        """
        清算閾值敏感度分析
        
        分析不同清算閾值下頭寸的安全邊際
        
        Args:
            position: 借款頭寸
            threshold_range: 清算閾值範圍
            step: 閾值變化步長
            
        Returns:
            DataFrame，包含不同閾值下的健康因子
        """
        current_prices = self.get_current_prices()
        collateral_value = position.get_collateral_value(current_prices, self.token_configs)
        borrow_value = position.get_borrow_value(current_prices, None)
        
        results = []
        thresholds = np.arange(threshold_range[0], threshold_range[1] + step, step)
        
        for threshold in thresholds:
            effective_collateral = collateral_value * threshold
            hf = effective_collateral / borrow_value if borrow_value > 0 else float('inf')
            
            results.append({
                'liquidation_threshold': threshold,
                'health_factor': float(hf),
                'distance_to_liquidation': float(hf - 1.0) if hf else None,
                'liquidation_risk': 'HIGH' if hf < 1.2 else ('MEDIUM' if hf < 1.5 else 'LOW')
            })
        
        return pd.DataFrame(results)

3.2 模擬器核心類

class LiquidationSimulator:
    """
    清算模擬器
    
    模擬清算事件的觸發、執行和影響
    """
    
    def __init__(
        self,
        token_configs: Dict[str, TokenConfig],
        liquidation_bonus: float = 0.1,  # 清算獎勵 10%
        protocol_fee: float = 0.05  # 協議費用 5%
    ):
        self.token_configs = token_configs
        self.liquidation_bonus = liquidation_bonus
        self.protocol_fee = protocol_fee
        self.hf_engine = HealthFactorEngine(token_configs)
        
        # 狀態追蹤
        self.positions: Dict[str, Position] = {}
        self.liquidation_events: List[LiquidationEvent] = []
        self.price_history: List[Dict[str, PriceData]] = []
    
    def add_position(self, position: Position):
        """添加借款頭寸"""
        self.positions[position.owner] = position
    
    def simulate_price_change(
        self,
        price_changes: Dict[str, float],  # 價格變化百分比
        simulate_liquidation: bool = True
    ) -> List[LiquidationEvent]:
        """
        模擬價格變化對頭寸的影響
        
        Args:
            price_changes: 價格變化，格式為 {symbol: change_percentage}
            simulate_liquidation: 是否觸發清算模擬
            
        Returns:
            觸發的清算事件列表
        """
        current_prices = self.hf_engine.get_current_prices()
        new_prices = {}
        
        # 更新價格
        for symbol, change_pct in price_changes.items():
            if symbol in current_prices:
                old_price = current_prices[symbol]
                new_price = old_price * Decimal(str(1 + change_pct))
                new_prices[symbol] = new_price
                
                # 更新引擎緩存
                self.hf_engine.update_price(PriceData(
                    timestamp=datetime.now(),
                    symbol=symbol,
                    price=new_price,
                    source='simulation'
                ))
        
        # 檢查清算觸發
        triggered_liquidations = []
        
        if simulate_liquidation:
            for owner, position in self.positions.items():
                hf = self.hf_engine.calculate_health_factor(position)
                
                if hf is not None and hf < 1:
                    # 觸發清算
                    event = self._execute_liquidation(
                        position=position,
                        prices=new_prices,
                        trigger_hf=hf
                    )
                    if event:
                        triggered_liquidations.append(event)
                        self.liquidation_events.append(event)
        
        # 記錄價格歷史
        self.price_history.append(new_prices)
        
        return triggered_liquidations
    
    def _execute_liquidation(
        self,
        position: Position,
        prices: Dict[str, Decimal],
        trigger_hf: Decimal
    ) -> Optional[LiquidationEvent]:
        """執行清算"""
        # 選擇價值最高的抵押資產進行清算
        best_collateral = None
        best_collateral_value = Decimal('0')
        
        for symbol, amount in position.collateral_assets.items():
            if symbol in prices:
                value = amount * prices[symbol] * self.token_configs[symbol].collateral_factor
                if value > best_collateral_value:
                    best_collateral_value = value
                    best_collateral = symbol
        
        if not best_collateral:
            return None
        
        # 選擇借款資產償還
        debt_to_repay = Decimal('0')
        debt_asset = None
        for symbol, amount in position.borrow_assets.items():
            if symbol in prices:
                value = amount * prices[symbol]
                debt_to_repay += value
                debt_asset = symbol
        
        # 計算結算金額
        # 清算人償還 debt_to_repay 獲得 collateral_amount * (1 + bonus)
        collateral_to_liquidate = min(
            position.collateral_assets[best_collateral],
            debt_to_repay / (prices[best_collateral] * Decimal(str(1 + self.liquidation_bonus)))
        )
        
        collateral_received = collateral_to_liquidate * Decimal(str(1 + self.liquidation_bonus))
        
        return LiquidationEvent(
            timestamp=datetime.now(),
            position_owner=position.owner,
            liquidated_collateral={best_collateral: collateral_to_liquidate},
            repaid_debt={debt_asset: debt_to_repay / prices[debt_asset]} if debt_asset else {},
            collateral_received={best_collateral: collateral_received},
            protocol_fee={best_collateral: collateral_received * Decimal(str(self.protocol_fee))},
            health_factor_at_liquidation=trigger_hf,
            price_data={symbol: PriceData(
                timestamp=datetime.now(),
                symbol=symbol,
                price=price,
                source='simulation'
            ) for symbol, price in prices.items()}
        )

4. 風險參數回測框架

4.1 回測引擎

class BacktestEngine:
    """
    回測引擎
    
    對歷史數據進行清算風險回測
    """
    
    def __init__(
        self,
        simulator: LiquidationSimulator,
        initial_capital: float = 100000.0  # 初始資金（USD）
    ):
        self.simulator = simulator
        self.initial_capital = Decimal(str(initial_capital))
        self.current_capital = self.initial_capital
        
        # 回測結果
        self.trades: List[Dict] = []
        self.equity_curve: List[Dict] = []
        self.liquidation_summary: Dict = {}
    
    def load_historical_data(
        self,
        data_source: str,
        start_date: datetime,
        end_date: datetime,
        symbols: List[str]
    ) -> pd.DataFrame:
        """
        加載歷史價格數據
        
        實際實現中可從 CoinGecko、The Graph 或自定義數據源獲取
        這裡使用模擬數據作為示例
        """
        # 模擬歷史數據
        dates = pd.date_range(start=start_date, end=end_date, freq='1H')
        data = {'timestamp': dates}
        
        for symbol in symbols:
            # 模擬價格走勢（隨機遊走）
            np.random.seed(42)  # 可重現性
            initial_price = 2000 if symbol == 'ETH' else 1  # ETH = 2000, USDC = 1
            returns = np.random.normal(0.0001, 0.02, len(dates))  # 日均收益 1%，波動率 2%
            prices = initial_price * np.exp(np.cumsum(returns))
            data[symbol] = prices
        
        return pd.DataFrame(data).set_index('timestamp')
    
    def run_backtest(
        self,
        historical_prices: pd.DataFrame,
        positions: List[Position],
        liquidation_threshold: float = 0.80,
        max_leverage: float = 2.0,
        rebalance_frequency: int = 24  # 小時
    ) -> Dict:
        """
        執行回測
        
        Args:
            historical_prices: 歷史價格數據
            positions: 初始頭寸列表
            liquidation_threshold: 清算閾值
            max_leverage: 最大槓桿
            rebalance_frequency: 再平衡頻率（小時）
            
        Returns:
            回測結果摘要
        """
        # 添加初始頭寸
        for position in positions:
            self.simulator.add_position(position)
        
        # 執行回測
        for i, (timestamp, row) in enumerate(historical_prices.iterrows()):
            # 更新價格
            price_changes = {}
            for symbol in row.index:
                if symbol in self.simulator.hf_engine._price_cache:
                    old_prices = self.simulator.hf_engine.get_current_prices()
                    if symbol in old_prices:
                        old_price = float(old_prices[symbol])
                        new_price = float(row[symbol])
                        change_pct = (new_price - old_price) / old_price
                        price_changes[symbol] = change_pct
                        
                        # 更新引擎
                        self.simulator.hf_engine.update_price(PriceData(
                            timestamp=timestamp,
                            symbol=symbol,
                            price=Decimal(str(new_price)),
                            source='backtest'
                        ))
            
            # 檢查清算
            liquidations = self.simulator.simulate_price_change(
                price_changes=price_changes,
                simulate_liquidation=True
            )
            
            # 記錄交易
            for liq in liquidations:
                self.trades.append({
                    'timestamp': timestamp,
                    'type': 'liquidation',
                    'owner': liq.position_owner,
                    'collateral_symbol': list(liq.liquidated_collateral.keys())[0],
                    'collateral_amount': float(list(liq.liquidated_collateral.values())[0]),
                    'collateral_usd': float(liq.total_collateral_usd),
                    'debt_repaid_usd': float(liq.total_debt_repaid_usd),
                    'profit': float(liq.liquidation_bonus),
                    'health_factor': float(liq.health_factor_at_liquidation)
                })
            
            # 記錄權益曲線
            if i % rebalance_frequency == 0:
                total_value = self.current_capital
                for position in self.simulator.positions.values():
                    prices = self.simulator.hf_engine.get_current_prices()
                    collateral_value = float(position.get_collateral_value(prices, self.simulator.token_configs))
                    total_value += Decimal(str(collateral_value))
                
                self.equity_curve.append({
                    'timestamp': timestamp,
                    'equity': float(total_value),
                    'returns': float((total_value - self.initial_capital) / self.initial_capital)
                })
        
        # 生成回測報告
        return self._generate_report()
    
    def _generate_report(self) -> Dict:
        """生成回測報告"""
        if not self.trades:
            return {
                'total_trades': 0,
                'total_pnl': 0.0,
                'win_rate': 0.0,
                'summary': 'No liquidations occurred during backtest period'
            }
        
        df_trades = pd.DataFrame(self.trades)
        
        report = {
            'total_trades': len(df_trades),
            'total_liquidated_collateral': df_trades['collateral_usd'].sum(),
            'total_debt_repaid': df_trades['debt_repaid_usd'].sum(),
            'total_pnl': df_trades['profit'].sum(),
            'average_profit_per_liquidation': df_trades['profit'].mean(),
            'win_rate': (df_trades['profit'] > 0).mean(),
            
            # 權益曲線分析
            'initial_capital': float(self.initial_capital),
            'final_capital': self.equity_curve[-1]['equity'] if self.equity_curve else float(self.initial_capital),
            'total_return': (self.equity_curve[-1]['equity'] / float(self.initial_capital) - 1) if self.equity_curve else 0.0,
            'sharpe_ratio': self._calculate_sharpe_ratio(),
            'max_drawdown': self._calculate_max_drawdown(),
            
            # 清算分析
            'liquidation_distribution': df_trades.groupby('collateral_symbol')['profit'].sum().to_dict(),
            'health_factor_at_liquidation_stats': {
                'mean': df_trades['health_factor'].mean(),
                'min': df_trades['health_factor'].min(),
                'max': df_trades['health_factor'].max(),
                'std': df_trades['health_factor'].std()
            }
        }
        
        return report
    
    def _calculate_sharpe_ratio(self) -> float:
        """計算夏普比率"""
        if len(self.equity_curve) < 2:
            return 0.0
        
        returns = [e['returns'] for e in self.equity_curve]
        if not returns:
            return 0.0
        
        mean_return = np.mean(returns)
        std_return = np.std(returns)
        
        if std_return == 0:
            return 0.0
        
        # 年化（假設每小時數據）
        return (mean_return * 8760) / (std_return * np.sqrt(8760))
    
    def _calculate_max_drawdown(self) -> float:
        """計算最大回撤"""
        if not self.equity_curve:
            return 0.0
        
        equity = [e['equity'] for e in self.equity_curve]
        peak = equity[0]
        max_dd = 0.0
        
        for value in equity:
            if value > peak:
                peak = value
            dd = (peak - value) / peak
            if dd > max_dd:
                max_dd = dd
        
        return max_dd

5. 歷史情境重現系統

5.1 情境定義與重現

class HistoricalScenarioReplay:
    """
    歷史情境重現系統
    
    重現重大市場事件，分析清算機制的表現
    """
    
    # 預定義的歷史情境
    SCENARIOS = {
        '2021_05_19_crash': {
            'name': '2021年5月19日加密貨幣崩潰',
            'description': '比特幣和以太坊在24小時內暴跌超過50%',
            'peak_date': '2021-05-19',
            'trough_date': '2021-05-19',
            'recovery_date': '2021-05-20',
            'price_changes': {
                'BTC': -49.5,
                'ETH': -58.2,
                'USDC': 0.1,  # 曾短暫脫鉤
                'DAI': 0.2
            }
        },
        '2022_11_ftx_collapse': {
            'name': '2022年11月FTX崩潰',
            'description': 'FTX破產引發市場恐慌',
            'peak_date': '2022-11-08',
            'trough_date': '2022-11-09',
            'recovery_date': '2023-01-01',
            'price_changes': {
                'BTC': -26.8,
                'ETH': -29.8,
                'SOL': -42.5,
                'FTT': -92.0
            }
        },
        '2023_03_svb_crisis': {
            'name': '2023年3月SVB銀行危機',
            'description': '銀行業危機導致市場動盪',
            'peak_date': '2023-03-10',
            'trough_date': '2023-03-13',
            'recovery_date': '2023-03-20',
            'price_changes': {
                'BTC': -15.2,
                'ETH': -12.8,
                'USDC': 3.5,  # 脫鉤後恢復
                'DAI': 0.5
            }
        },
        '2024_08_eth_correction': {
            'name': '2024年8月以太坊回調',
            'description': '主要山寨幣季節回調',
            'peak_date': '2024-08-25',
            'trough_date': '2024-08-30',
            'recovery_date': '2024-09-15',
            'price_changes': {
                'ETH': -32.5,
                'BTC': -18.2,
                'LINK': -25.3,
                'UNI': -28.7
            }
        }
    }
    
    def __init__(self, simulator: LiquidationSimulator):
        self.simulator = simulator
    
    def replay_scenario(
        self,
        scenario_name: str,
        positions: List[Position],
        simulate_partial_liquidation: bool = True,
        include_flash_crashes: bool = False
    ) -> Dict:
        """
        重現指定的歷史情境
        
        Args:
            scenario_name: 情境名稱
            positions: 要測試的頭寸列表
            simulate_partial_liquidation: 是否模擬部分清算
            include_flash_crashes: 是否包含閃電崩盤（更劇烈的波動）
            
        Returns:
            情境重現結果
        """
        if scenario_name not in self.SCENARIOS:
            raise ValueError(f"Unknown scenario: {scenario_name}")
        
        scenario = self.SCENARIOS[scenario_name]
        
        # 準備頭寸
        for position in positions:
            self.simulator.add_position(position)
        
        # 生成價格路徑
        price_path = self._generate_price_path(
            scenario=scenario,
            include_flash_crashes=include_flash_crashes
        )
        
        # 執行模擬
        results = []
        cumulative_hf = {}
        
        for step, prices in enumerate(price_path):
            # 更新模擬器價格
            for symbol, price_change in prices.items():
                self.simulator.hf_engine.update_price(PriceData(
                    timestamp=datetime.now(),
                    symbol=symbol,
                    price=Decimal(str(1 + price_change)),
                    source='scenario_replay'
                ))
            
            # 檢查頭寸狀態
            step_results = {
                'step': step,
                'prices': prices
            }
            
            for owner, position in self.simulator.positions.items():
                hf = self.simulator.hf_engine.calculate_health_factor(position)
                
                if owner not in cumulative_hf:
                    cumulative_hf[owner] = []
                
                cumulative_hf[owner].append(float(hf) if hf else float('inf'))
                
                step_results[f'{owner}_hf'] = float(hf) if hf else None
                step_results[f'{owner}_liquidated'] = hf is not None and hf < 1
            
            results.append(step_results)
        
        # 觸發清算模擬
        liquidations = self.simulator.simulate_price_change(
            price_changes={symbol: prices[-1] for symbol, prices in price_path[-1].items()} if price_path else {},
            simulate_liquidation=True
        )
        
        return {
            'scenario': scenario,
            'price_path': price_path,
            'simulation_results': results,
            'liquidations': [
                {
                    'timestamp': liq.timestamp,
                    'owner': liq.position_owner,
                    'collateral_symbol': list(liq.liquidated_collateral.keys())[0],
                    'collateral_amount': float(list(liq.liquidated_collateral.values())[0]),
                    'health_factor': float(liq.health_factor_at_liquidation),
                    'bonus': float(liq.liquidation_bonus)
                }
                for liq in liquidations
            ],
            'cumulative_health_factor': cumulative_hf,
            'summary': self._generate_scenario_summary(
                scenario, results, liquidations, cumulative_hf
            )
        }
    
    def _generate_price_path(
        self,
        scenario: Dict,
        include_flash_crashes: bool,
        steps: int = 24
    ) -> List[Dict[str, float]]:
        """生成價格路徑"""
        price_path = []
        
        # 根據情境生成價格變化
        base_changes = scenario['price_changes']
        
        for step in range(steps):
            step_prices = {}
            progress = step / steps
            
            for symbol, total_change in base_changes.items():
                # 根據進度分配價格變化
                if progress < 0.5:
                    # 前半段：主要下跌
                    change = total_change * (progress * 2)
                else:
                    # 後半段：低位震盪或反彈
                    change = total_change + (total_change * -0.3 * ((progress - 0.5) * 2))
                
                # 添加隨機波動
                noise = np.random.normal(0, abs(total_change) * 0.05)
                step_prices[symbol] = change + noise
                
                # 閃電崩盤模擬
                if include_flash_crashes and step == steps // 3:
                    step_prices[symbol] -= abs(total_change) * 0.1
            
            price_path.append(step_prices)
        
        return price_path
    
    def _generate_scenario_summary(
        self,
        scenario: Dict,
        results: List[Dict],
        liquidations: List[LiquidationEvent],
        cumulative_hf: Dict[str, List[float]]
    ) -> Dict:
        """生成情境摘要"""
        total_positions = len(cumulative_hf)
        liquidated_positions = sum(
            1 for owner, hfs in cumulative_hf.items()
            if any(hf < 1 for hf in hfs if hf != float('inf'))
        )
        
        # 計算平均最低健康因子
        min_hfs = [
            min(hfs) for hfs in cumulative_hf.values()
            if any(hf != float('inf') for hf in hfs)
        ]
        avg_min_hf = np.mean(min_hfs) if min_hfs else None
        
        return {
            'scenario_name': scenario['name'],
            'total_positions_tested': total_positions,
            'positions_liquidated': liquidated_positions,
            'liquidation_rate': liquidated_positions / total_positions if total_positions > 0 else 0,
            'total_liquidation_events': len(liquidations),
            'average_min_health_factor': float(avg_min_hf) if avg_min_hf else None,
            'risk_assessment': self._assess_scenario_risk(
                liquidated_positions, total_positions, avg_min_hf
            )
        }
    
    def _assess_scenario_risk(
        self,
        liquidated: int,
        total: int,
        avg_min_hf: float
    ) -> str:
        """評估情境風險"""
        if total == 0:
            return "UNKNOWN"
        
        liquidation_rate = liquidated / total
        
        if liquidation_rate > 0.5:
            return "CRITICAL"
        elif liquidation_rate > 0.2:
            return "HIGH"
        elif liquidation_rate > 0.05:
            return "MEDIUM"
        elif avg_min_hf and avg_min_hf < 1.2:
            return "MEDIUM"  # 接近清算線
        else:
            return "LOW"

6. 使用範例：完整回測流程

def run_complete_backtest():
    """
    完整回測流程示例
    """
    # 1. 初始化配置
    token_configs = {
        'ETH': TokenConfig(
            symbol='ETH',
            address='0x0000000000000000000000000000000000000000',
            decimals=18,
            oracle_address='0x0000000000000000000000000000000000000000',
            volatility_30d=0.05,
            liquidity_score=0.95,
            is_collateral=True,
            liquidation_threshold=0.80,
            collateral_factor=0.80
        ),
        'USDC': TokenConfig(
            symbol='USDC',
            address='0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48',
            decimals=6,
            oracle_address='0x0000000000000000000000000000000000000000',
            volatility_30d=0.001,
            liquidity_score=0.99,
            is_collateral=False
        ),
        'WBTC': TokenConfig(
            symbol='WBTC',
            address='0x2260FAC5E5542a773Aa44fBCfeDf7C193bc2C599',
            decimals=8,
            oracle_address='0x0000000000000000000000000000000000000000',
            volatility_30d=0.04,
            liquidity_score=0.85,
            is_collateral=True,
            liquidation_threshold=0.75,
            collateral_factor=0.70
        )
    }
    
    # 2. 創建模擬器
    simulator = LiquidationSimulator(
        token_configs=token_configs,
        liquidation_bonus=0.10,
        protocol_fee=0.05
    )
    
    # 3. 創建測試頭寸
    positions = [
        Position(
            owner='user_1',
            collateral_assets={'ETH': Decimal('10')},
            borrow_assets={'USDC': Decimal('10000')}  # 借款 10000 USDC
        ),
        Position(
            owner='user_2',
            collateral_assets={'WBTC': Decimal('0.5')},
            borrow_assets={'ETH': Decimal('5')}  # 借款 5 ETH
        ),
        Position(
            owner='user_3',
            collateral_assets={'ETH': Decimal('20'), 'WBTC': Decimal('0.2')},
            borrow_assets={'USDC': Decimal('50000')}  # 借款 50000 USDC
        )
    ]
    
    # 4. 初始化回測引擎
    backtest = BacktestEngine(
        simulator=simulator,
        initial_capital=100000.0
    )
    
    # 5. 模擬 2021 年 5 月 19 日情境
    scenario_replay = HistoricalScenarioReplay(simulator)
    
    results = scenario_replay.replay_scenario(
        scenario_name='2021_05_19_crash',
        positions=positions,
        simulate_partial_liquidation=True,
        include_flash_crashes=True
    )
    
    print("=" * 60)
    print(f"情境: {results['scenario']['name']}")
    print(f"描述: {results['scenario']['description']}")
    print("=" * 60)
    print()
    print("摘要:")
    print(f"  測試頭寸數: {results['summary']['total_positions_tested']}")
    print(f"  被清算頭寸數: {results['summary']['positions_liquidated']}")
    print(f"  清算率: {results['summary']['liquidation_rate']:.1%}")
    print(f"  清算事件數: {results['summary']['total_liquidation_events']}")
    print(f"  平均最低健康因子: {results['summary']['average_min_health_factor']:.3f}")
    print(f"  風險評估: {results['summary']['risk_assessment']}")
    print()
    
    if results['liquidations']:
        print("清算詳情:")
        for liq in results['liquidations']:
            print(f"  - {liq['owner']}: "
                  f"{liq['collateral_symbol']} x {liq['collateral_amount']:.4f}, "
                  f"HF={liq['health_factor']:.3f}, "
                  f"收益={liq['bonus']:.2f} USD")
    
    return results


if __name__ == '__main__':
    results = run_complete_backtest()

7. 結論

本文提供了完整的 DeFi 清算機制模擬器教學，涵蓋：

1. 健康因子引擎：即時計算、預測和監控頭寸安全性
2. 清算模擬器：模擬清算觸發條件和結算過程
3. 回測框架：對歷史數據進行風險參數回測
4. 歷史情境重現：重現重大市場事件，分析清算機制表現

透過這些工具，讀者可以：

• 評估不同市場條件下的清算風險
• 優化清算閾值參數
• 開發和回測清算策略
• 理解清算機制的經濟學原理

延伸閱讀

• Aave Protocol V3 Technical Paper
• Compound Finance Whitepaper
• MakerDAO Governance and Risk Management
• Flashbots MEV Research

本指南內容僅供教育目的。在進行任何 DeFi 操作前，請詳細閱讀相關協議文檔並諮詢專業意見。