Dex Pool Analysis

#!/usr/bin/env python3
"""Analyze all DEX pools for a Solana token and rank them by execution quality.

Fetches pool data from DexScreener (free, no API key required), computes
health scores, flags risks, and ranks pools for optimal trade execution.

Usage:
    python scripts/analyze_pools.py <TOKEN_MINT>
    python scripts/analyze_pools.py --demo

Dependencies:
    uv pip install httpx

Environment Variables:
    None required (DexScreener API is free and keyless).
"""

import argparse
import json
import math
import sys
import time
from dataclasses import dataclass, field
from typing import Optional

try:
    import httpx
except ImportError:
    print("Missing dependency. Install with: uv pip install httpx")
    sys.exit(1)


# ── Configuration ───────────────────────────────────────────────────

DEXSCREENER_BASE = "https://api.dexscreener.com"
REQUEST_TIMEOUT = 15.0

PROGRAM_IDS = {
    "raydium_v4": "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8",
    "raydium_clmm": "CAMMCzo5YL8w4VFF8KVHrK22GGUsp5VTaW7grrKgrWqK",
    "orca_whirlpool": "whirLbMiicVdio4qvUfM5KAg6Ct8VwpYzGff3uctyCc",
    "meteora_dlmm": "LBUZKhRxPF3XUpBCjp4YzTKgLccjZhTSDM9YuVaPwxo",
    "meteora_dynamic": "Eo7WjKq67rjJQSZxS6z3YkapzY3eMj6Xy8X5EQVn5UaB",
    "pumpswap": "PSwapMdSai8tjrEXcxFeQth87xC4rRsa4VA5mhGhXkP",
}

# Fee rates by DEX (used when DexScreener doesn't report fees)
DEFAULT_FEE_RATES: dict[str, float] = {
    "raydium": 0.0025,
    "raydium_clmm": 0.0025,
    "orca": 0.003,
    "meteora": 0.003,
    "pumpswap": 0.0025,
}


# ── Data Models ─────────────────────────────────────────────────────


@dataclass
class PoolInfo:
    """Parsed pool information with computed metrics."""

    pool_address: str
    dex_id: str
    dex_name: str
    pool_type: str
    base_token: str
    quote_token: str
    base_symbol: str
    quote_symbol: str
    price_usd: float
    tvl_usd: float
    volume_24h: float
    volume_6h: float
    volume_1h: float
    price_change_24h: float
    pool_age_hours: float
    fee_rate: float
    txn_count_24h: int
    buys_24h: int
    sells_24h: int
    url: str

    # Computed metrics
    volume_efficiency: float = 0.0
    fee_apr_estimate: float = 0.0
    health_score: float = 0.0
    execution_score: float = 0.0
    red_flags: list[str] = field(default_factory=list)


# ── Pool Classification ────────────────────────────────────────────


def classify_pool_type(dex_id: str) -> str:
    """Classify pool type from DexScreener dex ID.

    Args:
        dex_id: DexScreener DEX identifier string.

    Returns:
        Pool type classification string.
    """
    dex_lower = dex_id.lower()
    if "raydium" in dex_lower and "clmm" in dex_lower:
        return "clmm"
    if "raydium" in dex_lower and "cpmm" in dex_lower:
        return "cpmm"
    if "raydium" in dex_lower:
        return "constant_product"
    if "orca" in dex_lower or "whirlpool" in dex_lower:
        return "whirlpool"
    if "meteora" in dex_lower and "dlmm" in dex_lower:
        return "dlmm"
    if "meteora" in dex_lower:
        return "dynamic"
    if "pump" in dex_lower:
        return "constant_product"
    return "unknown"


def estimate_fee_rate(dex_id: str) -> float:
    """Estimate the swap fee rate based on DEX identifier.

    Args:
        dex_id: DexScreener DEX identifier string.

    Returns:
        Estimated fee rate as a decimal (e.g., 0.0025 for 0.25%).
    """
    dex_lower = dex_id.lower()
    for key, rate in DEFAULT_FEE_RATES.items():
        if key in dex_lower:
            return rate
    return 0.003  # Conservative default


# ── Health Scoring ──────────────────────────────────────────────────


def compute_health_score(pool: PoolInfo) -> float:
    """Compute a 0-100 health score for a pool.

    Components (each 0-20 points):
    - TVL adequacy: logarithmic scale, peaks at $1M+
    - Volume efficiency: sweet spot 0.5-3.0 V/TVL ratio
    - Pool maturity: older pools score higher (max at 72h)
    - Transaction diversity: buy/sell balance
    - TVL stability: penalize negative price change as proxy

    Args:
        pool: PoolInfo with all metrics populated.

    Returns:
        Health score from 0 to 100.
    """
    # TVL score (0-20): log scale, peaks at $1M+
    if pool.tvl_usd <= 0:
        tvl_score = 0.0
    else:
        tvl_score = min(20.0, max(0.0, 5.0 * math.log10(pool.tvl_usd) - 10.0))

    # Volume score (0-20): V/TVL sweet spot 0.5-3.0
    v_tvl = pool.volume_efficiency
    if v_tvl < 5.0:
        volume_score = min(20.0, max(0.0, v_tvl * 10.0))
    else:
        volume_score = max(0.0, 20.0 - (v_tvl - 5.0) * 4.0)

    # Age score (0-20): max at 72 hours
    age_score = min(20.0, pool.pool_age_hours / 72.0 * 20.0)

    # Transaction diversity score (0-20): balanced buys/sells
    total_txns = pool.buys_24h + pool.sells_24h
    if total_txns > 0:
        buy_ratio = pool.buys_24h / total_txns
        # Perfect balance = 0.5, score drops for imbalance
        balance = 1.0 - abs(buy_ratio - 0.5) * 2.0
        txn_score = min(20.0, balance * 15.0 + min(5.0, total_txns / 200.0 * 5.0))
    else:
        txn_score = 0.0

    # Stability score (0-20): penalize extreme price moves
    abs_change = abs(pool.price_change_24h)
    stability_score = max(0.0, 20.0 - abs_change * 0.4)

    return tvl_score + volume_score + age_score + txn_score + stability_score


