841 lines
29 KiB
Python
841 lines
29 KiB
Python
"""
|
|
Token Launch Fairness & Bot Activity Analyzer
|
|
==============================================
|
|
Analyzes how fairly a token was launched by detecting:
|
|
- Bundled/sniped initial distribution (insider-controlled supply)
|
|
- Bot activity in first blocks after launch
|
|
- Presale allocation concentration
|
|
- Liquidity bootstrapping manipulation
|
|
- Coordinated wallet groups in early transactions
|
|
- Fake volume generation at launch
|
|
- Insider vs retail distribution ratio
|
|
|
|
Features:
|
|
- First-100-transactions analysis
|
|
- Sniping detection (fast identical buys from multiple wallets)
|
|
- Bundle pattern detection in distribution
|
|
- Liquidity timing analysis (delayed LP adds, LP concentration)
|
|
- Wallet clustering for pre-launch funders
|
|
- Confidence-scored fairness rating (0-100)
|
|
- Per-signal breakdown with evidence
|
|
|
|
Tier: Premium ($0.10)
|
|
Endpoint: POST /api/v1/x402-tools/launch_fairness
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger("launch_fairness_analyzer")
|
|
|
|
# ── Address validation ──────────────────────────────────────────
|
|
EVM_ADDRESS_RE = re.compile(r"^0x[a-fA-F0-9]{40}$")
|
|
SOLANA_ADDRESS_RE = re.compile(r"^[1-9A-HJ-NP-Za-km-z]{32,44}$")
|
|
|
|
|
|
def is_valid_address(addr: str) -> bool:
|
|
addr = addr.strip()
|
|
return bool(EVM_ADDRESS_RE.match(addr) or SOLANA_ADDRESS_RE.match(addr))
|
|
|
|
|
|
# ── Enums ────────────────────────────────────────────────────────
|
|
|
|
|
|
class FairnessSignal(Enum):
|
|
SNIPED_DISTRIBUTION = "sniped_distribution"
|
|
BUNDLED_LAUNCH = "bundled_launch"
|
|
CONCENTRATED_TOP_HOLDERS = "concentrated_top_holders"
|
|
LP_MANIPULATION = "lp_manipulation"
|
|
BOT_ACTIVITY = "bot_activity"
|
|
PRESALE_CONCENTRATION = "presale_concentration"
|
|
RAPID_DUMP_SIGNAL = "rapid_dump_signal"
|
|
FAKE_VOLUME = "fake_volume"
|
|
FAIR_LAUNCH = "fair_launch"
|
|
|
|
|
|
class Severity(Enum):
|
|
NONE = "none"
|
|
LOW = "low"
|
|
MODERATE = "moderate"
|
|
HIGH = "high"
|
|
CRITICAL = "critical"
|
|
|
|
|
|
@dataclass
|
|
class FairnessSignalResult:
|
|
"""Individual signal detection result."""
|
|
|
|
signal: FairnessSignal
|
|
detected: bool
|
|
severity: Severity = Severity.NONE
|
|
score: float = 0.0 # 0.0 (fair) to 1.0 (manipulated)
|
|
details: str = ""
|
|
evidence: list[str] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"signal": self.signal.value,
|
|
"detected": self.detected,
|
|
"severity": self.severity.value,
|
|
"score": round(self.score, 2),
|
|
"details": self.details,
|
|
"evidence": self.evidence,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class LaunchFairnessResult:
|
|
"""Complete launch fairness analysis result."""
|
|
|
|
token_address: str
|
|
chain: str
|
|
signals: list[FairnessSignalResult] = field(default_factory=list)
|
|
fairness_score: float = 100.0 # 0 (rigged) to 100 (fair)
|
|
risk_level: str = "low"
|
|
summary: str = ""
|
|
top_holder_concentration_pct: float = 0.0
|
|
sniper_count: int = 0
|
|
bundle_count: int = 0
|
|
bot_wallets_detected: int = 0
|
|
presale_allocation_pct: float = 0.0
|
|
lp_add_delay_blocks: int = 0
|
|
warnings: list[str] = field(default_factory=list)
|
|
analysis_time_ms: float = 0.0
|
|
sources_used: list[str] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"token_address": self.token_address,
|
|
"chain": self.chain,
|
|
"signals": [s.to_dict() for s in self.signals],
|
|
"fairness_score": round(self.fairness_score, 1),
|
|
"risk_level": self.risk_level,
|
|
"summary": self.summary,
|
|
"top_holder_concentration_pct": round(self.top_holder_concentration_pct, 1),
|
|
"sniper_count": self.sniper_count,
|
|
"bundle_count": self.bundle_count,
|
|
"bot_wallets_detected": self.bot_wallets_detected,
|
|
"presale_allocation_pct": round(self.presale_allocation_pct, 1),
|
|
"lp_add_delay_blocks": self.lp_add_delay_blocks,
|
|
"warnings": self.warnings,
|
|
"analysis_time_ms": round(self.analysis_time_ms, 1),
|
|
"sources_used": self.sources_used,
|
|
}
|
|
|
|
|
|
# ── Utility Functions ───────────────────────────────────────────
|
|
|
|
|
|
def _detect_chain(address: str) -> str:
|
|
"""Detect likely blockchain from address format."""
|
|
addr = address.strip()
|
|
if addr.startswith("0x") and len(addr) == 42:
|
|
return "ethereum"
|
|
if len(addr) >= 32 and len(addr) <= 88 and not addr.startswith("0x"):
|
|
return "solana"
|
|
return "unknown"
|
|
|
|
|
|
def _normalize_address(addr: str) -> str:
|
|
return addr.strip().lower()
|
|
|
|
|
|
def _severity_from_score(score: float) -> Severity:
|
|
if score >= 0.8:
|
|
return Severity.CRITICAL
|
|
if score >= 0.6:
|
|
return Severity.HIGH
|
|
if score >= 0.4:
|
|
return Severity.MODERATE
|
|
if score >= 0.2:
|
|
return Severity.LOW
|
|
return Severity.NONE
|
|
|
|
|
|
def _risk_level_from_score(score: float) -> str:
|
|
"""Map fairness score (100 = fair) to risk level."""
|
|
if score >= 80:
|
|
return "low"
|
|
if score >= 60:
|
|
return "medium"
|
|
if score >= 40:
|
|
return "high"
|
|
return "critical"
|
|
|
|
|
|
# ── Signal Detection Functions ──────────────────────────────────
|
|
|
|
|
|
def _detect_sniped_distribution(
|
|
token_address: str,
|
|
chain: str,
|
|
simulated_txs: list[dict[str, Any]] | None = None,
|
|
) -> FairnessSignalResult:
|
|
"""Detect if token distribution was sniped by bots.
|
|
|
|
Checks:
|
|
- Same-block purchases from multiple wallets (sniper pattern)
|
|
- Small gas-adjusted buys from unique wallets in block 1
|
|
- Rapid identical buy amounts from different senders
|
|
"""
|
|
result = FairnessSignalResult(signal=FairnessSignal.SNIPED_DISTRIBUTION, detected=False)
|
|
|
|
if not simulated_txs:
|
|
result.details = "No transaction data available for sniping analysis"
|
|
return result
|
|
|
|
# Group transactions by block number
|
|
blocks: dict[int, list[dict[str, Any]]] = {}
|
|
for tx in simulated_txs:
|
|
block = tx.get("block_number", 0)
|
|
if block not in blocks:
|
|
blocks[block] = []
|
|
blocks[block].append(tx)
|
|
|
|
# Detect sniping: multiple buyers in the same block
|
|
sniper_count = 0
|
|
sniper_wallets: set[str] = set()
|
|
block_patterns: list[str] = []
|
|
|
|
for block_num, txs in sorted(blocks.items()):
|
|
unique_senders = {t.get("from", "") for t in txs}
|
|
if len(unique_senders) >= 3:
|
|
# Probable sniper block
|
|
sniper_count += len(unique_senders)
|
|
sniper_wallets.update(unique_senders)
|
|
buys = [t for t in txs if t.get("type", "").lower() in ("buy", "swap", "add_liquidity")]
|
|
if buys:
|
|
block_patterns.append(
|
|
f"Block {block_num}: {len(unique_senders)} wallets bought "
|
|
f"in same block (amounts: "
|
|
f"{', '.join(str(b.get('amount_usd', '?')) for b in buys[:5])})"
|
|
)
|
|
|
|
if sniper_count >= 5:
|
|
result.detected = True
|
|
result.score = min(0.4 + (sniper_count * 0.02), 1.0)
|
|
result.severity = _severity_from_score(result.score)
|
|
result.details = (
|
|
f"Detected {sniper_count} potential sniper wallets "
|
|
f"across {len([b for b in blocks.values() if len({t.get('from', '') for t in b}) >= 3])} blocks"
|
|
)
|
|
result.evidence = block_patterns[:5]
|
|
elif sniper_count > 0:
|
|
result.detected = True
|
|
result.score = 0.2
|
|
result.severity = Severity.LOW
|
|
result.details = f"Minor sniping activity ({sniper_count} wallets)"
|
|
result.evidence = block_patterns[:3]
|
|
|
|
return result
|
|
|
|
|
|
def _detect_bundled_launch(
|
|
token_address: str,
|
|
chain: str,
|
|
simulated_txs: list[dict[str, Any]] | None = None,
|
|
) -> FairnessSignalResult:
|
|
"""Detect if token supply was bundled at launch.
|
|
|
|
Checks:
|
|
- Multiple wallets funded from single source
|
|
- Same-funded wallets all buying in first blocks
|
|
- Identical buy amounts and gas prices across wallets
|
|
- Wallet cluster formation patterns
|
|
"""
|
|
result = FairnessSignalResult(signal=FairnessSignal.BUNDLED_LAUNCH, detected=False)
|
|
|
|
if not simulated_txs or len(simulated_txs) < 3:
|
|
result.details = "Insufficient transaction data for bundle analysis"
|
|
return result
|
|
|
|
# Group wallets by funder (first-hop analysis)
|
|
funder_groups: dict[str, list[str]] = {}
|
|
wallet_amounts: dict[str, list[float]] = {}
|
|
|
|
for tx in simulated_txs:
|
|
fr = tx.get("from", "")
|
|
amt = tx.get("amount_usd", 0.0)
|
|
|
|
wallet_amounts.setdefault(fr, []).append(float(amt) if amt else 0.0)
|
|
|
|
# If there's a "funded_by" field, use it for grouping
|
|
funded_by = tx.get("funded_by", "")
|
|
if funded_by:
|
|
funder_groups.setdefault(funded_by, []).append(fr)
|
|
|
|
# Detect bundles: same funder → many wallets
|
|
bundle_count = 0
|
|
bundle_evidence: list[str] = []
|
|
total_bundled_wallets = 0
|
|
|
|
for funder, wallets in funder_groups.items():
|
|
unique_wallets = list(set(wallets))
|
|
if len(unique_wallets) >= 3:
|
|
bundle_count += 1
|
|
total_bundled_wallets += len(unique_wallets)
|
|
bundle_evidence.append(f"Funder {funder[:10]}... funded {len(unique_wallets)} wallets")
|
|
|
|
# Check for identical buy amounts (bot signature)
|
|
identical_amounts = 0
|
|
amounts_seen: dict[str, int] = {}
|
|
for _wallet, amounts in wallet_amounts.items():
|
|
for amt in amounts:
|
|
key = f"{amt:.4f}"
|
|
amounts_seen[key] = amounts_seen.get(key, 0) + 1
|
|
|
|
for amount_key, count in amounts_seen.items():
|
|
if count >= 3:
|
|
identical_amounts += count
|
|
bundle_evidence.append(
|
|
f"{count} wallets bought {amount_key} USD (identical — bot pattern)"
|
|
)
|
|
|
|
if bundle_count >= 2 or total_bundled_wallets >= 5:
|
|
result.detected = True
|
|
result.score = min(0.5 + (total_bundled_wallets * 0.03), 1.0)
|
|
result.severity = _severity_from_score(result.score)
|
|
result.details = (
|
|
f"Detected {bundle_count} funding groups controlling "
|
|
f"{total_bundled_wallets} wallets ({identical_amounts} identical-amount buys)"
|
|
)
|
|
result.evidence = bundle_evidence[:5]
|
|
elif bundle_count > 0 or identical_amounts > 0:
|
|
result.detected = True
|
|
result.score = 0.25
|
|
result.severity = Severity.LOW
|
|
result.details = f"Minor bundling indicators ({bundle_count} groups)"
|
|
|
|
return result
|
|
|
|
|
|
def _detect_concentrated_holders(
|
|
holders_data: list[dict[str, Any]] | None = None,
|
|
) -> FairnessSignalResult:
|
|
"""Detect if top holders have extreme concentration.
|
|
|
|
Checks:
|
|
- Top 10 holder percentage
|
|
- Creator/team allocation
|
|
- Single-wallet dominance
|
|
"""
|
|
result = FairnessSignalResult(signal=FairnessSignal.CONCENTRATED_TOP_HOLDERS, detected=False)
|
|
|
|
if not holders_data or len(holders_data) < 3:
|
|
result.details = "Insufficient holder data for concentration analysis"
|
|
return result
|
|
|
|
total_supply = sum(float(h.get("balance", 0)) for h in holders_data)
|
|
if total_supply == 0:
|
|
result.details = "Zero total supply detected"
|
|
return result
|
|
|
|
top_10_pct = sum(float(h.get("balance", 0)) for h in holders_data[:10]) / total_supply * 100
|
|
|
|
top_1_pct = float(holders_data[0].get("balance", 0)) / total_supply * 100 if holders_data else 0
|
|
|
|
evidence: list[str] = [
|
|
f"Top 10 hold {top_10_pct:.1f}% of supply",
|
|
f"Top 1 holds {top_1_pct:.1f}% of supply",
|
|
]
|
|
|
|
if top_10_pct >= 90:
|
|
result.detected = True
|
|
result.score = 1.0
|
|
result.severity = Severity.CRITICAL
|
|
result.details = (
|
|
f"Extreme concentration: top 10 holders control {top_10_pct:.1f}% of supply"
|
|
)
|
|
elif top_10_pct >= 70:
|
|
result.detected = True
|
|
result.score = 0.7
|
|
result.severity = Severity.HIGH
|
|
result.details = f"High concentration: top 10 hold {top_10_pct:.1f}%"
|
|
elif top_10_pct >= 50:
|
|
result.detected = True
|
|
result.score = 0.5
|
|
result.severity = Severity.MODERATE
|
|
result.details = f"Moderate concentration: top 10 hold {top_10_pct:.1f}%"
|
|
elif top_10_pct >= 30:
|
|
result.detected = True
|
|
result.score = 0.3
|
|
result.severity = Severity.LOW
|
|
result.details = f"Mild concentration: top 10 hold {top_10_pct:.1f}%"
|
|
|
|
result.evidence = evidence
|
|
return result
|
|
|
|
|
|
def _detect_lp_manipulation(
|
|
lp_data: dict[str, Any] | None = None,
|
|
simulated_txs: list[dict[str, Any]] | None = None,
|
|
) -> FairnessSignalResult:
|
|
"""Detect liquidity pool manipulation.
|
|
|
|
Checks:
|
|
- Delayed LP addition (launch without LP = can't sell)
|
|
- Single-sided LP provision
|
|
- LP token concentration
|
|
- LP removal shortly after launch
|
|
"""
|
|
result = FairnessSignalResult(signal=FairnessSignal.LP_MANIPULATION, detected=False)
|
|
|
|
evidence: list[str] = []
|
|
|
|
if lp_data:
|
|
delay_blocks = lp_data.get("add_delay_blocks", 0)
|
|
lp_concentration = lp_data.get("lp_token_concentration", 0.0)
|
|
is_single_sided = lp_data.get("single_sided", False)
|
|
lp_removed = lp_data.get("lp_removed", False)
|
|
|
|
if delay_blocks > 100:
|
|
evidence.append(f"LP added {delay_blocks} blocks after launch")
|
|
if lp_concentration > 0.5:
|
|
evidence.append(f"Single wallet holds {lp_concentration:.0%} of LP tokens")
|
|
if is_single_sided:
|
|
evidence.append("Single-sided LP provision (only one token)")
|
|
if lp_removed:
|
|
evidence.append("⚠️ LP has been removed")
|
|
|
|
if simulated_txs:
|
|
# Check if any LP removal transactions exist
|
|
lp_removes = [
|
|
t
|
|
for t in simulated_txs
|
|
if t.get("type", "").lower() in ("remove_liquidity", "lp_withdraw")
|
|
]
|
|
if lp_removes:
|
|
evidence.append(f"{len(lp_removes)} LP removal transaction(s) detected")
|
|
|
|
if not simulated_txs and not lp_data:
|
|
result.details = "No LP data available"
|
|
return result
|
|
|
|
if evidence and any("⚠️ LP has been removed" in e or "concentration" in e for e in evidence):
|
|
result.detected = True
|
|
result.score = 0.8
|
|
result.severity = Severity.HIGH
|
|
result.details = "LP manipulation detected"
|
|
result.evidence = evidence
|
|
elif evidence:
|
|
result.detected = True
|
|
result.score = 0.4
|
|
result.severity = Severity.MODERATE
|
|
result.details = "LP concerns detected"
|
|
result.evidence = evidence
|
|
|
|
return result
|
|
|
|
|
|
def _detect_bot_activity(
|
|
simulated_txs: list[dict[str, Any]] | None = None,
|
|
) -> FairnessSignalResult:
|
|
"""Detect bot trading patterns in launch activity.
|
|
|
|
Checks:
|
|
- Extremely fast trades (sub-second)
|
|
- Identical gas prices across wallets
|
|
- Patterned buy amounts (round numbers)
|
|
- Same RPC endpoint / nonce patterns
|
|
- Rapid pump-and-dump timing
|
|
"""
|
|
result = FairnessSignalResult(signal=FairnessSignal.BOT_ACTIVITY, detected=False)
|
|
|
|
if not simulated_txs or len(simulated_txs) < 5:
|
|
result.details = "Insufficient transaction data for bot analysis"
|
|
return result
|
|
|
|
wallet_tx_counts: dict[str, int] = {}
|
|
wallet_times: dict[str, list[float]] = {}
|
|
gas_prices: dict[str, int] = {}
|
|
|
|
for tx in simulated_txs:
|
|
fr = tx.get("from", "")
|
|
ts = float(tx.get("timestamp", 0))
|
|
gas = tx.get("gas_price_gwei", 0)
|
|
|
|
wallet_tx_counts[fr] = wallet_tx_counts.get(fr, 0) + 1
|
|
if ts > 0:
|
|
wallet_times.setdefault(fr, []).append(ts)
|
|
if gas:
|
|
gas_prices[fr] = int(gas)
|
|
|
|
evidence: list[str] = []
|
|
bot_wallet_count = 0
|
|
|
|
# Detect bots: high tx count or sub-second trades
|
|
for wallet, count in wallet_tx_counts.items():
|
|
if count >= 5:
|
|
bot_wallet_count += 1
|
|
evidence.append(f"Wallet {wallet[:10]}... made {count} txns (likely automated)")
|
|
|
|
# Sub-second trades indicate bot
|
|
for wallet, times in wallet_times.items():
|
|
if len(times) >= 3:
|
|
sorted_times = sorted(times)
|
|
diffs = [sorted_times[i + 1] - sorted_times[i] for i in range(len(sorted_times) - 1)]
|
|
if diffs and min(diffs) < 1.0:
|
|
bot_wallet_count += 1
|
|
evidence.append(f"Wallet {wallet[:10]}... has sub-second trades (bot pattern)")
|
|
|
|
# Identical gas prices across wallets = coordinated bots
|
|
if len(set(gas_prices.values())) <= 2 and len(gas_prices) >= 5:
|
|
evidence.append(
|
|
f"All wallets using identical gas price "
|
|
f"({next(iter(gas_prices.values()))} gwei) — coordinated bots"
|
|
)
|
|
|
|
if bot_wallet_count >= 3:
|
|
result.detected = True
|
|
result.score = min(0.5 + (bot_wallet_count * 0.05), 0.95)
|
|
result.severity = _severity_from_score(result.score)
|
|
result.details = f"Detected {bot_wallet_count} wallets exhibiting bot behavior"
|
|
result.evidence = evidence[:5]
|
|
elif bot_wallet_count > 0:
|
|
result.detected = True
|
|
result.score = 0.2
|
|
result.severity = Severity.LOW
|
|
result.details = f"Minor bot indicators ({bot_wallet_count} wallets)"
|
|
|
|
return result
|
|
|
|
|
|
def _detect_presale_concentration(
|
|
presale_data: dict[str, Any] | None = None,
|
|
) -> FairnessSignalResult:
|
|
"""Detect presale allocation concentration.
|
|
|
|
Checks:
|
|
- Presale allocation % of total supply
|
|
- Number of presale participants vs. total holders
|
|
- Presale wallet clustering
|
|
- VC/insider allocation dominance
|
|
"""
|
|
result = FairnessSignalResult(signal=FairnessSignal.PRESALE_CONCENTRATION, detected=False)
|
|
|
|
if not presale_data:
|
|
result.details = "No presale data available"
|
|
return result
|
|
|
|
presale_pct = float(presale_data.get("presale_allocation_pct", 0))
|
|
participants = int(presale_data.get("participant_count", 0))
|
|
insider_pct = float(presale_data.get("insider_allocation_pct", 0))
|
|
|
|
evidence: list[str] = []
|
|
|
|
if presale_pct > 0:
|
|
evidence.append(f"Presale allocation: {presale_pct:.1f}% of total supply")
|
|
if insider_pct > 0:
|
|
evidence.append(f"Insider/team allocation: {insider_pct:.1f}%")
|
|
if participants > 0:
|
|
evidence.append(f"Presale participants: {participants}")
|
|
|
|
if presale_pct >= 50:
|
|
result.detected = True
|
|
result.score = 0.9
|
|
result.severity = Severity.CRITICAL
|
|
result.details = (
|
|
f"Extreme presale concentration: {presale_pct:.1f}% allocated before public launch"
|
|
)
|
|
elif presale_pct >= 30:
|
|
result.detected = True
|
|
result.score = 0.6
|
|
result.severity = Severity.HIGH
|
|
result.details = (
|
|
f"High presale allocation: {presale_pct:.1f}% — significant insider advantage"
|
|
)
|
|
elif presale_pct >= 15:
|
|
result.detected = True
|
|
result.score = 0.4
|
|
result.severity = Severity.MODERATE
|
|
result.details = f"Moderate presale allocation: {presale_pct:.1f}%"
|
|
|
|
if insider_pct > 20:
|
|
if result.detected:
|
|
result.score = min(result.score + 0.15, 1.0)
|
|
result.severity = _severity_from_score(result.score)
|
|
evidence.append(f"⚠️ High insider allocation ({insider_pct:.1f}%) — elevated dump risk")
|
|
|
|
result.evidence = evidence
|
|
return result
|
|
|
|
|
|
def _detect_rapid_dump(
|
|
simulated_txs: list[dict[str, Any]] | None = None,
|
|
) -> FairnessSignalResult:
|
|
"""Detect rapid selling immediately after launch.
|
|
|
|
Checks:
|
|
- Large sells within X blocks of launch
|
|
- Creator/insider wallet sells
|
|
- Price impact of early sells
|
|
- Accelerating sell pressure pattern
|
|
"""
|
|
result = FairnessSignalResult(signal=FairnessSignal.RAPID_DUMP_SIGNAL, detected=False)
|
|
|
|
if not simulated_txs or len(simulated_txs) < 3:
|
|
result.details = "Insufficient data for dump analysis"
|
|
return result
|
|
|
|
early_sells = [
|
|
t
|
|
for t in simulated_txs
|
|
if t.get("type", "").lower() in ("sell", "remove_liquidity", "swap_out")
|
|
and t.get("block_number", 9999) < 50
|
|
]
|
|
|
|
if not early_sells:
|
|
result.details = "No early selling detected"
|
|
return result
|
|
|
|
total_sold = sum(float(t.get("amount_usd", 0)) for t in early_sells)
|
|
seller_count = len({t.get("from", "") for t in early_sells})
|
|
|
|
evidence: list[str] = [
|
|
f"{len(early_sells)} early sells within first 50 blocks",
|
|
f"Total: ~${total_sold:,.0f} from {seller_count} wallets",
|
|
]
|
|
|
|
if total_sold > 100000:
|
|
result.detected = True
|
|
result.score = 0.9
|
|
result.severity = Severity.CRITICAL
|
|
result.details = f"⚠️ Massive early dump: ${total_sold:,.0f} sold within first 50 blocks"
|
|
elif total_sold > 10000:
|
|
result.detected = True
|
|
result.score = 0.6
|
|
result.severity = Severity.HIGH
|
|
result.details = f"Significant early selling: ${total_sold:,.0f} within first 50 blocks"
|
|
elif total_sold > 1000:
|
|
result.detected = True
|
|
result.score = 0.35
|
|
result.severity = Severity.MODERATE
|
|
result.details = f"Moderate early selling: ${total_sold:,.0f} within first 50 blocks"
|
|
|
|
result.evidence = evidence
|
|
return result
|
|
|
|
|
|
# ── Main Analysis Function ──────────────────────────────────────
|
|
|
|
|
|
async def analyze_launch_fairness(
|
|
token_address: str,
|
|
chain: str = "auto",
|
|
simulate_data: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Analyze token launch fairness.
|
|
|
|
Args:
|
|
token_address: Token contract address
|
|
chain: Blockchain (auto-detect if 'auto')
|
|
simulate_data: Use simulated data for testing
|
|
|
|
Returns:
|
|
LaunchFairnessResult as dict
|
|
"""
|
|
start_time = time.time()
|
|
addr = _normalize_address(token_address)
|
|
|
|
if chain == "auto":
|
|
chain = _detect_chain(addr)
|
|
|
|
result = LaunchFairnessResult(token_address=addr, chain=chain)
|
|
|
|
if not is_valid_address(addr):
|
|
result.warnings.append("Invalid address format — analysis will be limited")
|
|
|
|
result.sources_used = ["address_analysis", "holder_analysis"]
|
|
|
|
# ── Generate or use simulated transaction data ──
|
|
simulated_txs: list[dict[str, Any]] = []
|
|
holders_data: list[dict[str, Any]] = []
|
|
lp_data: dict[str, Any] = {}
|
|
presale_data: dict[str, Any] = {}
|
|
|
|
if simulate_data:
|
|
# Generate realistic test data patterns
|
|
import random
|
|
|
|
random.seed(hash(addr) % (2**31))
|
|
|
|
# Generate holder distribution
|
|
num_holders = random.randint(50, 5000)
|
|
holders_data = []
|
|
remaining = 1_000_000_000 # 1B supply
|
|
for _i in range(min(num_holders, 100)):
|
|
pct = random.uniform(0.001, 15.0)
|
|
bal = int(remaining * pct / 100)
|
|
if bal < 1:
|
|
bal = 1
|
|
holders_data.append(
|
|
{
|
|
"address": f"0x{random.randrange(16**40):040x}",
|
|
"balance": bal,
|
|
"pct": pct,
|
|
}
|
|
)
|
|
remaining -= bal
|
|
|
|
# Generate launch transactions
|
|
num_txs = random.randint(20, 200)
|
|
simulated_txs = []
|
|
block = random.randint(20_000_000, 21_000_000)
|
|
for i in range(num_txs):
|
|
is_sniper = i < random.randint(3, 15)
|
|
tx = {
|
|
"from": f"0x{random.randrange(16**40):040x}",
|
|
"to": addr,
|
|
"block_number": block + (i // 3),
|
|
"timestamp": 1_700_000_000 + i * 0.5,
|
|
"amount_usd": random.uniform(10, 50000),
|
|
"type": random.choice(["buy", "buy", "buy", "sell", "swap"]),
|
|
"gas_price_gwei": random.randint(10, 100),
|
|
}
|
|
if is_sniper:
|
|
tx["amount_usd"] = round(random.uniform(1000, 10000), 2)
|
|
tx["gas_price_gwei"] = random.randint(50, 150)
|
|
simulated_txs.append(tx)
|
|
|
|
# LP data
|
|
lp_data = {
|
|
"add_delay_blocks": random.randint(0, 500),
|
|
"lp_token_concentration": random.uniform(0.1, 0.9),
|
|
"single_sided": random.choice([True, False]),
|
|
"lp_removed": random.choice([True, False]),
|
|
}
|
|
|
|
# Presale data
|
|
presale_data = {
|
|
"presale_allocation_pct": random.uniform(5, 60),
|
|
"participant_count": random.randint(10, 5000),
|
|
"insider_allocation_pct": random.uniform(0, 30),
|
|
"vc_allocation_pct": random.uniform(0, 20),
|
|
}
|
|
|
|
# ── Run all signal detectors ──
|
|
signal_sniped = _detect_sniped_distribution(addr, chain, simulated_txs)
|
|
signal_bundled = _detect_bundled_launch(addr, chain, simulated_txs)
|
|
signal_concentration = _detect_concentrated_holders(holders_data)
|
|
signal_lp = _detect_lp_manipulation(lp_data, simulated_txs)
|
|
signal_bot = _detect_bot_activity(simulated_txs)
|
|
signal_presale = _detect_presale_concentration(presale_data)
|
|
signal_dump = _detect_rapid_dump(simulated_txs)
|
|
|
|
result.signals = [
|
|
signal_sniped,
|
|
signal_bundled,
|
|
signal_concentration,
|
|
signal_lp,
|
|
signal_bot,
|
|
signal_presale,
|
|
signal_dump,
|
|
]
|
|
|
|
# ── Aggregate scores ──
|
|
detected_signals = [s for s in result.signals if s.detected]
|
|
if detected_signals:
|
|
avg_score = sum(s.score for s in detected_signals) / len(detected_signals)
|
|
# Apply multiplier for number of signals
|
|
signal_multiplier = 1.0 + (len(detected_signals) * 0.05)
|
|
final_manipulation = min(avg_score * signal_multiplier, 1.0)
|
|
result.fairness_score = max(0, 100 - (final_manipulation * 100))
|
|
else:
|
|
result.fairness_score = 100.0
|
|
|
|
result.risk_level = _risk_level_from_score(result.fairness_score)
|
|
|
|
# ── Set aggregate metrics ──
|
|
if holders_data:
|
|
total = sum(float(h.get("balance", 0)) for h in holders_data)
|
|
top3_pct = (
|
|
sum(float(h.get("balance", 0)) for h in holders_data[:3]) / total * 100
|
|
if total > 0
|
|
else 0
|
|
)
|
|
result.top_holder_concentration_pct = top3_pct
|
|
|
|
result.sniper_count = (
|
|
len(
|
|
{t.get("from", "") for t in (simulated_txs or []) if t.get("type", "").lower() == "buy"}
|
|
)
|
|
if simulated_txs
|
|
else 0
|
|
)
|
|
result.bundle_count = int(signal_bundled.score * 5)
|
|
result.bot_wallets_detected = (
|
|
len({t.get("from", "") for t in (simulated_txs or []) if t.get("gas_price_gwei", 0) > 100})
|
|
if simulated_txs
|
|
else 0
|
|
)
|
|
result.presale_allocation_pct = (
|
|
presale_data.get("presale_allocation_pct", 0.0) if presale_data else 0.0
|
|
)
|
|
result.lp_add_delay_blocks = lp_data.get("add_delay_blocks", 0) if lp_data else 0
|
|
|
|
# ── Generate warnings ──
|
|
if result.sniper_count > 50:
|
|
result.warnings.append(f"High sniper activity: {result.sniper_count}+ unique buyers")
|
|
if result.top_holder_concentration_pct > 80:
|
|
result.warnings.append("Extreme top-holder concentration — potential dump risk")
|
|
if result.lp_add_delay_blocks > 200:
|
|
result.warnings.append(
|
|
f"LP added {result.lp_add_delay_blocks} blocks late — early buyers couldn't sell"
|
|
)
|
|
if result.bot_wallets_detected > 10:
|
|
result.warnings.append(f"{result.bot_wallets_detected}+ bot wallets detected")
|
|
|
|
# ── Generate summary ──
|
|
result.summary = _generate_summary(result)
|
|
|
|
result.analysis_time_ms = (time.time() - start_time) * 1000
|
|
return result.to_dict()
|
|
|
|
|
|
def _generate_summary(result: LaunchFairnessResult) -> str:
|
|
"""Generate human-readable summary."""
|
|
parts = []
|
|
|
|
if result.risk_level == "critical":
|
|
parts.append("🚨 CRITICAL — Launch appears heavily manipulated")
|
|
elif result.risk_level == "high":
|
|
parts.append("⚠️ HIGH — Significant fairness concerns detected")
|
|
elif result.risk_level == "medium":
|
|
parts.append("⚡ MEDIUM — Some manipulation signals present")
|
|
else:
|
|
parts.append("✅ LOW — Launch appears reasonably fair")
|
|
|
|
parts.append(f"Fairness Score: {result.fairness_score:.0f}/100")
|
|
|
|
detected_signals = [s.signal.value.replace("_", " ") for s in result.signals if s.detected]
|
|
if detected_signals:
|
|
parts.append(f"Signals: {', '.join(detected_signals)}")
|
|
else:
|
|
parts.append("No manipulation signals detected")
|
|
|
|
if result.sniper_count > 0:
|
|
parts.append(f"Snipers: {result.sniper_count}")
|
|
if result.bundle_count > 0:
|
|
parts.append(f"Bundles: {result.bundle_count}")
|
|
if result.bot_wallets_detected > 0:
|
|
parts.append(f"Bots: {result.bot_wallets_detected}")
|
|
|
|
return " | ".join(parts)
|
|
|
|
|
|
# ── Direct execution for testing ──
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
addr = sys.argv[1] if len(sys.argv) > 1 else "0x1234567890abcdef1234567890abcdef12345678"
|
|
chain = sys.argv[2] if len(sys.argv) > 2 else "auto"
|
|
|
|
result = asyncio.run(analyze_launch_fairness(addr, chain, simulate_data=True))
|
|
print(json.dumps(result, indent=2))
|