mcpField
S
sybil-detection

sybil-detection

github.com/agiprolabs数据与数据库456445 次安装8 周前更新
代码仓库

关于 sybil-detection

面向 Solana 代币市场的协同钱包与虚假交易分析技能,通过资金来源、共同交易和捆绑交易等信号识别女巫集群及人为制造的活跃度。

主要功能

  1. 01追踪持有者钱包的资金来源
  2. 02检测共同交易与捆绑买入
  3. 03分析洗售循环和创建者关联
  4. 04计算女巫风险指标与综合评分
  5. 05提供集群检测和资金追踪脚本

适用场景

  1. 01评估代币持有者是否由关联钱包组成
  2. 02检查早期买入是否存在捆绑或协同
  3. 03识别交易量和持有者数量中的虚假信号

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/agiprolabs/claude-trading-skills --skill sybil-detection

安装后在 Claude Code 中输入/sybil-detection确认已加载。

包含的文件

除 SKILL.md 外,安装时一起下载的参考资料和脚本。

references/
bundler_detection.md
clustering_methods.md
scripts/
detect_sybils.py
funding_tracer.py

详细介绍

Sybil 检测 — 协同钱包与虚假活动分析

Solana 代币市场中的 Sybil 攻击,是指同一实体操作多个钱包,制造有机活动的假象。本技能涵盖协同钱包集群、刷量交易、捆绑交易和虚增持有者数量的检测,这对评估代币指标反映的是真实需求还是人为制造的信号至关重要。

Sybil 检测为何重要

Solana 上的代币市场充斥着人为制造的信号:

  • 虚增持有者数量:500 个“持有者”实际上是 10 个实体,每个实体拥有 50 个钱包
  • 虚假交易量:通过受自己控制的钱包之间刷量交易,模拟需求
  • 人为制造的社会认同:许多钱包持有少量代币,营造广泛分布的假象
  • 为 Rug 做准备:创建者将供应量分散到多个钱包中,然后协同卖出
  • 捆绑式发行:PumpFun 代币中,创建者通过 Jito bundle 在首个 slot 买入

一个有 1,000 名持有者、其中 80% 的资金来自 3 个钱包的代币,与一个有 1,000 名资金来源各异的持有者的代币有本质区别。Sybil 检测可以区分真实需求与表演。

检测类别

1. 资金来源分析

沿每个持有者钱包向上追溯 1–2 跳,查找向其发送 SOL 的钱包:

import httpx

def trace_funding_source(wallet: str, api_key: str, max_hops: int = 2) -> list[str]:
    """Trace SOL funding sources for a wallet via Helius parsed transactions."""
    url = f"https://api.helius.xyz/v0/addresses/{wallet}/transactions"
    resp = httpx.get(url, params={"api-key": api_key, "type": "TRANSFER", "limit": 50})
    transfers = resp.json()

    funders = []
    for tx in transfers:
        for transfer in tx.get("nativeTransfers", []):
            if transfer["toUserAccount"] == wallet and transfer["amount"] > 0.001 * 1e9:
                funders.append(transfer["fromUserAccount"])
    return funders

关键信号:

  • 3 个或更多持有者钱包由同一来源注资 = 集群
  • 注资发生在代币创建后 24 小时内 = 高度可疑
  • 注资金额相同(例如,每个钱包 0.05 SOL)= 自动化分发
2. 协同交易模式

几乎同时买入同一代币的钱包,很可能存在协同:

def detect_co_trades(buy_events: list[dict], slot_window: int = 3) -> list[list[str]]:
    """Group wallets that bought within the same slot window."""
    buy_events.sort(key=lambda x: x["slot"])
    clusters = []
    current_cluster = [buy_events[0]]

    for i in range(1, len(buy_events)):
        if buy_events[i]["slot"] - current_cluster[0]["slot"] <= slot_window:
            current_cluster.append(buy_events[i])
        else:
            if len(current_cluster) >= 3:
                clusters.append([b["wallet"] for b in current_cluster])
            current_cluster = [buy_events[i]]

    if len(current_cluster) >= 3:
        clusters.append([b["wallet"] for b in current_cluster])
    return clusters

解读:

  • 同一 slot、不同交易 = 存在协同(由机器人驱动)
  • 同一笔交易 = 已捆绑(确定为 Sybil)
  • 代币创建后的前 3 个 slot = 发行狙击集群
3. 捆绑交易

将多笔买入打包到单笔 Solana 交易或 Jito bundle 中:

def check_bundle_ratio(early_buys: list[dict], bundle_window_slots: int = 5) -> dict:
    """Calculate the ratio of bundled vs independent early buys."""
    bundled = [b for b in early_buys if b.get("is_bundled", False)]
    first_slot = min(b["slot"] for b in early_buys) if early_buys else 0
    early = [b for b in early_buys if b["slot"] - first_slot <= bundle_window_slots]

    return {
        "total_early_buys": len(early),
        "bundled_buys": len(bundled),
        "bundle_ratio": len(bundled) / max(len(early), 1),
        "bundled_supply_pct": sum(b["amount"] for b in bundled) / max(sum(b["amount"] for b in early), 1),
    }

有关 PumpFun 特有模式和 Jito bundle 运作机制,请参阅 references/bundler_detection.md。

4. 刷量交易检测

同一实体通过多个钱包买入和卖出,以虚增交易量:

信号:

  • 钱包 A 买入代币,再转给钱包 B,由钱包 B 卖出 — 形成循环流转
  • 多个钱包来回交易,但净持仓没有变化
  • 交易量集中在存在注资关联的钱包对之间