def check_red_flags(pool: PoolInfo) -> list[str]:
    """Check pool for warning signs and risk indicators.

    Args:
        pool: PoolInfo with all metrics populated.

    Returns:
        List of red flag description strings.
    """
    flags: list[str] = []

    if pool.tvl_usd < 1_000:
        flags.append("CRITICAL: TVL below $1,000 — extreme slippage risk")
    elif pool.tvl_usd < 10_000:
        flags.append("WARNING: TVL below $10,000 — high slippage on moderate trades")

    if pool.pool_age_hours < 1:
        flags.append("CRITICAL: Pool is less than 1 hour old — sniper risk")
    elif pool.pool_age_hours < 24:
        flags.append("WARNING: Pool is less than 24 hours old — limited history")

    if pool.volume_efficiency > 10:
        flags.append("WARNING: V/TVL > 10 — possible wash trading")
    elif pool.volume_efficiency > 5:
        flags.append("NOTICE: V/TVL > 5 — unusually high turnover")

    if pool.volume_24h == 0 and pool.pool_age_hours > 6:
        flags.append("WARNING: Zero 24h volume — pool may be dead")

    total_txns = pool.buys_24h + pool.sells_24h
    if total_txns > 0:
        buy_ratio = pool.buys_24h / total_txns
        if buy_ratio > 0.9:
            flags.append("WARNING: >90% buys — possible coordinated pump")
        elif buy_ratio < 0.1:
            flags.append("WARNING: >90% sells — possible dump in progress")

    if pool.price_change_24h < -50:
        flags.append("CRITICAL: Price dropped >50% in 24h")
    elif pool.price_change_24h < -20:
        flags.append("WARNING: Price dropped >20% in 24h")

    return flags


# ── Execution Ranking ───────────────────────────────────────────────


def estimate_slippage(pool: PoolInfo, trade_size_usd: float) -> float:
    """Estimate slippage percentage for a trade of the given size.

    Concentrated liquidity pools have better capital efficiency, resulting
    in lower slippage per TVL dollar.

    Args:
        pool: PoolInfo with pool type and TVL.
        trade_size_usd: Size of the trade in USD.

    Returns:
        Estimated slippage as a percentage (e.g., 1.0 = 1%).
    """
    if pool.tvl_usd <= 0:
        return 100.0

    base_slippage = (trade_size_usd / pool.tvl_usd) * 100.0

    # Concentrated liquidity pools have better capital efficiency
    if pool.pool_type in ("clmm", "whirlpool", "dlmm"):
        return base_slippage * 0.3  # ~3x more efficient
    return base_slippage


def rank_pools_for_execution(
    pools: list[PoolInfo], trade_size_usd: float
) -> list[PoolInfo]:
    """Rank pools by execution quality for a given trade size.

    Scoring weights:
    - 50%: Execution cost (fee + estimated slippage) — lower is better
    - 30%: Health score — higher is better
    - 20%: Fee rate — lower is better

    Args:
        pools: List of PoolInfo objects to rank.
        trade_size_usd: Trade size in USD for slippage estimation.

    Returns:
        Pools sorted by execution score, best first.
    """
    for pool in pools:
        slippage_pct = estimate_slippage(pool, trade_size_usd)
        fee_cost = pool.fee_rate * 100.0  # As percentage
        total_cost = slippage_pct + fee_cost

        # Lower cost is better — invert for scoring
        cost_score = 1.0 / max(total_cost, 0.001) * 10.0

        # Normalize health (0-100 → 0-10)
        health_norm = pool.health_score / 10.0

        # Lower fee is better — invert for scoring
        fee_score = 1.0 / max(pool.fee_rate, 0.0001)

        pool.execution_score = cost_score * 0.5 + health_norm * 0.3 + fee_score * 0.0002

    pools.sort(key=lambda p: p.execution_score, reverse=True)
    return pools


# ── Data Fetching ───────────────────────────────────────────────────


def fetch_pools(token_mint: str) -> list[PoolInfo]:
    """Fetch all Solana pools for a token from DexScreener.

    Args:
        token_mint: Solana token mint address.

    Returns:
        List of PoolInfo objects.

    Raises:
        httpx.HTTPStatusError: On non-2xx response from DexScreener.
    """
    url = f"{DEXSCREENER_BASE}/tokens/v1/solana/{token_mint}"
    with httpx.Client(timeout=REQUEST_TIMEOUT) as client:
        resp = client.get(url)
        resp.raise_for_status()
        raw_pools = resp.json()

    if not isinstance(raw_pools, list):
        print(f"Unexpected response format: {type(raw_pools)}")
        return []

    pools: list[PoolInfo] = []
    now_ms = int(time.time() * 1000)

    for p in raw_pools:
        # Skip non-Solana pools
        if p.get("chainId", "") != "solana":
            continue

        created_at = p.get("pairCreatedAt", now_ms)
        age_hours = max(0, (now_ms - created_at) / 3_600_000)

        liquidity = p.get("liquidity") or {}
        volume = p.get("volume") or {}
        price_change = p.get("priceChange") or {}
        txns = p.get("txns") or {}
        txns_24h = txns.get("h24") or {}

        dex_id = p.get("dexId", "unknown")

        pool_info = PoolInfo(
            pool_address=p.get("pairAddress", ""),
            dex_id=dex_id,
            dex_name=p.get("dexId", "unknown"),
            pool_type=classify_pool_type(dex_id),
            base_token=p.get("baseToken", {}).get("address", ""),
            quote_token=p.get("quoteToken", {}).get("address", ""),
            base_symbol=p.get("baseToken", {}).get("symbol", "???"),
            quote_symbol=p.get("quoteToken", {}).get("symbol", "???"),
            price_usd=float(p.get("priceUsd", 0) or 0),
            tvl_usd=float(liquidity.get("usd", 0) or 0),
            volume_24h=float(volume.get("h24", 0) or 0),
            volume_6h=float(volume.get("h6", 0) or 0),
            volume_1h=float(volume.get("h1", 0) or 0),
            price_change_24h=float(price_change.get("h24", 0) or 0),
            pool_age_hours=age_hours,
            fee_rate=estimate_fee_rate(dex_id),
            txn_count_24h=int(txns_24h.get("buys", 0) or 0)
            + int(txns_24h.get("sells", 0) or 0),
            buys_24h=int(txns_24h.get("buys", 0) or 0),
            sells_24h=int(txns_24h.get("sells", 0) or 0),
            url=p.get("url", ""),
        )

        # Compute derived metrics
        pool_info.volume_efficiency = (
            pool_info.volume_24h / pool_info.tvl_usd if pool_info.tvl_usd > 0 else 0.0
        )
        pool_info.fee_apr_estimate = (
            pool_info.volume_efficiency * pool_info.fee_rate * 365 * 100
        )
        pool_info.health_score = compute_health_score(pool_info)
        pool_info.red_flags = check_red_flags(pool_info)

        pools.append(pool_info)

    return pools


