liquidity-analysis Checked

See references/data_sources.md for complete endpoint documentation and usage examples.

Core Analysis Pipeline

### Step 1: Identify Pools

Fetch all pools for a token. Most Solana tokens have multiple pools across Raydium, Orca, and Meteora.

import httpx

def get_pools(mint: str) -> list[dict]:
    """Fetch all DEX pools for a token from DexScreener."""
    resp = httpx.get(f"https://api.dexscreener.com/tokens/v1/solana/{mint}")
    resp.raise_for_status()
    pairs = resp.json()
    return [p for p in pairs if p.get("liquidity", {}).get("usd", 0) > 0]

### Step 2: Measure Depth

For each pool, extract liquidity metrics:

def extract_depth(pool: dict) -> dict:
    """Extract liquidity metrics from a DexScreener pool."""
    return {
        "dex": pool.get("dexId", "unknown"),
        "liquidity_usd": pool.get("liquidity", {}).get("usd", 0),
        "volume_24h": pool.get("volume", {}).get("h24", 0),
        "pool_age_hours": _pool_age_hours(pool.get("pairCreatedAt", 0)),
        "pair_address": pool.get("pairAddress", ""),
    }

### Step 3: Estimate Slippage

Use Jupiter quotes at multiple sizes to build an empirical slippage curve. This captures real routing across all pools:

import httpx

SOL_MINT = "So11111111111111111111111111111111111111112"
LAMPORTS = 1_000_000_000

async def estimate_slippage(token_mint: str, sol_amounts: list[float]) -> list[dict]:
    """Query Jupiter for slippage at multiple trade sizes.

    Args:
        token_mint: Token mint address to buy.
        sol_amounts: List of SOL amounts to test (e.g., [0.1, 0.5, 1, 5, 10]).

    Returns:
        List of dicts with sol_amount, output_tokens, price_per_token, slippage_bps.
    """
    results = []
    base_price = None
    async with httpx.AsyncClient() as client:
        for sol in sol_amounts:
            lamports = int(sol * LAMPORTS)
            resp = await client.get(
                "https://api.jup.ag/quote/v1",
                params={
                    "inputMint": SOL_MINT,
                    "outputMint": token_mint,
                    "amount": str(lamports),
                    "slippageBps": 5000,
                },
            )
            if resp.status_code != 200:
                continue
            data = resp.json()
            out_amount = int(data["outAmount"])
            price = sol / out_amount if out_amount > 0 else 0
            if base_price is None:
                base_price = price
            slippage_bps = int((price - base_price) / base_price * 10000) if base_price > 0 else 0
            results.append({
                "sol_amount": sol,
                "output_tokens": out_amount,
                "price_per_token": price,
                "slippage_bps": max(0, slippage_bps),
            })
    return results

### Step 4: Assess Concentration

For CLMM pools (Orca Whirlpool, Raydium CLMM, Meteora DLMM), liquidity may be concentrated in a narrow range. Check if the current price is within the active range and how deep liquidity extends:

def assess_concentration(pools: list[dict]) -> dict:
    """Assess concentration risk from pool data."""
    clmm_pools = [p for p in pools if p.get("dexId") in ("raydium", "orca") and "clmm" in p.get("labels", [])]
    cpmm_pools = [p for p in pools if p not in clmm_pools]

    total_clmm = sum(p.get("liquidity", {}).get("usd", 0) for p in clmm_pools)
    total_cpmm = sum(p.get("liquidity", {}).get("usd", 0) for p in cpmm_pools)
    total = total_clmm + total_cpmm

    return {
        "clmm_ratio": total_clmm / total if total > 0 else 0,
        "cpmm_liquidity": total_cpmm,
        "clmm_liquidity": total_clmm,
        "concentration_risk": "high" if total_clmm / total > 0.8 and total > 0 else "low",
    }

### Step 5: Compute Liquidity Score

Composite score from 0 (dangerous) to 100 (deep, safe liquidity):

def compute_liquidity_score(
    total_liquidity_usd: float,
    pool_count: int,
    largest_pool_pct: float,
    oldest_pool_hours: float,
    max_slippage_bps_at_1sol: int,
) -> int:
    """Compute composite liquidity score (0-100).

    Components:
        Depth (40%): log-scaled TVL from $1K (0) to $1M+ (40)
        Diversity (15%): more pools = more resilient
        Concentration (15%): penalty if one pool dominates
        Age (15%): older pools are more reliable
        Slippage (15%): lower slippage = better
    """
    import math
    # Depth: 0-40 points
    depth = min(40, int(40 * math.log10(max(total_liquidity_usd, 1)) / 6))

    # Diversity: 0-15 points
    diversity = min(15, pool_count * 3)

    # Concentration: 0-15 points (penalty for single-pool dominance)
    concentration = int(15 * (1 - largest_pool_pct))

    # Age: 0-15 points (7+ days = full marks)
    age = min(15, int(15 * oldest_pool_hours / 168))

    # Slippage: 0-15 points
    slippage = max(0, 15 - max_slippage_bps_at_1sol // 10)

    return max(0, min(100, depth + diversity + concentration + age + slippage))

Risk Flags

Flag these conditions before entering any position:

| Flag | Condition | Risk Level |

|------|-----------|