sybil-detection Checked

if len(current_cluster) >= 3:

clusters.append([b["wallet"] for b in current_cluster])

return clusters

**Interpretation:**
- Same slot, different transactions = coordinated (bot-driven)
- Same transaction = bundled (definite sybil)
- First 3 slots after token creation = launch sniping cluster

### 3. Bundled Transactions

Multiple buys packed into a single Solana transaction or 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),

}

See `references/bundler_detection.md` for PumpFun-specific patterns and Jito bundle mechanics.

### 4. Wash Trading Detection

Same entity buying and selling through multiple wallets to inflate volume:

**Signals:**
- Wallet A buys token, transfers to Wallet B, Wallet B sells — circular flow
- Multiple wallets trading back and forth with no net position change
- Volume concentrated in wallet pairs with funding links

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. Creator Network Analysis

Identify wallets controlled by the token creator:

- Creator wallet's funding history reveals other wallets it funded
- Those wallets holding token supply = insider distribution
- Creator selling from "different" wallets = disguised dump

## Key Metrics

| Metric | Formula | Healthy | Suspicious | Critical |
|--------|---------|---------|------------|----------|
| Unique funder ratio | unique_funders / total_holders | > 0.8 | 0.4-0.8 | < 0.4 |
| Funding cluster size | max(cluster_sizes) | < 5 | 5-20 | > 20 |
| Co-trade score | wallets_in_first_3_slots / total_holders | < 0.1 | 0.1-0.3 | > 0.3 |
| Bundle ratio | bundled_buys / total_early_buys | < 0.1 | 0.1-0.4 | > 0.4 |
| Bundled supply % | bundled_token_amount / total_supply_sold | < 5% | 5-20% | > 20% |
| Transfer density | internal_transfers / total_transfers | < 0.1 | 0.1-0.3 | > 0.3 |
| Wash trade pairs | reciprocal_pairs / total_holder_pairs | 0 | 1-3 pairs | > 3 pairs |

## Composite Risk Score

Combine individual signals into a single sybil risk score (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}

## Data Sources

| Source | What It Provides | Auth Required |
|--------|-----------------|---------------|
| Helius parsed transactions | Funding history, transfer details, parsed instruction data | API key (free tier: 30 req/s) |
| SolanaTracker API | Bundler detection, holder lists, token metadata | API key |
| Solana RPC (getSignaturesForAddress) | Raw transaction signatures for any wallet | RPC URL |
| Solana RPC (getTokenLargestAccounts) | Top holders by balance | RPC URL |
| DexScreener | Basic token/pair data for cross-referencing | None |

## Workflow: Evaluate a To