# ── Demo Data ───────────────────────────────────────────────────────


def generate_demo_pools() -> list[PoolInfo]:
    """Generate synthetic multi-pool data for demo mode.

    Creates a realistic scenario: a token with pools across multiple DEXes
    at different stages of maturity and health.

    Returns:
        List of synthetic PoolInfo objects.
    """
    demo_pools_raw = [
        {
            "name": "Raydium V4 (Main)",
            "pool_address": "DemoPool1111111111111111111111111111111111111",
            "dex_id": "raydium",
            "pool_type": "constant_product",
            "tvl_usd": 250_000,
            "volume_24h": 500_000,
            "volume_6h": 120_000,
            "volume_1h": 18_000,
            "price_change_24h": 5.2,
            "pool_age_hours": 720,
            "fee_rate": 0.0025,
            "buys_24h": 1200,
            "sells_24h": 980,
        },
        {
            "name": "Orca Whirlpool (CL)",
            "pool_address": "DemoPool2222222222222222222222222222222222222",
            "dex_id": "orca",
            "pool_type": "whirlpool",
            "tvl_usd": 180_000,
            "volume_24h": 420_000,
            "volume_6h": 100_000,
            "volume_1h": 15_000,
            "price_change_24h": 5.1,
            "pool_age_hours": 480,
            "fee_rate": 0.003,
            "buys_24h": 800,
            "sells_24h": 750,
        },
        {
            "name": "Meteora DLMM",
            "pool_address": "DemoPool3333333333333333333333333333333333333",
            "dex_id": "meteora_dlmm",
            "pool_type": "dlmm",
            "tvl_usd": 85_000,
            "volume_24h": 200_000,
            "volume_6h": 55_000,
            "volume_1h": 8_000,
            "price_change_24h": 4.8,
            "pool_age_hours": 168,
            "fee_rate": 0.003,
            "buys_24h": 400,
            "sells_24h": 380,
        },
        {
            "name": "Raydium CLMM",
            "pool_address": "DemoPool4444444444444444444444444444444444444",
            "dex_id": "raydium_clmm",
            "pool_type": "clmm",
            "tvl_usd": 60_000,
            "volume_24h": 150_000,
            "volume_6h": 35_000,
            "volume_1h": 5_000,
            "price_change_24h": 5.5,
            "pool_age_hours": 96,
            "fee_rate": 0.0025,
            "buys_24h": 300,
            "sells_24h": 280,
        },
        {
            "name": "PumpSwap (New)",
            "pool_address": "DemoPool5555555555555555555555555555555555555",
            "dex_id": "pumpswap",
            "pool_type": "constant_product",
            "tvl_usd": 12_000,
            "volume_24h": 180_000,
            "volume_6h": 80_000,
            "volume_1h": 25_000,
            "price_change_24h": 45.0,
            "pool_age_hours": 3,
            "fee_rate": 0.0025,
            "buys_24h": 2500,
            "sells_24h": 500,
        },
        {
            "name": "Dead Pool (Raydium V4)",
            "pool_address": "DemoPool6666666666666666666666666666666666666",
            "dex_id": "raydium",
            "pool_type": "constant_product",
            "tvl_usd": 500,
            "volume_24h": 0,
            "volume_6h": 0,
            "volume_1h": 0,
            "price_change_24h": -85.0,
            "pool_age_hours": 2400,
            "fee_rate": 0.0025,
            "buys_24h": 0,
            "sells_24h": 0,
        },
    ]

    pools: list[PoolInfo] = []
    for d in demo_pools_raw:
        pool = PoolInfo(
            pool_address=d["pool_address"],
            dex_id=d["dex_id"],
            dex_name=d["name"],
            pool_type=d["pool_type"],
            base_token="DemoMint1111111111111111111111111111111111111",
            quote_token="So11111111111111111111111111111111111111112",
            base_symbol="DEMO",
            quote_symbol="SOL",
            price_usd=0.00042,
            tvl_usd=d["tvl_usd"],
            volume_24h=d["volume_24h"],
            volume_6h=d["volume_6h"],
            volume_1h=d["volume_1h"],
            price_change_24h=d["price_change_24h"],
            pool_age_hours=d["pool_age_hours"],
            fee_rate=d["fee_rate"],
            txn_count_24h=d["buys_24h"] + d["sells_24h"],
            buys_24h=d["buys_24h"],
            sells_24h=d["sells_24h"],
            url=f"https://dexscreener.com/solana/{d['pool_address']}",
        )

        pool.volume_efficiency = (
            pool.volume_24h / pool.tvl_usd if pool.tvl_usd > 0 else 0.0
        )
        pool.fee_apr_estimate = pool.volume_efficiency * pool.fee_rate * 365 * 100
        pool.health_score = compute_health_score(pool)
        pool.red_flags = check_red_flags(pool)
        pools.append(pool)

    return pools


