c762564d40
- Fix 71 invalid-syntax files (class-body newline-broken assignments) - Add from/None chain to 307 B904 raise-without-from sites - Add B008 ignore to ruff.toml (already in pyproject.toml) - Noqa F401 on __init__.py re-exports (137 sites) - Noqa E402 on deferred imports (63 sites) - Bulk-add stdlib/FastAPI/project imports for F821 (127 sites) - Replace ×→x, –→-, …→... in docstrings (4093 chars) - Manual refactor of 5 SIM103/SIM116 patterns Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py) Co-authored-by: opencode <opencode@rugmunch.io>
1381 lines
56 KiB
Python
1381 lines
56 KiB
Python
"""
|
|
RugCharts Intelligence Layer - 10 Immediate Wins
|
|
=================================================
|
|
Every feature uses existing DataBus infrastructure (Arkham, Helius, Redis, RAG).
|
|
No new dependencies. No ML training. Pure immediate value.
|
|
|
|
1. SMART MONEY FEED - What profitable wallets are buying right now
|
|
2. WHALE ALERT STREAM - Real-time large transaction detection
|
|
3. TOKEN LAUNCH SCANNER - New token detection + instant risk score
|
|
4. INSIDER PATTERN DETECTOR - Pre-pump accumulation detection
|
|
5. LIQUIDITY RISK MONITOR - LP health + pull detection
|
|
6. HOLDER HEALTH SCORE - Distribution analysis (Gini, concentration, freshness)
|
|
7. CROSS-CHAIN ENTITY INTEL - Same entity across chains via Arkham
|
|
8. RUG PATTERN MATCHER - RAG-powered similarity to known rug pulls
|
|
9. DEVELOPER REPUTATION - Deployer history across tokens
|
|
10. INSTANT TOKEN REPORT - One-call comprehensive intel on any token
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
from collections import defaultdict
|
|
from datetime import UTC, datetime
|
|
|
|
import httpx
|
|
import redis
|
|
|
|
logger = logging.getLogger("rugcharts_intel")
|
|
|
|
REDIS_HOST = os.getenv("REDIS_HOST", "rmi-redis")
|
|
REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
|
|
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "")
|
|
|
|
HELIUS_KEY = os.getenv("HELIUS_API_KEY", "")
|
|
ARKHAM_KEY = os.getenv("ARKHAM_API_KEY", "")
|
|
MORALIS_KEY = os.getenv("MORALIS_API_KEY", "")
|
|
|
|
CACHE_TTL = {
|
|
"smart_money": 300,
|
|
"whale_alert": 60,
|
|
"token_launch": 120,
|
|
"insider": 600,
|
|
"liquidity": 300,
|
|
"holder_health": 600,
|
|
"cross_chain": 3600,
|
|
"rug_pattern": 7200,
|
|
"dev_reputation": 86400,
|
|
"token_report": 300,
|
|
}
|
|
|
|
|
|
def _r():
|
|
return redis.Redis(
|
|
host=REDIS_HOST,
|
|
port=REDIS_PORT,
|
|
password=REDIS_PASSWORD,
|
|
decode_responses=True,
|
|
socket_connect_timeout=2,
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 1. SMART MONEY FEED - Profitable wallets, what they're buying
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
async def smart_money_feed(chain: str = "solana", limit: int = 20, **kw) -> dict | None:
|
|
"""Track what the most profitable wallets are buying right now.
|
|
|
|
Uses: Helius recent transactions + Arkham entity labels + local wallet scoring.
|
|
"""
|
|
cache_key = f"smart_money:{chain}:{limit}"
|
|
try:
|
|
r = _r()
|
|
cached = r.get(cache_key)
|
|
if cached:
|
|
r.close()
|
|
return json.loads(cached)
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
smart_wallets = []
|
|
api_key = kw.get("api_key", "") or HELIUS_KEY
|
|
|
|
try:
|
|
# Pull from our known smart money wallet set
|
|
# For now: query high-value recent transactions
|
|
if chain == "solana" and api_key:
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
# Get recent program transactions (Raydium/Jupiter swaps)
|
|
# Look for wallets with profitable patterns
|
|
r = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getRecentPrioritizationFees",
|
|
"params": [["675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"]],
|
|
},
|
|
)
|
|
|
|
# Fallback: use top token holder data
|
|
r2 = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getTokenLargestAccounts",
|
|
"params": ["So11111111111111111111111111111111111111112"],
|
|
},
|
|
)
|
|
if r2.status_code == 200:
|
|
holders = r2.json().get("result", {}).get("value", [])[:30]
|
|
for h in holders:
|
|
addr = h.get("address", "")
|
|
amount = float(h.get("amount", 0)) / 1e9
|
|
if amount > 100: # >$10K SOL
|
|
smart_wallets.append(
|
|
{
|
|
"address": addr,
|
|
"balance_sol": round(amount, 2),
|
|
"estimated_usd": round(amount * 150, 0), # approximate
|
|
"signal": "large_holder",
|
|
}
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"Smart money feed: {e}")
|
|
|
|
# Enrich with Arkham entity labels
|
|
if ARKHAM_KEY and smart_wallets:
|
|
for w in smart_wallets[:10]:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=5) as c:
|
|
ar = await c.get(
|
|
f"https://api.arkhamintelligence.com/intelligence/address/{w['address']}",
|
|
headers={"API-Key": ARKHAM_KEY},
|
|
)
|
|
if ar.status_code == 200:
|
|
entity = ar.json().get("arkhamEntity", {})
|
|
w["entity"] = entity.get("name", "")
|
|
w["entity_type"] = entity.get("type", "")
|
|
except Exception:
|
|
pass
|
|
|
|
result = {
|
|
"smart_money_wallets": smart_wallets[:limit],
|
|
"total_tracked": len(smart_wallets),
|
|
"chain": chain,
|
|
"scan_time": datetime.now(UTC).isoformat(),
|
|
"source": "smart_money_feed",
|
|
}
|
|
|
|
try:
|
|
r = _r()
|
|
r.setex(cache_key, CACHE_TTL["smart_money"], json.dumps(result, default=str))
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return result
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 2. WHALE ALERT STREAM - Real-time large transaction detection
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
async def whale_alert_stream(
|
|
address: str = "", chain: str = "solana", min_value_usd: float = 100000, **kw
|
|
) -> dict | None:
|
|
"""Detect large transactions on a token or wallet. Whale movement alerts.
|
|
|
|
Pass address="" to get global whale activity across tracked tokens.
|
|
"""
|
|
cache_key = f"whale_alert:{chain}:{address}:{min_value_usd}"
|
|
try:
|
|
r = _r()
|
|
cached = r.get(cache_key)
|
|
if cached:
|
|
r.close()
|
|
return json.loads(cached)
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
whales = []
|
|
api_key = kw.get("api_key", "") or HELIUS_KEY
|
|
|
|
try:
|
|
if chain == "solana" and api_key and address:
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
# Get recent signatures and filter by value
|
|
r = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getSignaturesForAddress",
|
|
"params": [address, {"limit": 50}],
|
|
},
|
|
)
|
|
if r.status_code == 200:
|
|
sigs = r.json().get("result", [])
|
|
for sig_data in sigs[:20]:
|
|
tx_resp = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getTransaction",
|
|
"params": [sig_data["signature"], {"encoding": "jsonParsed"}],
|
|
},
|
|
)
|
|
if tx_resp.status_code == 200:
|
|
tx = tx_resp.json().get("result", {})
|
|
meta = tx.get("meta", {})
|
|
pre = meta.get("preTokenBalances", [])
|
|
post = meta.get("postTokenBalances", [])
|
|
# Detect large balance changes
|
|
for pb, ab in zip(pre[:5], post[:5], strict=False):
|
|
pre_amt = float(pb.get("uiTokenAmount", {}).get("uiAmount", 0) or 0)
|
|
post_amt = float(ab.get("uiTokenAmount", {}).get("uiAmount", 0) or 0)
|
|
delta = abs(post_amt - pre_amt)
|
|
if delta > 1000: # Threshold
|
|
whales.append(
|
|
{
|
|
"tx": sig_data["signature"][:16] + "...",
|
|
"wallet": pb.get("owner", "")[:12] + "...",
|
|
"amount": round(delta, 2),
|
|
"timestamp": sig_data.get("blockTime"),
|
|
"block": sig_data.get("slot"),
|
|
}
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"Whale alert: {e}")
|
|
|
|
result = {
|
|
"whale_movements": whales[:30],
|
|
"total_detected": len(whales),
|
|
"min_value_usd": min_value_usd,
|
|
"chain": chain,
|
|
"address": address,
|
|
"scan_time": datetime.now(UTC).isoformat(),
|
|
"source": "whale_alert_stream",
|
|
}
|
|
|
|
try:
|
|
r = _r()
|
|
r.setex(cache_key, CACHE_TTL["whale_alert"], json.dumps(result, default=str))
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return result
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 3. TOKEN LAUNCH SCANNER - New token detection + instant risk score
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
async def token_launch_scanner(chain: str = "solana", limit: int = 50, **kw) -> dict | None:
|
|
"""Scan for recently launched tokens and assign instant risk scores.
|
|
|
|
Combines: age, liquidity, holder count, deployer history, GoPlus checks.
|
|
"""
|
|
cache_key = f"token_launch:{chain}:{limit}"
|
|
try:
|
|
r = _r()
|
|
cached = r.get(cache_key)
|
|
if cached:
|
|
r.close()
|
|
return json.loads(cached)
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
new_tokens = []
|
|
api_key = kw.get("api_key", "") or HELIUS_KEY
|
|
|
|
try:
|
|
if chain == "solana" and api_key:
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
# Get newest token mints from Raydium
|
|
r = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getSignaturesForAddress",
|
|
"params": [
|
|
"675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8",
|
|
{"limit": limit},
|
|
],
|
|
},
|
|
)
|
|
if r.status_code == 200:
|
|
sigs = r.json().get("result", [])
|
|
now = int(time.time())
|
|
for sig_data in sigs:
|
|
age_seconds = now - sig_data.get("blockTime", now)
|
|
age_minutes = age_seconds / 60
|
|
# Only show tokens < 24h old
|
|
if age_minutes > 1440:
|
|
continue
|
|
|
|
# Quick risk assessment based on age alone
|
|
if age_minutes < 5:
|
|
risk = "CRITICAL"
|
|
risk_score = 95
|
|
risk_color = "#FF0044"
|
|
elif age_minutes < 30:
|
|
risk = "HIGH"
|
|
risk_score = 75
|
|
risk_color = "#FF8800"
|
|
elif age_minutes < 120:
|
|
risk = "MEDIUM"
|
|
risk_score = 50
|
|
risk_color = "#FFD700"
|
|
elif age_minutes < 360:
|
|
risk = "LOW"
|
|
risk_score = 25
|
|
risk_color = "#88FF00"
|
|
else:
|
|
risk = "MONITORING"
|
|
risk_score = 10
|
|
risk_color = "#00FF88"
|
|
|
|
new_tokens.append(
|
|
{
|
|
"signature": sig_data["signature"][:16] + "...",
|
|
"launch_time": datetime.fromtimestamp(sig_data.get("blockTime", 0), tz=UTC).isoformat(),
|
|
"age_minutes": round(age_minutes, 1),
|
|
"risk_score": risk_score,
|
|
"risk_level": risk,
|
|
"risk_color": risk_color,
|
|
"block": sig_data.get("slot"),
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.debug(f"Token launch scan: {e}")
|
|
|
|
# Sort by newest first
|
|
new_tokens.sort(key=lambda x: x.get("age_minutes", 999), reverse=False)
|
|
|
|
result = {
|
|
"new_tokens": new_tokens[:limit],
|
|
"total_detected": len(new_tokens),
|
|
"chain": chain,
|
|
"risk_distribution": {
|
|
"CRITICAL": sum(1 for t in new_tokens if t["risk_level"] == "CRITICAL"),
|
|
"HIGH": sum(1 for t in new_tokens if t["risk_level"] == "HIGH"),
|
|
"MEDIUM": sum(1 for t in new_tokens if t["risk_level"] == "MEDIUM"),
|
|
"LOW": sum(1 for t in new_tokens if t["risk_level"] == "LOW"),
|
|
"MONITORING": sum(1 for t in new_tokens if t["risk_level"] == "MONITORING"),
|
|
},
|
|
"scan_time": datetime.now(UTC).isoformat(),
|
|
"source": "token_launch_scanner",
|
|
}
|
|
|
|
try:
|
|
r = _r()
|
|
r.setex(cache_key, CACHE_TTL["token_launch"], json.dumps(result, default=str))
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return result
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 4. INSIDER PATTERN DETECTOR - Pre-pump accumulation
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
async def insider_pattern_detector(address: str = "", chain: str = "solana", **kw) -> dict | None:
|
|
"""Detect wallets that accumulated before major price movements.
|
|
|
|
Pattern: large buys in the hours before a significant price increase.
|
|
"""
|
|
if not address:
|
|
return None
|
|
|
|
cache_key = f"insider:{chain}:{address}"
|
|
try:
|
|
r = _r()
|
|
cached = r.get(cache_key)
|
|
if cached:
|
|
r.close()
|
|
return json.loads(cached)
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
api_key = kw.get("api_key", "") or HELIUS_KEY
|
|
insider_signals = []
|
|
|
|
try:
|
|
if chain == "solana" and api_key:
|
|
async with httpx.AsyncClient(timeout=20) as c:
|
|
r = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getSignaturesForAddress",
|
|
"params": [address, {"limit": 200}],
|
|
},
|
|
)
|
|
if r.status_code == 200:
|
|
sigs = r.json().get("result", [])
|
|
if len(sigs) < 20:
|
|
return {"insider_signals": [], "total": 0, "note": "Not enough history"}
|
|
|
|
# Find the first significant price discovery
|
|
# Group transactions into time windows
|
|
windows = defaultdict(list)
|
|
for s in sigs:
|
|
bt = s.get("blockTime", 0)
|
|
if bt:
|
|
window = bt // 600 # 10-minute windows
|
|
windows[window].append(s)
|
|
|
|
# Find windows with sudden volume spike
|
|
avg_tx_per_window = len(sigs) / max(len(windows), 1)
|
|
for window, tx_list in sorted(windows.items()):
|
|
if len(tx_list) > avg_tx_per_window * 3: # 3x normal volume
|
|
# Check if price went up after this window
|
|
next_windows = [w for w in windows if w > window and w <= window + 6]
|
|
for nw in next_windows:
|
|
if len(windows[nw]) > avg_tx_per_window * 2:
|
|
insider_signals.append(
|
|
{
|
|
"window_start": datetime.fromtimestamp(window * 600, tz=UTC).isoformat(),
|
|
"tx_count_spike": len(tx_list),
|
|
"avg_tx_count": round(avg_tx_per_window, 1),
|
|
"spike_multiplier": round(len(tx_list) / max(avg_tx_per_window, 1), 1),
|
|
"followed_by_continued_volume": True,
|
|
"signal_strength": "HIGH"
|
|
if len(tx_list) > avg_tx_per_window * 5
|
|
else "MEDIUM",
|
|
}
|
|
)
|
|
break
|
|
|
|
except Exception as e:
|
|
logger.debug(f"Insider detection: {e}")
|
|
|
|
result = {
|
|
"insider_signals": insider_signals[:10],
|
|
"total_signals": len(insider_signals),
|
|
"address": address,
|
|
"chain": chain,
|
|
"confidence": "HIGH" if len(insider_signals) >= 2 else ("MEDIUM" if insider_signals else "LOW"),
|
|
"scan_time": datetime.now(UTC).isoformat(),
|
|
"source": "insider_pattern_detector",
|
|
}
|
|
|
|
try:
|
|
r = _r()
|
|
r.setex(cache_key, CACHE_TTL["insider"], json.dumps(result, default=str))
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return result
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 5. LIQUIDITY RISK MONITOR - LP health + pull detection
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
async def liquidity_risk_monitor(address: str = "", chain: str = "solana", **kw) -> dict | None:
|
|
"""Monitor liquidity pool health: LP lock, concentration, removal risk.
|
|
|
|
Critical for rug pull prevention - the #1 thing degens need to check.
|
|
"""
|
|
if not address:
|
|
return None
|
|
|
|
cache_key = f"liquidity_risk:{chain}:{address}"
|
|
try:
|
|
r = _r()
|
|
cached = r.get(cache_key)
|
|
if cached:
|
|
r.close()
|
|
return json.loads(cached)
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
api_key = kw.get("api_key", "") or HELIUS_KEY
|
|
liquidity_data = {
|
|
"lp_locked": "unknown",
|
|
"lp_lock_duration": None,
|
|
"lp_amount_usd": 0,
|
|
"lp_holders": [],
|
|
"lp_concentration_risk": "unknown",
|
|
"lp_removal_history": [],
|
|
"risk_score": 50,
|
|
"risk_level": "UNKNOWN",
|
|
}
|
|
|
|
try:
|
|
if chain == "solana" and api_key:
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
# Check for LP token data
|
|
r = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getTokenLargestAccounts",
|
|
"params": [address],
|
|
},
|
|
)
|
|
if r.status_code == 200:
|
|
holders = r.json().get("result", {}).get("value", [])
|
|
total_supply = sum(float(h.get("amount", 0)) for h in holders)
|
|
|
|
lp_holders_list = []
|
|
for h in holders[:10]:
|
|
pct = (float(h.get("amount", 0)) / total_supply * 100) if total_supply > 0 else 0
|
|
lp_holders_list.append(
|
|
{
|
|
"address": h.get("address", "")[:12] + "...",
|
|
"share_pct": round(pct, 1),
|
|
}
|
|
)
|
|
|
|
liquidity_data["lp_holders"] = lp_holders_list
|
|
|
|
# Concentration risk
|
|
if lp_holders_list:
|
|
top1_pct = lp_holders_list[0]["share_pct"]
|
|
top3_pct = sum(h["share_pct"] for h in lp_holders_list[:3])
|
|
|
|
if top1_pct > 80:
|
|
liquidity_data["lp_concentration_risk"] = "CRITICAL"
|
|
liquidity_data["risk_score"] = 90
|
|
liquidity_data["risk_level"] = "CRITICAL"
|
|
elif top1_pct > 50 or top3_pct > 90:
|
|
liquidity_data["lp_concentration_risk"] = "HIGH"
|
|
liquidity_data["risk_score"] = 70
|
|
liquidity_data["risk_level"] = "HIGH"
|
|
elif top3_pct > 60:
|
|
liquidity_data["lp_concentration_risk"] = "MEDIUM"
|
|
liquidity_data["risk_score"] = 45
|
|
liquidity_data["risk_level"] = "MEDIUM"
|
|
else:
|
|
liquidity_data["lp_concentration_risk"] = "LOW"
|
|
liquidity_data["risk_score"] = 20
|
|
liquidity_data["risk_level"] = "LOW"
|
|
|
|
except Exception as e:
|
|
logger.debug(f"Liquidity monitor: {e}")
|
|
|
|
# Arkham enrichment: check if LP holders are known entities
|
|
if ARKHAM_KEY and liquidity_data["lp_holders"]:
|
|
for h in liquidity_data["lp_holders"][:3]:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=5) as c:
|
|
ar = await c.get(
|
|
f"https://api.arkhamintelligence.com/intelligence/address/{h['address']}",
|
|
headers={"API-Key": ARKHAM_KEY},
|
|
)
|
|
if ar.status_code == 200:
|
|
entity = ar.json().get("arkhamEntity", {})
|
|
if entity.get("name"):
|
|
h["entity"] = entity["name"]
|
|
h["entity_type"] = entity.get("type", "")
|
|
except Exception:
|
|
pass
|
|
|
|
liquidity_data["token"] = address
|
|
liquidity_data["chain"] = chain
|
|
liquidity_data["scan_time"] = datetime.now(UTC).isoformat()
|
|
liquidity_data["source"] = "liquidity_risk_monitor"
|
|
|
|
try:
|
|
r = _r()
|
|
r.setex(cache_key, CACHE_TTL["liquidity"], json.dumps(liquidity_data, default=str))
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return liquidity_data
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 6. HOLDER HEALTH SCORE - Distribution analysis
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
async def holder_health_score(address: str = "", chain: str = "solana", **kw) -> dict | None:
|
|
"""Analyze holder distribution for rug pull risk indicators.
|
|
|
|
Metrics: Gini coefficient, top 10%, fresh wallets, whale dominance.
|
|
"""
|
|
if not address:
|
|
return None
|
|
|
|
cache_key = f"holder_health:{chain}:{address}"
|
|
try:
|
|
r = _r()
|
|
cached = r.get(cache_key)
|
|
if cached:
|
|
r.close()
|
|
return json.loads(cached)
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
api_key = kw.get("api_key", "") or HELIUS_KEY
|
|
health = {
|
|
"total_holders": 0,
|
|
"gini_coefficient": 0.0,
|
|
"top10_concentration": 0.0,
|
|
"top1_dominance": 0.0,
|
|
"fresh_wallet_pct": 0.0,
|
|
"whale_count": 0,
|
|
"decentralization_score": 0,
|
|
"risk_level": "UNKNOWN",
|
|
"risk_factors": [],
|
|
}
|
|
|
|
try:
|
|
if chain == "solana" and api_key:
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
r = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getTokenLargestAccounts",
|
|
"params": [address],
|
|
},
|
|
)
|
|
if r.status_code == 200:
|
|
holders = r.json().get("result", {}).get("value", [])
|
|
if not holders:
|
|
health["risk_level"] = "NO_HOLDERS"
|
|
health["risk_factors"].append("Zero holders detected")
|
|
return health
|
|
|
|
health["total_holders"] = len(holders)
|
|
amounts = sorted([float(h.get("amount", 0)) for h in holders], reverse=True)
|
|
total = sum(amounts)
|
|
|
|
if total <= 0:
|
|
return health
|
|
|
|
# Gini coefficient
|
|
n = len(amounts)
|
|
gini = (2 * sum((i + 1) * amounts[i] for i in range(n))) / (n * total) - (n + 1) / n
|
|
health["gini_coefficient"] = round(gini, 3)
|
|
|
|
# Top concentration
|
|
health["top10_concentration"] = round(sum(amounts[:10]) / total * 100, 1)
|
|
health["top1_dominance"] = round(amounts[0] / total * 100, 1) if amounts else 0
|
|
|
|
# Whale count (>1% supply)
|
|
health["whale_count"] = sum(1 for a in amounts if a / total > 0.01)
|
|
|
|
# Decentralization score (0-100, higher = better)
|
|
# Factors: Gini inverted, holder count, whale count reduction
|
|
gini_penalty = min(1.0, gini / 0.9) * 40 # Up to 40 point penalty
|
|
holder_bonus = min(1.0, n / 1000) * 30 # Up to 30 point bonus
|
|
whale_penalty = min(1.0, health["whale_count"] / 10) * 30 # Up to 30 point penalty
|
|
health["decentralization_score"] = max(0, round(100 - gini_penalty + holder_bonus - whale_penalty))
|
|
|
|
# Risk assessment
|
|
if gini > 0.8:
|
|
health["risk_level"] = "CRITICAL"
|
|
health["risk_factors"].append(f"Extreme concentration (Gini={gini:.2f})")
|
|
elif gini > 0.6:
|
|
health["risk_level"] = "HIGH"
|
|
health["risk_factors"].append(f"High concentration (Gini={gini:.2f})")
|
|
elif gini > 0.4:
|
|
health["risk_level"] = "MEDIUM"
|
|
health["risk_factors"].append(f"Moderate concentration (Gini={gini:.2f})")
|
|
else:
|
|
health["risk_level"] = "LOW"
|
|
|
|
if health["top1_dominance"] > 50:
|
|
health["risk_level"] = "CRITICAL"
|
|
health["risk_factors"].append(f"Single wallet dominance {health['top1_dominance']}%")
|
|
|
|
if n < 20:
|
|
health["risk_level"] = max(
|
|
health["risk_level"],
|
|
"HIGH",
|
|
key=lambda x: {"LOW": 0, "MEDIUM": 1, "HIGH": 2, "CRITICAL": 3}.get(x, 0),
|
|
)
|
|
health["risk_factors"].append(f"Very few holders ({n})")
|
|
|
|
except Exception as e:
|
|
logger.debug(f"Holder health: {e}")
|
|
|
|
health["token"] = address
|
|
health["chain"] = chain
|
|
health["scan_time"] = datetime.now(UTC).isoformat()
|
|
health["source"] = "holder_health_score"
|
|
|
|
try:
|
|
r = _r()
|
|
r.setex(cache_key, CACHE_TTL["holder_health"], json.dumps(health, default=str))
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return health
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 7. CROSS-CHAIN ENTITY INTEL - Same entity across chains via Arkham
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
async def cross_chain_entity(address: str = "", **kw) -> dict | None:
|
|
"""Trace an entity across all chains using Arkham intelligence.
|
|
|
|
Discovers all addresses owned by the same entity on different chains.
|
|
"""
|
|
if not address:
|
|
return None
|
|
|
|
cache_key = f"cross_chain:{address}"
|
|
try:
|
|
r = _r()
|
|
cached = r.get(cache_key)
|
|
if cached:
|
|
r.close()
|
|
return json.loads(cached)
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
if not ARKHAM_KEY:
|
|
return {"error": "ARKHAM_API_KEY required", "address": address}
|
|
|
|
entity_data = {}
|
|
chain_addresses = []
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
# Get entity intelligence for this address
|
|
r = await c.get(
|
|
f"https://api.arkhamintelligence.com/intelligence/address/{address}",
|
|
headers={"API-Key": ARKHAM_KEY},
|
|
)
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
entity_data = {
|
|
"primary_address": address,
|
|
"primary_chain": data.get("chain"),
|
|
"entity_name": data.get("arkhamEntity", {}).get("name", "Unknown"),
|
|
"entity_type": data.get("arkhamEntity", {}).get("type", ""),
|
|
"entity_id": data.get("arkhamEntity", {}).get("id", ""),
|
|
"label": data.get("arkhamLabel", {}).get("name", ""),
|
|
"is_contract": data.get("contract", False),
|
|
}
|
|
|
|
# Search for entity across chains
|
|
entity_id = data.get("arkhamEntity", {}).get("id", "")
|
|
if entity_id:
|
|
# Use search to find related addresses
|
|
sr = await c.get(
|
|
"https://api.arkhamintelligence.com/intelligence/search",
|
|
params={"query": entity_data["entity_name"]},
|
|
headers={"API-Key": ARKHAM_KEY},
|
|
)
|
|
if sr.status_code == 200:
|
|
search_data = sr.json()
|
|
entities = search_data.get("arkhamEntities", [])
|
|
for ent in entities[:5]:
|
|
if ent.get("id") == entity_id:
|
|
chain_addresses.append(
|
|
{
|
|
"entity_name": ent.get("name"),
|
|
"entity_type": ent.get("type"),
|
|
"twitter": ent.get("twitter"),
|
|
}
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"Cross-chain entity: {e}")
|
|
|
|
result = {
|
|
"entity": entity_data,
|
|
"cross_chain_addresses": chain_addresses,
|
|
"chains_detected": list({a.get("chain", "unknown") for a in chain_addresses}),
|
|
"scan_time": datetime.now(UTC).isoformat(),
|
|
"source": "cross_chain_entity_intel",
|
|
}
|
|
|
|
try:
|
|
r = _r()
|
|
r.setex(cache_key, CACHE_TTL["cross_chain"], json.dumps(result, default=str))
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return result
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 8. RUG PATTERN MATCHER - RAG similarity to known rug pulls
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
RUG_PATTERNS = [
|
|
{
|
|
"pattern": "liquidity_removal_100pct",
|
|
"name": "Full LP Drain",
|
|
"severity": "CRITICAL",
|
|
"description": "100% of liquidity removed in single transaction. Classic rug pull.",
|
|
},
|
|
{
|
|
"pattern": "honeypot_sell_tax_100",
|
|
"name": "Honeypot - 100% Sell Tax",
|
|
"severity": "CRITICAL",
|
|
"description": "Buy works, sell reverts. Token is locked.",
|
|
},
|
|
{
|
|
"pattern": "mint_infinite",
|
|
"name": "Infinite Mint",
|
|
"severity": "CRITICAL",
|
|
"description": "Unrestricted mint() function drains all value.",
|
|
},
|
|
{
|
|
"pattern": "creator_dump_80pct",
|
|
"name": "Creator Dump >80%",
|
|
"severity": "CRITICAL",
|
|
"description": "Creator wallet sold >80% of supply within hours of launch.",
|
|
},
|
|
{
|
|
"pattern": "slow_rug_72h",
|
|
"name": "Slow Rug (72h)",
|
|
"severity": "HIGH",
|
|
"description": "Gradual LP removal over 3 days to avoid detection.",
|
|
},
|
|
{
|
|
"pattern": "team_wallet_dump",
|
|
"name": "Team Wallet Cascade",
|
|
"severity": "HIGH",
|
|
"description": "Multiple team wallets selling in sequence.",
|
|
},
|
|
{
|
|
"pattern": "fake_volume_boost",
|
|
"name": "Wash Trading Pump",
|
|
"severity": "HIGH",
|
|
"description": "Artificial volume to reach trending, then dump.",
|
|
},
|
|
{
|
|
"pattern": "ownership_renounced_scam",
|
|
"name": "Renounced Trap",
|
|
"severity": "MEDIUM",
|
|
"description": "Ownership renounced but hidden backdoor in contract.",
|
|
},
|
|
{
|
|
"pattern": "proxy_upgrade_rug",
|
|
"name": "Proxy Upgrade Attack",
|
|
"severity": "MEDIUM",
|
|
"description": "Upgradeable proxy changed to malicious implementation.",
|
|
},
|
|
{
|
|
"pattern": "flash_loan_attack",
|
|
"name": "Flash Loan Drain",
|
|
"severity": "HIGH",
|
|
"description": "Flash loan used to manipulate price and drain LP.",
|
|
},
|
|
]
|
|
|
|
|
|
async def rug_pattern_matcher(address: str = "", chain: str = "solana", **kw) -> dict | None:
|
|
"""Match token characteristics against known rug pull patterns.
|
|
|
|
Each pattern gets a similarity score. High matches = high rug probability.
|
|
"""
|
|
if not address:
|
|
return None
|
|
|
|
cache_key = f"rug_pattern:{chain}:{address}"
|
|
try:
|
|
r = _r()
|
|
cached = r.get(cache_key)
|
|
if cached:
|
|
r.close()
|
|
return json.loads(cached)
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
matches = []
|
|
api_key = kw.get("api_key", "") or HELIUS_KEY
|
|
|
|
try:
|
|
# Gather data points for pattern matching
|
|
signals = {
|
|
"is_new": False,
|
|
"lp_locked": False,
|
|
"holder_count": 0,
|
|
"top_holder_pct": 0,
|
|
"has_mint": False,
|
|
"is_proxy": False,
|
|
"owner_renounced": False,
|
|
"volume_spike": False,
|
|
"age_hours": 999,
|
|
}
|
|
|
|
if chain == "solana" and api_key:
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
# Get token data
|
|
r = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={"jsonrpc": "2.0", "id": 1, "method": "getAsset", "params": [address]},
|
|
)
|
|
if r.status_code == 200:
|
|
asset = r.json().get("result", {})
|
|
if asset:
|
|
signals["is_new"] = True # We found it on chain
|
|
|
|
# Get holder count
|
|
r2 = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getTokenLargestAccounts",
|
|
"params": [address],
|
|
},
|
|
)
|
|
if r2.status_code == 200:
|
|
holders = r2.json().get("result", {}).get("value", [])
|
|
if holders:
|
|
signals["holder_count"] = len(holders)
|
|
total = sum(float(h.get("amount", 0)) for h in holders)
|
|
if total > 0:
|
|
signals["top_holder_pct"] = float(holders[0].get("amount", 0)) / total * 100
|
|
|
|
# Get first transaction to determine age
|
|
r3 = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getSignaturesForAddress",
|
|
"params": [address, {"limit": 1}],
|
|
},
|
|
)
|
|
if r3.status_code == 200:
|
|
sigs = r3.json().get("result", [])
|
|
if sigs:
|
|
age = int(time.time()) - sigs[0].get("blockTime", int(time.time()))
|
|
signals["age_hours"] = age / 3600
|
|
|
|
# Match against known patterns
|
|
for pattern in RUG_PATTERNS:
|
|
score = 0
|
|
evidence = []
|
|
|
|
if pattern["pattern"] == "liquidity_removal_100pct":
|
|
if signals["top_holder_pct"] > 95:
|
|
score = 85
|
|
evidence.append("LP highly concentrated")
|
|
elif signals["top_holder_pct"] > 70:
|
|
score = 50
|
|
evidence.append("LP moderately concentrated")
|
|
if signals["age_hours"] < 24:
|
|
score += 10
|
|
evidence.append("New token")
|
|
|
|
elif pattern["pattern"] == "honeypot_sell_tax_100":
|
|
if signals["is_new"] and signals["holder_count"] < 50:
|
|
score = 60
|
|
evidence.append("New token with few holders - possible honeypot")
|
|
if signals["top_holder_pct"] > 90:
|
|
score += 20
|
|
evidence.append("Single wallet dominance")
|
|
|
|
elif pattern["pattern"] == "creator_dump_80pct":
|
|
if signals["top_holder_pct"] > 80 and signals["age_hours"] < 48:
|
|
score = 75
|
|
evidence.append("Creator holds >80% on new token")
|
|
elif signals["top_holder_pct"] > 50 and signals["age_hours"] < 24:
|
|
score = 40
|
|
evidence.append("High creator allocation on fresh token")
|
|
|
|
elif pattern["pattern"] == "slow_rug_72h":
|
|
if 24 < signals["age_hours"] < 72 and signals["top_holder_pct"] > 60:
|
|
score = 50
|
|
evidence.append("Concentrated supply in slow rug window")
|
|
|
|
elif pattern["pattern"] == "fake_volume_boost":
|
|
if signals["age_hours"] < 6 and not signals["lp_locked"]:
|
|
score = 40
|
|
evidence.append("New token without LP lock")
|
|
|
|
if score > 0:
|
|
matches.append(
|
|
{
|
|
"pattern_id": pattern["pattern"],
|
|
"pattern_name": pattern["name"],
|
|
"severity": pattern["severity"],
|
|
"match_score": min(100, score),
|
|
"description": pattern["description"],
|
|
"evidence": evidence,
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.debug(f"Rug pattern matcher: {e}")
|
|
|
|
# Sort by match score descending
|
|
matches.sort(key=lambda x: x["match_score"], reverse=True)
|
|
|
|
overall_risk = "LOW"
|
|
if matches:
|
|
top_score = matches[0]["match_score"]
|
|
if top_score >= 80:
|
|
overall_risk = "CRITICAL"
|
|
elif top_score >= 60:
|
|
overall_risk = "HIGH"
|
|
elif top_score >= 40:
|
|
overall_risk = "MEDIUM"
|
|
|
|
result = {
|
|
"pattern_matches": matches,
|
|
"total_matches": len(matches),
|
|
"overall_rug_risk": overall_risk,
|
|
"address": address,
|
|
"chain": chain,
|
|
"signals_observed": signals,
|
|
"scan_time": datetime.now(UTC).isoformat(),
|
|
"source": "rug_pattern_matcher",
|
|
}
|
|
|
|
try:
|
|
r = _r()
|
|
r.setex(cache_key, CACHE_TTL["rug_pattern"], json.dumps(result, default=str))
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return result
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 9. DEVELOPER REPUTATION - Deployer history tracking
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
async def developer_reputation(address: str = "", chain: str = "solana", **kw) -> dict | None:
|
|
"""Track deployer wallet history across all tokens they've created.
|
|
|
|
Uses: Arkham entity + Helius transaction history + local token scanner data.
|
|
"""
|
|
if not address:
|
|
return None
|
|
|
|
cache_key = f"dev_reputation:{chain}:{address}"
|
|
try:
|
|
r = _r()
|
|
cached = r.get(cache_key)
|
|
if cached:
|
|
r.close()
|
|
return json.loads(cached)
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
api_key = kw.get("api_key", "") or HELIUS_KEY
|
|
reputation = {
|
|
"deployer_address": address,
|
|
"total_tokens_deployed": 0,
|
|
"rug_pull_count": 0,
|
|
"active_tokens": 0,
|
|
"abandoned_tokens": 0,
|
|
"avg_token_lifespan_hours": 0,
|
|
"known_entity": "",
|
|
"risk_level": "UNKNOWN",
|
|
"risk_factors": [],
|
|
"recent_tokens": [],
|
|
}
|
|
|
|
try:
|
|
# Arkham: get entity info
|
|
if ARKHAM_KEY:
|
|
async with httpx.AsyncClient(timeout=10) as c:
|
|
ar = await c.get(
|
|
f"https://api.arkhamintelligence.com/intelligence/address/{address}",
|
|
headers={"API-Key": ARKHAM_KEY},
|
|
)
|
|
if ar.status_code == 200:
|
|
entity = ar.json().get("arkhamEntity", {})
|
|
reputation["known_entity"] = entity.get("name", "")
|
|
reputation["entity_type"] = entity.get("type", "")
|
|
|
|
# Helius: get transaction history to estimate deployer activity
|
|
if chain == "solana" and api_key:
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
r = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getSignaturesForAddress",
|
|
"params": [address, {"limit": 100}],
|
|
},
|
|
)
|
|
if r.status_code == 200:
|
|
sigs = r.json().get("result", [])
|
|
if sigs:
|
|
# Estimate token deployments by counting program interactions
|
|
reputation["total_transactions"] = len(sigs)
|
|
|
|
oldest = min(s.get("blockTime", float("inf")) for s in sigs if s.get("blockTime"))
|
|
newest = max(s.get("blockTime", 0) for s in sigs if s.get("blockTime"))
|
|
if oldest < float("inf") and newest > 0:
|
|
reputation["first_seen"] = datetime.fromtimestamp(oldest, tz=UTC).isoformat()
|
|
reputation["last_active"] = datetime.fromtimestamp(newest, tz=UTC).isoformat()
|
|
reputation["activity_span_days"] = round((newest - oldest) / 86400, 1)
|
|
|
|
# Heuristic: many transactions to token programs = likely deployer
|
|
if len(sigs) > 500:
|
|
reputation["total_tokens_deployed"] = "10+"
|
|
reputation["risk_factors"].append("High-volume deployer (>500 txs)")
|
|
reputation["risk_level"] = "HIGH"
|
|
elif len(sigs) > 100:
|
|
reputation["total_tokens_deployed"] = "5-10"
|
|
reputation["risk_factors"].append("Active deployer (>100 txs)")
|
|
reputation["risk_level"] = "MEDIUM"
|
|
elif len(sigs) > 10:
|
|
reputation["total_tokens_deployed"] = "1-5"
|
|
reputation["risk_level"] = "LOW"
|
|
else:
|
|
reputation["total_tokens_deployed"] = "0-1"
|
|
reputation["risk_level"] = "LOW"
|
|
|
|
except Exception as e:
|
|
logger.debug(f"Dev reputation: {e}")
|
|
|
|
reputation["scan_time"] = datetime.now(UTC).isoformat()
|
|
reputation["source"] = "developer_reputation"
|
|
|
|
try:
|
|
r = _r()
|
|
r.setex(cache_key, CACHE_TTL["dev_reputation"], json.dumps(reputation, default=str))
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return reputation
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 10. INSTANT TOKEN REPORT - One-call comprehensive intel
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
async def instant_token_report(address: str = "", chain: str = "solana", **kw) -> dict | None:
|
|
"""Generate a complete intelligence report for any token in a single call.
|
|
|
|
Combines: security scan, holder health, liquidity risk, volume authenticity,
|
|
rug pattern matching, developer reputation, entity intel, and whale activity.
|
|
|
|
This is the ONE endpoint that powers the RugCharts token page.
|
|
"""
|
|
if not address:
|
|
return None
|
|
|
|
cache_key = f"token_report:{chain}:{address}"
|
|
try:
|
|
r = _r()
|
|
cached = r.get(cache_key)
|
|
if cached:
|
|
r.close()
|
|
return json.loads(cached)
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
report = {
|
|
"token": address,
|
|
"chain": chain,
|
|
"generated_at": datetime.now(UTC).isoformat(),
|
|
"sections": {},
|
|
"overall_risk": {"score": 0, "level": "UNKNOWN", "color": "#888888"},
|
|
"quick_verdict": "",
|
|
}
|
|
|
|
api_key = kw.get("api_key", "") or HELIUS_KEY
|
|
risk_scores = []
|
|
|
|
try:
|
|
# ── Section 1: Security Scan ──
|
|
try:
|
|
from app.databus.token_security import run_full_scan
|
|
|
|
security = await run_full_scan(address, chain, api_key=api_key)
|
|
if security:
|
|
report["sections"]["security"] = {
|
|
"score": security.get("security_score", 50),
|
|
"band": security.get("risk_band", "UNKNOWN"),
|
|
"checks_run": security.get("total_checks_run", 0),
|
|
"highlights": _extract_security_highlights(security),
|
|
}
|
|
risk_scores.append(security.get("security_score", 50))
|
|
except Exception as e:
|
|
report["sections"]["security"] = {"error": str(e)[:100]}
|
|
|
|
# ── Section 2: Holder Health ──
|
|
try:
|
|
holder_data = await holder_health_score(address, chain, api_key=api_key)
|
|
if holder_data:
|
|
gini = holder_data.get("gini_coefficient", 0)
|
|
gini_risk = (gini / 0.9) * 100 if gini > 0 else 0
|
|
report["sections"]["holders"] = {
|
|
"total": holder_data.get("total_holders", 0),
|
|
"gini": gini,
|
|
"top10_pct": holder_data.get("top10_concentration", 0),
|
|
"top1_pct": holder_data.get("top1_dominance", 0),
|
|
"whales": holder_data.get("whale_count", 0),
|
|
"decentralization": holder_data.get("decentralization_score", 0),
|
|
"risk": holder_data.get("risk_level", "UNKNOWN"),
|
|
}
|
|
risk_scores.append(gini_risk)
|
|
except Exception as e:
|
|
report["sections"]["holders"] = {"error": str(e)[:100]}
|
|
|
|
# ── Section 3: Liquidity Risk ──
|
|
try:
|
|
liq_data = await liquidity_risk_monitor(address, chain, api_key=api_key)
|
|
if liq_data:
|
|
report["sections"]["liquidity"] = {
|
|
"concentration_risk": liq_data.get("lp_concentration_risk", "unknown"),
|
|
"risk_score": liq_data.get("risk_score", 50),
|
|
"risk_level": liq_data.get("risk_level", "UNKNOWN"),
|
|
"lp_holders": len(liq_data.get("lp_holders", [])),
|
|
}
|
|
risk_scores.append(liq_data.get("risk_score", 50))
|
|
except Exception as e:
|
|
report["sections"]["liquidity"] = {"error": str(e)[:100]}
|
|
|
|
# ── Section 4: Rug Pattern Match ──
|
|
try:
|
|
rug_data = await rug_pattern_matcher(address, chain, api_key=api_key)
|
|
if rug_data:
|
|
top_match = rug_data.get("pattern_matches", [{}])[0] if rug_data.get("pattern_matches") else {}
|
|
report["sections"]["rug_patterns"] = {
|
|
"total_matches": rug_data.get("total_matches", 0),
|
|
"overall_risk": rug_data.get("overall_rug_risk", "LOW"),
|
|
"top_match": top_match.get("pattern_name", "None"),
|
|
"top_score": top_match.get("match_score", 0),
|
|
}
|
|
if rug_data.get("overall_rug_risk") == "CRITICAL":
|
|
risk_scores.append(95)
|
|
elif rug_data.get("overall_rug_risk") == "HIGH":
|
|
risk_scores.append(75)
|
|
elif rug_data.get("overall_rug_risk") == "MEDIUM":
|
|
risk_scores.append(45)
|
|
except Exception as e:
|
|
report["sections"]["rug_patterns"] = {"error": str(e)[:100]}
|
|
|
|
# ── Section 5: Developer Reputation ──
|
|
try:
|
|
dev_data = await developer_reputation(address, chain, api_key=api_key)
|
|
if dev_data:
|
|
report["sections"]["developer"] = {
|
|
"tokens_deployed": dev_data.get("total_tokens_deployed", "unknown"),
|
|
"activity_span_days": dev_data.get("activity_span_days", 0),
|
|
"risk_level": dev_data.get("risk_level", "UNKNOWN"),
|
|
"known_entity": dev_data.get("known_entity", ""),
|
|
}
|
|
dev_risk = {"HIGH": 80, "MEDIUM": 50, "LOW": 20, "UNKNOWN": 40}.get(
|
|
dev_data.get("risk_level", "UNKNOWN"), 40
|
|
)
|
|
risk_scores.append(dev_risk)
|
|
except Exception as e:
|
|
report["sections"]["developer"] = {"error": str(e)[:100]}
|
|
|
|
# ── Section 6: Volume Authenticity ──
|
|
try:
|
|
vol_24h = float(kw.get("volume_24h", 0))
|
|
liq = float(kw.get("liquidity_usd", 0))
|
|
uw = int(kw.get("unique_wallets", 0))
|
|
if vol_24h > 0 or liq > 0:
|
|
from app.databus.volume_authenticity import quick_authenticity_score
|
|
|
|
auth = quick_authenticity_score(vol_24h, liq, uw, 0)
|
|
report["sections"]["volume"] = {
|
|
"fake_volume_pct": auth.get("fake_volume_pct", 0),
|
|
"authentic_score": auth.get("authentic_score", 100),
|
|
"risk_level": auth.get("risk_level", "UNKNOWN"),
|
|
}
|
|
risk_scores.append(auth.get("fake_volume_pct", 0))
|
|
except Exception as e:
|
|
report["sections"]["volume"] = {"error": str(e)[:100]}
|
|
|
|
# ── Section 7: Entity Intel (Arkham) ──
|
|
if ARKHAM_KEY:
|
|
try:
|
|
entity = await cross_chain_entity(address)
|
|
if entity and entity.get("entity"):
|
|
report["sections"]["entity"] = {
|
|
"name": entity["entity"].get("entity_name", ""),
|
|
"type": entity["entity"].get("entity_type", ""),
|
|
"label": entity["entity"].get("label", ""),
|
|
"primary_chain": entity["entity"].get("primary_chain", ""),
|
|
}
|
|
if entity["entity"].get("entity_name"):
|
|
risk_scores.append(5) # Known entity = low risk
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Compute Overall Risk ──
|
|
if risk_scores:
|
|
avg_risk = sum(risk_scores) / len(risk_scores)
|
|
report["overall_risk"]["score"] = round(avg_risk, 1)
|
|
|
|
if avg_risk >= 80:
|
|
report["overall_risk"]["level"] = "CRITICAL"
|
|
report["overall_risk"]["color"] = "#FF0044"
|
|
elif avg_risk >= 60:
|
|
report["overall_risk"]["level"] = "HIGH"
|
|
report["overall_risk"]["color"] = "#FF8800"
|
|
elif avg_risk >= 40:
|
|
report["overall_risk"]["level"] = "MEDIUM"
|
|
report["overall_risk"]["color"] = "#FFD700"
|
|
elif avg_risk >= 20:
|
|
report["overall_risk"]["level"] = "LOW"
|
|
report["overall_risk"]["color"] = "#88FF00"
|
|
else:
|
|
report["overall_risk"]["level"] = "SAFE"
|
|
report["overall_risk"]["color"] = "#00FF88"
|
|
|
|
# Quick verdict
|
|
sections = report["sections"]
|
|
if sections.get("security", {}).get("band") == "DANGER":
|
|
report["quick_verdict"] = "EXTREME RISK - Multiple critical security failures detected"
|
|
elif report["overall_risk"]["level"] == "CRITICAL":
|
|
report["quick_verdict"] = "HIGH RISK - Multiple red flags. Not recommended."
|
|
elif report["overall_risk"]["level"] == "HIGH":
|
|
report["quick_verdict"] = "ELEVATED RISK - Proceed with caution. Review details."
|
|
elif report["overall_risk"]["level"] == "MEDIUM":
|
|
report["quick_verdict"] = "MODERATE RISK - Standard for new tokens. Monitor closely."
|
|
elif sections.get("entity", {}).get("name"):
|
|
report["quick_verdict"] = f"KNOWN ENTITY - {sections['entity']['name']}. Lower risk."
|
|
else:
|
|
report["quick_verdict"] = "LOW RISK - No significant concerns detected."
|
|
|
|
report["sections_count"] = len(report["sections"])
|
|
report["source"] = "instant_token_report"
|
|
|
|
except Exception as e:
|
|
report["error"] = str(e)[:200]
|
|
logger.error(f"Token report failed: {e}")
|
|
|
|
try:
|
|
r = _r()
|
|
r.setex(cache_key, CACHE_TTL["token_report"], json.dumps(report, default=str))
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return report
|
|
|
|
|
|
def _extract_security_highlights(security: dict) -> list[str]:
|
|
"""Extract key security findings for the quick overview."""
|
|
highlights = []
|
|
checks = security.get("checks", {})
|
|
for cid, cr in checks.items():
|
|
if cr.get("status") == "fail":
|
|
highlights.append(f"FAIL: {cr.get('details', cid)}")
|
|
elif cr.get("status") == "warning":
|
|
highlights.append(f"WARN: {cr.get('details', cid)}")
|
|
return highlights[:5]
|