def detect_wash_cycles(transfers: list[dict], holder_set: set[str]) -> list[tuple]:
    """Find circular transfer patterns among known holders."""
    # Build directed graph of transfers between holders
    edges: dict[tuple, float] = {}
    for t in transfers:
        if t["from"] in holder_set and t["to"] in holder_set:
            key = (t["from"], t["to"])
            edges[key] = edges.get(key, 0) + t["amount"]

    # Find reciprocal pairs (A->B and B->A both exist)
    wash_pairs = []
    for (a, b), vol_ab in edges.items():
        vol_ba = edges.get((b, a), 0)
        if vol_ba > 0:
            wash_pairs.append((a, b, vol_ab, vol_ba))
    return wash_pairs
5. 创建者网络分析

识别由代币创建者控制的钱包:

  • 创建者钱包的注资历史可揭示其曾向其他哪些钱包注资
  • 这些钱包持有代币供应量 = 内部人士分发
  • 创建者通过“不同”的钱包卖出 = 伪装抛售

关键指标

指标公式健康可疑严重
独立注资方比例unique_funders / total_holders> 0.80.4-0.8< 0.4
注资集群规模max(cluster_sizes)< 55-20> 20
协同交易评分wallets_in_first_3_slots / total_holders< 0.10.1-0.3> 0.3
捆绑比例bundled_buys / total_early_buys< 0.10.1-0.4> 0.4
捆绑供应量占比bundled_token_amount / total_supply_sold< 5%5-20%> 20%
转账密度internal_transfers / total_transfers< 0.10.1-0.3> 0.3
刷量交易对reciprocal_pairs / total_holder_pairs01-3 对> 3 对

综合风险评分

将各项信号合并为单一的 Sybil 风险评分(0–100):

def compute_sybil_score(metrics: dict) -> dict:
    """Compute composite sybil risk score from individual metrics."""
    weights = {
        "funding_cluster": 25,    # Wallets from same funder
        "co_trade": 20,           # Coordinated buy timing
        "bundle_ratio": 20,       # Bundled early transactions
        "unique_funder": 15,      # Diversity of funding sources
        "transfer_density": 10,   # Internal transfers between holders
        "wash_trade": 10,         # Circular trading patterns
    }

    scores = {}
    # Each sub-score normalized to 0-1, then weighted
    scores["funding_cluster"] = min(metrics.get("max_cluster_size", 0) / 20, 1.0)
    scores["co_trade"] = min(metrics.get("co_trade_pct", 0) / 0.3, 1.0)
    scores["bundle_ratio"] = min(metrics.get("bundle_ratio", 0) / 0.5, 1.0)
    scores["unique_funder"] = 1.0 - min(metrics.get("unique_funder_ratio", 1.0), 1.0)
    scores["transfer_density"] = min(metrics.get("transfer_density", 0) / 0.3, 1.0)
    scores["wash_trade"] = min(metrics.get("wash_pairs", 0) / 5, 1.0)

    composite = sum(scores[k] * weights[k] for k in weights)
    risk_level = "LOW" if composite < 30 else "MEDIUM" if composite < 60 else "HIGH"

    return {"score": round(composite, 1), "risk_level": risk_level, "components": scores}

数据来源

来源提供内容是否需要身份验证
Helius 解析后的交易注资历史、转账详情、解析后的指令数据API 密钥(免费层级:30 req/s)
SolanaTracker API捆绑者检测、持有者列表、代币元数据API 密钥
Solana RPC (getSignaturesForAddress)任意钱包的原始交易签名RPC URL
Solana RPC (getTokenLargestAccounts)按余额排列的顶级持有者RPC URL
DexScreener用于交叉引用的基本代币/交易对数据无

评估代币的工作流

# 1. Get top holders
holders = get_top_holders(token_mint, rpc_url)

# 2. Trace funding sources for each holder
funding_map = {}
for wallet in holders[:30]:  # Top 30 is usually sufficient
    funding_map[wallet] = trace_funding_source(wallet, helius_key)

# 3. Cluster by common funder
clusters = cluster_by_funder(funding_map)

# 4. Check co-trade timing
early_buys = get_early_buy_events(token_mint, helius_key)
co_trade_groups = detect_co_trades(early_buys)

# 5. Check for bundles
bundle_stats = check_bundle_ratio(early_buys)

# 6. Check wash trading
transfers = get_token_transfers(token_mint, helius_key)
wash_pairs = detect_wash_cycles(transfers, set(holders))

# 7. Compute composite score
metrics = {
    "max_cluster_size": max(len(c) for c in clusters) if clusters else 0,
    "co_trade_pct": sum(len(g) for g in co_trade_groups) / len(holders),
    "bundle_ratio": bundle_stats["bundle_ratio"],
    "unique_funder_ratio": len(set(f for fs in funding_map.values() for f in fs)) / len(holders),
    "transfer_density": len(wash_pairs) / max(len(holders), 1),
    "wash_pairs": len(wash_pairs),
}
result = compute_sybil_score(metrics)
print(f"Sybil Risk: {result['risk_level']} ({result['score']}/100)")

与其他技能集成

  • token-holder-analysis:使用持有者列表作为输入;女巫检测会补充集群上下文
  • helius-api:已解析交易历史的主要数据源
  • jito-bundles:详细的捆绑检测和 MEV 上下文
  • liquidity-analysis:结合女巫评分分析——低流动性 + 高女巫评分 = 极高风险
  • whale-tracking:区分真实巨鲸与女巫集群汇总数据

文件

文件描述
references/clustering_methods.md资金来源聚类、协同交易时间分析和基于图的检测方法
references/bundler_detection.md捆绑交易检测、PumpFun 模式、Jito 捆绑机制
scripts/detect_sybils.py完整的女巫检测流程:持有者 -> 资金来源 -> 集群 -> 风险评分
scripts/funding_tracer.py追踪一组钱包的资金来源,并按共同祖先分组

评论

支持 Markdown