# ── Display ─────────────────────────────────────────────────────────


def print_pool_summary(pool: PoolInfo, rank: int) -> None:
    """Print a formatted summary of a single pool.

    Args:
        pool: PoolInfo to display.
        rank: Display rank number.
    """
    print(f"\n{'='*70}")
    print(f"  #{rank}  {pool.dex_name} — {pool.base_symbol}/{pool.quote_symbol}")
    print(f"{'='*70}")
    print(f"  Pool Type:      {pool.pool_type}")
    print(f"  Pool Address:   {pool.pool_address[:20]}...")
    print(f"  Price (USD):    ${pool.price_usd:.8f}")
    print(f"  TVL:            ${pool.tvl_usd:,.0f}")
    print(f"  Volume (24h):   ${pool.volume_24h:,.0f}")
    print(f"  Volume (1h):    ${pool.volume_1h:,.0f}")
    print(f"  V/TVL Ratio:    {pool.volume_efficiency:.2f}")
    print(f"  Fee Rate:       {pool.fee_rate*100:.2f}%")
    print(f"  Fee APR Est:    {pool.fee_apr_estimate:.1f}%")
    print(f"  Pool Age:       {pool.pool_age_hours:.0f} hours ({pool.pool_age_hours/24:.1f} days)")
    print(f"  Txns (24h):     {pool.txn_count_24h} (buys: {pool.buys_24h}, sells: {pool.sells_24h})")
    print(f"  Price Δ (24h):  {pool.price_change_24h:+.1f}%")
    print(f"  Health Score:   {pool.health_score:.1f}/100")
    print(f"  Exec Score:     {pool.execution_score:.2f}")

    if pool.red_flags:
        print(f"\n  Red Flags:")
        for flag in pool.red_flags:
            print(f"    - {flag}")
    else:
        print(f"\n  Red Flags:      None")


def print_execution_recommendation(
    pools: list[PoolInfo], trade_size_usd: float
) -> None:
    """Print execution recommendation based on ranked pools.

    Args:
        pools: Ranked list of pools (best first).
        trade_size_usd: Trade size used for ranking.
    """
    print(f"\n{'='*70}")
    print(f"  EXECUTION ANALYSIS (trade size: ${trade_size_usd:,.0f})")
    print(f"{'='*70}")

    viable = [p for p in pools if p.tvl_usd >= trade_size_usd * 50]

    if not viable:
        print("\n  No pools have sufficient TVL for this trade size.")
        print(f"  Minimum TVL needed: ${trade_size_usd * 50:,.0f}")
        print("  Consider reducing trade size or splitting across pools.")
        return

    best = viable[0]
    slippage = estimate_slippage(best, trade_size_usd)
    total_cost = slippage + best.fee_rate * 100

    print(f"\n  Recommended Pool: {best.dex_name}")
    print(f"  Pool Type:        {best.pool_type}")
    print(f"  Est. Slippage:    {slippage:.3f}%")
    print(f"  Fee:              {best.fee_rate*100:.2f}%")
    print(f"  Total Est. Cost:  {total_cost:.3f}% (${trade_size_usd * total_cost / 100:,.2f})")

    if best.red_flags:
        print(f"\n  Caution — flags on recommended pool:")
        for flag in best.red_flags:
            print(f"    - {flag}")

    if len(viable) > 1:
        print(f"\n  Alternative pools ({len(viable)-1}):")
        for alt in viable[1:3]:
            alt_slip = estimate_slippage(alt, trade_size_usd)
            alt_cost = alt_slip + alt.fee_rate * 100
            print(f"    - {alt.dex_name}: ~{alt_cost:.3f}% total cost")


def print_report(pools: list[PoolInfo], trade_size_usd: float) -> None:
    """Print full analysis report for all pools.

    Args:
        pools: List of PoolInfo objects (will be ranked internally).
        trade_size_usd: Trade size for execution analysis.
    """
    if not pools:
        print("No pools found.")
        return

    print(f"\n{'#'*70}")
    print(f"  DEX POOL ANALYSIS REPORT")
    print(f"  Token: {pools[0].base_symbol}/{pools[0].quote_symbol}")
    print(f"  Pools found: {len(pools)}")
    print(f"{'#'*70}")

    # Rank pools for execution
    ranked = rank_pools_for_execution(pools, trade_size_usd)

    # Print each pool
    for i, pool in enumerate(ranked, 1):
        print_pool_summary(pool, i)

    # Execution recommendation
    print_execution_recommendation(ranked, trade_size_usd)

    # Summary statistics
    total_tvl = sum(p.tvl_usd for p in pools)
    total_vol = sum(p.volume_24h for p in pools)
    critical_flags = sum(
        1 for p in pools for f in p.red_flags if f.startswith("CRITICAL")
    )

    print(f"\n{'='*70}")
    print(f"  SUMMARY")
    print(f"{'='*70}")
    print(f"  Total TVL across pools:    ${total_tvl:,.0f}")
    print(f"  Total Volume (24h):        ${total_vol:,.0f}")
    print(f"  Aggregate V/TVL:           {total_vol / max(total_tvl, 1):.2f}")
    print(f"  Pools with critical flags: {critical_flags}/{len(pools)}")
    print(f"  Pool types: {', '.join(sorted(set(p.pool_type for p in pools)))}")

    print(f"\n  Note: This analysis provides information only, not financial advice.")
    print(f"  Always verify pool data on-chain before executing trades.\n")


