1235 lines
52 KiB
Python
1235 lines
52 KiB
Python
"""
|
|
Rug Pull Imminence Predictor (RIP)
|
|
=====================================
|
|
AI-powered early warning system that predicts imminent rug pulls BEFORE they happen.
|
|
|
|
Architecture:
|
|
Fuses 7 signal categories into one predictive score:
|
|
1. LP HEALTH (25%) — Lock duration, depth changes, mint/burn events
|
|
2. DEPLOYER RISK (20%) — Previous rugs, funding patterns, dev wallet activity
|
|
3. SMART MONEY FLOW (15%) — Insider sells, dev exits, deployer→CEX transfers
|
|
4. BUNDLE PATTERN (15%) — Coordinated launch, identical amounts, common funder
|
|
5. SOCIAL VELOCITY (10%) — Sudden hype / shill spikes vs real engagement
|
|
6. CONTRACT RISK (10%) — Honeypot, ownership, proxy patterns
|
|
7. LIQUIDITY MIGRATION (5%) — LP moves from known to unknown, withdrawal patterns
|
|
|
|
Each signal has configurable weights and threshold-based scoring.
|
|
Produces: RIPScore (0-100), Imminence (LOW/MEDIUM/HIGH/CRITICAL), and
|
|
a human-readable risk narrative.
|
|
|
|
Competitive differentiator:
|
|
- No competitor predicts WHEN a rug will happen (DexScreener, Birdeye, TokenSniffer
|
|
only scan current state)
|
|
- Combines all RMI modules (smart money + defi auditor + bundle detector) into one
|
|
predictive model
|
|
- Real-time alerting via webhook for monitored tokens
|
|
|
|
Usage:
|
|
from app.rug_imminence_predictor import RugImminencePredictor
|
|
predictor = RugImminencePredictor()
|
|
result = await predictor.predict("0x1234...", chain="base")
|
|
print(result.score, result.verdict, result.narrative())
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime, timedelta
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Enums & Types
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
|
|
class ImminenceLevel(Enum):
|
|
LOW = "low"
|
|
MEDIUM = "medium"
|
|
HIGH = "high"
|
|
CRITICAL = "critical"
|
|
|
|
@property
|
|
def emoji(self) -> str:
|
|
return {"low": "🟢", "medium": "🟡", "high": "🟠", "critical": "🔴"}[self.value]
|
|
|
|
@property
|
|
def numeric(self) -> int:
|
|
return {"low": 1, "medium": 2, "high": 3, "critical": 4}[self.value]
|
|
|
|
|
|
@dataclass
|
|
class SignalCategory:
|
|
"""A single signal category with weight, score, and evidence."""
|
|
|
|
name: str
|
|
weight: float # 0-1, all weights sum to 1
|
|
score: float = 0.0 # 0-100
|
|
confidence: float = 0.0 # 0-1 (data availability)
|
|
evidence: list[str] = field(default_factory=list)
|
|
flags: list[str] = field(default_factory=list)
|
|
|
|
def weighted_contribution(self) -> float:
|
|
return self.score * self.weight
|
|
|
|
|
|
@dataclass
|
|
class RIPResult:
|
|
"""Complete rug imminence prediction result."""
|
|
|
|
token_address: str
|
|
chain: str
|
|
token_name: str = ""
|
|
token_symbol: str = ""
|
|
|
|
# Core prediction
|
|
score: float = 0.0 # 0-100
|
|
imminence: ImminenceLevel = ImminenceLevel.LOW
|
|
|
|
# Signal breakdown
|
|
signals: dict[str, SignalCategory] = field(default_factory=dict)
|
|
|
|
# Metadata
|
|
warnings: list[str] = field(default_factory=list)
|
|
recommendations: list[str] = field(default_factory=list)
|
|
last_updated: str = ""
|
|
scan_duration_ms: float = 0.0
|
|
|
|
# Data sources consulted
|
|
sources_used: list[str] = field(default_factory=list)
|
|
sources_failed: list[str] = field(default_factory=list)
|
|
|
|
def summary(self) -> str:
|
|
"""One-line summary for alerts."""
|
|
emoji = self.imminence.emoji
|
|
return (
|
|
f"{emoji} RIP {self.score:.0f}/100 — {self.token_symbol or self.token_address[:10]} "
|
|
f"on {self.chain}: {self.imminence.value.upper()} imminence. "
|
|
f"{len(self.warnings)} warnings, {len(self.recommendations)} recommendations."
|
|
)
|
|
|
|
def narrative(self, detailed: bool = False) -> str:
|
|
"""Human-readable narrative of the prediction."""
|
|
lines = [
|
|
f"{self.imminence.emoji} **Rug Pull Imminence Report**",
|
|
f"Token: {self.token_name or self.token_symbol or self.token_address[:18]}...",
|
|
f"Chain: {self.chain.upper()}",
|
|
f"Score: **{self.score:.0f}/100** — {self.imminence.value.upper()} imminence",
|
|
f"Scanned: {self.last_updated} ({self.scan_duration_ms:.0f}ms)",
|
|
]
|
|
|
|
if self.warnings:
|
|
lines.append("\n⚠️ **Warnings:**")
|
|
for w in self.warnings[:5]:
|
|
lines.append(f" • {w}")
|
|
|
|
if self.recommendations:
|
|
lines.append("\n💡 **Recommendations:**")
|
|
for r in self.recommendations[:5]:
|
|
lines.append(f" • {r}")
|
|
|
|
if detailed and self.signals:
|
|
lines.append("\n📊 **Signal Breakdown:**")
|
|
for name, sig in sorted(
|
|
self.signals.items(),
|
|
key=lambda x: x[1].weighted_contribution(),
|
|
reverse=True,
|
|
):
|
|
lines.append(
|
|
f" {name}: {sig.score:.0f}/100 (weight: {sig.weight:.0%}) "
|
|
f"→ contribution: {sig.weighted_contribution():.0f}"
|
|
)
|
|
for e in sig.evidence[:3]:
|
|
lines.append(f" └ {e}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""JSON-serializable dict."""
|
|
return {
|
|
"token_address": self.token_address,
|
|
"chain": self.chain,
|
|
"token_name": self.token_name,
|
|
"token_symbol": self.token_symbol,
|
|
"score": round(self.score, 1),
|
|
"imminence": self.imminence.value,
|
|
"verdict": self.imminence.value.upper(),
|
|
"warnings": self.warnings,
|
|
"recommendations": self.recommendations,
|
|
"last_updated": self.last_updated,
|
|
"scan_duration_ms": round(self.scan_duration_ms, 1),
|
|
"signals": {
|
|
k: {
|
|
"name": v.name,
|
|
"score": round(v.score, 1),
|
|
"weight": v.weight,
|
|
"confidence": round(v.confidence, 2),
|
|
"contribution": round(v.weighted_contribution(), 1),
|
|
"evidence": v.evidence,
|
|
"flags": v.flags,
|
|
}
|
|
for k, v in self.signals.items()
|
|
},
|
|
"sources_used": self.sources_used,
|
|
"sources_failed": self.sources_failed,
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Default signal weights (sum to 1.0)
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
DEFAULT_WEIGHTS = {
|
|
"lp_health": 0.25,
|
|
"deployer_risk": 0.20,
|
|
"smart_money_flow": 0.15,
|
|
"bundle_pattern": 0.15,
|
|
"social_velocity": 0.10,
|
|
"contract_risk": 0.10,
|
|
"liquidity_migration": 0.05,
|
|
}
|
|
|
|
SIGNAL_NAMES = {
|
|
"lp_health": "LP Health",
|
|
"deployer_risk": "Deployer Risk",
|
|
"smart_money_flow": "Smart Money Flow",
|
|
"bundle_pattern": "Bundle Pattern",
|
|
"social_velocity": "Social Velocity",
|
|
"contract_risk": "Contract Risk",
|
|
"liquidity_migration": "Liquidity Migration",
|
|
}
|
|
|
|
# Known safe LP lockers — loaded from env var for configurability.
|
|
# Format: comma-separated hex addresses (e.g., 0x1234...,0x5678...)
|
|
# Falls back to well-known lockers if env var is not set.
|
|
_KNOWN_LP_LOCKERS: list[str] | None = None
|
|
|
|
|
|
def _get_known_lp_lockers() -> list[str]:
|
|
"""Get configured LP lockers from env, or use well-known defaults."""
|
|
global _KNOWN_LP_LOCKERS
|
|
if _KNOWN_LP_LOCKERS is not None:
|
|
return _KNOWN_LP_LOCKERS
|
|
|
|
env_lockers = os.getenv("RIP_KNOWN_LP_LOCKERS", "")
|
|
if env_lockers.strip():
|
|
_KNOWN_LP_LOCKERS = [a.strip() for a in env_lockers.split(",") if a.strip()]
|
|
logger.info(f"Loaded {len(_KNOWN_LP_LOCKERS)} LP lockers from environment")
|
|
else:
|
|
_KNOWN_LP_LOCKERS = [
|
|
"0x407993575c91ce7643a4d4cCACc9A98c36eE1BBE", # Unicrypt
|
|
"0x663a5c229c09b049e36dCc11a9B0d4a8Eb9db214", # Unicrypt v2
|
|
"0x74de5d12FC0fC274C1b0C6E9F58E60e0b4B7eCb1", # Team Finance
|
|
"0xE2fE530C047f2d85298b07D9333C05737f1435FB", # Mudra
|
|
"0x6b8DA0E0E1d7b8B0E0E1d7b8B0E0E1d7b8B0E0E1d", # PinkLock (example)
|
|
"0x4e59b44847b379578588920cA78FbF26c0B4956C", # DXlock
|
|
"0xD152f549545093347A162Dce210e7293f1452150", # TrustSwap
|
|
]
|
|
logger.info("Using default LP lockers list")
|
|
return _KNOWN_LP_LOCKERS
|
|
|
|
|
|
# Risk thresholds
|
|
SCORE_THRESHOLDS = {
|
|
ImminenceLevel.LOW: (0, 30),
|
|
ImminenceLevel.MEDIUM: (30, 55),
|
|
ImminenceLevel.HIGH: (55, 75),
|
|
ImminenceLevel.CRITICAL: (75, 101),
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Main Predictor
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
|
|
class RugImminencePredictor:
|
|
"""Predicts imminent rug pulls by fusing 7 signal categories."""
|
|
|
|
def __init__(self, weights: dict[str, float] | None = None):
|
|
self.weights = weights or dict(DEFAULT_WEIGHTS)
|
|
self._verify_weights()
|
|
self._cache: dict[str, RIPResult] = {}
|
|
self._cache_ttl = 300 # 5 minutes
|
|
logger.info(f"RugImminencePredictor initialized with {len(self.weights)} signals")
|
|
|
|
def _verify_weights(self):
|
|
"""Ensure weights sum to ~1.0."""
|
|
total = sum(self.weights.values())
|
|
if abs(total - 1.0) > 0.01:
|
|
logger.warning(f"Signal weights sum to {total:.2f}, not 1.0. Normalizing.")
|
|
for k in self.weights:
|
|
self.weights[k] /= total
|
|
|
|
def _cache_key(self, address: str, chain: str) -> str:
|
|
return f"{chain}:{address.lower()}"
|
|
|
|
def _get_cached(self, address: str, chain: str) -> RIPResult | None:
|
|
"""Get cached result if still valid (compares expiry timestamps properly)."""
|
|
key = self._cache_key(address, chain)
|
|
result = self._cache.get(key)
|
|
if result:
|
|
try:
|
|
expiry = datetime.fromisoformat(result.last_updated)
|
|
if datetime.now(UTC) < expiry:
|
|
return result
|
|
except (ValueError, TypeError):
|
|
pass
|
|
# Expired or invalid timestamp — remove from cache
|
|
del self._cache[key]
|
|
return None
|
|
|
|
def _set_cache(self, result: RIPResult):
|
|
"""Cache result with TTL as absolute expiry timestamp."""
|
|
key = self._cache_key(result.token_address, result.chain)
|
|
result.last_updated = (datetime.now(UTC) + timedelta(seconds=self._cache_ttl)).isoformat()
|
|
self._cache[key] = result
|
|
|
|
# Evict old entries if cache grows too large (prevent DoS via memory exhaustion)
|
|
if len(self._cache) > 500:
|
|
# Remove oldest 25% of entries
|
|
datetime.now(UTC)
|
|
to_evict = sorted(
|
|
self._cache.items(),
|
|
key=lambda x: x[1].last_updated,
|
|
)[: len(self._cache) // 4]
|
|
for k, _ in to_evict:
|
|
del self._cache[k]
|
|
logger.info(
|
|
f"Cache eviction: removed {len(to_evict)} entries, {len(self._cache)} remaining"
|
|
)
|
|
|
|
async def predict(
|
|
self,
|
|
address: str,
|
|
chain: str = "base",
|
|
force_refresh: bool = False,
|
|
) -> RIPResult:
|
|
"""
|
|
Full rug imminence prediction for a token.
|
|
|
|
Runs all 7 signal analyzers in parallel, then fuses results
|
|
into a weighted score with imminence classification.
|
|
|
|
Raises:
|
|
ValueError: If address or chain is invalid.
|
|
"""
|
|
# ── Input validation ──────────────────────────────────
|
|
address = address.strip()
|
|
chain = chain.strip().lower()
|
|
|
|
valid_chains = {
|
|
"base",
|
|
"solana",
|
|
"ethereum",
|
|
"bsc",
|
|
"polygon",
|
|
"arbitrum",
|
|
"optimism",
|
|
"avalanche",
|
|
"fantom",
|
|
"gnosis",
|
|
}
|
|
if chain not in valid_chains:
|
|
raise ValueError(
|
|
f"Unsupported chain '{chain}'. Supported: {', '.join(sorted(valid_chains))}"
|
|
)
|
|
|
|
# Validate address format (EVM hex or Solana base58)
|
|
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]}...'. "
|
|
"Expected 0x-prefixed EVM address (42 chars) or Solana base58 address."
|
|
)
|
|
|
|
t0 = time.time()
|
|
|
|
# Check cache
|
|
if not force_refresh:
|
|
cached = self._get_cached(address, chain)
|
|
if cached:
|
|
return cached
|
|
|
|
# Build result container
|
|
result = RIPResult(
|
|
token_address=address,
|
|
chain=chain,
|
|
last_updated=datetime.now(UTC).isoformat(),
|
|
)
|
|
|
|
# Initialize signal categories
|
|
for sig_id, weight in self.weights.items():
|
|
result.signals[sig_id] = SignalCategory(
|
|
name=SIGNAL_NAMES.get(sig_id, sig_id),
|
|
weight=weight,
|
|
)
|
|
|
|
# Run all signal analyzers in parallel
|
|
analyzers = [
|
|
self._analyze_lp_health(result),
|
|
self._analyze_deployer_risk(result),
|
|
self._analyze_smart_money_flow(result),
|
|
self._analyze_bundle_pattern(result),
|
|
self._analyze_social_velocity(result),
|
|
self._analyze_contract_risk(result),
|
|
self._analyze_liquidity_migration(result),
|
|
]
|
|
await asyncio.gather(*analyzers, return_exceptions=True)
|
|
|
|
# Fuse signals into final score
|
|
result.score = sum(sig.weighted_contribution() for sig in result.signals.values())
|
|
|
|
# Determine imminence level
|
|
for level, (lo, hi) in SCORE_THRESHOLDS.items():
|
|
if lo <= result.score < hi:
|
|
result.imminence = level
|
|
break
|
|
|
|
# Generate warnings and recommendations
|
|
self._generate_warnings(result)
|
|
self._generate_recommendations(result)
|
|
|
|
result.scan_duration_ms = (time.time() - t0) * 1000
|
|
self._set_cache(result)
|
|
return result
|
|
|
|
# ── Signal Analyzers ───────────────────────────────────────
|
|
|
|
async def _analyze_lp_health(self, result: RIPResult) -> None:
|
|
"""
|
|
Signal 1: LP Health (25% weight).
|
|
|
|
Checks:
|
|
- LP lock status (locked vs unlocked)
|
|
- LP lock duration remaining
|
|
- Recent LP changes (additions/removals)
|
|
- Liquidity depth vs volume ratio
|
|
- Single-sided LP additions (danger signal)
|
|
"""
|
|
sig = result.signals["lp_health"]
|
|
try:
|
|
# Try DexScreener for pool data
|
|
from app.caching_shield.service_mcp import get_service_mcp
|
|
|
|
get_service_mcp()
|
|
|
|
dex_data = None
|
|
result.sources_used.append("dexscreener")
|
|
try:
|
|
dex_data = await self._fetch_dexscreener_pool(result.token_address, result.chain)
|
|
except Exception:
|
|
result.sources_failed.append("dexscreener")
|
|
|
|
if dex_data and dex_data.get("pairs"):
|
|
pair = dex_data["pairs"][0]
|
|
liquidity_usd = (
|
|
float(pair.get("liquidity", {}).get("usd", 0))
|
|
if isinstance(pair.get("liquidity"), dict)
|
|
else float(pair.get("liquidity", 0))
|
|
)
|
|
volume_24h = (
|
|
float(pair.get("volume", {}).get("h24", 0))
|
|
if isinstance(pair.get("volume"), dict)
|
|
else float(pair.get("volume", 0))
|
|
)
|
|
txns_24h = (
|
|
pair.get("txns", {}).get("h24", {})
|
|
if isinstance(pair.get("txns"), dict)
|
|
else {}
|
|
)
|
|
buys = int(txns_24h.get("buys", 0)) if isinstance(txns_24h, dict) else 0
|
|
sells = int(txns_24h.get("sells", 0)) if isinstance(txns_24h, dict) else 0
|
|
|
|
# Low liquidity = high risk
|
|
if liquidity_usd < 1000:
|
|
sig.score += 40
|
|
sig.flags.append("critically_low_liquidity")
|
|
sig.evidence.append(f"Liquidity ${liquidity_usd:.0f} < $1K — extreme risk")
|
|
elif liquidity_usd < 10000:
|
|
sig.score += 20
|
|
sig.flags.append("low_liquidity")
|
|
sig.evidence.append(f"Liquidity ${liquidity_usd:.0f} < $10K — low")
|
|
elif liquidity_usd > 100000:
|
|
sig.score -= 15
|
|
sig.evidence.append(f"Liquidity ${liquidity_usd:.0f} — healthy")
|
|
|
|
# Volume vs liquidity ratio (high ratio = potential manipulation)
|
|
if liquidity_usd > 0 and volume_24h > 0:
|
|
vol_liq_ratio = volume_24h / liquidity_usd
|
|
if vol_liq_ratio > 10:
|
|
sig.score += 15
|
|
sig.flags.append("high_volume_liquidity_ratio")
|
|
sig.evidence.append(
|
|
f"Volume/Liquidity ratio {vol_liq_ratio:.1f}x — potential wash trading"
|
|
)
|
|
elif vol_liq_ratio < 0.1:
|
|
sig.score += 10
|
|
sig.flags.append("low_activity")
|
|
sig.evidence.append("Volume/Liquidity ratio < 0.1x — dead token")
|
|
|
|
# Sell/buy ratio > 2 = distribution
|
|
if buys > 0 and sells > 0:
|
|
sell_buy_ratio = sells / buys
|
|
if sell_buy_ratio > 2:
|
|
sig.score += 15
|
|
sig.flags.append("sell_pressure")
|
|
sig.evidence.append(
|
|
f"Sell/Buy ratio {sell_buy_ratio:.1f}x — distribution pressure"
|
|
)
|
|
elif sell_buy_ratio < 0.3:
|
|
sig.score += 5
|
|
sig.flags.append("buy_pressure")
|
|
sig.evidence.append(
|
|
f"Buy pressure dominates ({sell_buy_ratio:.1f}x) — potential manipulation"
|
|
)
|
|
|
|
# Try to get LP lock info (via contract analysis)
|
|
try:
|
|
lock_info = await self._check_lp_lock(result.token_address, result.chain)
|
|
if lock_info.get("locked"):
|
|
sig.score -= 20
|
|
sig.evidence.append(
|
|
f"LP locked until {lock_info.get('unlock_date', 'unknown')}"
|
|
)
|
|
elif lock_info.get("unlocked") is True:
|
|
sig.score += 25
|
|
sig.flags.append("unlocked_liquidity")
|
|
sig.evidence.append("LP UNLOCKED — liquidity can be pulled at any time")
|
|
else:
|
|
sig.score += 10
|
|
sig.flags.append("unknown_lock_status")
|
|
sig.evidence.append("LP lock status unknown — assume unlocked")
|
|
except Exception:
|
|
sig.score += 5 # Penalize for missing data
|
|
result.sources_failed.append("lp_lock_check")
|
|
|
|
else:
|
|
sig.score += 15 # No DexScreener data = suspicious
|
|
sig.evidence.append("No DEX pool data found — token may have no real liquidity")
|
|
|
|
sig.confidence = min(1.0, sig.score / 50) if sig.score > 0 else 0.3
|
|
|
|
except Exception as e:
|
|
logger.warning(f"LP health analysis failed for {result.token_address}: {e}")
|
|
sig.evidence.append(f"Analysis error: {str(e)[:60]}")
|
|
sig.confidence = 0.1
|
|
result.sources_failed.append("lp_health_analysis")
|
|
|
|
async def _analyze_deployer_risk(self, result: RIPResult) -> None:
|
|
"""
|
|
Signal 2: Deployer Risk (20% weight).
|
|
|
|
Checks:
|
|
- Deployer wallet reputation (previous rugs)
|
|
- Deployer funding source (CEX vs private wallet)
|
|
- Deployer activity (recent deployments, exits)
|
|
- Time since deployer creation (new wallet = risky)
|
|
"""
|
|
sig = result.signals["deployer_risk"]
|
|
try:
|
|
deployer = await self._resolve_deployer(result.token_address, result.chain)
|
|
if not deployer:
|
|
sig.score += 10 # Unknown deployer = risk
|
|
sig.evidence.append("Could not resolve deployer address")
|
|
sig.confidence = 0.3
|
|
result.sources_failed.append("deployer_resolution")
|
|
return
|
|
|
|
result.sources_used.append("deployer_analysis")
|
|
|
|
# Check deployer age (new wallets are risky)
|
|
age_info = await self._check_wallet_age(deployer, result.chain)
|
|
if age_info:
|
|
sig.evidence.append(
|
|
f"Deployer wallet created {age_info.get('age_days', '?')} days ago"
|
|
)
|
|
if age_info.get("age_days", 999) < 7:
|
|
sig.score += 25
|
|
sig.flags.append("new_deployer_wallet")
|
|
sig.evidence.append(
|
|
"Deployer wallet < 7 days old — HIGH risk of disposable wallet"
|
|
)
|
|
elif age_info.get("age_days", 999) < 30:
|
|
sig.score += 15
|
|
sig.flags.append("recent_deployer_wallet")
|
|
else:
|
|
sig.score -= 15
|
|
sig.evidence.append("Established deployer wallet (>30 days)")
|
|
|
|
# Check deployer token history
|
|
history = await self._check_deployer_history(deployer, result.chain)
|
|
total_tokens = history.get("total_deployed", 0)
|
|
rug_count = history.get("rug_tokens", 0)
|
|
active_tokens = history.get("active_tokens", 0)
|
|
|
|
if rug_count > 0:
|
|
sig.score += 30
|
|
sig.flags.append("known_rug_deployer")
|
|
sig.evidence.append(
|
|
f"Deployer has {rug_count} previous rug pull(s) out of {total_tokens} tokens"
|
|
)
|
|
elif rug_count == 0 and total_tokens > 0:
|
|
sig.score -= 10
|
|
sig.evidence.append(
|
|
f"Clean deployer history ({total_tokens} tokens, no known rugs)"
|
|
)
|
|
|
|
if total_tokens > 5:
|
|
sig.score += 10
|
|
sig.flags.append("serial_deployer")
|
|
sig.evidence.append(
|
|
f"Deployer created {total_tokens} tokens — potential serial deployer"
|
|
)
|
|
|
|
if active_tokens == 0 and total_tokens > 0:
|
|
sig.score += 10
|
|
sig.evidence.append("All previously deployed tokens are inactive/dead")
|
|
|
|
sig.confidence = min(0.9, 0.4 + (total_tokens / 20))
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Deployer risk analysis failed: {e}")
|
|
sig.evidence.append(f"Analysis error: {str(e)[:60]}")
|
|
sig.confidence = 0.2
|
|
result.sources_failed.append("deployer_risk_analysis")
|
|
|
|
async def _analyze_smart_money_flow(self, result: RIPResult) -> None:
|
|
"""
|
|
Signal 3: Smart Money Flow (15% weight).
|
|
|
|
Checks:
|
|
- Insider sells detected by smart_money_tracker
|
|
- Deployer→CEX transfers (cashing out)
|
|
- Whale position reductions
|
|
- Abnormal holder count changes
|
|
"""
|
|
sig = result.signals["smart_money_flow"]
|
|
try:
|
|
# Try smart money tracker for relevant moves
|
|
try:
|
|
from app.smart_money_tracker import SmartMoneyTracker
|
|
|
|
tracker = SmartMoneyTracker()
|
|
moves = await tracker.scan_for_token(
|
|
result.token_address, result.chain, lookback_hours=24
|
|
)
|
|
result.sources_used.append("smart_money_tracker")
|
|
|
|
if moves:
|
|
insider_sells = [
|
|
m
|
|
for m in moves
|
|
if m.get("type") == "sell" and m.get("category") == "insider"
|
|
]
|
|
whale_sells = [
|
|
m for m in moves if m.get("type") == "sell" and m.get("category") == "whale"
|
|
]
|
|
deployer_moves = [
|
|
m for m in moves if m.get("type") == "transfer" and m.get("to_exchange")
|
|
]
|
|
|
|
if insider_sells:
|
|
total_insider = sum(m.get("value_usd", 0) for m in insider_sells)
|
|
sig.score += 20
|
|
sig.flags.append("insider_selling")
|
|
sig.evidence.append(
|
|
f"{len(insider_sells)} insider sell(s) detected (${total_insider:,.0f} total)"
|
|
)
|
|
|
|
if whale_sells:
|
|
sig.score += 10
|
|
sig.flags.append("whale_selling")
|
|
sig.evidence.append(f"{len(whale_sells)} whale(s) reducing position")
|
|
|
|
if deployer_moves:
|
|
sig.score += 25
|
|
sig.flags.append("deployer_cashing_out")
|
|
sig.evidence.append(
|
|
f"Deployer sent funds to exchange ({len(deployer_moves)} txns)"
|
|
)
|
|
except (ImportError, AttributeError) as e:
|
|
logger.debug(f"Smart money tracker unavailable: {e}")
|
|
result.sources_failed.append("smart_money_tracker")
|
|
|
|
# Check holder count changes
|
|
try:
|
|
holder_data = await self._fetch_holder_data(result.token_address, result.chain)
|
|
if holder_data:
|
|
result.sources_used.append("holder_data")
|
|
prev_count = holder_data.get("holders_24h_ago", 0)
|
|
curr_count = holder_data.get("holders_now", 0)
|
|
|
|
if prev_count > 0 and curr_count < prev_count * 0.8:
|
|
sig.score += 15
|
|
sig.flags.append("holder_exodus")
|
|
sig.evidence.append(
|
|
f"Holders dropped {(1 - curr_count / prev_count) * 100:.0f}% in 24h"
|
|
)
|
|
|
|
if holder_data.get("top10_pct", 0) > 80:
|
|
sig.score += 15
|
|
sig.flags.append("concentrated_holdings")
|
|
sig.evidence.append(
|
|
f"Top 10 holders own {holder_data['top10_pct']:.0f}% of supply"
|
|
)
|
|
except Exception:
|
|
result.sources_failed.append("holder_data")
|
|
|
|
sig.confidence = min(0.8, 0.3 + (sig.score / 50))
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Smart money flow analysis failed: {e}")
|
|
sig.evidence.append(f"Analysis error: {str(e)[:60]}")
|
|
sig.confidence = 0.15
|
|
result.sources_failed.append("smart_money_flow_analysis")
|
|
|
|
async def _analyze_bundle_pattern(self, result: RIPResult) -> None:
|
|
"""
|
|
Signal 4: Bundle Pattern (15% weight).
|
|
|
|
Checks:
|
|
- Atomic block co-occurrence
|
|
- Common funder for initial buyers
|
|
- Identical purchase amounts
|
|
- Distribution anomaly
|
|
"""
|
|
sig = result.signals["bundle_pattern"]
|
|
try:
|
|
from app.bundle_detector import BundleDetector
|
|
|
|
detector = BundleDetector()
|
|
detection = await detector.detect(result.token_address, result.chain)
|
|
result.sources_used.append("bundle_detector")
|
|
|
|
if detection.is_bundled:
|
|
sig.score += detection.confidence * 50
|
|
sig.flags.append("bundled_launch")
|
|
sig.evidence.append(f"Bundle detected (confidence: {detection.confidence:.0%})")
|
|
|
|
if detection.common_funder_score > 0.7:
|
|
sig.score += 10
|
|
sig.evidence.append("All initial buyers funded by same wallet")
|
|
|
|
if detection.top10_holder_pct > 90:
|
|
sig.score += 10
|
|
sig.evidence.append(
|
|
f"Top 10 holders: {detection.top10_holder_pct:.0f}% — extreme concentration"
|
|
)
|
|
|
|
if detection.identical_amount_count > 3:
|
|
sig.score += 10
|
|
sig.evidence.append(
|
|
f"{detection.identical_amount_count} identical purchases — coordinated"
|
|
)
|
|
else:
|
|
sig.score -= 10 # Clean launch = lower risk
|
|
sig.evidence.append("No bundle patterns detected")
|
|
|
|
sig.confidence = detection.confidence if detection.is_bundled else 0.6
|
|
|
|
except ImportError:
|
|
logger.debug("BundleDetector not available")
|
|
sig.evidence.append("Bundle detection unavailable")
|
|
sig.confidence = 0.3
|
|
result.sources_failed.append("bundle_detector")
|
|
except Exception as e:
|
|
logger.warning(f"Bundle analysis failed: {e}")
|
|
sig.evidence.append(f"Analysis error: {str(e)[:60]}")
|
|
sig.confidence = 0.2
|
|
result.sources_failed.append("bundle_pattern_analysis")
|
|
|
|
async def _analyze_social_velocity(self, result: RIPResult) -> None:
|
|
"""
|
|
Signal 5: Social Velocity (10% weight).
|
|
|
|
Checks:
|
|
- Sudden social mentions spike (hype vs baseline)
|
|
- Shill detection (same message across channels)
|
|
- Sentiment polarity change
|
|
- Influencer/KOL mentions correlation
|
|
"""
|
|
sig = result.signals["social_velocity"]
|
|
try:
|
|
# Try social velocity analyzer
|
|
try:
|
|
from app.scanners.social_velocity import SocialVelocityAnalyzer
|
|
|
|
analyzer = SocialVelocityAnalyzer()
|
|
velocity = await analyzer.analyze(result.token_address)
|
|
result.sources_used.append("social_velocity_analyzer")
|
|
|
|
if velocity:
|
|
mention_spike = velocity.get("mention_spike_pct", 0)
|
|
shill_score = velocity.get("shill_score", 0)
|
|
sentiment_change = velocity.get("sentiment_shift", 0)
|
|
|
|
if mention_spike > 500:
|
|
sig.score += 15
|
|
sig.flags.append("massive_mention_spike")
|
|
sig.evidence.append(
|
|
f"Social mentions spiked {mention_spike:.0f}% — suspicious coordinated hype"
|
|
)
|
|
elif mention_spike > 200:
|
|
sig.score += 8
|
|
sig.flags.append("mention_spike")
|
|
sig.evidence.append(f"Social mentions up {mention_spike:.0f}%")
|
|
|
|
if shill_score > 0.7:
|
|
sig.score += 10
|
|
sig.flags.append("coordinated_shilling")
|
|
sig.evidence.append(
|
|
"Coordinated shilling detected (same messages across channels)"
|
|
)
|
|
|
|
if sentiment_change < -0.3:
|
|
sig.score += 10
|
|
sig.flags.append("sentiment_crash")
|
|
sig.evidence.append(f"Sentiment dropped {sentiment_change:.0%}")
|
|
except (ImportError, AttributeError) as e:
|
|
logger.debug(f"Social velocity analyzer unavailable: {e}")
|
|
result.sources_failed.append("social_velocity_analyzer")
|
|
|
|
# Check for recent FUD/panic signals
|
|
try:
|
|
from app.scanners.sentiment_analyzer import SentimentAnalyzer
|
|
|
|
sentiment = SentimentAnalyzer()
|
|
sent_result = await sentiment.analyze_token_sentiment(
|
|
result.token_symbol or result.token_address
|
|
)
|
|
result.sources_used.append("sentiment_analyzer")
|
|
|
|
if sent_result and sent_result.get("fear_index", 0) > 70:
|
|
sig.score += 5
|
|
sig.flags.append("high_fear")
|
|
sig.evidence.append("High fear sentiment in community")
|
|
except (ImportError, AttributeError):
|
|
result.sources_failed.append("sentiment_analyzer")
|
|
|
|
sig.confidence = min(0.7, 0.3 + (sig.score / 40))
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Social velocity analysis failed: {e}")
|
|
sig.evidence.append(f"Analysis error: {str(e)[:60]}")
|
|
sig.confidence = 0.15
|
|
result.sources_failed.append("social_velocity_analysis")
|
|
|
|
async def _analyze_contract_risk(self, result: RIPResult) -> None:
|
|
"""
|
|
Signal 6: Contract Risk (10% weight).
|
|
|
|
Checks:
|
|
- Honeypot detection (can you sell?)
|
|
- Ownership status (renounced vs owned)
|
|
- Proxy patterns (upgradable = dangerous)
|
|
- High-risk functions (mint, blacklist, fees)
|
|
- Verified source code
|
|
"""
|
|
sig = result.signals["contract_risk"]
|
|
try:
|
|
# Try honeypot detector
|
|
try:
|
|
from app.scanners.honeypot_detector import HoneypotDetector
|
|
|
|
hp = HoneypotDetector()
|
|
hp_result = await hp.detect(result.token_address, result.chain)
|
|
result.sources_used.append("honeypot_detector")
|
|
|
|
if hp_result and hp_result.get("is_honeypot", False):
|
|
sig.score += 30
|
|
sig.flags.append("honeypot")
|
|
sig.evidence.append(
|
|
f"HONEYPOT DETECTED: {hp_result.get('reason', 'Cannot sell')}"
|
|
)
|
|
elif hp_result and hp_result.get("suspicious", False):
|
|
sig.score += 15
|
|
sig.flags.append("suspicious_contract")
|
|
sig.evidence.append(
|
|
f"Suspicious contract behavior: {hp_result.get('reason', '')}"
|
|
)
|
|
else:
|
|
sig.score -= 10
|
|
sig.evidence.append("No honeypot detected")
|
|
except (ImportError, AttributeError):
|
|
result.sources_failed.append("honeypot_detector")
|
|
|
|
# Check ownership status
|
|
try:
|
|
from app.scanners.contract_authority import ContractAuthorityScanner
|
|
|
|
auth = ContractAuthorityScanner()
|
|
authority = await auth.scan(result.token_address, result.chain)
|
|
result.sources_used.append("contract_authority")
|
|
|
|
if authority:
|
|
if authority.get("ownership_renounced", False):
|
|
sig.score -= 15
|
|
sig.evidence.append("Ownership renounced — lower risk")
|
|
else:
|
|
sig.score += 15
|
|
sig.flags.append("active_ownership")
|
|
sig.evidence.append("Ownership NOT renounced — owner can manipulate")
|
|
|
|
if authority.get("has_proxy", False):
|
|
sig.score += 10
|
|
sig.flags.append("upgradable_proxy")
|
|
sig.evidence.append("Upgradable proxy contract — implementation can change")
|
|
except (ImportError, AttributeError):
|
|
result.sources_failed.append("contract_authority")
|
|
|
|
sig.confidence = min(0.85, 0.4 + (sig.score / 40))
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Contract risk analysis failed: {e}")
|
|
sig.evidence.append(f"Analysis error: {str(e)[:60]}")
|
|
sig.confidence = 0.2
|
|
result.sources_failed.append("contract_risk_analysis")
|
|
|
|
async def _analyze_liquidity_migration(self, result: RIPResult) -> None:
|
|
"""
|
|
Signal 7: Liquidity Migration (5% weight).
|
|
|
|
Checks:
|
|
- LP moves between pools
|
|
- LP withdrawal patterns
|
|
- Large liquidity removal events
|
|
- LP token transfers to exchanges
|
|
"""
|
|
sig = result.signals["liquidity_migration"]
|
|
try:
|
|
# Check recent LP changes via on-chain data
|
|
try:
|
|
from app.scanners.liquidity_verifier import LiquidityVerifier
|
|
|
|
verifier = LiquidityVerifier()
|
|
lp_status = await verifier.verify(
|
|
result.token_address, result.chain, lookback_hours=24
|
|
)
|
|
result.sources_used.append("liquidity_verifier")
|
|
|
|
if lp_status:
|
|
if lp_status.get("lp_removed", False):
|
|
sig.score += 30
|
|
sig.flags.append("lp_removed")
|
|
amount = lp_status.get("amount_removed_usd", 0)
|
|
sig.evidence.append(f"LP REMOVED! ${amount:,.0f} withdrawn in last 24h")
|
|
elif lp_status.get("lp_decreased", False):
|
|
sig.score += 20
|
|
sig.flags.append("lp_decreasing")
|
|
sig.evidence.append("LP decreased significantly in last 24h")
|
|
|
|
if lp_status.get("single_sided_add", False):
|
|
sig.score += 10
|
|
sig.flags.append("single_sided_lp")
|
|
sig.evidence.append("Single-sided LP addition — potential manipulation")
|
|
|
|
if lp_status.get("lp_to_cex", False):
|
|
sig.score += 25
|
|
sig.flags.append("lp_to_exchange")
|
|
sig.evidence.append("LP tokens transferred to CEX — preparing to exit")
|
|
|
|
else:
|
|
sig.evidence.append("No recent LP migration detected")
|
|
except (ImportError, AttributeError):
|
|
result.sources_failed.append("liquidity_verifier")
|
|
|
|
# Fallback: check via DexScreener if verifier unavailable
|
|
if not result.signals["liquidity_migration"].evidence:
|
|
try:
|
|
dex_data = await self._fetch_dexscreener_pool(
|
|
result.token_address, result.chain
|
|
)
|
|
if dex_data and dex_data.get("pairs"):
|
|
pair = dex_data["pairs"][0]
|
|
liq_change = (
|
|
pair.get("liquidity", {}).get("change_24h", 0)
|
|
if isinstance(pair.get("liquidity"), dict)
|
|
else 0
|
|
)
|
|
if liq_change < -50:
|
|
sig.score += 15
|
|
sig.flags.append("liquidity_drop")
|
|
sig.evidence.append(f"Liquidity dropped {abs(liq_change):.0f}% in 24h")
|
|
except Exception:
|
|
pass
|
|
|
|
sig.confidence = min(0.7, 0.3 + (sig.score / 40))
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Liquidity migration analysis failed: {e}")
|
|
sig.evidence.append(f"Analysis error: {str(e)[:60]}")
|
|
sig.confidence = 0.15
|
|
result.sources_failed.append("liquidity_migration_analysis")
|
|
|
|
# ── Helper Methods ─────────────────────────────────────────
|
|
|
|
def _generate_warnings(self, result: RIPResult) -> None:
|
|
"""Generate human-readable warnings from flags."""
|
|
flag_warnings = {
|
|
"honeypot": "⚠️ HONEYPOT: Token cannot be sold — users are trapped",
|
|
"unlocked_liquidity": "⚠️ LP UNLOCKED: Creator can pull all liquidity at any moment",
|
|
"lp_removed": "⚠️ LP REMOVED: Liquidity has been withdrawn — imminent rug",
|
|
"known_rug_deployer": "⚠️ KNOWN RUG DEPLOYER: This wallet has rugged before",
|
|
"deployer_cashing_out": "⚠️ DEPLOYER CASHING OUT: Sending funds to exchanges",
|
|
"insider_selling": "⚠️ INSIDER SELLING: Team/insiders are dumping",
|
|
"bundled_launch": "⚠️ BUNDLED LAUNCH: Coordinated token launch",
|
|
"coordinated_shilling": "⚠️ COORDINATED SHILLING: Artificial social hype",
|
|
"active_ownership": "⚠️ ACTIVE OWNERSHIP: Owner can modify contract",
|
|
"upgradable_proxy": "⚠️ UPGRADABLE PROXY: Contract can be changed",
|
|
"new_deployer_wallet": "⚠️ NEW DEPLOYER: Disposable wallet, < 7 days old",
|
|
"single_sided_lp": "⚠️ SINGLE-SIDED LP: Potential price manipulation",
|
|
"lp_to_exchange": "⚠️ LP TO CEX: Liquidity tokens sent to exchange",
|
|
"sell_pressure": "⚠️ SELL PRESSURE: More sells than buys",
|
|
"massive_mention_spike": "⚠️ HYPE SPIKE: Social mentions up 500%+ — suspicious",
|
|
"serial_deployer": "⚠️ SERIAL DEPLOYER: Created 5+ tokens",
|
|
"concentrated_holdings": "⚠️ CONCENTRATED HOLDINGS: Top 10 own >80% of supply",
|
|
"holder_exodus": "⚠️ HOLDER EXODUS: Large drop in holder count",
|
|
}
|
|
|
|
seen = set()
|
|
for sig in result.signals.values():
|
|
for flag in sig.flags:
|
|
if flag in flag_warnings and flag not in seen:
|
|
result.warnings.append(flag_warnings[flag])
|
|
seen.add(flag)
|
|
|
|
if not result.warnings:
|
|
if result.score < 30:
|
|
result.warnings.append("✅ No imminent rug indicators detected")
|
|
else:
|
|
result.warnings.append("🔍 Some risk signals present — monitor closely")
|
|
|
|
def _generate_recommendations(self, result: RIPResult) -> None:
|
|
"""Generate actionable recommendations from flags."""
|
|
flag_recos = {
|
|
"unlocked_liquidity": "Sell immediately if holding — LP can be pulled any moment",
|
|
"lp_removed": "EXIT NOW — liquidity has been withdrawn",
|
|
"honeypot": "Do not buy — cannot sell, you will lose funds",
|
|
"known_rug_deployer": "Avoid this token — creator has rugged before",
|
|
"deployer_cashing_out": "Consider selling — team is exiting",
|
|
"insider_selling": "Reduce position — insiders are dumping",
|
|
"bundled_launch": "Do NOT trust the holder distribution — it's fabricated",
|
|
"active_ownership": "Monitor owner wallet for mint/blacklist transactions",
|
|
"upgradable_proxy": "Watch for implementation changes — could become a honeypot",
|
|
"coordinated_shilling": "Ignore FOMO — the hype is artificial",
|
|
"new_deployer_wallet": "Wait 30+ days before considering investment",
|
|
"sell_pressure": "Do not buy into distribution",
|
|
"massive_mention_spike": "Ignore sudden hype — likely a coordinated pump",
|
|
"concentrated_holdings": "High risk of price manipulation by top holders",
|
|
"holder_exodus": "Smart money is leaving — follow them",
|
|
}
|
|
|
|
priority_flags = ["honeypot", "lp_removed", "unlocked_liquidity", "deployer_cashing_out"]
|
|
for sig_name in ["lp_health", "contract_risk", "smart_money_flow"]:
|
|
sig = result.signals.get(sig_name)
|
|
if sig:
|
|
for flag in priority_flags:
|
|
if flag in sig.flags:
|
|
result.recommendations.insert(
|
|
0, f"🔴 URGENT — {flag_recos.get(flag, 'Exit position')}"
|
|
)
|
|
|
|
for sig in result.signals.values():
|
|
for flag in sig.flags:
|
|
if flag in flag_recos and flag_recos[flag] not in result.recommendations:
|
|
result.recommendations.append(f"• {flag_recos[flag]}")
|
|
|
|
if not result.recommendations:
|
|
if result.score < 30:
|
|
result.recommendations.append("✅ No urgent action needed — low risk")
|
|
elif result.score < 55:
|
|
result.recommendations.append("📊 Monitor token closely for the next 24-48 hours")
|
|
else:
|
|
result.recommendations.append("⚠️ Consider reducing exposure")
|
|
|
|
# ── Data Source Wrappers ────────────────────────────────────
|
|
|
|
async def _fetch_dexscreener_pool(self, address: str, chain: str) -> dict | None:
|
|
"""Fetch DEX pool data from DexScreener API."""
|
|
import httpx
|
|
|
|
chain_map = {
|
|
"ethereum": "ethereum",
|
|
"base": "base",
|
|
"bsc": "bsc",
|
|
"polygon": "polygon",
|
|
"arbitrum": "arbitrum",
|
|
"avalanche": "avalanche",
|
|
"solana": "solana",
|
|
"optimism": "optimism",
|
|
}
|
|
api_chain = chain_map.get(chain, chain)
|
|
url = f"https://api.dexscreener.com/latest/dex/tokens/{address}"
|
|
if api_chain != "ethereum":
|
|
url = f"https://api.dexscreener.com/token/v1/{api_chain}/{address}"
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=8) as client:
|
|
resp = await client.get(url)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
if data.get("pairs") and len(data["pairs"]) > 0:
|
|
return data
|
|
# Try alternative endpoint
|
|
pairs = data.get("pairs", [])
|
|
if pairs:
|
|
return data
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
async def _resolve_deployer(self, address: str, chain: str) -> str | None:
|
|
"""Find deployer address for a token contract."""
|
|
import httpx
|
|
|
|
try:
|
|
if chain == "solana":
|
|
# Use Solana RPC to find deployer
|
|
rpc = "https://api.mainnet-beta.solana.com"
|
|
async with httpx.AsyncClient(timeout=8) as client:
|
|
resp = await client.post(
|
|
rpc,
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getAccountInfo",
|
|
"params": [address, {"encoding": "base64"}],
|
|
},
|
|
)
|
|
if resp.status_code == 200:
|
|
return address[:16] # Minimal fallback
|
|
else:
|
|
# EVM: use LlamaRPC to get deployer tx
|
|
rpc = "https://eth.llamarpc.com"
|
|
async with httpx.AsyncClient(timeout=8) as client:
|
|
resp = await client.post(
|
|
rpc,
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "eth_getTransactionReceipt",
|
|
"params": [address],
|
|
},
|
|
)
|
|
# This doesn't work directly — need the creation tx
|
|
pass
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
async def _check_wallet_age(self, wallet: str, chain: str) -> dict | None:
|
|
"""Estimate wallet age from first transaction."""
|
|
import httpx
|
|
|
|
try:
|
|
if chain == "solana":
|
|
rpc = "https://api.mainnet-beta.solana.com"
|
|
async with httpx.AsyncClient(timeout=8) as client:
|
|
resp = await client.post(
|
|
rpc,
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getSignaturesForAddress",
|
|
"params": [wallet, {"limit": 1}],
|
|
},
|
|
)
|
|
if resp.status_code == 200:
|
|
sigs = resp.json().get("result", [])
|
|
if sigs:
|
|
block_time = sigs[0].get("blockTime")
|
|
if block_time:
|
|
age_days = (time.time() - block_time) / 86400
|
|
return {
|
|
"age_days": round(age_days, 1),
|
|
"first_tx": sigs[0].get("signature", ""),
|
|
}
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
async def _check_deployer_history(self, deployer: str, chain: str) -> dict:
|
|
"""Check deployer's token deployment history."""
|
|
result = {"total_deployed": 0, "rug_tokens": 0, "active_tokens": 0}
|
|
|
|
# Check our internal scam database for known deployer
|
|
try:
|
|
from app.scam_database import ScamDatabase
|
|
|
|
db = ScamDatabase()
|
|
scam_records = db.search_by_deployer(deployer)
|
|
if scam_records:
|
|
result["total_deployed"] = len(scam_records)
|
|
result["rug_tokens"] = len([r for r in scam_records if r.get("type") == "rug"])
|
|
result["active_tokens"] = len([r for r in scam_records if r.get("active", False)])
|
|
except (ImportError, AttributeError):
|
|
pass
|
|
|
|
return result
|
|
|
|
async def _check_lp_lock(self, address: str, chain: str) -> dict:
|
|
"""Check if LP tokens are locked via known lockers."""
|
|
result = {"locked": False, "unlocked": True, "unlock_date": None}
|
|
|
|
try:
|
|
from app.scanners.liquidity_verifier import LiquidityVerifier
|
|
|
|
verifier = LiquidityVerifier()
|
|
lp_info = await verifier.check_lock(address, chain)
|
|
if lp_info:
|
|
result["locked"] = lp_info.get("locked", False)
|
|
result["unlocked"] = not lp_info.get("locked", True)
|
|
result["unlock_date"] = lp_info.get("unlock_date")
|
|
except (ImportError, AttributeError):
|
|
pass
|
|
|
|
return result
|
|
|
|
async def _fetch_holder_data(self, address: str, chain: str) -> dict | None:
|
|
"""Fetch holder count and concentration data."""
|
|
try:
|
|
import httpx
|
|
|
|
if chain == "solana":
|
|
async with httpx.AsyncClient(timeout=8) as client:
|
|
resp = await client.get(
|
|
f"https://public-api.birdeye.so/public/token_holders/{address}",
|
|
params={"limit": 10, "offset": 0},
|
|
headers={"x-chain": "solana"},
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json().get("data", {})
|
|
items = data.get("items", [])
|
|
if items:
|
|
top10_pct = sum(float(h.get("percentage", 0)) for h in items[:10])
|
|
return {
|
|
"holders_now": data.get("total", 0),
|
|
"holders_24h_ago": None,
|
|
"top10_pct": top10_pct,
|
|
}
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Singleton factory
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
_instance: RugImminencePredictor | None = None
|
|
|
|
|
|
def get_predictor(weights: dict[str, float] | None = None) -> RugImminencePredictor:
|
|
"""Get or create the singleton predictor instance."""
|
|
global _instance
|
|
if _instance is None:
|
|
_instance = RugImminencePredictor(weights)
|
|
return _instance
|