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  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_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_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  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  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  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 = 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_upper:
            value_hodl = (sqrt_pb - sqrt_pa)
        else:
            value_hodl_token0 = (sqrt_pb - sqrt_p0) * price_current / price_initial
            value_hodl_token1 = (sqrt_p0 - sqrt_pa) * sqrt_p0
            value_hodl = value_hodl_token0 + value_hodl_token1
        
        if value_hodl == 0:
            return 0.0
        
        return (value_current - value_hodl) / value_hodl
    
    @staticmethod
    def il_with_fees(
        price_ratio: float,
        fee_tier: float,
        volume_to_tvl_daily: float,
        days: int,
    ) -> dict:
        """
        Calculate net IL after fee compensation.
        
        Args:
            price_ratio: Current price / initial price
            fee_tier: Pool fee tier (e.g., 0.003)
            volume_to_tvl_daily: Daily volume/TVL ratio
            days: Number of days position has been open
        
        Returns:
            Dict with il, fees_earned, net_pnl (all as percentages)
        """
        il_pct = ImpermanentLossCalculator.v2_il_percentage(price_ratio)
        daily_fee_yield = fee_tier * volume_to_tvl_daily * 100
        total_fees = daily_fee_yield * days
        net_pnl = total_fees - il_pct
        
        return {
            "impermanent_loss_pct": round(il_pct, 4),
            "fees_earned_pct": round(total_fees, 4),
            "net_pnl_pct": round(net_pnl, 4),
            "days_to_breakeven": round(il_pct / daily_fee_yield, 1) if daily_fee_yield > 0 else float("inf"),
            "daily_fee_yield_pct": round(daily_fee_yield, 4),
            "annualized_fee_yield_pct": round(daily_fee_yield * 365, 2),
            "compensated": net_pnl >= 0,
        }
    
    @staticmethod
    def il_table(price_changes: List[float] = None) -> pd.DataFrame:
        """
        Generate an IL reference table for common price changes.
        
        Returns DataFrame with columns: price_change_pct, price_ratio, il_pct
        """
        if price_changes is None:
            price_changes = [-90, -80, -70, -60, -50, -40, -30, -25, -20, -15, -10, -5,
                             0, 5, 10, 15, 20, 25, 30, 40, 50, 60, 70, 80, 90, 100,
                             150, 200, 300, 400, 500]
        rows = []
        for pct in price_changes:
            ratio = 1 + pct / 100
            if ratio  float:
        """
        Calculate the daily volume needed to offset IL with fees.
        
        Returns required daily volume in USD.
        """
        il_pct = ImpermanentLossCalculator.v2_il_percentage(price_ratio) / 100
        il_usd = il_pct * tvl
        if days == 0 or fee_tier == 0:
            return float("inf")
        required_daily_fees = il_usd / days
        required_daily_volume = required_daily_fees / fee_tier
        return required_daily_volume

---

On-Chain Analytics

On-Chain Analytics

class OnChainAnalytics:
    """
    On-chain data analysis for trading intelligence.
    
    Tracks:
    - Exchange netflow (bullish/bearish indicator)
    - Whale activity and accumulation
    - Network health metrics
    - Active address trends
    - Token holder distribution
    - Smart money flows
    """
    
    @staticmethod
    def exchange_netflow(
        inflow_usd: float,
        outflow_usd: float,
    ) -> dict:
        """
        Analyze exchange netflow.
        
        Positive netflow (more inflow) = bearish (coins moving to exchange to sell)
        Negative netflow (more outflow) = bullish (coins leaving exchange = accumulation)
        
        Args:
            inflow_usd: Total USD value flowing INTO exchanges
            outflow_usd: Total USD value flowing OUT of exchanges
        """
        netflow = inflow_usd - outflow_usd
        total_flow = inflow_usd + outflow_usd
        
        if total_flow == 0:
            bias = "neutral"
            strength = 0.0
        else:
            ratio = netflow / total_flow
            if ratio > 0.1:
                bias = "strongly_bearish"
                strength = min(abs(ratio) * 5, 1.0)
            elif ratio > 0.03:
                bias = "bearish"
                strength = min(abs(ratio) * 5, 1.0)
            elif ratio  0 else 0,
            "bias": bias,
            "strength": round(strength, 2),
            "interpretation": (
                "Coins flowing TO exchanges — sell pressure likely"
                if netflow > 0
                else "Coins flowing FROM exchanges — accumulation signal"
            ),
        }
    
    @staticmethod
    def

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [mahmoud20138](https://github.com/mahmoud20138)
- **Source:** [mahmoud20138/Tradecraft](https://github.com/mahmoud20138/Tradecraft)
- **License:** MIT

Install and usage instructions live in the source repository linked above.