# ── Main ────────────────────────────────────────────────────────────


def main() -> None:
    """Entry point: parse arguments and run pool analysis."""
    parser = argparse.ArgumentParser(
        description="Analyze DEX pools for a Solana token"
    )
    parser.add_argument(
        "token_mint",
        nargs="?",
        default=None,
        help="Solana token mint address to analyze",
    )
    parser.add_argument(
        "--demo",
        action="store_true",
        help="Run with synthetic demo data (no API calls)",
    )
    parser.add_argument(
        "--trade-size",
        type=float,
        default=500.0,
        help="Trade size in USD for execution analysis (default: 500)",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Output results as JSON",
    )

    args = parser.parse_args()

    if args.demo:
        print("Running in demo mode with synthetic pool data...\n")
        pools = generate_demo_pools()
    elif args.token_mint:
        print(f"Fetching pools for {args.token_mint}...")
        try:
            pools = fetch_pools(args.token_mint)
        except httpx.HTTPStatusError as e:
            print(f"API error: {e.response.status_code} — {e.response.text[:200]}")
            sys.exit(1)
        except httpx.RequestError as e:
            print(f"Network error: {e}")
            sys.exit(1)
    else:
        parser.print_help()
        print("\nExamples:")
        print("  python scripts/analyze_pools.py --demo")
        print("  python scripts/analyze_pools.py <TOKEN_MINT_ADDRESS>")
        sys.exit(1)

    if args.json:
        output = []
        for p in pools:
            output.append({
                "pool_address": p.pool_address,
                "dex": p.dex_name,
                "pool_type": p.pool_type,
                "tvl_usd": p.tvl_usd,
                "volume_24h": p.volume_24h,
                "volume_efficiency": round(p.volume_efficiency, 4),
                "fee_apr_estimate": round(p.fee_apr_estimate, 2),
                "health_score": round(p.health_score, 2),
                "red_flags": p.red_flags,
            })
        print(json.dumps(output, indent=2))
    else:
        print_report(pools, args.trade_size)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Monitor DEX pool metrics over time and detect liquidity events.

Tracks TVL changes, volume trends, and new pool creation for a Solana token.
In demo mode, simulates a pool lifecycle from creation through maturity,
including liquidity addition, removal events, and volume spikes.

Usage:
    python scripts/pool_monitor.py <TOKEN_MINT> --interval 60
    python scripts/pool_monitor.py --demo

Dependencies:
    uv pip install httpx

Environment Variables:
    None required (DexScreener API is free and keyless).
