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       │
└──────────────┴──────────────┴──────────────┴──────────────┘

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)