liquidity-analysis

流动性分析 — Solana 代币的 DEX 深度评估

流动性分析在每笔交易前回答三个关键问题：我能以合理的价格买入吗？ 需要时我能卖出吗？ 这个池子安全吗？ 如果不进行分析，你可能会面临过高的滑点、无法退出以及 Rug Pull 风险。

流动性分析为何重要

仓位规模：最大仓位受可用流动性限制。若池子的 TVL 为 20K 美元，在其中建立 10K 美元的仓位会显著影响价格。经验法则：交易规模应低于池子深度的 2%，以将滑点控制在 1% 以下。

执行成本：滑点是直接成本。买入 5 SOL 时，0.3% 和 3% 的滑点之间的差额，会在每次买入和卖出时造成真实的资金损失。

Rug 风险检测：流动性稀薄、只有单个池子、LP 代币未锁定以及新建池子都是警示信号。流动性分析能在你入场前发现这些问题。

退出规划：入场时的流动性可能与退出时不同。如果 LP 未锁定且由单个钱包持有，随时都可能被撤走。

核心概念

总锁仓价值（TVL）

存入池子的资产总价值。对于一个包含 100 SOL 和 100 万 TOKEN（每个价值 0.01 美元）的 SOL/TOKEN 池，TVL = 100 * SOL 价格 + 100 万 * 0.01 美元。仅凭 TVL 并不足够——你还需要了解当前价格区间内的深度。

流动性深度

在价格变动 X% 前可以交易多少资产。在恒定乘积 AMM 中，深度均匀分布。在集中流动性（CLMM）中，深度会随价格区间变化——当前价格附近较深，活跃区间之外则较浅或为零。

集中度因子（CLMM）

集中流动性池将资金集中在狭窄的价格区间内，在该区间内提供更深的流动性，但区间之外则没有流动性。一个 TVL 为 50K 美元、集中在 +/-5% 区间内的池子，在该区间内提供的深度与一个 500K 美元的恒定乘积池相同，但超出该区间后深度为零。

滑点曲线

滑点并非线性变化。绘制滑点与交易规模的关系图，会得到一条小额交易时较平缓、大额交易时较陡峭的曲线。曲线形状取决于池子类型、TVL 和集中度。

池子构成

流动性提供者的情况很重要。已锁定的 LP 代币无法被提取（更安全）。单边流动性意味着池子不平衡。池龄反映稳定性——TVL 稳定且创建时间超过 7 天的池子更可靠。

数据来源

以下四种互补的数据来源，从免费到全面：

来源	需要身份验证	最适合	局限性
DexScreener	无	快速查询池子、获取 liquidity.usd	不提供链上池子详情
Jupiter Quote API	无	测量任意规模交易的实际滑点	汇总多个池子的情况
Birdeye	API 密钥	详细的池子数据、交易历史	免费层级有速率限制
链上数据	仅需 RPC	LP 锁定状态、精确储备量	需要了解相关程序

完整的端点文档和用法示例请参阅 references/data_sources.md。

核心分析流程

第 1 步：识别池子

获取某个代币的所有池子。大多数 Solana 代币在 Raydium、Orca 和 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]

第 2 步：测量深度

提取每个池子的流动性指标：

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", ""),
    }

第 3 步：估算滑点

使用多个规模的 Jupiter 报价，构建实际滑点曲线。这样可以反映跨所有池子的真实路由情况：

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

第 4 步：评估集中度

对于 CLMM 池子（Orca Whirlpool、Raydium CLMM、Meteora DLMM），流动性可能集中在狭窄的区间内。检查当前价格是否处于活跃区间内，以及流动性深度能延伸多远：

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",
    }

第 5 步：计算流动性评分

综合评分范围为 0（危险）到 100（流动性充足且安全）：

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))

风险标记

在建立任何仓位前，标记以下情况：

标记	条件	风险等级
单一池	仅存在 1 个 DEX 池	高
流动性稀薄	总 TVL < $10,000	严重
新建池	池创建时间 < 2 小时	高
LP 未锁定	LP 代币未销毁或锁定	中
成交量不匹配	成交量 >> TVL（刷量交易）	中
价格偏差	不同池之间的价格差异 >5%	高
流动性集中于 CLMM	>80% 的流动性集中在窄区间 CLMM 中	中