"""

import argparse
import math
import random
import sys
import time
from dataclasses import dataclass, field
from typing import Optional

try:
    import httpx
except ImportError:
    print("Missing dependency. Install with: uv pip install httpx")
    sys.exit(1)


# ── Configuration ───────────────────────────────────────────────────

DEXSCREENER_BASE = "https://api.dexscreener.com"
REQUEST_TIMEOUT = 15.0

# Alert thresholds
TVL_DROP_ALERT_PCT = -15.0       # Alert if TVL drops more than 15%
TVL_SPIKE_ALERT_PCT = 50.0       # Alert if TVL spikes more than 50%
VOLUME_SPIKE_MULT = 3.0          # Alert if volume > 3x average
NEW_POOL_ALERT = True            # Alert on new pool detection


# ── Data Models ─────────────────────────────────────────────────────


@dataclass
class PoolSnapshot:
    """A point-in-time snapshot of pool metrics."""

    timestamp: float
    pool_address: str
    dex_name: str
    tvl_usd: float
    volume_1h: float
    volume_24h: float
    price_usd: float
    buys_1h: int
    sells_1h: int


@dataclass
class PoolAlert:
    """An alert generated from pool monitoring."""

    timestamp: float
    pool_address: str
    dex_name: str
    alert_type: str  # "tvl_drop", "tvl_spike", "volume_spike", "new_pool", "dead_pool"
    severity: str    # "critical", "warning", "info"
    message: str
    details: dict = field(default_factory=dict)


@dataclass
class MonitorState:
    """Tracking state for a monitored pool."""

    pool_address: str
    dex_name: str
    snapshots: list[PoolSnapshot] = field(default_factory=list)
    alerts: list[PoolAlert] = field(default_factory=list)
    first_seen: float = 0.0
    last_tvl: float = 0.0
    avg_volume_1h: float = 0.0


# ── Alert Detection ────────────────────────────────────────────────


def detect_alerts(
    state: MonitorState, current: PoolSnapshot
) -> list[PoolAlert]:
    """Compare current snapshot against state to detect alerts.

    Checks for:
    - TVL drops exceeding threshold
    - TVL spikes exceeding threshold
    - Volume spikes vs rolling average
    - Dead pools (zero activity)

    Args:
        state: Current monitoring state for this pool.
        current: Latest pool snapshot.

    Returns:
        List of new alerts detected.
    """
    alerts: list[PoolAlert] = []

    if state.last_tvl > 0:
        tvl_change_pct = ((current.tvl_usd - state.last_tvl) / state.last_tvl) * 100

        # TVL drop alert
        if tvl_change_pct < TVL_DROP_ALERT_PCT:
            severity = "critical" if tvl_change_pct < -50 else "warning"
            alerts.append(PoolAlert(
                timestamp=current.timestamp,
                pool_address=current.pool_address,
                dex_name=current.dex_name,
                alert_type="tvl_drop",
                severity=severity,
                message=f"TVL dropped {tvl_change_pct:.1f}% "
                        f"(${state.last_tvl:,.0f} -> ${current.tvl_usd:,.0f})",
                details={
                    "previous_tvl": state.last_tvl,
                    "current_tvl": current.tvl_usd,
                    "change_pct": tvl_change_pct,
                },
            ))

        # TVL spike alert
        if tvl_change_pct > TVL_SPIKE_ALERT_PCT:
            alerts.append(PoolAlert(
                timestamp=current.timestamp,
                pool_address=current.pool_address,
                dex_name=current.dex_name,
                alert_type="tvl_spike",
                severity="info",
                message=f"TVL increased {tvl_change_pct:.1f}% "
                        f"(${state.last_tvl:,.0f} -> ${current.tvl_usd:,.0f})",
                details={
                    "previous_tvl": state.last_tvl,
                    "current_tvl": current.tvl_usd,
                    "change_pct": tvl_change_pct,
                },
            ))

    # Volume spike vs average
    if state.avg_volume_1h > 0 and current.volume_1h > 0:
        volume_mult = current.volume_1h / state.avg_volume_1h
        if volume_mult > VOLUME_SPIKE_MULT:
            alerts.append(PoolAlert(
                timestamp=current.timestamp,
                pool_address=current.pool_address,
                dex_name=current.dex_name,
                alert_type="volume_spike",
                severity="warning",
                message=f"Volume spike: {volume_mult:.1f}x average "
                        f"(${current.volume_1h:,.0f} vs avg ${state.avg_volume_1h:,.0f})",
                details={
                    "current_volume_1h": current.volume_1h,
                    "avg_volume_1h": state.avg_volume_1h,
                    "multiplier": volume_mult,
                },
            ))

    # Dead pool detection
    if (
        len(state.snapshots) >= 3
        and all(s.volume_1h == 0 for s in state.snapshots[-3:])
        and current.volume_1h == 0
    ):
        alerts.append(PoolAlert(
            timestamp=current.timestamp,
            pool_address=current.pool_address,
            dex_name=current.dex_name,
            alert_type="dead_pool",
            severity="warning",
            message="Pool has had zero volume for 3+ consecutive checks",
            details={"consecutive_zero_checks": len(state.snapshots)},
        ))

    return alerts


def update_state(state: MonitorState, snapshot: PoolSnapshot) -> None:
    """Update monitoring state with a new snapshot.

    Args:
        state: Current monitoring state to update.
        snapshot: New snapshot to incorporate.
    """
    state.snapshots.append(snapshot)
    state.last_tvl = snapshot.tvl_usd

    # Rolling average of 1h volume (last 10 snapshots)
    recent = state.snapshots[-10:]
    volumes = [s.volume_1h for s in recent if s.volume_1h > 0]
    state.avg_volume_1h = sum(volumes) / len(volumes) if volumes else 0.0


# ── Data Fetching ───────────────────────────────────────────────────


def fetch_pool_snapshots(token_mint: str) -> list[PoolSnapshot]:
    """Fetch current pool data from DexScreener and create snapshots.

    Args:
        token_mint: Solana token mint address.

    Returns:
        List of PoolSnapshot objects, one per pool.

    Raises:
        httpx.HTTPStatusError: On non-2xx response.
    """
    url = f"{DEXSCREENER_BASE}/tokens/v1/solana/{token_mint}"
    now = time.time()

    with httpx.Client(timeout=REQUEST_TIMEOUT) as client:
        resp = client.get(url)
        resp.raise_for_status()
        raw_pools = resp.json()

    if not isinstance(raw_pools, list):
        return []

    snapshots: list[PoolSnapshot] = []
    for p in raw_pools:
        if p.get("chainId", "") != "solana":
            continue

        volume = p.get("volume") or {}
        liquidity = p.get("liquidity") or {}
        txns = p.get("txns") or {}
        txns_1h = txns.get("h1") or {}

        snapshots.append(PoolSnapshot(
            timestamp=now,
            pool_address=p.get("pairAddress", ""),
            dex_name=p.get("dexId", "unknown"),
            tvl_usd=float(liquidity.get("usd", 0) or 0),
            volume_1h=float(volume.get("h1", 0) or 0),
            volume_24h=float(volume.get("h24", 0) or 0),
            price_usd=float(p.get("priceUsd", 0) or 0),
            buys_1h=int(txns_1h.get("buys", 0) or 0),
            sells_1h=int(txns_1h.get("sells", 0) or 0),
        ))

    return snapshots


# ── Demo Simulation ─────────────────────────────────────────────────


def simulate_pool_lifecycle() -> list[tuple[float, list[PoolSnapshot]]]:
    """Simulate a token's pool lifecycle for demo mode.

    Simulates 20 time steps covering:
    - T0-T2: Pool creation on Raydium V4 (PumpFun graduation)
    - T3-T5: Early trading with high volume, TVL growth
    - T6-T8: Orca Whirlpool pool created (new pool event)
    - T9-T12: Stable period, normal trading
    - T13-T15: Liquidity removal event on Raydium (rug warning)
    - T16-T18: Volume dies on Raydium, moves to Orca
    - T19: Final state

    Returns:
        List of (timestamp, snapshots) tuples representing time steps.
    """
    random.seed(42)  # Reproducible demo
    base_time = time.time()
    steps: list[tuple[float, list[PoolSnapshot]]] = []

    # Pool lifecycle parameters
    raydium_tvl_curve = [
        12_000, 18_000, 35_000, 55_000, 80_000,  # Growth phase
        95_000, 100_000, 105_000, 110_000, 108_000,  # Stable
        105_000, 100_000, 95_000, 40_000, 25_000,  # Rug event
        15_000, 8_000, 5_000, 3_000, 2_000,  # Decline
    ]

    # Orca pool appears at step 6
    orca_tvl_curve = [
        0, 0, 0, 0, 0, 0,
        50_000, 65_000, 75_000, 80_000,  # Creation + growth
        85_000, 90_000, 95_000, 100_000, 110_000,  # Absorbs volume
        120_000, 130_000, 135_000, 140_000, 145_000,  # Dominant
    ]

    for step in range(20):
        t = base_time + step * 3600  # 1 hour between steps
        snapshots: list[PoolSnapshot] = []

        # Raydium V4 pool
        ray_tvl = raydium_tvl_curve[step]
        ray_vol_base = ray_tvl * random.uniform(0.3, 1.5)
        if step >= 13:
            ray_vol_base *= 0.2  # Volume dies after rug

        snapshots.append(PoolSnapshot(
            timestamp=t,
            pool_address="RayPool1111111111111111111111111111111111111",
            dex_name="raydium",
            tvl_usd=ray_tvl * random.uniform(0.95, 1.05),
            volume_1h=ray_vol_base * random.uniform(0.02, 0.08),
            volume_24h=ray_vol_base,
            price_usd=0.00042 * (1 + step * 0.02 + random.uniform(-0.05, 0.05)),
            buys_1h=int(random.uniform(10, 80)),
            sells_1h=int(random.uniform(8, 70)),
        ))

        # Orca Whirlpool (appears at step 6)
        orca_tvl = orca_tvl_curve[step]
        if orca_tvl > 0:
            orca_vol_base = orca_tvl * random.uniform(0.5, 2.0)
            snapshots.append(PoolSnapshot(
                timestamp=t,
                pool_address="OrcaPool2222222222222222222222222222222222222",
                dex_name="orca_whirlpool",
                tvl_usd=orca_tvl * random.uniform(0.95, 1.05),
                volume_1h=orca_vol_base * random.uniform(0.03, 0.08),
                volume_24h=orca_vol_base,
                price_usd=0.00042 * (1 + step * 0.02 + random.uniform(-0.05, 0.05)),
                buys_1h=int(random.uniform(15, 90)),
                sells_1h=int(random.uniform(12, 85)),
            ))

        steps.append((t, snapshots))

    return steps


# ── Display ─────────────────────────────────────────────────────────


def format_timestamp(ts: float) -> str:
    """Format a Unix timestamp to a readable string.

    Args:
        ts: Unix timestamp.

    Returns:
        Formatted time string.
    """
    return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts))


def print_snapshot_report(
    snapshots: list[PoolSnapshot], states: dict[str, MonitorState], step: int
) -> None:
    """Print a monitoring report for the current time step.

    Args:
        snapshots: Current pool snapshots.
        states: Monitoring states keyed by pool address.
        step: Current step number.
    """
    if not snapshots:
        print(f"  Step {step}: No pool data available")
        return

    ts = snapshots[0].timestamp
    print(f"\n{'─'*60}")
    print(f"  Step {step} | {format_timestamp(ts)}")
    print(f"{'─'*60}")

    total_tvl = 0.0
    total_vol = 0.0

    for snap in snapshots:
        total_tvl += snap.tvl_usd
        total_vol += snap.volume_1h

        state = states.get(snap.pool_address)
        tvl_change = ""
        if state and state.last_tvl > 0 and len(state.snapshots) > 0:
            prev = state.last_tvl
            change_pct = ((snap.tvl_usd - prev) / prev) * 100
            tvl_change = f" ({change_pct:+.1f}%)"

        print(
            f"  {snap.dex_name:20s} | "
            f"TVL: ${snap.tvl_usd:>10,.0f}{tvl_change:>10s} | "
            f"Vol(1h): ${snap.volume_1h:>8,.0f} | "
            f"Txns: {snap.buys_1h + snap.sells_1h:>4d}"
        )

    print(f"  {'':20s} | Total TVL: ${total_tvl:>10,.0f} | Total Vol: ${total_vol:>8,.0f}")


def print_alert(alert: PoolAlert) -> None:
    """Print a formatted alert message.

    Args:
        alert: PoolAlert to display.
    """
    severity_prefix = {
        "critical": "!! CRITICAL",
        "warning": "!  WARNING ",
        "info": "i  INFO    ",
    }
    prefix = severity_prefix.get(alert.severity, "   UNKNOWN ")
    print(f"\n  [{prefix}] {format_timestamp(alert.timestamp)}")
    print(f"  Pool: {alert.dex_name} ({alert.pool_address[:16]}...)")
    print(f"  Type: {alert.alert_type}")
    print(f"  {alert.message}")


def print_monitoring_summary(states: dict[str, MonitorState]) -> None:
    """Print final monitoring summary with all alerts.

    Args:
        states: All monitoring states.
    """
    all_alerts: list[PoolAlert] = []
    for state in states.values():
        all_alerts.extend(state.alerts)

    print(f"\n{'='*60}")
    print(f"  MONITORING SUMMARY")
    print(f"{'='*60}")
    print(f"  Pools monitored: {len(states)}")
    print(f"  Total alerts: {len(all_alerts)}")

    # Count by severity
    by_severity: dict[str, int] = {}
    by_type: dict[str, int] = {}
    for a in all_alerts:
        by_severity[a.severity] = by_severity.get(a.severity, 0) + 1
        by_type[a.alert_type] = by_type.get(a.alert_type, 0) + 1

    if by_severity:
        print(f"\n  By severity:")
        for sev, count in sorted(by_severity.items()):
            print(f"    {sev}: {count}")

    if by_type:
        print(f"\n  By type:")
        for atype, count in sorted(by_type.items()):
            print(f"    {atype}: {count}")

    # Per-pool summary
    print(f"\n  Per-pool summary:")
    for addr, state in states.items():
        if state.snapshots:
            first = state.snapshots[0]
            last = state.snapshots[-1]
            tvl_change = (
                ((last.tvl_usd - first.tvl_usd) / first.tvl_usd * 100)
                if first.tvl_usd > 0 else 0
            )
            print(
                f"    {state.dex_name:20s} | "
                f"TVL: ${first.tvl_usd:>10,.0f} -> ${last.tvl_usd:>10,.0f} "
                f"({tvl_change:+.1f}%) | "
                f"Alerts: {len(state.alerts)}"
            )

    print(f"\n  Note: This monitoring report is for informational purposes only.")
    print(f"  It does not constitute financial advice.\n")


# ── Monitor Loop ────────────────────────────────────────────────────


def run_monitor(token_mint: str, interval_seconds: int, max_checks: int) -> None:
    """Run the pool monitor in live mode.

    Args:
        token_mint: Solana token mint address.
        interval_seconds: Seconds between checks.
        max_checks: Maximum number of checks before stopping.
    """
    states: dict[str, MonitorState] = {}
    known_pools: set[str] = set()

    print(f"Monitoring pools for {token_mint}")
    print(f"Interval: {interval_seconds}s | Max checks: {max_checks}")
    print(f"Press Ctrl+C to stop.\n")

    for check_num in range(1, max_checks + 1):
        try:
            snapshots = fetch_pool_snapshots(token_mint)
        except (httpx.HTTPStatusError, httpx.RequestError) as e:
            print(f"  Fetch error at check {check_num}: {e}")
            time.sleep(interval_seconds)
            continue

        # Detect new pools
        for snap in snapshots:
            if snap.pool_address not in known_pools:
                known_pools.add(snap.pool_address)
                if check_num > 1 and NEW_POOL_ALERT:
                    alert = PoolAlert(
                        timestamp=snap.timestamp,
                        pool_address=snap.pool_address,
                        dex_name=snap.dex_name,
                        alert_type="new_pool",
                        severity="info",
                        message=f"New pool detected on {snap.dex_name} "
                                f"with TVL ${snap.tvl_usd:,.0f}",
                    )
                    print_alert(alert)
                    if snap.pool_address in states:
                        states[snap.pool_address].alerts.append(alert)

            # Initialize state if needed
            if snap.pool_address not in states:
                states[snap.pool_address] = MonitorState(
                    pool_address=snap.pool_address,
                    dex_name=snap.dex_name,
                    first_seen=snap.timestamp,
                    last_tvl=snap.tvl_usd,
                )

        # Detect alerts and update states
        for snap in snapshots:
            state = states[snap.pool_address]
            alerts = detect_alerts(state, snap)
            for alert in alerts:
                print_alert(alert)
                state.alerts.append(alert)
            update_state(state, snap)

        # Print report
        print_snapshot_report(snapshots, states, check_num)

        if check_num < max_checks:
            try:
                time.sleep(interval_seconds)
            except KeyboardInterrupt:
                print("\n\nMonitoring stopped by user.")
                break

    print_monitoring_summary(states)


def run_demo() -> None:
    """Run the pool monitor with simulated lifecycle data.

    Simulates 20 hours of pool activity including pool creation,
    liquidity events, and volume shifts.
    """
    print("Running demo simulation: Token pool lifecycle over 20 time steps")
    print("Simulating: PumpFun graduation -> growth -> liquidity event -> decline\n")

    lifecycle = simulate_pool_lifecycle()
    states: dict[str, MonitorState] = {}
    known_pools: set[str] = set()

    for step, (ts, snapshots) in enumerate(lifecycle, 1):
        # Detect new pools
        for snap in snapshots:
            if snap.pool_address not in known_pools:
                known_pools.add(snap.pool_address)
                if step > 1:
                    alert = PoolAlert(
                        timestamp=snap.timestamp,
                        pool_address=snap.pool_address,
                        dex_name=snap.dex_name,
                        alert_type="new_pool",
                        severity="info",
                        message=f"New pool detected on {snap.dex_name} "
                                f"with TVL ${snap.tvl_usd:,.0f}",
                    )
                    print_alert(alert)
                    states.setdefault(snap.pool_address, MonitorState(
                        pool_address=snap.pool_address,
                        dex_name=snap.dex_name,
                        first_seen=snap.timestamp,
                    )).alerts.append(alert)

            if snap.pool_address not in states:
                states[snap.pool_address] = MonitorState(
                    pool_address=snap.pool_address,
                    dex_name=snap.dex_name,
                    first_seen=snap.timestamp,
                    last_tvl=snap.tvl_usd,
                )

        # Detect alerts
        for snap in snapshots:
            state = states[snap.pool_address]
            alerts = detect_alerts(state, snap)
            for alert in alerts:
                print_alert(alert)
                state.alerts.append(alert)
            update_state(state, snap)

        # Print step report
        print_snapshot_report(snapshots, states, step)

    print_monitoring_summary(states)


# ── Main ────────────────────────────────────────────────────────────


def main() -> None:
    """Entry point: parse arguments and run pool monitor."""
    parser = argparse.ArgumentParser(
        description="Monitor DEX pool metrics and detect liquidity events"
    )
    parser.add_argument(
        "token_mint",
        nargs="?",
        default=None,
        help="Solana token mint address to monitor",
    )
    parser.add_argument(
        "--demo",
        action="store_true",
        help="Run with simulated lifecycle data (no API calls)",
    )
    parser.add_argument(
        "--interval",
        type=int,
        default=60,
        help="Seconds between checks in live mode (default: 60)",
    )
    parser.add_argument(
        "--max-checks",
        type=int,
        default=60,
        help="Maximum number of checks before stopping (default: 60)",
    )

    args = parser.parse_args()

    if args.demo:
        run_demo()
    elif args.token_mint:
        run_monitor(args.token_mint, args.interval, args.max_checks)
    else:
        parser.print_help()
        print("\nExamples:")
        print("  python scripts/pool_monitor.py --demo")
        print("  python scripts/pool_monitor.py <TOKEN_MINT> --interval 30")
        sys.exit(1)


if __name__ == "__main__":
    main()