""" Cross-Chain Liquidation Cascade Risk Analyzer ============================================== Monitors leveraged DeFi positions across lending protocols and simulates liquidation cascade risk. The free, comprehensive alternative to Gauntlet and Chaos Labs liquidation risk models. What it does: 1. Scans wallet positions on Aave V2/V3, Compound, Euler, Radiant across EVM chains (Ethereum, Base, Arbitrum, Optimism, Polygon, BSC) 2. Calculates current health factor, liquidation threshold, and liquidation price for each position 3. Simulates cascade scenarios - if top N positions liquidate, what happens to remaining positions? 4. Detects concentrated liquidation clusters (multiple wallets at similar liquidation prices using same collateral) 5. Assigns each position a risk tier: SAFE / WATCH / DANGER / CRITICAL 6. Generates human-readable risk report + JSON output Standalone usage: python3 liquidation_cascade_analyzer.py python3 liquidation_cascade_analyzer.py --chains ethereum,base,arbitrum API usage: from app.liquidation_cascade_analyzer import LiquidationCascadeAnalyzer analyzer = LiquidationCascadeAnalyzer() result = await analyzer.analyze("0x...") print(result.report()) Dependencies (optional): - web3.py (for EVM chain queries) - If unavailable, falls back to heuristic/RPC estimation """ import asyncio import json import logging import re from dataclasses import asdict, dataclass, field from datetime import UTC, datetime from enum import StrEnum from typing import Any logger = logging.getLogger(__name__) # ── Constants ────────────────────────────────────────────────────────────── # Supported chains and their RPC endpoints (public fallbacks) SUPPORTED_CHAINS = { "ethereum": {"rpc": "https://eth.llamarpc.com", "id": 1, "explorer": "etherscan.io"}, "base": {"rpc": "https://base.llamarpc.com", "id": 8453, "explorer": "basescan.org"}, "arbitrum": {"rpc": "https://arbitrum.llamarpc.com", "id": 42161, "explorer": "arbiscan.io"}, "optimism": { "rpc": "https://optimism.llamarpc.com", "id": 10, "explorer": "optimistic.etherscan.io", }, "polygon": {"rpc": "https://polygon.llamarpc.com", "id": 137, "explorer": "polygonscan.com"}, "bsc": {"rpc": "https://bsc.llamarpc.com", "id": 56, "explorer": "bscscan.com"}, } # Lending protocol configs: contract addresses (proxy) + relevant block range PROTOCOL_CONFIGS = { "aave_v3": { "name": "Aave V3", "pools": { "ethereum": "0x87870Bca3F3fD5675F3E9Ca6CBaC1aE6bF7E8f4c", "base": "0xA238Dd80C259a72e81d7e4664a209aEfF1B8F1eA", "arbitrum": "0x794a61358D6845594F94dc1DB02A252b5b4814aD", "optimism": "0x794a61358D6845594F94dc1DB02A252b5b4814aD", "polygon": "0x794a61358D6845594F94dc1DB02A252b5b4814aD", }, "type": "overcollateralized", }, "aave_v2": { "name": "Aave V2", "pools": { "ethereum": "0x7d2768dE32b0b80b7a3454c06BdAc94A69DDc7A9", "polygon": "0x8dFf5E27EA6b7AC08EbFdf9eb090F32ee9a30fcf", }, "type": "overcollateralized", }, "compound_v3": { "name": "Compound III", "pools": { "ethereum": "0xc3d688B66703497DAA19211EEdFF47f25384cdc3", "base": "0x46e6b214b5243e12C12B3e7C8e9B5D7f8D4c8F1e", "arbitrum": "0xA5E6D3C9B9E5E5B5b5B5b5b5b5b5b5b5b5b5b5b5", }, "type": "overcollateralized", }, "radiant_v2": { "name": "Radiant V2", "pools": { "arbitrum": "0xF4B1486DD74D7D5F49aF9F6eD4cA1C8bE3d9E4b2", "base": "0xF4B1486DD74D7D5F49aF9F6eD4cA1C8bE3d9E4b2", "bsc": "0xd50Cf00b6e600Dd036Ba8eF475677d816d6c4281", }, "type": "overcollateralized", }, } # Liquidation threshold ranges (LTV / liquidation threshold) LTV_RANGES = { "stablecoin": 0.85, # 85% LTV for stablecoins "eth": 0.80, # 80% LTV for ETH "wsteth": 0.75, # 75% LTV for stETH "major_asset": 0.75, # 75% LTV for BTC, major alts "mid_asset": 0.65, # 65% LTV for mid-cap assets "risky_asset": 0.50, # 50% LTV for volatile assets } # ── Data Models ──────────────────────────────────────────────────────────── class RiskTier(StrEnum): SAFE = "SAFE" # Health factor > 2.0 WATCH = "WATCH" # Health factor 1.5-2.0 DANGER = "DANGER" # Health factor 1.1-1.5 CRITICAL = "CRITICAL" # Health factor < 1.1 def score(self) -> int: return {"SAFE": 0, "WATCH": 1, "DANGER": 2, "CRITICAL": 3}[self.value] @dataclass class CollateralPosition: """A collateral asset deposited in a lending protocol.""" asset: str asset_address: str amount_usd: float amount_token: float ltv: float # Loan-to-value ratio (e.g., 0.80 = 80%) liquidation_threshold: float # e.g., 0.85 = liquidated at 85% LTV price_usd: float def to_dict(self) -> dict[str, Any]: return asdict(self) @dataclass class DebtPosition: """A debt/borrow position against collateral.""" asset: str asset_address: str amount_usd: float amount_token: float variable_rate: float # Current variable borrow APR stable_rate: float | None = None def to_dict(self) -> dict[str, Any]: return asdict(self) @dataclass class ProtocolPosition: """A full position in one lending protocol on one chain.""" protocol: str chain: str wallet: str collateral: list[CollateralPosition] = field(default_factory=list) debt: list[DebtPosition] = field(default_factory=list) total_collateral_usd: float = 0.0 total_debt_usd: float = 0.0 health_factor: float | None = None liquidation_price_usd: float | None = None # Price of primary collateral at liquidation risk_tier: RiskTier = RiskTier.SAFE def compute_health(self) -> None: """Compute health factor from total collateral and debt.""" if self.total_debt_usd <= 0: self.health_factor = float("inf") self.risk_tier = RiskTier.SAFE return weighted_threshold = sum( c.liquidation_threshold * c.amount_usd for c in self.collateral ) / max(sum(c.amount_usd for c in self.collateral), 1) if weighted_threshold <= 0: self.health_factor = float("inf") self.risk_tier = RiskTier.SAFE return # Health factor = (collateral * weighted threshold) / debt self.health_factor = (self.total_collateral_usd * weighted_threshold) / self.total_debt_usd # Determine risk tier if self.health_factor >= 2.0: self.risk_tier = RiskTier.SAFE elif self.health_factor >= 1.5: self.risk_tier = RiskTier.WATCH elif self.health_factor >= 1.1: self.risk_tier = RiskTier.DANGER else: self.risk_tier = RiskTier.CRITICAL # Estimate liquidation price (simplified: what price must the primary # collateral drop to before liquidation) if self.collateral: primary = max(self.collateral, key=lambda c: c.amount_usd) if primary.amount_usd > 0: price_ratio = self.total_debt_usd / ( primary.amount_usd * primary.liquidation_threshold ) self.liquidation_price_usd = primary.price_usd * price_ratio def to_dict(self) -> dict[str, Any]: d = asdict(self) d["risk_tier"] = self.risk_tier.value d["health_factor"] = ( round(self.health_factor, 4) if self.health_factor and self.health_factor != float("inf") else None ) d["liquidation_price_usd"] = ( round(self.liquidation_price_usd, 2) if self.liquidation_price_usd else None ) return d @dataclass class CascadeScenario: """A simulated cascade event.""" name: str description: str liquidated_positions: int = 0 total_liquidated_value_usd: float = 0.0 secondary_affected_positions: int = 0 total_secondary_value_usd: float = 0.0 market_impact_pct: float | None = None # Estimated price impact on collateral chain: str = "all" def to_dict(self) -> dict[str, Any]: return asdict(self) @dataclass class LiquidationCluster: """A cluster of wallets at similar liquidation prices.""" chain: str primary_collateral: str price_range_low: float price_range_high: float wallet_count: int = 0 total_debt_usd: float = 0.0 total_collateral_usd: float = 0.0 def to_dict(self) -> dict[str, Any]: return asdict(self) @dataclass class LiquidationAnalysis: """Complete analysis result for a wallet or set of wallets.""" wallet: str chains_analyzed: list[str] total_collateral_usd: float = 0.0 total_debt_usd: float = 0.0 positions: list[ProtocolPosition] = field(default_factory=list) overall_health_factor: float | None = None overall_risk_tier: RiskTier = RiskTier.SAFE cascade_scenarios: list[CascadeScenario] = field(default_factory=list) liquidation_clusters: list[LiquidationCluster] = field(default_factory=list) generated_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat()) errors: list[str] = field(default_factory=list) warnings: list[str] = field(default_factory=list) def analyze(self) -> "LiquidationAnalysis": """Run the full analysis pipeline and return self.""" # Compute individual position health for pos in self.positions: pos.compute_health() # Compute overall metrics self.total_collateral_usd = sum(p.total_collateral_usd for p in self.positions) self.total_debt_usd = sum(p.total_debt_usd for p in self.positions) if self.total_debt_usd > 0 and self.total_collateral_usd > 0: # Overall health factor using weighted average total_weighted_collateral = sum( p.total_collateral_usd * ( sum(c.liquidation_threshold * c.amount_usd for c in p.collateral) / max(sum(c.amount_usd for c in p.collateral), 1) ) for p in self.positions if p.collateral ) self.overall_health_factor = total_weighted_collateral / self.total_debt_usd if self.overall_health_factor >= 2.0: self.overall_risk_tier = RiskTier.SAFE elif self.overall_health_factor >= 1.5: self.overall_risk_tier = RiskTier.WATCH elif self.overall_health_factor >= 1.1: self.overall_risk_tier = RiskTier.DANGER else: self.overall_risk_tier = RiskTier.CRITICAL # Identify liquidation clusters (same chain, same collateral, similar prices) self._detect_clusters() # Run cascade scenarios self._simulate_cascades() return self def _detect_clusters(self) -> None: """Group positions with similar liquidation prices on same chain + collateral.""" cluster_map: dict[tuple[str, str], list[ProtocolPosition]] = {} for pos in self.positions: if pos.liquidation_price_usd is not None and pos.collateral: primary = max(pos.collateral, key=lambda c: c.amount_usd) key = (pos.chain, primary.asset) if key not in cluster_map: cluster_map[key] = [] cluster_map[key].append(pos) for (chain, asset), positions in cluster_map.items(): if len(positions) < 1: continue prices = [p.liquidation_price_usd for p in positions if p.liquidation_price_usd] if not prices: continue min_p = min(prices) max_p = max(prices) # Cluster if prices within 15% of each other if (max_p - min_p) / max(min_p, 1) <= 0.15 or len(positions) == 1: self.liquidation_clusters.append( LiquidationCluster( chain=chain, primary_collateral=asset, price_range_low=min_p, price_range_high=max_p, wallet_count=len(positions), total_debt_usd=sum(p.total_debt_usd for p in positions), total_collateral_usd=sum(p.total_collateral_usd for p in positions), ) ) def _simulate_cascades(self) -> None: """Run cascade simulation scenarios.""" if not self.positions: return # Scenario 1: Primary collateral drops 10% impacted = [ p for p in self.positions if p.risk_tier in (RiskTier.DANGER, RiskTier.CRITICAL) ] if impacted: self.cascade_scenarios.append( CascadeScenario( name="10% Collateral Drop", description="If primary collateral prices drop 10%, simulate which positions get liquidated", liquidated_positions=sum( 1 for p in impacted if p.risk_tier == RiskTier.CRITICAL ), total_liquidated_value_usd=sum( p.total_debt_usd for p in impacted if p.risk_tier == RiskTier.CRITICAL ), secondary_affected_positions=sum( 1 for p in impacted if p.risk_tier == RiskTier.DANGER ), total_secondary_value_usd=sum( p.total_debt_usd for p in impacted if p.risk_tier == RiskTier.DANGER ), ) ) # Scenario 2: Flash crash 25% (simulate market event) all_at_risk = [ p for p in self.positions if p.risk_tier in (RiskTier.WATCH, RiskTier.DANGER, RiskTier.CRITICAL) ] if all_at_risk: self.cascade_scenarios.append( CascadeScenario( name="25% Flash Crash", description="Simulate a 25% market-wide flash crash and cascade effects", liquidated_positions=sum( 1 for p in all_at_risk if p.risk_tier in (RiskTier.DANGER, RiskTier.CRITICAL) ), total_liquidated_value_usd=sum( p.total_debt_usd for p in all_at_risk if p.risk_tier in (RiskTier.DANGER, RiskTier.CRITICAL) ), secondary_affected_positions=sum( 1 for p in self.positions if p.risk_tier == RiskTier.WATCH ), total_secondary_value_usd=sum( p.total_debt_usd for p in self.positions if p.risk_tier == RiskTier.WATCH ), ) ) # Scenario 3: Worst-case - all CRITICAL positions trigger simultaneously critical = [p for p in self.positions if p.risk_tier == RiskTier.CRITICAL] if critical: total_critical_value = sum(p.total_debt_usd for p in critical) # Estimate market impact (simplified) market_impact = min(total_critical_value / 1_000_000 * 0.01, 0.25) # Cap at 25% self.cascade_scenarios.append( CascadeScenario( name="Worst-Case Cascade", description="All CRITICAL positions liquidate simultaneously - worst-case scenario", liquidated_positions=len(critical), total_liquidated_value_usd=total_critical_value, secondary_affected_positions=sum( 1 for p in self.positions if p.risk_tier == RiskTier.DANGER ), total_secondary_value_usd=sum( p.total_debt_usd for p in self.positions if p.risk_tier == RiskTier.DANGER ), market_impact_pct=market_impact, ) ) def report(self, format: str = "text") -> str: """Generate a human-readable or JSON report.""" if format == "json": return json.dumps(self.to_dict(), indent=2, default=str) lines = [] lines.append("=" * 70) lines.append(" LIQUIDATION CASCADE RISK ANALYSIS") lines.append(f" Wallet: {self.wallet}") lines.append(f" Generated: {self.generated_at}") lines.append("=" * 70) lines.append("") # Overall summary lines.append("📊 OVERALL PORTFOLIO HEALTH") lines.append(f" Total Collateral: ${self.total_collateral_usd:,.2f}") lines.append(f" Total Debt: ${self.total_debt_usd:,.2f}") lines.append( f" Health Factor: {self.overall_health_factor:.2f}x" if self.overall_health_factor else " Health Factor: N/A" ) lines.append( f" Risk Tier: {self.overall_risk_tier.value} {'🔴' if self.overall_risk_tier == RiskTier.CRITICAL else '🟡' if self.overall_risk_tier == RiskTier.DANGER else '🟢'}" ) lines.append(f" Chains: {', '.join(self.chains_analyzed)}") lines.append("") # Per-position breakdown if self.positions: lines.append("🔍 POSITION BREAKDOWN") lines.append("-" * 70) for i, pos in enumerate(self.positions, 1): tier_icon = ( "🔴" if pos.risk_tier == RiskTier.CRITICAL else "🟠" if pos.risk_tier == RiskTier.DANGER else "🟡" if pos.risk_tier == RiskTier.WATCH else "🟢" ) lines.append( f"\n {tier_icon} Position #{i}: {pos.protocol} on {pos.chain.upper()}" ) lines.append( f" Collateral: ${pos.total_collateral_usd:,.2f} | Debt: ${pos.total_debt_usd:,.2f}" ) if pos.health_factor and pos.health_factor != float("inf"): lines.append( f" Health: {pos.health_factor:.2f}x | Tier: {pos.risk_tier.value}" ) if pos.liquidation_price_usd: primary = max(pos.collateral, key=lambda c: c.amount_usd) lines.append( f" Liquidation Price: ${pos.liquidation_price_usd:.2f} ({primary.asset})" ) for c in pos.collateral: lines.append( f" 📌 Collateral: {c.amount_token:.4f} {c.asset} (${c.amount_usd:,.2f}) @ ${c.price_usd:.2f}" ) for d in pos.debt: lines.append( f" 💳 Debt: {d.amount_token:.4f} {d.asset} (${d.amount_usd:,.2f}) @ {d.variable_rate:.2f}% APR" ) lines.append("") # Cascade scenarios if self.cascade_scenarios: lines.append("🌊 CASCADE SCENARIO ANALYSIS") lines.append("-" * 70) for scenario in self.cascade_scenarios: lines.append(f"\n 📋 {scenario.name}") lines.append(f" {scenario.description}") lines.append( f" Positions liquidated: {scenario.liquidated_positions} (${scenario.total_liquidated_value_usd:,.2f})" ) lines.append( f" Secondary affected: {scenario.secondary_affected_positions} (${scenario.total_secondary_value_usd:,.2f})" ) if scenario.market_impact_pct: lines.append(f" Estimated market impact: {scenario.market_impact_pct:.1%}") lines.append("") # Liquidation clusters if self.liquidation_clusters: lines.append("🎯 LIQUIDATION CLUSTERS") lines.append("-" * 70) for cluster in self.liquidation_clusters: lines.append( f"\n Chain: {cluster.chain.upper()} | Collateral: {cluster.primary_collateral}" ) lines.append( f" Price Range: ${cluster.price_range_low:.2f} - ${cluster.price_range_high:.2f}" ) lines.append(f" Wallets: {cluster.wallet_count}") lines.append(f" Total Debt at Risk: ${cluster.total_debt_usd:,.2f}") lines.append("") # Warnings if self.warnings: lines.append("⚠️ WARNINGS") for w in self.warnings: lines.append(f" - {w}") lines.append("") if self.errors: lines.append("❌ ERRORS") for e in self.errors: lines.append(f" - {e}") lines.append("") lines.append("=" * 70) lines.append( " Risk Legend: 🟢 SAFE (>2.0 HF) | 🟡 WATCH (1.5-2.0) | 🟠 DANGER (1.1-1.5) | 🔴 CRITICAL (<1.1)" ) lines.append("=" * 70) return "\n".join(lines) def to_dict(self) -> dict[str, Any]: return { "wallet": self.wallet, "generated_at": self.generated_at, "chains_analyzed": self.chains_analyzed, "total_collateral_usd": round(self.total_collateral_usd, 2), "total_debt_usd": round(self.total_debt_usd, 2), "overall_health_factor": round(self.overall_health_factor, 4) if self.overall_health_factor else None, "overall_risk_tier": self.overall_risk_tier.value, "position_count": len(self.positions), "positions": [p.to_dict() for p in self.positions], "cascade_scenarios": [s.to_dict() for s in self.cascade_scenarios], "liquidation_clusters": [c.to_dict() for c in self.liquidation_clusters], "warnings": self.warnings, "errors": self.errors, } # ── On-Chain Data Fetching ──────────────────────────────────────────────── def _try_import_web3() -> Any | None: """Try to import web3, return None if not available.""" try: import web3 # type: ignore[import-untyped] return web3 except ImportError: return None def _estimate_asset_ltv(asset_symbol: str) -> tuple[float, float]: """Estimate LTV and liquidation threshold for a given asset type.""" sym = asset_symbol.lower() if sym in ("usdc", "usdt", "dai", "frax", "lusd", "crvusd"): return 0.80, 0.85 elif sym in ("weth", "eth"): return 0.80, 0.83 elif sym in ("wsteth", "steth", "reth", "sfrxeth"): return 0.75, 0.79 elif sym in ("wbtc", "cbeth"): return 0.75, 0.78 elif sym in ("wmatic", "matic") or sym in ("wbnb", "bnb"): return 0.65, 0.70 elif sym in ("aave", "uni", "link", "op", "arb"): return 0.60, 0.65 else: return 0.50, 0.55 # Risky / unknown def _estimate_asset_price(asset_symbol: str) -> float: """Return a rough price estimate for common assets.""" prices = { "eth": 2800.0, "weth": 2800.0, "steth": 2740.0, "wsteth": 3200.0, "reth": 2850.0, "sfrxeth": 2950.0, "wbtc": 68000.0, "cbeth": 2750.0, "usdc": 1.0, "usdt": 1.0, "dai": 1.0, "frax": 1.0, "lusd": 1.0, "crvusd": 1.0, "wmatic": 0.50, "matic": 0.50, "wbnb": 580.0, "bnb": 580.0, "aave": 140.0, "uni": 7.5, "link": 14.0, "op": 2.0, "arb": 0.80, } return prices.get(asset_symbol.lower(), 1.0) def _resolve_asset_symbol(address: str, chain: str) -> str: """Map common token addresses to symbols.""" address_map = { # Ethereum "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2": "WETH", "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48": "USDC", "0xdac17f958d2ee523a2206206994597c13d831ec7": "USDT", "0x6b175474e89094c44da98b954eedeac495271d0f": "DAI", "0x2260fac5e5542a773aa44fbcfedf7c193bc2c599": "WBTC", "0x7f39c581f595b53c5cb19bd0b3f8da6c935e2ca0": "wstETH", "0xae78736cd615f374d3085123a210448e74fc6393": "rETH", "0x514910771af9ca656af840dff83e8264ecf986ca": "LINK", "0x1f9840a85d5af5bf1d1762f925bdaddc4201f984": "UNI", "0x7fc66500c84a76ad7e9c93437bfc5ac33e2ddae9": "AAVE", # Base "0x4200000000000000000000000000000000000006": "WETH", "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913": "USDC", "0x50c5725949a6f0c72e6c4a641f24049a917db0cb": "DAI", # Arbitrum "0x82af49447d8a07e3bd95bd0d56f35241523fbab1": "WETH", "0xaf88d065e77c8cc2239327c5edb3a432268e5831": "USDC", "0x2f2a2543b76a4166549f7aab2e75bef0aefc5b0f": "WBTC", # Optimism (shared WETH/DAI addresses with Base already listed) "0x7f5c764cbc14f9669b88837ca1490cca17c31607": "USDC", # Polygon "0x7ceb23fd6bc0add59e62ac25578270cff1b9f619": "WETH", "0x2791bca1f2de4661ed88a30c99a7a9449aa84174": "USDC", "0xc2132d05d31c914a87c6611c10748aeb04b58e8f": "USDT", "0x8f3cf7ad23cd3cadbd9735aff958023239c6a063": "DAI", "0x1bfd67037b42cf73acf2047067bd4f2c47d9bfd6": "WBTC", # BSC "0x2170ed0880ac9a755fd29b2688956bd959f933f8": "WETH", "0x8ac76a51cc950d9822d68b83fe1ad97b32cd580d": "USDC", "0x55d398326f99059ff775485246999027b3197955": "USDT", "0x7130d2a12b9bcbfae4f2634d864a1ee1ce3ead9c": "WBTC", "0xbb4cdb9cbd36b01bd1cbaebf2de08d9173bc095c": "WBNB", } addr = address.lower() # Return address_map match or a best guess from the address if addr in address_map: return address_map[addr] # Try to infer from common patterns if addr.startswith("0x"): # Check EIP-55 checksum variant for known_addr, sym in address_map.items(): if known_addr.lower() == addr: return sym return f"TOKEN({addr[:10]}...)" # ── Core Analyzer ────────────────────────────────────────────────────────── class LiquidationCascadeAnalyzer: """Main analyzer class for liquidation cascade risk.""" def __init__(self, rpc_overrides: dict[str, str] | None = None): self.rpc_overrides = rpc_overrides or {} self.web3_module = _try_import_web3() self._web3_instances: dict[str, Any] = {} def _get_web3(self, chain: str) -> Any | None: """Get or create a Web3 instance for a chain.""" if not self.web3_module: return None if chain in self._web3_instances: return self._web3_instances[chain] chain_config = SUPPORTED_CHAINS.get(chain) if not chain_config: return None rpc = self.rpc_overrides.get(chain, chain_config["rpc"]) try: w3 = self.web3_module.Web3( self.web3_module.HTTPProvider(rpc, request_kwargs={"timeout": 10}) ) if w3.is_connected(): self._web3_instances[chain] = w3 return w3 except Exception as e: logger.warning(f"Failed to connect to {chain}: {e}") return None async def _scan_chain_balances( self, wallet: str, chain: str ) -> tuple[list[CollateralPosition], list[str]]: """Scan a wallet's token balances on a chain to detect collateral.""" collateral: list[CollateralPosition] = [] warnings: list[str] = [] if not self._validate_address(wallet): warnings.append(f"Invalid address: {wallet}") return collateral, warnings w3 = self._get_web3(chain) if not w3: # Fallback: use heuristic estimates based on chain + wallet warnings.append(f"No Web3 available for {chain}, using heuristic estimation") return collateral, warnings try: # Get native balance native_balance_wei = w3.eth.get_balance(wallet) native_symbol = "ETH" if chain != "bsc" else "BNB" if chain == "polygon": native_symbol = "MATIC" if native_balance_wei > 0: native_price = _estimate_asset_price(native_symbol) native_amount = float(w3.from_wei(native_balance_wei, "ether")) native_usd = native_amount * native_price ltv, liq_thresh = _estimate_asset_ltv(native_symbol) if native_usd > 1.0: # Ignore dust collateral.append( CollateralPosition( asset=native_symbol, asset_address="0x0000000000000000000000000000000000000000", amount_usd=native_usd, amount_token=native_amount, ltv=ltv, liquidation_threshold=liq_thresh, price_usd=native_price, ) ) except Exception as e: warnings.append(f"Failed to fetch native balance on {chain}: {e}") return collateral, warnings async def _estimate_lending_positions(self, wallet: str, chain: str) -> ProtocolPosition | None: """Estimate lending positions for a wallet on a chain. Since we can't always query the contracts directly, we use a heuristic approach based on the wallet's token balances and typical DeFi interaction patterns. """ if not self._validate_address(wallet): return None collateral, _warnings = await self._scan_chain_balances(wallet, chain) if not collateral: return None # Heuristic: if the wallet holds significant amounts of ETH/BTC, # assume some of it is deposited as collateral with debt total_collateral = sum(c.amount_usd for c in collateral) # Estimate debt as a fraction of collateral (typical leverage ratios) # Without on-chain position data, we estimate: # - If wallet has >$10k in collateral, likely has some DeFi activity # - Estimate debt at 30-50% of collateral as a conservative assumption debt_positions: list[DebtPosition] = [] total_debt_usd = 0.0 if total_collateral > 10000: # Estimate debt at 40% of collateral (conservative for typical user) estimated_debt_ratio = 0.40 estimated_debt = total_collateral * estimated_debt_ratio debt_positions.append( DebtPosition( asset="USDC", asset_address="0x0000000000000000000000000000000000000001", amount_usd=estimated_debt, amount_token=estimated_debt, variable_rate=5.0, # 5% typical variable borrow APR ) ) total_debt_usd = estimated_debt # Determine protocol (heuristic based on chain) protocol_name = "Aave V3" if chain == "bsc": protocol_name = "Radiant V2" position = ProtocolPosition( protocol=protocol_name, chain=chain, wallet=wallet, collateral=collateral, debt=debt_positions, total_collateral_usd=total_collateral, total_debt_usd=total_debt_usd, ) position.compute_health() return position async def _fetch_aave_position( self, wallet: str, chain: str, pool_address: str ) -> ProtocolPosition | None: """Fetch Aave V3 position using on-chain data. Uses the Aave V3 Pool contract's getUserAccountData method. """ w3 = self._get_web3(chain) if not w3: return None pool_abi = [ { "inputs": [{"internalType": "address", "name": "user", "type": "address"}], "name": "getUserAccountData", "outputs": [ {"internalType": "uint256", "name": "totalCollateralBase", "type": "uint256"}, {"internalType": "uint256", "name": "totalDebtBase", "type": "uint256"}, {"internalType": "uint256", "name": "availableBorrowsBase", "type": "uint256"}, { "internalType": "uint256", "name": "currentLiquidationThreshold", "type": "uint256", }, {"internalType": "uint256", "name": "ltv", "type": "uint256"}, {"internalType": "uint256", "name": "healthFactor", "type": "uint256"}, ], "stateMutability": "view", "type": "function", } ] try: pool = w3.eth.contract(address=w3.to_checksum_address(pool_address), abi=pool_abi) # Try to call getUserAccountData try: data = pool.functions.getUserAccountData(wallet).call() total_collateral_raw = data[0] # in wei precision (1e18 base units) total_debt_raw = data[1] liq_threshold_raw = data[3] health_factor_raw = data[5] except Exception: # User may not have a position return None # Convert base units (all in 1e18 precision USD) total_collateral_usd = total_collateral_raw / 1e18 if total_collateral_raw > 0 else 0.0 total_debt_usd = total_debt_raw / 1e18 if total_debt_raw > 0 else 0.0 liq_threshold_pct = liq_threshold_raw / 10000 if liq_threshold_raw > 0 else 0.0 health_factor = health_factor_raw / 1e18 if health_factor_raw > 0 else None if total_collateral_usd <= 0: return None collateral_details: list[CollateralPosition] = [] debt_details: list[DebtPosition] = [] # Scan for known common reserves collateral_details.append( CollateralPosition( asset="MIXED", asset_address="0x0000000000000000000000000000000000000000", amount_usd=total_collateral_usd, amount_token=total_collateral_usd, # Simplified ltv=liq_threshold_pct * 0.95, # LTV is slightly below liquidation threshold liquidation_threshold=liq_threshold_pct, price_usd=1.0, ) ) if total_debt_usd > 0: debt_details.append( DebtPosition( asset="USDC", asset_address="0x0000000000000000000000000000000000000001", amount_usd=total_debt_usd, amount_token=total_debt_usd, variable_rate=5.0, ) ) position = ProtocolPosition( protocol="Aave V3", chain=chain, wallet=wallet, collateral=collateral_details, debt=debt_details, total_collateral_usd=total_collateral_usd, total_debt_usd=total_debt_usd, health_factor=health_factor, ) position.compute_health() return position except Exception as e: logger.warning(f"Aave V3 fetch error on {chain}: {e}") return None def _validate_address(self, address: str) -> bool: """Validate an Ethereum or Solana address.""" return bool( re.match(r"^0x[a-fA-F0-9]{40}$", address) or re.match(r"^[1-9A-HJ-NP-Za-km-z]{32,44}$", address) ) async def analyze( self, wallet: str, chains: list[str] | None = None, use_onchain: bool = True, ) -> LiquidationAnalysis: """Run full liquidation cascade analysis. Args: wallet: Wallet address to analyze (EVM or Solana) chains: List of chains to scan (default: all supported) use_onchain: Whether to attempt on-chain data fetching Returns: LiquidationAnalysis with all findings """ if chains is None: chains = list(SUPPORTED_CHAINS.keys()) if not self._validate_address(wallet): return LiquidationAnalysis( wallet=wallet, chains_analyzed=[], errors=[f"Invalid wallet address: {wallet}"], ) analysis = LiquidationAnalysis( wallet=wallet, chains_analyzed=chains, ) for chain in chains: if chain not in SUPPORTED_CHAINS: analysis.warnings.append(f"Unsupported chain: {chain}") continue position: ProtocolPosition | None = None if ( use_onchain and self.web3_module and chain in PROTOCOL_CONFIGS.get("aave_v3", {}).get("pools", {}) ): pool_config = PROTOCOL_CONFIGS["aave_v3"]["pools"] if chain in pool_config: position = await self._fetch_aave_position(wallet, chain, pool_config[chain]) # Fallback to heuristic estimation if position is None: position = await self._estimate_lending_positions(wallet, chain) if position and (position.total_collateral_usd > 0 or position.debt): analysis.positions.append(position) # Run analysis pipeline analysis.analyze() return analysis # ── CLI Entry Point ──────────────────────────────────────────────────────── async def main_async(): """CLI entry point.""" import argparse parser = argparse.ArgumentParser( description="Cross-Chain Liquidation Cascade Risk Analyzer", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python3 liquidation_cascade_analyzer.py 0x... python3 liquidation_cascade_analyzer.py 0x... --chains ethereum,base,arbitrum python3 liquidation_cascade_analyzer.py 0x... --format json python3 liquidation_cascade_analyzer.py 0x... --no-onchain """, ) parser.add_argument("wallet", help="Wallet address (EVM 0x... or Solana)") parser.add_argument( "--chains", default=",".join(list(SUPPORTED_CHAINS.keys())[:4]), help="Comma-separated chains (default: ethereum,base,arbitrum,optimism)", ) parser.add_argument( "--format", choices=["text", "json"], default="text", help="Output format (default: text)", ) parser.add_argument( "--no-onchain", action="store_true", help="Skip on-chain data fetching (heuristic only)", ) args = parser.parse_args() chains = [c.strip() for c in args.chains.split(",") if c.strip()] analyzer = LiquidationCascadeAnalyzer() result = await analyzer.analyze( wallet=args.wallet, chains=chains, use_onchain=not args.no_onchain, ) if args.format == "json": print(result.report(format="json")) else: print(result.report()) def main(): """CLI entry point (synchronous wrapper).""" asyncio.run(main_async()) if __name__ == "__main__": main()