def detect_risk_flags(pools: list[dict]) -> list[str]:
    """Detect liquidity risk flags from pool data."""
    flags = []
    if len(pools) < 2:
        flags.append("SINGLE_POOL: Only 1 pool exists — exit may be difficult")

    total_liq = sum(p.get("liquidity", {}).get("usd", 0) for p in pools)
    if total_liq < 10_000:
        flags.append(f"THIN_LIQUIDITY: Total TVL ${total_liq:,.0f} < $10,000")

    for p in pools:
        age_ms = p.get("pairCreatedAt", 0)
        if age_ms > 0:
            import time
            age_hours = (time.time() * 1000 - age_ms) / 3_600_000
            if age_hours < 2:
                flags.append(f"NEW_POOL: {p.get('dexId')} pool is {age_hours:.1f}h old")

    volumes = [p.get("volume", {}).get("h24", 0) for p in pools]
    liqs = [p.get("liquidity", {}).get("usd", 0) for p in pools]
    for v, l, p in zip(volumes, liqs, pools):
        if l > 0 and v / l > 10:
            flags.append(f"VOLUME_MISMATCH: {p.get('dexId')} volume/TVL = {v/l:.1f}x")

    prices = [float(p.get("priceUsd", 0)) for p in pools if float(p.get("priceUsd", 0)) > 0]
    if len(prices) >= 2:
        deviation = (max(prices) - min(prices)) / min(prices)
        if deviation > 0.05:
            flags.append(f"PRICE_DEVIATION: {deviation:.1%} across pools")

    return flags

根据流动性确定仓位规模

最大仓位规模应确保滑点低于你的阈值：

交易类型	最大滑点	最大仓位占 TVL 的比例
短线交易	0.5% (50 bps)	1%
波段交易	2% (200 bps)	2-5%
中长线持仓	5% (500 bps)	5-10%

def max_position_from_liquidity(
    total_liquidity_usd: float,
    max_slippage_pct: float = 1.0,
    trade_type: str = "swing",
) -> float:
    """Estimate maximum position size in USD based on liquidity.

    Uses rule-of-thumb: max_position = tvl_fraction * total_liquidity.
    For constant-product AMM, 1% of TVL produces ~2% slippage.

    Args:
        total_liquidity_usd: Total liquidity across all pools.
        max_slippage_pct: Maximum acceptable slippage percentage.
        trade_type: "scalp", "swing", or "position".

    Returns:
        Maximum position size in USD.
    """
    fractions = {"scalp": 0.01, "swing": 0.03, "position": 0.07}
    base_fraction = fractions.get(trade_type, 0.03)
    adjusted = base_fraction * (max_slippage_pct / 2.0)
    return total_liquidity_usd * adjusted

滑点估算

关于滑点计算的详细数学内容，包括恒定乘积公式、CLMM 模型和经验曲线拟合，请参阅 references/slippage_curves.md。

恒定乘积 AMM 的关键公式：

slippage = Δx / (x + Δx)

其中，Δx 是交易规模，x 是池中输入代币的储备量。在储备量为 100 SOL 的池中交易 1 SOL，滑点 = 1/101 = 0.99%。

池类型

Solana DEX 使用不同的 AMM 设计，流动性特征也各不相同。有关详细介绍，包括：

• 恒定乘积（Raydium V4、Orca Legacy）：流动性均匀，滑点可预测
• 集中流动性（Raydium CLMM、Orca Whirlpool）：当前价格附近流动性深，区间外为零
• 动态 AMM（Meteora DLMM）：自适应费用，基于区间的流动性

请参阅 references/pool_types.md。

与其他技能集成

token-holder-analysis：入场前检查 LP 代币持有者分布。如果一个钱包持有 >50% 的 LP 代币且这些代币未锁定，退出风险较高。

position-sizing：将 max_position_from_liquidity() 的输出作为上限，输入仓位规模模型。

slippage-modeling：将此技能集中的经验滑点曲线用作执行成本模型的输入。

birdeye-api：获取详细池数据，包括交易历史和 LP 事件。

dexscreener-api：免费发现池并获取基本流动性指标。

工作流示例

# Full liquidity assessment for a token
import httpx

TOKEN = "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263"  # BONK

# 1. Get pools
resp = httpx.get(f"https://api.dexscreener.com/tokens/v1/solana/{TOKEN}")
pools = [p for p in resp.json() if p.get("liquidity", {}).get("usd", 0) > 0]

# 2. Analyze
total_liq = sum(p["liquidity"]["usd"] for p in pools)
largest = max(p["liquidity"]["usd"] for p in pools)
largest_pct = largest / total_liq if total_liq > 0 else 1.0

# 3. Risk flags
flags = detect_risk_flags(pools)

# 4. Score
score = compute_liquidity_score(total_liq, len(pools), largest_pct, 1000, 50)

# 5. Position sizing
max_pos = max_position_from_liquidity(total_liq, max_slippage_pct=1.0, trade_type="swing")

print(f"Total Liquidity: ${total_liq:,.0f}")
print(f"Pools: {len(pools)}")
print(f"Score: {score}/100")
print(f"Max Position (swing, 1% slip): ${max_pos:,.0f}")
for f in flags:
    print(f"  WARNING: {f}")

文件

文件	描述
references/slippage_curves.md	恒定乘积池和 CLMM 池的滑点计算、经验曲线拟合
references/pool_types.md	Solana 上的 AMM 设计：恒定乘积、集中流动性、动态流动性
references/data_sources.md	获取流动性数据的 API 端点和链上方法
scripts/analyze_liquidity.py	全面的流动性评估，包含评分和风险标记
scripts/pool_comparison.py	比较某个代币在不同 DEX 上的池