crypto-defi-trading

> **Skill:** Crypto Defi Trading | **Domain:** trading | **Category:** asset-class | **Level:** advanced > **Tags:** `trading`, `asset-class`, `crypto`, `defi`, `dex`, `mev`, `yield-farming`, `bitcoin` --- ## DEX Analysis Engine # DEX Analysis Engine ## Overview Complete decentralized exchange analysis covering Uniswap V2/V3, SushiSwap, Curve, and other AMM protocols. Analyzes pool states, liquidity distributions, price impact, and optimal routing across DEXes. ## Architecture ``` ┌───────────────────────────────────────────────────────────┐ │ DEX Analysis Engine │ ├──────────────┬──────────────┬──────────────┬──────────────┤ │ Pool State │ Liquidity │ Price Impact │ Cross-DEX │ │ Analyzer │ Distribution │ Calculator │ Router │ └──────────────┴──────────────┴──────────────┴──────────────┘ ``` ```python import numpy as np import pandas as pd from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple from datetime import datetime, timezone import math # ═════════════════════════════════════════════════════════════ # CORE DATA TYPES # ═════════════════════════════════════════════════════════════ @dataclass class Token: """Represents an ERC-20 token.""" address: str symbol: str decimals: int = 18 name: str = "" def format_amount(self, raw_amount: int) -> float: """Convert raw token amount to human-readable.""" return raw_amount / (10 ** self.decimals) def to_raw(self, amount: float) -> int: """Convert human-readable amount to raw.""" return int(amount * (10 ** self.decimals)) @dataclass class PoolState: """State of an AMM liquidity pool.""" pool_address: str token_0: Token token_1: Token reserve_0: float reserve_1: float fee_tier: float # e.g., 0.003 for 0.3% total_liquidity: float price: float # token_1 per token_0 volume_24h: float = 0.0 fee_revenue_24h: float = 0.0 tvl_usd: float = 0.0 tick_current: Optional[int] = None # Uniswap V3 sqrt_price_x96: Optional[int] = None # Uniswap V3 @property def fee_apr(self) -> float: """Annualized fee APR based on 24h volume.""" if self.tvl_usd == 0: return 0.0 daily_fee_rate = self.fee_revenue_24h / self.tvl_usd return daily_fee_rate * 365 * 100 @property def volume_to_tvl(self) -> float: """Volume/TVL ratio — higher = more capital efficient.""" if self.tvl_usd == 0: return 0.0 return self.volume_24h / self.tvl_usd @dataclass class LiquidityPosition: """A liquidity provider's position.""" pool_address: str owner: str liquidity: float token_0_amount: float token_1_amount: float lower_tick: Optional[int] = None # V3 range upper_tick: Optional[int] = None # V3 range fees_earned_0: float = 0.0 fees_earned_1: float = 0.0 opened_at: Optional[datetime] = None @property def is_in_range(self) -> bool: """Check if a V3 position is currently in range (needs current tick).""" if self.lower_tick is None or self.upper_tick is None: return True # V2 positions are always in range # Caller must check against current tick return True # ═════════════════════════════════════════════════════════════ # UNISWAP V2 ANALYZER # ═════════════════════════════════════════════════════════════ class UniswapV2Analyzer: """ Uniswap V2 constant product AMM analyzer. Core formula: x * y = k Price: p = y / x Output amount: dy = (y * dx * (1 - fee)) / (x + dx * (1 - fee)) """ @staticmethod def get_price(reserve_0: float, reserve_1: float) -> float: """Calculate spot price (token1 per token0).""" if reserve_0 == 0: return 0.0 return reserve_1 / reserve_0 @staticmethod def get_output_amount( amount_in: float, reserve_in: float, reserve_out: float, fee: float = 0.003, ) -> float: """ Calculate output amount for a swap. Args: amount_in: Amount of input token reserve_in: Reserve of input token reserve_out: Reserve of output token fee: Fee tier (e.g., 0.003 for 0.3%) """ if reserve_in == 0 or reserve_out == 0: return 0.0 amount_in_with_fee = amount_in * (1 - fee) numerator = amount_in_with_fee * reserve_out denominator = reserve_in + amount_in_with_fee return numerator / denominator @staticmethod def get_price_impact( amount_in: float, reserve_in: float, reserve_out: float, fee: float = 0.003, ) -> float: """ Calculate price impact of a trade as a percentage. Returns: Price impact as a decimal (e.g., 0.02 = 2% impact) """ if reserve_in == 0 or reserve_out == 0: return 1.0 spot_price = reserve_out / reserve_in output = UniswapV2Analyzer.get_output_amount( amount_in, reserve_in, reserve_out, fee ) if amount_in == 0: return 0.0 exec_price = output / amount_in impact = 1 - (exec_price / spot_price) return abs(impact) @staticmethod def get_k(reserve_0: float, reserve_1: float) -> float: """Calculate the constant product k.""" return reserve_0 * reserve_1 @staticmethod def optimal_liquidity( amount_0: float, reserve_0: float, reserve_1: float, ) -> Tuple[float, float]: """ Calculate optimal token amounts for adding liquidity. Given an amount of token0, returns the required amount of token1 to maintain the pool ratio. """ if reserve_0 == 0: return amount_0, 0.0 amount_1 = amount_0 * reserve_1 / reserve_0 return amount_0, amount_1 @staticmethod def lp_share( liquidity_added: float, total_liquidity: float, ) -> float: """Calculate LP share percentage.""" total = total_liquidity + liquidity_added if total == 0: return 0.0 return liquidity_added / total # ═════════════════════════════════════════════════════════════ # UNISWAP V3 CONCENTRATED LIQUIDITY ANALYZER # ═════════════════════════════════════════════════════════════ class UniswapV3Analyzer: """ Uniswap V3 concentrated liquidity analyzer. V3 uses ticks and concentrated positions. Liquidity is provided within price ranges instead of across the full curve. """ TICK_BASE = 1.0001 MIN_TICK = -887272 MAX_TICK = 887272 Q96 = 2 ** 96 @staticmethod def tick_to_price(tick: int) -> float: """Convert a tick to a price.""" return UniswapV3Analyzer.TICK_BASE ** tick @staticmethod def price_to_tick(price: float) -> int: """Convert a price to the nearest tick.""" if price <= 0: return UniswapV3Analyzer.MIN_TICK return int(math.log(price) / math.log(UniswapV3Analyzer.TICK_BASE)) @staticmethod def sqrt_price_x96_to_price(sqrt_price_x96: int, decimals_0: int = 18, decimals_1: int = 18) -> float: """Convert sqrtPriceX96 to human-readable price.""" price = (sqrt_price_x96 / UniswapV3Analyzer.Q96) ** 2 return price * (10 ** (decimals_0 - decimals_1)) @staticmethod def liquidity_for_amounts( sqrt_price_current: float, sqrt_price_lower: float, sqrt_price_upper: float, amount_0: float, amount_1: float, ) -> float: """ Calculate liquidity for given token amounts and price range. Based on the Uniswap V3 whitepaper formulas. """ if sqrt_price_current <= sqrt_price_lower: # Below range — all in token0 if amount_0 == 0: return 0.0 return amount_0 * sqrt_price_lower * sqrt_price_upper / (sqrt_price_upper - sqrt_price_lower) elif sqrt_price_current >= sqrt_price_upper: # Above range — all in token1 if amount_1 == 0: return 0.0 return amount_1 / (sqrt_price_upper - sqrt_price_lower) else: # In range — need both tokens liq_0 = amount_0 * sqrt_price_current * sqrt_price_upper / (sqrt_price_upper - sqrt_price_current) liq_1 = amount_1 / (sqrt_price_current - sqrt_price_lower) return min(liq_0, liq_1) @staticmethod def amounts_for_liquidity( liquidity: float, sqrt_price_current: float, sqrt_price_lower: float, sqrt_price_upper: float, ) -> Tuple[float, float]: """Calculate token amounts for a given liquidity and price range.""" if sqrt_price_current <= sqrt_price_lower: amount_0 = liquidity * (sqrt_price_upper - sqrt_price_lower) / (sqrt_price_lower * sqrt_price_upper) amount_1 = 0.0 elif sqrt_price_current >= sqrt_price_upper: amount_0 = 0.0 amount_1 = liquidity * (sqrt_price_upper - sqrt_price_lower) else: amount_0 = liquidity * (sqrt_price_upper - sqrt_price_current) / (sqrt_price_current * sqrt_price_upper) amount_1 = liquidity * (sqrt_price_current - sqrt_price_lower) return amount_0, amount_1 @staticmethod def fee_growth_in_range( fee_growth_global_0: float, fee_growth_global_1: float, fee_growth_outside_lower_0: float, fee_growth_outside_lower_1: float, fee_growth_outside_upper_0: float, fee_growth_outside_upper_1: float, tick_current: int, tick_lower: int, tick_upper: int, ) -> Tuple[float, float]: """Calculate accumulated fees within a position's range.""" if tick_current >= tick_lower: fee_below_0 = fee_growth_outside_lower_0 fee_below_1 = fee_growth_outside_lower_1 else: fee_below_0 = fee_growth_global_0 - fee_growth_outside_lower_0 fee_below_1 = fee_growth_global_1 - fee_growth_outside_lower_1 if tick_current < tick_upper: fee_above_0 = fee_growth_outside_upper_0 fee_above_1 = fee_growth_outside_upper_1 else: fee_above_0 = fee_growth_global_0 - fee_growth_outside_upper_0 fee_above_1 = fee_growth_global_1 - fee_growth_outside_upper_1 fee_in_range_0 = fee_growth_global_0 - fee_below_0 - fee_above_0 fee_in_range_1 = fee_growth_global_1 - fee_below_1 - fee_above_1 return fee_in_range_0, fee_in_range_1 @staticmethod def capital_efficiency( tick_lower: int, tick_upper: int, ) -> float: """ Calculate capital efficiency multiplier vs V2 full range. Narrower ranges = higher efficiency but more IL risk. """ price_lower = UniswapV3Analyzer.tick_to_price(tick_lower) price_upper = UniswapV3Analyzer.tick_to_price(tick_upper) if price_lower <= 0 or price_upper <= price_lower: return 1.0 sqrt_lower = math.sqrt(price_lower) sqrt_upper = math.sqrt(price_upper) # Full range efficiency relative to concentrated position return 1.0 / (1.0 - sqrt_lower / sqrt_upper) ``` --- ## Impermanent Loss Calculator # Impermanent Loss Calculator ```python class ImpermanentLossCalculator: """ Complete impermanent loss analysis for AMM liquidity provision. Covers: - Standard IL formula for V2 constant product AMMs - V3 concentrated liquidity IL - IL with fee compensation - Break-even analysis - Multi-asset IL - IL hedging strategies """ @staticmethod def v2_impermanent_loss(price_ratio: float) -> float: """ Calculate impermanent loss for Uniswap V2 (constant product). Args: price_ratio: Current price / initial price (e.g., 1.5 = 50% increase) Returns: IL as a negative decimal (e.g., -0.0566 = -5.66% loss vs HODL) """ if price_ratio <= 0: return -1.0 sqrt_ratio = math.sqrt(price_ratio) il = 2 * sqrt_ratio / (1 + price_ratio) - 1 return il @staticmethod def v2_il_percentage(price_ratio: float) -> float: """IL as a positive percentage (convenience).""" return abs(ImpermanentLossCalculator.v2_impermanent_loss(price_ratio)) * 100 @staticmethod def v3_impermanent_loss( price_initial: float, price_current: float, price_lower: float, price_upper: float, ) -> float: """ Calculate impermanent loss for Uniswap V3 concentrated position. Concentrated liquidity amplifies both fees earned AND impermanent loss. IL can be significantly worse than V2 for narrow ranges. Args: price_initial: Price when position was opened price_current: Current price price_lower: Lower bound of liquidity range price_upper: Upper bound of liquidity range """ if price_current <= 0 or price_initial <= 0: return -1.0 # Clamp prices to range p0 = max(min(price_initial, price_upper), price_lower) p1 = max(min(price_current, price_upper), price_lower) sqrt_p0 = math.sqrt(p0) sqrt_p1 = math.sqrt(p1) sqrt_pa = math.sqrt(price_lower) sqrt_pb = math.sqrt(price_upper) # Value at current price if price_current <= price_lower: # All in token0 value_current = sqrt_pb - sqrt_pa elif price_current >= price_upper: # All in token1 value_current = (sqrt_pb - sqrt_pa) * price_current / sqrt_pb else: value_current = (sqrt_p1 - sqrt_pa) * sqrt_p1 + (sqrt_pb - sqrt_p1) # Value if just held if price_initial <= price_lower: value_hodl = (sqrt_pb - sqrt_pa) * price_current / price_initial elif price_initial >= price_upper: value_hodl = (sqrt_pb - sqrt_pa) else: