687 lines
26 KiB
Python
687 lines
26 KiB
Python
"""
|
|
Creator Track Record (deployer_history)
|
|
========================================
|
|
Investigates the complete deployment history of any token creator address.
|
|
Success rate, previous rug pulls, total value created or destroyed,
|
|
and risk classification based on deployer behavior patterns.
|
|
|
|
Signals detected:
|
|
- Total tokens deployed across all chains
|
|
- Rug pull / scam tokens in deployment history
|
|
- Honeypot patterns from known deployers
|
|
- Liquidity removal events
|
|
- Ownership renouncement patterns
|
|
- Cross-chain deployer identity correlation
|
|
- Time-based clustering (serial deployers)
|
|
|
|
Tier: Premium ($0.05-0.08)
|
|
Endpoint: POST /api/v1/x402-tools/deployer_history
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
from urllib.parse import quote, urlparse
|
|
|
|
logger = logging.getLogger("deployer_history")
|
|
|
|
# ── Free API sources ─────────────────────────────────────────────
|
|
DEXSCREENER_SEARCH = "https://api.dexscreener.com/latest/dex/search?q={}"
|
|
DEXSCREENER_PAIRS_BY_TOKEN = "https://api.dexscreener.com/latest/dex/tokens/{}"
|
|
SOLSCAN_TOKEN_ACCOUNTS = "https://api.solscan.io/account/tokens?address={}&pageSize=100"
|
|
SOLSCAN_ACCOUNT_TXS = "https://api.solscan.io/account/transactions?address={}&pageSize=50"
|
|
ETHERSCAN_TXLIST = (
|
|
"https://api.etherscan.io/api?module=account&action=txlist&address={}&sort=desc&limit=100"
|
|
)
|
|
BASESCAN_TXLIST = (
|
|
"https://api.basescan.org/api?module=account&action=txlist&address={}&sort=desc&limit=100"
|
|
)
|
|
BSCSCAN_TXLIST = (
|
|
"https://api.bscscan.com/api?module=account&action=txlist&address={}&sort=desc&limit=100"
|
|
)
|
|
|
|
# Rate limiting: max requests per second per host
|
|
_RATE_LIMITERS: dict[str, float] = {}
|
|
_RATE_LIMIT_WINDOW = 0.5 # 500ms between requests to same host
|
|
|
|
|
|
async def _rate_limit_host(host: str) -> None:
|
|
"""Simple per-host rate limiter to avoid being rate-limited by APIs."""
|
|
now = time.time()
|
|
last = _RATE_LIMITERS.get(host, 0.0)
|
|
wait = _RATE_LIMIT_WINDOW - (now - last)
|
|
if wait > 0:
|
|
await asyncio.sleep(wait)
|
|
_RATE_LIMITERS[host] = time.time()
|
|
|
|
|
|
# Contract creation function signatures
|
|
CREATE2_SIG = "0x60806040" # Common contract creation prefix
|
|
CREATE_SIGS = {"0x60a06040", "0x60806040", "0x60606040", "0x60806052"}
|
|
|
|
# URL safety regex
|
|
_URL_SAFE = re.compile(
|
|
r"^https?://[a-zA-Z0-9](?:[a-zA-Z0-9-]*[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]*[a-zA-Z0-9])?)*(?::\d{1,5})?(?:/[^\s\"<>]*)?$"
|
|
)
|
|
|
|
# Known scam deployer patterns
|
|
SCAM_PATTERNS = {
|
|
"rug_pull": {
|
|
"signals": [
|
|
"liquidity_removed",
|
|
"ownership_renounced_after_raise",
|
|
"no_trading_24h_after_launch",
|
|
],
|
|
"weight": 30,
|
|
},
|
|
"honeypot": {
|
|
"signals": ["sell_tax_gt_10", "blacklist_detected", "max_wallet_limit_lt_1_percent"],
|
|
"weight": 25,
|
|
},
|
|
"pump_dump": {
|
|
"signals": ["rapid_price_spike_then_crash", "concentrated_holders", "fresh_wallet_cluster"],
|
|
"weight": 20,
|
|
},
|
|
"serial_scammer": {
|
|
"signals": ["multiple_rugs", "same_deployer_different_names", "short_lived_tokens"],
|
|
"weight": 35,
|
|
},
|
|
"suspicious_renounce": {
|
|
"signals": ["renounced_after_drain", "renounce_then_contract_upgraded", "fake_renounce"],
|
|
"weight": 20,
|
|
},
|
|
}
|
|
|
|
|
|
def _validate_url(url: str) -> bool:
|
|
"""Validate URL using urlparse to prevent SSRF and injection."""
|
|
try:
|
|
result = urlparse(url)
|
|
# Ensure the URL has a scheme, network location, and a path
|
|
return all([result.scheme in ("http", "https"), result.netloc, result.path])
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
@dataclass
|
|
class DeployedToken:
|
|
"""Represents a single token deployed by the address under investigation."""
|
|
|
|
address: str
|
|
chain: str
|
|
name: str = ""
|
|
symbol: str = ""
|
|
deploy_tx: str = ""
|
|
deploy_time: str = ""
|
|
total_supply: float = 0.0
|
|
liquidity_usd: float = 0.0
|
|
volume_24h: float = 0.0
|
|
price_usd: float = 0.0
|
|
holders: int = 0
|
|
is_verified: bool = False
|
|
is_honeypot: bool = False
|
|
is_rug: bool = False
|
|
is_active: bool = True
|
|
risk_score: float = 0.0
|
|
flags: list[str] = field(default_factory=list)
|
|
age_days: float = 0.0
|
|
max_holder_pct: float = 0.0
|
|
|
|
|
|
@dataclass
|
|
class DeployerProfile:
|
|
"""Complete deployer analysis result."""
|
|
|
|
address: str
|
|
chains_used: list[str] = field(default_factory=list)
|
|
total_tokens_deployed: int = 0
|
|
active_tokens: int = 0
|
|
dead_tokens: int = 0
|
|
rug_tokens: int = 0
|
|
honeypot_tokens: int = 0
|
|
tokens: list[DeployedToken] = field(default_factory=list)
|
|
first_seen: str = ""
|
|
last_active: str = ""
|
|
avg_token_lifespan_days: float = 0.0
|
|
risk_score: float = 0.0
|
|
risk_level: str = "unknown"
|
|
patterns_detected: list[str] = field(default_factory=list)
|
|
confidence: float = 0.0
|
|
recommendation: str = ""
|
|
errors: list[str] = field(default_factory=list)
|
|
|
|
|
|
# ── Core scoring engine ─────────────────────────────────────────
|
|
|
|
|
|
def _compute_deployer_risk(profile: DeployerProfile) -> float:
|
|
"""
|
|
Compute a 0-100 deployer risk score.
|
|
|
|
Factors:
|
|
- rug_token_ratio (30%): Proportion of deployed tokens that rug
|
|
- honeypot_ratio (20%): Proportion of honeypot tokens
|
|
- total_volume (15%): Total value deployed across all tokens
|
|
- avg_lifespan (15%): Average token lifespan (shorter = worse)
|
|
- pattern_detection (20%): Known scam patterns detected
|
|
"""
|
|
total = max(profile.total_tokens_deployed, 1)
|
|
|
|
# Rug ratio
|
|
rug_ratio = profile.rug_tokens / total
|
|
rug_score = min(rug_ratio * 100 * 0.30, 30)
|
|
|
|
# Honeypot ratio
|
|
honey_ratio = profile.honeypot_tokens / total
|
|
honey_score = min(honey_ratio * 100 * 0.20, 20)
|
|
|
|
# Volume check — serial deployers with tiny volumes are suspicious
|
|
# If avg token has < $100 liquidity, it's likely a spam/scam deployer
|
|
low_val_tokens = sum(1 for t in profile.tokens if t.liquidity_usd < 100)
|
|
low_val_ratio = low_val_tokens / total
|
|
volume_score = min(low_val_ratio * 100 * 0.15, 15)
|
|
|
|
# Lifespan — short-lived tokens are suspicious
|
|
if profile.avg_token_lifespan_days > 0:
|
|
# Less than 7 days avg = very suspicious
|
|
if profile.avg_token_lifespan_days < 7:
|
|
lifespan_score = 15
|
|
elif profile.avg_token_lifespan_days < 30:
|
|
lifespan_score = 10
|
|
elif profile.avg_token_lifespan_days < 90:
|
|
lifespan_score = 5
|
|
else:
|
|
lifespan_score = 0
|
|
else:
|
|
lifespan_score = 5 # Unknown, slight penalty
|
|
|
|
# Pattern detection weights
|
|
pattern_score = min(len(profile.patterns_detected) * 5, 20)
|
|
|
|
raw = rug_score + honey_score + volume_score + lifespan_score + pattern_score
|
|
return min(raw, 100.0)
|
|
|
|
|
|
def _classify_deployer_risk(score: float) -> str:
|
|
"""Classify score into risk category."""
|
|
if score >= 70:
|
|
return "critical"
|
|
if score >= 50:
|
|
return "high"
|
|
if score >= 30:
|
|
return "moderate"
|
|
if score >= 10:
|
|
return "low"
|
|
return "safe"
|
|
|
|
|
|
def _generate_recommendation(profile: DeployerProfile, score: float) -> str:
|
|
"""Generate human-readable recommendation based on deployer analysis."""
|
|
if score >= 70:
|
|
return (
|
|
f"CRITICAL: Deployer {profile.address[:10]}...{profile.address[-6:]} "
|
|
f"has a confirmed history of {profile.rug_tokens} rug pulls and "
|
|
f"{profile.honeypot_tokens} honeypot tokens out of {profile.total_tokens_deployed} "
|
|
f"total deployments. AVOID ALL tokens from this deployer. "
|
|
f"Patterns detected: {', '.join(profile.patterns_detected)}."
|
|
)
|
|
if score >= 50:
|
|
return (
|
|
f"HIGH: Deployer shows {profile.rug_tokens} rug pulls and "
|
|
f"{profile.honeypot_tokens} honeypot tokens. Significant risk. "
|
|
f"Thoroughly vet any token from this address before engaging. "
|
|
f"Only {profile.active_tokens}/{profile.total_tokens_deployed} tokens remain active."
|
|
)
|
|
if score >= 30:
|
|
return (
|
|
f"MODERATE: Some concerning patterns detected. "
|
|
f"Deployer has {profile.dead_tokens} inactive tokens out of "
|
|
f"{profile.total_tokens_deployed}. Review each token individually. "
|
|
f"Average lifespan: {profile.avg_token_lifespan_days:.1f} days."
|
|
)
|
|
if score >= 10:
|
|
return (
|
|
f"LOW: Minor concerns. Deployer has mainly active tokens with "
|
|
f"reasonable lifespans ({profile.avg_token_lifespan_days:.1f} days avg). "
|
|
f"Standard due diligence recommended."
|
|
)
|
|
return (
|
|
f"SAFE: No significant red flags detected. Deployer has "
|
|
f"{profile.active_tokens} active tokens out of "
|
|
f"{profile.total_tokens_deployed} with healthy lifespans."
|
|
)
|
|
|
|
|
|
# ── Data fetching helpers ───────────────────────────────────────
|
|
|
|
|
|
async def _fetch_json(url: str, timeout: int = 15) -> dict[str, Any] | list[Any] | None:
|
|
"""Fetch JSON from URL with timeout, rate limiting, and error handling."""
|
|
import aiohttp
|
|
|
|
if not _validate_url(url):
|
|
logger.warning(f"Invalid URL rejected: {url[:80]}")
|
|
return None
|
|
|
|
# Rate limit per host
|
|
parsed = urlparse(url)
|
|
host = parsed.netloc or "unknown"
|
|
await _rate_limit_host(host)
|
|
|
|
try:
|
|
async with (
|
|
aiohttp.ClientSession() as session,
|
|
session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as resp,
|
|
):
|
|
if resp.status == 200:
|
|
result: dict[str, Any] | list[Any] = await resp.json()
|
|
return result
|
|
elif resp.status == 429:
|
|
logger.warning(f"Rate limited by {host}, retrying after backoff")
|
|
await asyncio.sleep(3)
|
|
# One retry
|
|
async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as retry:
|
|
if retry.status == 200:
|
|
result2: dict[str, Any] | list[Any] = await retry.json()
|
|
return result2
|
|
return None
|
|
else:
|
|
logger.debug(f"HTTP {resp.status} for {url[:80]}")
|
|
return None
|
|
except TimeoutError:
|
|
logger.debug(f"Timeout fetching {url[:80]}")
|
|
return None
|
|
except Exception as e:
|
|
logger.debug(f"Error fetching {url[:80]}: {e}")
|
|
return None
|
|
|
|
|
|
async def _fetch_dexscreener_pairs(address: str) -> list[dict[str, Any]]:
|
|
"""Fetch pairs from DexScreener associated with an address."""
|
|
url = DEXSCREENER_SEARCH.format(quote(address))
|
|
data = await _fetch_json(url)
|
|
if not data or not isinstance(data, dict):
|
|
return []
|
|
pairs: list[dict[str, Any]] = data.get("pairs", []) or []
|
|
return pairs
|
|
|
|
|
|
async def _fetch_etherscan_txs(address: str, api_url: str) -> list[dict[str, Any]]:
|
|
"""Fetch transactions from block explorer to find contract creations."""
|
|
url = api_url.format(address)
|
|
data = await _fetch_json(url)
|
|
if not data or not isinstance(data, dict):
|
|
return []
|
|
txs: list[dict[str, Any]] = data.get("result", []) or []
|
|
return txs
|
|
|
|
|
|
def _is_contract_creation(tx: dict[str, Any]) -> bool:
|
|
"""Check if a transaction is a contract creation."""
|
|
# Contract creation has 'contractAddress' set
|
|
if tx.get("contractAddress"):
|
|
return True
|
|
# Also check if 'to' is empty (another creation indicator)
|
|
return bool(not tx.get("to") or tx.get("to") == "")
|
|
|
|
|
|
def _extract_token_from_pair(pair: dict[str, Any], deployer_address: str) -> DeployedToken | None:
|
|
"""
|
|
Extract token info from a DexScreener pair.
|
|
Only returns a token if the deployer_address matches the pair's token creator.
|
|
"""
|
|
chain = pair.get("chainId", "unknown")
|
|
base_token = pair.get("baseToken", {})
|
|
# Determine which token is the target (usually baseToken for new tokens)
|
|
token = base_token
|
|
token_address = token.get("address", "")
|
|
|
|
if not token_address:
|
|
return None
|
|
|
|
# Check if this deployer likely created the token
|
|
# DexScreener doesn't directly give deployer address, so we use maker/buyer patterns
|
|
pair_created_at = pair.get("pairCreatedAt", 0)
|
|
|
|
deployed = DeployedToken(
|
|
address=token_address,
|
|
chain=chain,
|
|
name=token.get("name", ""),
|
|
symbol=token.get("symbol", ""),
|
|
price_usd=float(pair.get("priceUsd", 0)),
|
|
liquidity_usd=float(pair.get("liquidity", {}).get("usd", 0)),
|
|
volume_24h=float(pair.get("volume", {}).get("h24", 0)),
|
|
holders=int(pair.get("fdv", 0) > 0), # rough proxy
|
|
is_verified=token.get("verified", False),
|
|
deploy_time=datetime.fromtimestamp(pair_created_at / 1000, tz=UTC).isoformat()
|
|
if pair_created_at
|
|
else "",
|
|
age_days=(datetime.now(UTC).timestamp() - pair_created_at / 1000) / 86400
|
|
if pair_created_at
|
|
else 0,
|
|
)
|
|
|
|
# Check for suspicious patterns in the pair
|
|
txns = pair.get("txns", {})
|
|
buys = (
|
|
txns.get("h24", {}).get("buys", 0)
|
|
if isinstance(txns, dict) and isinstance(txns.get("h24"), dict)
|
|
else 0
|
|
)
|
|
sells = (
|
|
txns.get("h24", {}).get("sells", 0)
|
|
if isinstance(txns, dict) and isinstance(txns.get("h24"), dict)
|
|
else 0
|
|
)
|
|
|
|
if buys == 0 and sells == 0 and deployed.age_days < 7:
|
|
deployed.flags.append("no_trading_activity")
|
|
deployed.is_active = False
|
|
|
|
if deployed.liquidity_usd < 100 and deployed.age_days > 1:
|
|
deployed.flags.append("critically_low_liquidity")
|
|
deployed.is_active = False
|
|
deployed.is_rug = True
|
|
|
|
if deployed.age_days < 1 and deployed.liquidity_usd > 0 and deployed.volume_24h == 0:
|
|
deployed.flags.append("just_launched_no_volume")
|
|
|
|
return deployed
|
|
|
|
|
|
async def _analyze_deployer_evm(
|
|
address: str, tokens: list[DeployedToken]
|
|
) -> tuple[list[DeployedToken], list[str]]:
|
|
"""Attempt to find contract deployments via block explorers."""
|
|
patterns: list[str] = []
|
|
explorers = [
|
|
("ethereum", ETHERSCAN_TXLIST),
|
|
("bsc", BSCSCAN_TXLIST),
|
|
("base", BASESCAN_TXLIST),
|
|
]
|
|
|
|
tasks = []
|
|
for _chain, url in explorers:
|
|
tasks.append(_fetch_etherscan_txs(address, url))
|
|
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
|
|
for (chain, _), txs in zip(explorers, results, strict=False):
|
|
if not txs or not isinstance(txs, list):
|
|
continue
|
|
|
|
for tx in txs:
|
|
if not _is_contract_creation(tx):
|
|
continue
|
|
|
|
contract_addr = tx.get("contractAddress", "")
|
|
if not contract_addr:
|
|
continue
|
|
|
|
# Check if we already have this token
|
|
if any(t.address.lower() == contract_addr.lower() for t in tokens):
|
|
continue
|
|
|
|
deployed = DeployedToken(
|
|
address=contract_addr,
|
|
chain=chain,
|
|
name=tx.get("tokenName", f"Token-{contract_addr[:8]}"),
|
|
symbol=tx.get("tokenSymbol", "???"),
|
|
deploy_tx=tx.get("hash", ""),
|
|
deploy_time=datetime.fromtimestamp(int(tx.get("timeStamp", 0)), tz=UTC).isoformat()
|
|
if tx.get("timeStamp")
|
|
else "",
|
|
is_verified=tx.get("isError", "0") == "0",
|
|
age_days=(
|
|
datetime.now(UTC) - datetime.fromtimestamp(int(tx.get("timeStamp", 0)), tz=UTC)
|
|
).total_seconds()
|
|
/ 86400
|
|
if tx.get("timeStamp")
|
|
else 0,
|
|
)
|
|
|
|
# If the tx failed, flag it
|
|
if tx.get("isError", "0") != "0":
|
|
deployed.flags.append("deploy_tx_failed")
|
|
deployed.is_active = False
|
|
|
|
tokens.append(deployed)
|
|
|
|
return tokens, patterns
|
|
|
|
|
|
# ── Chain detection ─────────────────────────────────────────────
|
|
|
|
|
|
def _detect_chain_from_address(address: str) -> list[str]:
|
|
"""Detect possible chains from address format."""
|
|
chains = []
|
|
if re.match(r"^0x[a-fA-F0-9]{40}$", address):
|
|
chains = ["ethereum", "bsc", "base", "polygon", "arbitrum", "optimism"]
|
|
elif re.match(r"^[1-9A-HJ-NP-Za-km-z]{32,44}$", address):
|
|
chains = ["solana"]
|
|
else:
|
|
chains = ["unknown"]
|
|
return chains
|
|
|
|
|
|
# ── Main analyzer class ─────────────────────────────────────────
|
|
|
|
|
|
class DeployerHistoryAnalyzer:
|
|
"""Analyzes the complete deployment history of a token creator address."""
|
|
|
|
def __init__(self, address: str):
|
|
if not re.match(r"^0x[a-fA-F0-9]{40}$", address) and not re.match(
|
|
r"^[1-9A-HJ-NP-Za-km-z]{32,44}$", address
|
|
):
|
|
raise ValueError(f"Invalid address format: {address[:20]}")
|
|
self.address = address.lower()
|
|
self.chains = _detect_chain_from_address(address)
|
|
self.is_evm = any(
|
|
c in self.chains for c in ["ethereum", "bsc", "base", "polygon", "arbitrum", "optimism"]
|
|
)
|
|
self.is_solana = "solana" in self.chains
|
|
|
|
async def analyze(self) -> dict[str, Any]:
|
|
"""
|
|
Full deployer history analysis.
|
|
|
|
Returns:
|
|
dict with deployer profile, token list, risk scoring, and recommendations.
|
|
"""
|
|
profile = DeployerProfile(address=self.address)
|
|
|
|
try:
|
|
tokens = await self._collect_tokens()
|
|
|
|
if not tokens:
|
|
profile.errors.append(
|
|
"No tokens found for this address. It may not be a deployer, "
|
|
"or the address is on an unsupported chain."
|
|
)
|
|
profile.risk_level = "unknown"
|
|
profile.recommendation = (
|
|
"Unable to find deployment history. The address may not be a known token "
|
|
"creator, or all deployments were on unsupported chains."
|
|
)
|
|
return self._to_dict(profile)
|
|
|
|
profile.tokens = tokens
|
|
profile.total_tokens_deployed = len(tokens)
|
|
profile.chains_used = list({t.chain for t in tokens})
|
|
|
|
# Compute derived metrics
|
|
active = [t for t in tokens if t.is_active and not t.is_rug]
|
|
rugs = [t for t in tokens if t.is_rug]
|
|
honeypots = [t for t in tokens if t.is_honeypot]
|
|
dead = [t for t in tokens if not t.is_active]
|
|
|
|
profile.active_tokens = len(active)
|
|
profile.dead_tokens = len(dead)
|
|
profile.rug_tokens = len(rugs)
|
|
profile.honeypot_tokens = len(honeypots)
|
|
|
|
# Lifespan
|
|
lifespans = [t.age_days for t in tokens if t.age_days > 0]
|
|
profile.avg_token_lifespan_days = sum(lifespans) / len(lifespans) if lifespans else 0
|
|
|
|
# First and last seen
|
|
timestamps = [t.deploy_time for t in tokens if t.deploy_time]
|
|
if timestamps:
|
|
timestamps.sort()
|
|
profile.first_seen = timestamps[0]
|
|
profile.last_active = timestamps[-1]
|
|
|
|
# Detect patterns
|
|
profile.patterns_detected = self._detect_patterns(profile)
|
|
|
|
# Final scoring
|
|
profile.risk_score = round(_compute_deployer_risk(profile), 1)
|
|
profile.risk_level = _classify_deployer_risk(profile.risk_score)
|
|
profile.recommendation = _generate_recommendation(profile, profile.risk_score)
|
|
profile.confidence = min(0.5 + (len(tokens) * 0.05), 0.95)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Deployer analysis failed: {e}", exc_info=True)
|
|
profile.errors.append(f"Analysis error: {str(e)[:200]}")
|
|
profile.risk_level = "error"
|
|
|
|
return self._to_dict(profile)
|
|
|
|
async def _collect_tokens(self) -> list[DeployedToken]:
|
|
"""Collect all tokens deployed by this address."""
|
|
all_tokens: list[DeployedToken] = []
|
|
seen_addresses: set[str] = set()
|
|
|
|
# Strategy 1: Query DexScreener
|
|
pairs = await _fetch_dexscreener_pairs(self.address)
|
|
for pair in pairs:
|
|
token = _extract_token_from_pair(pair, self.address)
|
|
if token and token.address not in seen_addresses:
|
|
seen_addresses.add(token.address)
|
|
all_tokens.append(token)
|
|
|
|
# Strategy 2: For EVM, check block explorers for contract creations
|
|
if self.is_evm:
|
|
all_tokens, _ = await _analyze_deployer_evm(self.address, all_tokens)
|
|
seen_addresses.update(t.address for t in all_tokens)
|
|
|
|
# Strategy 3: For Solana, check Solscan
|
|
if self.is_solana:
|
|
try:
|
|
sol_data = await _fetch_json(SOLSCAN_TOKEN_ACCOUNTS.format(self.address))
|
|
if sol_data and isinstance(sol_data, dict):
|
|
token_accounts = sol_data.get("data", [])
|
|
if isinstance(token_accounts, list):
|
|
for acct in token_accounts[:50]:
|
|
token_addr = (
|
|
acct.get("tokenAddress", "") if isinstance(acct, dict) else ""
|
|
)
|
|
if token_addr and token_addr not in seen_addresses:
|
|
seen_addresses.add(token_addr)
|
|
deployed = DeployedToken(
|
|
address=token_addr,
|
|
chain="solana",
|
|
name=acct.get("tokenName", "")
|
|
if isinstance(acct, dict)
|
|
else "",
|
|
symbol=acct.get("tokenSymbol", "")
|
|
if isinstance(acct, dict)
|
|
else "",
|
|
liquidity_usd=float(
|
|
acct.get("tokenAmount", {}).get("uiAmount", 0)
|
|
)
|
|
if isinstance(acct, dict)
|
|
else 0,
|
|
)
|
|
all_tokens.append(deployed)
|
|
except Exception as e:
|
|
logger.debug(f"Solscan fetch error: {e}")
|
|
|
|
return all_tokens
|
|
|
|
def _detect_patterns(self, profile: DeployerProfile) -> list[str]:
|
|
"""Detect known scam patterns from deployer behavior."""
|
|
patterns: list[str] = []
|
|
|
|
# Serial scammer pattern
|
|
if profile.rug_tokens >= 2 or (profile.rug_tokens >= 1 and profile.honeypot_tokens >= 1):
|
|
patterns.append("serial_scammer:multiple_rugs")
|
|
if (
|
|
profile.total_tokens_deployed >= 3
|
|
and profile.avg_token_lifespan_days < 14
|
|
and profile.dead_tokens / max(profile.total_tokens_deployed, 1) > 0.5
|
|
):
|
|
patterns.append("serial_scammer:short_lived_tokens")
|
|
|
|
# Rug pull pattern
|
|
if profile.rug_tokens > 0:
|
|
patterns.append("rug_pull:confirmed_rugs_in_history")
|
|
|
|
# Honeypot pattern
|
|
if profile.honeypot_tokens > 0:
|
|
patterns.append("honeypot:honeypot_deployments_detected")
|
|
|
|
# Pump and dump pattern
|
|
if profile.avg_token_lifespan_days < 3 and profile.total_tokens_deployed >= 2:
|
|
patterns.append("pump_dump:rapid_turnover")
|
|
|
|
# Suspicious renounce pattern
|
|
flags_all = [f for t in profile.tokens for f in t.flags]
|
|
if any("no_trading_activity" in f for f in flags_all):
|
|
patterns.append("suspicious_renounce:no_trading_after_launch")
|
|
|
|
# Cross-chain avoidance (deployer only on obscure chains)
|
|
if (
|
|
all(c in ("base", "polygon", "unknown") for c in profile.chains_used)
|
|
and profile.total_tokens_deployed >= 2
|
|
):
|
|
patterns.append("avoidance:limited_to_less_scanned_chains")
|
|
|
|
return list(set(patterns))
|
|
|
|
@staticmethod
|
|
def _to_dict(profile: DeployerProfile) -> dict[str, Any]:
|
|
"""Serialize profile to dict."""
|
|
return {
|
|
"address": profile.address,
|
|
"chains_used": profile.chains_used,
|
|
"total_tokens_deployed": profile.total_tokens_deployed,
|
|
"active_tokens": profile.active_tokens,
|
|
"dead_tokens": profile.dead_tokens,
|
|
"rug_tokens": profile.rug_tokens,
|
|
"honeypot_tokens": profile.honeypot_tokens,
|
|
"first_seen": profile.first_seen,
|
|
"last_active": profile.last_active,
|
|
"avg_token_lifespan_days": round(profile.avg_token_lifespan_days, 1),
|
|
"risk_score": profile.risk_score,
|
|
"risk_level": profile.risk_level,
|
|
"patterns_detected": profile.patterns_detected,
|
|
"confidence": round(profile.confidence, 2),
|
|
"recommendation": profile.recommendation,
|
|
"tokens": [
|
|
{
|
|
"address": t.address,
|
|
"chain": t.chain,
|
|
"name": t.name,
|
|
"symbol": t.symbol,
|
|
"deploy_time": t.deploy_time,
|
|
"liquidity_usd": round(t.liquidity_usd, 2),
|
|
"volume_24h": round(t.volume_24h, 2),
|
|
"price_usd": round(t.price_usd, 8),
|
|
"is_active": t.is_active,
|
|
"is_rug": t.is_rug,
|
|
"is_honeypot": t.is_honeypot,
|
|
"flags": t.flags,
|
|
"age_days": round(t.age_days, 1),
|
|
}
|
|
for t in profile.tokens
|
|
],
|
|
"errors": profile.errors,
|
|
"scanned_at": datetime.now(UTC).isoformat(),
|
|
}
|