1309 lines
48 KiB
Python
1309 lines
48 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
ULTIMATE CRYPTO EMBEDDER — Rug Munch Intelligence
|
|
===================================================
|
|
Multi-head embedding system purpose-built for crypto scam/rug detection.
|
|
|
|
Architecture:
|
|
PRIMARY: OpenRouter NVIDIA Nemotron (2048d, multimodal, FREE)
|
|
FALLBACK: Local BGE-small-en-v1.5 (384-dim) — runs on CPU, zero API, zero cost
|
|
FALLBACK: HuggingFace BGE-M3 (1024d) if token has inference permissions
|
|
SPECIALTY: Crypto-aware hashing — contract bytecode sim, tx pattern fingerprints
|
|
|
|
Embedding Heads:
|
|
SEMANTIC — token descriptions, scam narratives, news (OpenRouter/HF)
|
|
CODE — smart contract similarity (AST-aware + bytecode hash)
|
|
BEHAVIORAL — transaction patterns, wallet behavior vectors (numeric → float[])
|
|
ENTITY — wallet labels, cluster IDs, known scammer fingerprints
|
|
|
|
Collections:
|
|
wallet_profiles — wallet behavior + labels
|
|
token_analysis — token metadata + risk signals
|
|
scam_patterns — known rug/honeypot signatures
|
|
forensic_reports — investigation findings
|
|
market_intel — news, trends, alerts
|
|
contract_audits — smart contract code + audit results
|
|
known_scams — verified scam DB entries
|
|
"""
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ─── CONFIG ──────────────────────────────────────────────────────────
|
|
OPENROUTER_KEY = os.getenv("OPENROUTER_API_KEY", "")
|
|
HF_TOKEN = os.getenv("HUGGINGFACE_TOKEN", "")
|
|
REDIS_HOST = os.getenv("REDIS_HOST", "rmi-redis")
|
|
REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
|
|
REDIS_PW = os.getenv("REDIS_PASSWORD", "")
|
|
|
|
# Embedding dimensions by model
|
|
DIMS = {
|
|
"local/bge-small-en-v1.5": 384,
|
|
"openai/text-embedding-3-large": 3072,
|
|
"openai/text-embedding-3-small": 1536,
|
|
"BAAI/bge-m3": 1024,
|
|
"BAAI/bge-large-en-v1.5": 1024,
|
|
"BAAI/bge-small-en-v1.5": 384,
|
|
"nvidia/llama-nemotron-embed-vl-1b-v2:free": 2048,
|
|
"rmi/qwen3-embedding:4b": 2048, # deprecated — using bge-m3 1024d instead
|
|
"crypto_behavioral": 64,
|
|
"crypto_code_hash": 128,
|
|
}
|
|
|
|
# Default model per head — ONE embedder: bge-m3 1024d
|
|
HEAD_DEFAULTS = {
|
|
"semantic": "rmi/bge-m3", # OUR OWN — 1024d via Ollama, zero cost
|
|
"code": "rmi/bge-m3", # OUR OWN — 1024d via Ollama
|
|
"behavioral": "crypto_behavioral", # LOCAL — numeric vectors, zero cost
|
|
"entity": "crypto_behavioral", # LOCAL — numeric vectors, zero cost
|
|
}
|
|
|
|
# Primary embedding model: OUR OWN bge-m3 via Ollama (1024d)
|
|
OR_PRIMARY_EMBED = "rmi/bge-m3"
|
|
# OpenRouter fallback: NVIDIA Nemotron (2048d, free) — only if key works
|
|
OR_PAID_FALLBACK = "nvidia/llama-nemotron-embed-vl-1b-v2:free"
|
|
# Always-available local fallback: BGE-small (384d, CPU, zero cost)
|
|
OR_FALLBACK_EMBED = "local/bge-small-en-v1.5"
|
|
# NVIDIA NIM embed endpoint
|
|
NVIDIA_EMBED_URL = "https://integrate.api.nvidia.com/v1/embeddings"
|
|
|
|
COLLECTIONS = [
|
|
"wallet_profiles",
|
|
"token_analysis",
|
|
"scam_patterns",
|
|
"forensic_reports",
|
|
"market_intel",
|
|
"contract_audits",
|
|
"known_scams",
|
|
"news_articles",
|
|
"transaction_patterns",
|
|
"defi_hacks",
|
|
"rug_timeline",
|
|
"vuln_patterns",
|
|
"crime_reports",
|
|
]
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# CRYPTO-AWARE FEATURE EXTRACTORS (no API needed, runs locally)
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def extract_contract_features(code: str) -> np.ndarray:
|
|
"""
|
|
Extract structural features from smart contract code.
|
|
Returns a 128-dim vector encoding:
|
|
- Opcode/instruction frequency
|
|
- Common rug patterns (honeypot, maxTx, blacklist, mint, ownership)
|
|
- Complexity metrics (cyclomatic, nesting depth)
|
|
- Library/import fingerprint
|
|
"""
|
|
vec = np.zeros(128, dtype=np.float32)
|
|
|
|
if not code or len(code) < 10:
|
|
return vec
|
|
|
|
code_lower = code.lower()
|
|
|
|
# ── Rug pull keyword patterns (first 32 dims) ──
|
|
rug_patterns = [
|
|
# Honeypot indicators
|
|
"maxTx",
|
|
"maxtransaction",
|
|
"_maxTxAmount",
|
|
"maxSell",
|
|
"maxBuy",
|
|
"tradingEnabled",
|
|
"tradingActive",
|
|
"enableTrading",
|
|
"isTradingEnabled",
|
|
"blacklist",
|
|
"isBlacklisted",
|
|
"_blacklist",
|
|
"addToBlacklist",
|
|
# Mint/ownership abuse
|
|
"mint",
|
|
"onlyOwner",
|
|
"excludeFromFees",
|
|
"excludeFromReward",
|
|
"setTax",
|
|
"updateTax",
|
|
"setFee",
|
|
"taxFee",
|
|
"liquidityFee",
|
|
# Hidden functions
|
|
"sweep",
|
|
"drain",
|
|
"rug",
|
|
"pullLiquidity",
|
|
"removeLiquidity",
|
|
"manualSwap",
|
|
"manualSend",
|
|
"swapAndLiquify",
|
|
"swapAndLiquify",
|
|
# Proxy / upgrade traps
|
|
"upgradeTo",
|
|
"upgradeToAndCall",
|
|
"changeAdmin",
|
|
"selfdestruct",
|
|
"suicide",
|
|
"delegatecall",
|
|
# Unverified patterns
|
|
"transferOwnership",
|
|
"renounceOwnership",
|
|
"lock",
|
|
]
|
|
for i, pattern in enumerate(rug_patterns):
|
|
if pattern in code_lower:
|
|
vec[i] = 1.0
|
|
|
|
# ── Structural complexity (dims 32-40) ──
|
|
lines = code.count("\n") + 1
|
|
vec[32] = min(lines / 5000.0, 1.0) # normalized line count
|
|
vec[33] = min(code.count("function ") / 100.0, 1.0) # function count
|
|
vec[34] = min(code.count("require(") / 50.0, 1.0) # safety checks
|
|
vec[35] = min(code.count("modifier ") / 20.0, 1.0) # access control
|
|
vec[36] = min(code.count("event ") / 30.0, 1.0) # event emissions
|
|
vec[37] = min(code.count("assembly") / 10.0, 1.0) # low-level code
|
|
vec[38] = min(code.count("for (") + code.count("while ("), 1.0) / 20.0
|
|
vec[39] = min(code.count("if (") / 100.0, 1.0)
|
|
|
|
# ── Library/import fingerprint (dims 40-50) ──
|
|
imports = [
|
|
"openzeppelin",
|
|
"uniswap",
|
|
"pancake",
|
|
"sushiswap",
|
|
"chainlink",
|
|
"@openzeppelin",
|
|
"@uniswap",
|
|
"erc20",
|
|
"erc721",
|
|
"ownable",
|
|
]
|
|
for i, imp in enumerate(imports):
|
|
vec[40 + i] = 1.0 if imp in code_lower else 0.0
|
|
|
|
# ── Language fingerprint (dims 50-55) ──
|
|
lang_markers = {
|
|
"solidity": ["pragma solidity", "contract ", "address public"],
|
|
"rust": ["fn ", "pub fn", "struct ", "impl ", "#[account]"],
|
|
"move": ["module ", "public fun", "struct ", "acquires"],
|
|
}
|
|
idx = 50
|
|
for _lang, markers in lang_markers.items():
|
|
score = sum(1 for m in markers if m in code)
|
|
vec[idx] = min(score / len(markers), 1.0)
|
|
idx += 1
|
|
|
|
# ── Hash-based fingerprint (dims 56-127) ──
|
|
code_hash = hashlib.sha256(code.encode()).digest()
|
|
for i in range(72):
|
|
vec[56 + i] = (code_hash[i % 32] / 255.0) if i < 72 else 0.0
|
|
|
|
return vec
|
|
|
|
|
|
def extract_transaction_features(tx_data: dict) -> np.ndarray:
|
|
"""
|
|
Extract behavioral features from transaction history.
|
|
Returns a 64-dim vector capturing:
|
|
- Volume patterns (mean, std, burst frequency)
|
|
- Time patterns (hour-of-day, day-of-week clusters)
|
|
- Counterparty diversity
|
|
- Gas patterns
|
|
- Token diversity
|
|
"""
|
|
vec = np.zeros(64, dtype=np.float32)
|
|
|
|
txs = tx_data.get("transactions", []) if isinstance(tx_data, dict) else []
|
|
|
|
if not txs:
|
|
return vec
|
|
|
|
# ── Volume patterns (dims 0-9) ──
|
|
amounts = []
|
|
timestamps = []
|
|
counterparties = set()
|
|
programs = set()
|
|
for tx in txs:
|
|
amt = float(tx.get("amount", 0) or 0)
|
|
ts = tx.get("timestamp", 0) or tx.get("block_time", 0)
|
|
if amt > 0:
|
|
amounts.append(amt)
|
|
if ts:
|
|
timestamps.append(int(ts))
|
|
cp = tx.get("counterparty") or tx.get("from") or tx.get("to")
|
|
if cp:
|
|
counterparties.add(str(cp))
|
|
prog = tx.get("program") or tx.get("instruction")
|
|
if prog:
|
|
programs.add(str(prog))
|
|
|
|
if amounts:
|
|
arr = np.array(amounts, dtype=np.float32)
|
|
vec[0] = np.log1p(float(np.mean(arr))) / 20.0
|
|
vec[1] = min(float(np.std(arr)) / (float(np.mean(arr)) + 1), 1.0)
|
|
vec[2] = min(float(np.max(arr)) / (float(np.sum(arr)) + 1), 1.0)
|
|
vec[3] = min(len(amounts) / 1000.0, 1.0)
|
|
# Burst detection: consecutive txs within short windows
|
|
if len(amounts) > 2:
|
|
vec[4] = min(float(np.sum(arr[:5])) / (float(np.sum(arr)) + 1), 1.0)
|
|
|
|
# ── Temporal patterns (dims 10-19) ──
|
|
if len(timestamps) > 1:
|
|
ts_sorted = sorted(timestamps)
|
|
gaps = np.diff(ts_sorted)
|
|
vec[10] = min(float(np.mean(gaps)) / 3600.0, 1.0) # avg gap in hours
|
|
vec[11] = min(float(np.std(gaps)) / 3600.0, 1.0)
|
|
# Hour-of-day entropy
|
|
hours = [datetime.fromtimestamp(t).hour for t in ts_sorted]
|
|
hour_counts = np.bincount(hours, minlength=24)
|
|
hour_probs = hour_counts / (hour_counts.sum() + 1)
|
|
entropy = -np.sum(hour_probs * np.log(hour_probs + 1e-10))
|
|
vec[12] = min(entropy / 3.2, 1.0) # normalized entropy
|
|
# Burstiness: ratio of txs in 10% of time to total
|
|
window = max(1, (ts_sorted[-1] - ts_sorted[0]) // 10)
|
|
burst_txs = sum(1 for i in range(len(ts_sorted) - 1) if ts_sorted[i + 1] - ts_sorted[i] < max(60, window))
|
|
vec[13] = min(burst_txs / max(1, len(ts_sorted)), 1.0)
|
|
|
|
# ── Diversity metrics (dims 20-29) ──
|
|
vec[20] = min(len(counterparties) / 200.0, 1.0)
|
|
vec[21] = min(len(programs) / 50.0, 1.0)
|
|
vec[22] = min(len(counterparties) / max(1, len(txs)), 1.0) # uniqueness ratio
|
|
# Token interaction count
|
|
tokens_seen = set()
|
|
for tx in txs:
|
|
for field in ["token", "mint", "token_address", "contract"]:
|
|
if field in tx:
|
|
tokens_seen.add(str(tx[field]))
|
|
vec[23] = min(len(tokens_seen) / 100.0, 1.0)
|
|
|
|
# ── Risk signals (dims 30-39) ──
|
|
risk = tx_data.get("risk_signals", {}) if isinstance(tx_data, dict) else {}
|
|
vec[30] = 1.0 if risk.get("mixer_interaction") else 0.0
|
|
vec[31] = 1.0 if risk.get("sanctioned_counterparty") else 0.0
|
|
vec[32] = 1.0 if risk.get("flash_loan_usage") else 0.0
|
|
vec[33] = min(risk.get("wash_trading_score", 0) / 100.0, 1.0)
|
|
vec[34] = 1.0 if risk.get("sandwich_attack_target") else 0.0
|
|
vec[35] = 1.0 if risk.get("new_wallet", True) else 0.0 # default True = suspicious
|
|
vec[36] = min(risk.get("failed_tx_ratio", 0), 1.0)
|
|
vec[37] = min(risk.get("rug_pull_pattern_score", 0) / 100.0, 1.0)
|
|
|
|
# ── Hash fingerprint of counterparty graph (dims 40-63) ──
|
|
cp_sorted = sorted(counterparties)[:50]
|
|
cp_hash = hashlib.sha256(",".join(cp_sorted).encode()).digest()
|
|
for i in range(24):
|
|
vec[40 + i] = cp_hash[i] / 255.0
|
|
|
|
return vec
|
|
|
|
|
|
def extract_wallet_features(wallet_data: dict) -> np.ndarray:
|
|
"""
|
|
Extract wallet entity features for clustering/scam detection.
|
|
Returns a 64-dim vector.
|
|
"""
|
|
vec = np.zeros(64, dtype=np.float32)
|
|
|
|
if not wallet_data:
|
|
return vec
|
|
|
|
# ── Age & activity (dims 0-9) ──
|
|
created = wallet_data.get("created_at") or wallet_data.get("first_seen")
|
|
if created:
|
|
try:
|
|
if isinstance(created, str):
|
|
created_ts = datetime.fromisoformat(created.replace("Z", "+00:00")).timestamp()
|
|
else:
|
|
created_ts = float(created)
|
|
age_days = max(0, (time.time() - created_ts) / 86400)
|
|
vec[0] = min(age_days / 365.0, 1.0) # normalized age
|
|
vec[1] = 1.0 if age_days < 7 else 0.0 # newborn wallet (suspicious)
|
|
vec[2] = 1.0 if age_days < 30 else 0.0 # young wallet
|
|
except (ValueError, TypeError):
|
|
vec[0] = 0.1 # unknown age → slightly suspicious
|
|
|
|
# ── Balance/Value (dims 10-19) ──
|
|
balance = float(wallet_data.get("balance_usd", 0) or 0)
|
|
vec[10] = min(np.log1p(balance) / 15.0, 1.0) # log-scaled balance
|
|
txn_count = int(wallet_data.get("transaction_count", 0) or 0)
|
|
vec[11] = min(txn_count / 10000.0, 1.0)
|
|
vec[12] = min(float(wallet_data.get("avg_txn_value_usd", 0) or 0) / 100000.0, 1.0)
|
|
|
|
# ── Risk labels (dims 20-29) ──
|
|
labels = wallet_data.get("labels", []) or []
|
|
risk_tags = {
|
|
"scammer": 20,
|
|
"rug_puller": 21,
|
|
"honeypot_creator": 22,
|
|
"wash_trader": 23,
|
|
"mixer_user": 24,
|
|
"exploiter": 25,
|
|
"sanctioned": 26,
|
|
"phisher": 27,
|
|
"insider": 28,
|
|
"bot": 29,
|
|
}
|
|
for label in labels:
|
|
label_lower = str(label).lower().replace(" ", "_")
|
|
for tag, dim in risk_tags.items():
|
|
if tag in label_lower:
|
|
vec[dim] = 1.0
|
|
|
|
# ── Entity cluster hash (dims 30-39) ──
|
|
cluster = str(wallet_data.get("cluster_id", ""))
|
|
if cluster:
|
|
ch = hashlib.md5(cluster.encode()).digest()
|
|
for i in range(10):
|
|
vec[30 + i] = ch[i] / 255.0
|
|
|
|
# ── Chain activity fingerprint (dims 40-49) ──
|
|
chains = wallet_data.get("active_chains", []) or []
|
|
chain_list = [
|
|
"ethereum",
|
|
"solana",
|
|
"base",
|
|
"bsc",
|
|
"polygon",
|
|
"arbitrum",
|
|
"optimism",
|
|
"avalanche",
|
|
"fantom",
|
|
"tron",
|
|
]
|
|
for i, ch in enumerate(chain_list):
|
|
if ch in [c.lower() for c in chains]:
|
|
vec[40 + i] = 1.0
|
|
|
|
# ── Hash of associated addresses (dims 50-63) ──
|
|
assoc = str(wallet_data.get("associated_addresses", ""))
|
|
ah = hashlib.sha256(assoc.encode()).digest()
|
|
for i in range(14):
|
|
vec[50 + i] = ah[i] / 255.0
|
|
|
|
return vec
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# LOCAL BGE EMBEDDER (primary — always available, zero cost)
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class LocalBGEEmbedder:
|
|
"""
|
|
Local BAAI BGE-small-en-v1.5 embedder.
|
|
Runs on CPU, ~80MB RAM, 384-dim vectors.
|
|
Loaded lazily — first call initializes the model.
|
|
FALLBACK only. Primary is Ollama bge-m3.
|
|
"""
|
|
|
|
MODEL_NAME = "BAAI/bge-small-en-v1.5"
|
|
_model = None
|
|
|
|
@classmethod
|
|
def _get_model(cls):
|
|
if cls._model is None:
|
|
from sentence_transformers import SentenceTransformer
|
|
|
|
cls._model = SentenceTransformer(cls.MODEL_NAME)
|
|
logger.info(f"LocalBGEEmbedder loaded: {cls.MODEL_NAME} (384d)")
|
|
return cls._model
|
|
|
|
async def embed(self, texts: list[str]) -> list[list[float]]:
|
|
model = self._get_model()
|
|
embeddings = await asyncio.to_thread(lambda: model.encode(texts, normalize_embeddings=True).tolist())
|
|
return embeddings
|
|
|
|
async def embed_one(self, text: str) -> list[float]:
|
|
results = await self.embed([text])
|
|
return results[0]
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# PRIMARY EMBEDDER — Our own Ollama bge-m3 (1024d, zero external API)
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class OllamaBGEEmbedder:
|
|
"""Our own bge-m3 via Ollama. 1024d, always available, zero cost."""
|
|
|
|
OLLAMA_URL = "http://172.19.0.1:11434"
|
|
MODEL = "bge-m3"
|
|
DIMS = 1024
|
|
|
|
def __init__(self, ollama_url: str | None = None):
|
|
self.url = ollama_url or os.getenv("OLLAMA_URL", self.OLLAMA_URL)
|
|
|
|
async def embed(self, texts: list[str]) -> list[list[float]]:
|
|
import httpx
|
|
|
|
results = []
|
|
async with httpx.AsyncClient(timeout=3) as client:
|
|
for text in texts:
|
|
try:
|
|
resp = await client.post(
|
|
f"{self.url}/api/embeddings",
|
|
json={"model": self.MODEL, "prompt": text[:500]},
|
|
)
|
|
if resp.status_code == 200:
|
|
vec = resp.json().get("embedding", [])
|
|
if len(vec) == self.DIMS:
|
|
results.append(vec)
|
|
continue
|
|
except Exception:
|
|
pass
|
|
# Fallback: zero vector
|
|
results.append([0.0] * self.DIMS)
|
|
return results
|
|
|
|
async def embed_one(self, text: str) -> list[float]:
|
|
results = await self.embed([text])
|
|
return results[0]
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# API EMBEDDING PROVIDERS (fallbacks for higher quality)
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class OpenRouterEmbedder:
|
|
"""OpenRouter embedding API — NVIDIA Nemotron primary, bge-m3 fallback."""
|
|
|
|
BASE = "https://openrouter.ai/api/v1/embeddings"
|
|
MODEL = OR_PRIMARY_EMBED
|
|
FALLBACK = OR_FALLBACK_EMBED
|
|
|
|
def __init__(self, api_key: str | None = None):
|
|
self.api_key = api_key or OPENROUTER_KEY
|
|
if not self.api_key:
|
|
raise ValueError("OPENROUTER_API_KEY required")
|
|
|
|
async def embed(self, texts: list[str], model: str | None = None) -> list[list[float]]:
|
|
import httpx
|
|
|
|
model_id = model or self.MODEL
|
|
|
|
async def _try_embed(m: str) -> list[list[float]]:
|
|
headers = {
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
payload = {"model": m, "input": texts}
|
|
async with httpx.AsyncClient(timeout=3) as client:
|
|
resp = await client.post(self.BASE, headers=headers, json=payload)
|
|
if resp.status_code == 429:
|
|
raise RuntimeError("RATE_LIMITED")
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
return [d["embedding"] for d in sorted(data["data"], key=lambda x: x["index"])]
|
|
|
|
try:
|
|
return await _try_embed(model_id)
|
|
except Exception as e:
|
|
if "RATE_LIMITED" in str(e) and model_id != self.FALLBACK:
|
|
logger.warning(f"NVIDIA embed rate-limited, falling back to {self.FALLBACK}")
|
|
return await _try_embed(self.FALLBACK)
|
|
raise
|
|
|
|
async def embed_one(self, text: str, model: str | None = None) -> list[float]:
|
|
results = await self.embed([text], model)
|
|
return results[0]
|
|
|
|
|
|
class HuggingFaceEmbedder:
|
|
"""HuggingFace Inference API — free tier fallback."""
|
|
|
|
BASE = "https://api-inference.huggingface.co/models"
|
|
MODEL = "BAAI/bge-m3"
|
|
|
|
def __init__(self, token: str | None = None):
|
|
self.token = token or HF_TOKEN
|
|
if not self.token:
|
|
raise ValueError("HUGGINGFACE_TOKEN required")
|
|
|
|
async def embed(self, texts: list[str], model: str | None = None) -> list[list[float]]:
|
|
import httpx
|
|
|
|
model_id = model or self.MODEL
|
|
headers = {
|
|
"Authorization": f"Bearer {self.token}",
|
|
"Content-Type": "application/json",
|
|
"x-wait-for-model": "true",
|
|
}
|
|
url = f"{self.BASE}/{model_id}"
|
|
async with httpx.AsyncClient(timeout=60) as client:
|
|
# HF API expects either a single string or list
|
|
results: list[list[float]] = []
|
|
for text in texts:
|
|
resp = await client.post(url, headers=headers, json={"inputs": text})
|
|
if resp.status_code == 503: # model loading
|
|
await asyncio.sleep(3)
|
|
resp = await client.post(url, headers=headers, json={"inputs": text})
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
# HF feature-extraction returns nested lists: [[0.1, 0.2, ...]]
|
|
# Handle both shapes: list of floats or list of list of floats
|
|
if isinstance(data, list) and len(data) > 0:
|
|
if isinstance(data[0], list):
|
|
# Already [[float]] — take the first (mean-pooled usually)
|
|
results.append(data[0] if len(data) == 1 else data[0])
|
|
else:
|
|
# Flat [float] — wrap it
|
|
results.append(data)
|
|
else:
|
|
logger.warning(f"HF returned unexpected shape: {type(data)}")
|
|
results.append([0.0] * 1024)
|
|
return results
|
|
|
|
async def embed_one(self, text: str, model: str | None = None) -> list[float]:
|
|
results = await self.embed([text], model)
|
|
return results[0]
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# CONTRACT CODE EMBEDDER (specialized)
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class ContractCodeEmbedder:
|
|
"""
|
|
Specialized contract code embedder.
|
|
Uses structural features + semantic embedding of decompiled code.
|
|
Detects rug patterns even in obfuscated contracts.
|
|
"""
|
|
|
|
@staticmethod
|
|
async def embed(code: str, semantic_embedder=None) -> np.ndarray:
|
|
"""
|
|
Generate a combined 320-dim contract embedding:
|
|
- 128 dims: structural features (local)
|
|
- 192 dims: semantic understanding (API) — only for non-trivial code
|
|
"""
|
|
structural = extract_contract_features(code)
|
|
|
|
if semantic_embedder and len(code) > 50:
|
|
# Extract the meaningful parts (not just imports/boilerplate)
|
|
# Take first 4000 chars — enough for the key logic
|
|
code_snippet = code[:4000]
|
|
try:
|
|
semantic_vec = await semantic_embedder.embed_one(f"Smart contract code: {code_snippet[:3000]}")
|
|
# Truncate or pad to 192
|
|
semantic = np.array(semantic_vec[:192], dtype=np.float32)
|
|
if len(semantic) < 192:
|
|
semantic = np.pad(semantic, (0, 192 - len(semantic)))
|
|
except Exception as e:
|
|
logger.warning(f"Semantic contract embedding failed: {e}")
|
|
semantic = np.zeros(192, dtype=np.float32)
|
|
else:
|
|
semantic = np.zeros(192, dtype=np.float32)
|
|
|
|
return np.concatenate([structural, semantic])
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# MAIN CRYPTO EMBEDDER
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@dataclass
|
|
class EmbeddingResult:
|
|
vector: list[float]
|
|
dims: int
|
|
model: str
|
|
head: str
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
cached: bool = False
|
|
|
|
|
|
class CryptoEmbedder:
|
|
"""
|
|
Ultimate multi-head embedder for crypto scam detection.
|
|
|
|
Usage:
|
|
embedder = CryptoEmbedder()
|
|
await embedder.initialize()
|
|
|
|
# Token scam analysis
|
|
vec = await embedder.embed_token_scam(
|
|
name="FakeRugToken",
|
|
symbol="RUG",
|
|
description="Revolutionary defi...",
|
|
contract_code="pragma solidity...",
|
|
creator_address="0x...",
|
|
)
|
|
|
|
# Wallet investigation
|
|
vec = await embedder.embed_wallet(
|
|
address="0x...",
|
|
transactions=[...],
|
|
labels=["suspected_scammer"],
|
|
)
|
|
|
|
# Semantic search
|
|
results = await embedder.search("honeypot tokens with ownership renounce pattern")
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._local: LocalBGEEmbedder | None = None
|
|
self._ollama: OllamaBGEEmbedder | None = None
|
|
self._openrouter: OpenRouterEmbedder | None = None
|
|
self._huggingface: HuggingFaceEmbedder | None = None
|
|
self._redis = None
|
|
self._initialized = False
|
|
self._cache_hits = 0
|
|
self._cache_misses = 0
|
|
|
|
async def initialize(self):
|
|
"""Initialize embedding providers. Ollama qwen3-embedding is PRIMARY."""
|
|
or_key = os.environ.get("OPENROUTER_API_KEY", "") or OPENROUTER_KEY
|
|
hf_token = os.environ.get("HUGGINGFACE_TOKEN", "") or HF_TOKEN
|
|
|
|
# PRIMARY: Our own qwen3-embedding:4b via Ollama (2048d, #2 MTEB)
|
|
self._ollama = OllamaBGEEmbedder()
|
|
logger.info("CryptoEmbedder: Ollama qwen3-embedding:4b ready (PRIMARY, 2048d)")
|
|
|
|
self._local = LocalBGEEmbedder()
|
|
logger.info("CryptoEmbedder: Local BGE-small ready (fallback, 384d)")
|
|
|
|
if or_key:
|
|
self._openrouter = OpenRouterEmbedder(api_key=or_key)
|
|
logger.info("CryptoEmbedder: OpenRouter ready (fallback, 2048d NVIDIA Nemotron)")
|
|
|
|
if hf_token:
|
|
self._huggingface = HuggingFaceEmbedder(token=hf_token)
|
|
logger.info("CryptoEmbedder: HuggingFace ready (fallback, 1024d)")
|
|
|
|
self._initialized = True
|
|
logger.info("CryptoEmbedder initialized")
|
|
|
|
async def _get_redis(self):
|
|
import redis.asyncio as redis
|
|
|
|
if self._redis is None:
|
|
self._redis = redis.Redis(
|
|
host=REDIS_HOST,
|
|
port=REDIS_PORT,
|
|
password=REDIS_PW or None,
|
|
db=0,
|
|
decode_responses=True,
|
|
)
|
|
return self._redis
|
|
|
|
async def _semantic_embed(self, texts: list[str], head: str = "semantic") -> list[list[float]]:
|
|
"""
|
|
Semantic embedding pipeline.
|
|
PRIMARY: Our own qwen3-embedding:4b via Ollama (2048d, #2 MTEB)
|
|
FALLBACK: OpenRouter NVIDIA Nemotron (2048d, free)
|
|
FALLBACK: Local BGE-small (384d, zero cost)
|
|
LAST: Hash-based (2048d, deterministic)
|
|
"""
|
|
if not texts:
|
|
return []
|
|
|
|
# PRIMARY: Our own qwen3-embedding:4b (2048d, best quality, zero cost)
|
|
if self._ollama:
|
|
try:
|
|
return await self._ollama.embed(texts)
|
|
except Exception as e:
|
|
logger.warning(f"Ollama qwen3 embedding failed: {e}")
|
|
|
|
# Fallback: OpenRouter NVIDIA (2048d, free)
|
|
if self._openrouter:
|
|
try:
|
|
return await self._openrouter.embed(texts)
|
|
except Exception as e:
|
|
logger.warning(f"OpenRouter/NVIDIA embedding failed: {e}")
|
|
|
|
# Fallback: Local BGE (384d, always available, zero cost)
|
|
if self._local:
|
|
try:
|
|
return await self._local.embed(texts)
|
|
except Exception as e:
|
|
logger.warning(f"Local BGE embedding failed: {e}")
|
|
|
|
# Fallback: HuggingFace
|
|
if self._huggingface:
|
|
try:
|
|
return await self._huggingface.embed(texts)
|
|
except Exception as e:
|
|
logger.warning(f"HuggingFace embedding failed: {e}")
|
|
|
|
# Last resort: hash-based
|
|
logger.warning("Using hash-based fallback embeddings (low quality)")
|
|
return [self._hash_embed(text) for text in texts]
|
|
|
|
async def _semantic_embed_one(self, text: str, head: str = "semantic") -> list[float]:
|
|
# Check cache first
|
|
cache_key = await self._cache_key(head, text)
|
|
cached = await self._cache_get(cache_key)
|
|
if cached is not None:
|
|
return cached
|
|
|
|
results = await self._semantic_embed([text], head)
|
|
vector = results[0] if results else self._hash_embed(text)
|
|
|
|
# Store in cache (24h TTL for semantic embeddings)
|
|
await self._cache_set(cache_key, vector, ttl=86400)
|
|
return vector
|
|
|
|
def _hash_embed(self, text: str, dims: int = 1024) -> list[float]:
|
|
"""Deterministic hash-based embedding (emergency fallback). 1024d matches bge-m3."""
|
|
vec = np.zeros(dims, dtype=np.float32)
|
|
words = re.findall(r"\w+", text.lower())
|
|
for word in words:
|
|
h = int(hashlib.md5(word.encode()).hexdigest(), 16)
|
|
for i in range(min(8, dims)):
|
|
vec[(h + i) % dims] += 0.01
|
|
norm = np.linalg.norm(vec)
|
|
return (vec / norm).tolist() if norm > 0 else vec.tolist()
|
|
|
|
async def _cache_key(self, prefix: str, text: str) -> str:
|
|
h = hashlib.sha256(text.encode()).hexdigest()[:32]
|
|
return f"emb:{prefix}:{h}"
|
|
|
|
async def _cache_get(self, key: str) -> list[float] | None:
|
|
try:
|
|
r = await self._get_redis()
|
|
data = await r.get(key)
|
|
if data:
|
|
self._cache_hits += 1
|
|
return json.loads(data)
|
|
self._cache_misses += 1
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
async def _cache_set(self, key: str, vector: list[float], ttl: int = 86400):
|
|
try:
|
|
r = await self._get_redis()
|
|
await r.setex(key, ttl, json.dumps(vector))
|
|
except Exception:
|
|
pass
|
|
|
|
# ─── SPECIALIZED EMBED METHODS ────────────────────────────────
|
|
|
|
async def embed_token_scam(
|
|
self,
|
|
name: str,
|
|
symbol: str,
|
|
description: str = "",
|
|
contract_code: str = "",
|
|
creator_address: str = "",
|
|
chain: str = "solana",
|
|
metadata: dict | None = None,
|
|
) -> EmbeddingResult:
|
|
"""
|
|
Generate a comprehensive token scam embedding.
|
|
Combines semantic, code, and behavioral vectors.
|
|
"""
|
|
# Build rich semantic text
|
|
semantic_text = f"""Token Analysis: {name} ({symbol}) on {chain}.
|
|
Description: {description[:2000]}
|
|
Creator: {creator_address}
|
|
Metadata: {json.dumps(metadata or {})[:1000]}"""
|
|
|
|
# Get semantic embedding
|
|
semantic_vec = await self._semantic_embed_one(semantic_text, "semantic")
|
|
semantic_dim = len(semantic_vec)
|
|
|
|
# Get code features
|
|
code_vec = extract_contract_features(contract_code).tolist() if contract_code else [0.0] * 128
|
|
|
|
# Get behavioral features (from metadata)
|
|
behavioral = extract_transaction_features(metadata or {})
|
|
behavioral_vec = behavioral.tolist()
|
|
|
|
# Get wallet features for creator
|
|
wallet = extract_wallet_features(
|
|
{
|
|
"address": creator_address,
|
|
"labels": metadata.get("creator_labels", []) if metadata else [],
|
|
}
|
|
)
|
|
wallet_vec = wallet.tolist()
|
|
|
|
# Concatenate: semantic (d) + code (128) + behavioral (64) + wallet (64)
|
|
combined = semantic_vec + code_vec + behavioral_vec + wallet_vec
|
|
|
|
return EmbeddingResult(
|
|
vector=combined,
|
|
dims=len(combined),
|
|
model="crypto_multihead_v1",
|
|
head="token_scam",
|
|
metadata={
|
|
"name": name,
|
|
"symbol": symbol,
|
|
"chain": chain,
|
|
"semantic_dim": semantic_dim,
|
|
"has_code": bool(contract_code),
|
|
"has_behavioral": bool(metadata),
|
|
},
|
|
)
|
|
|
|
async def embed_wallet(
|
|
self,
|
|
address: str,
|
|
transactions: list[dict] | None = None,
|
|
labels: list[str] | None = None,
|
|
chain: str = "solana",
|
|
balance_usd: float = 0,
|
|
) -> EmbeddingResult:
|
|
"""
|
|
Generate a comprehensive wallet risk embedding.
|
|
"""
|
|
wallet_data = {
|
|
"address": address,
|
|
"labels": labels or [],
|
|
"balance_usd": balance_usd,
|
|
"transaction_count": len(transactions or []),
|
|
}
|
|
wallet_vec = extract_wallet_features(wallet_data).tolist()
|
|
|
|
behavioral_vec = extract_transaction_features({"transactions": transactions or []}).tolist()
|
|
|
|
# Semantic: labels + chain context
|
|
semantic_text = f"Wallet {address} on {chain}. Labels: {', '.join(labels or ['unknown'])}. Activity: {len(transactions or [])} transactions."
|
|
semantic_vec = await self._semantic_embed_one(semantic_text, "semantic")
|
|
|
|
combined = semantic_vec + wallet_vec + behavioral_vec
|
|
|
|
return EmbeddingResult(
|
|
vector=combined,
|
|
dims=len(combined),
|
|
model="crypto_multihead_v1",
|
|
head="wallet",
|
|
metadata={
|
|
"address": address,
|
|
"chain": chain,
|
|
"labels": labels or [],
|
|
"tx_count": len(transactions or []),
|
|
},
|
|
)
|
|
|
|
async def embed_scam_pattern(
|
|
self,
|
|
pattern_name: str,
|
|
description: str,
|
|
code_snippets: list[str] | None = None,
|
|
indicators: list[str] | None = None,
|
|
severity: str = "high",
|
|
) -> EmbeddingResult:
|
|
"""
|
|
Embed a known scam pattern for similarity matching.
|
|
"""
|
|
semantic_text = f"""SCAM PATTERN: {pattern_name}
|
|
Severity: {severity}
|
|
Description: {description}
|
|
Indicators: {", ".join(indicators or [])}
|
|
Code patterns: {"; ".join(code_snippets or [])[:3000]}"""
|
|
|
|
semantic_vec = await self._semantic_embed_one(semantic_text, "semantic")
|
|
len(semantic_vec)
|
|
|
|
# Code features from snippets
|
|
all_code = "\n".join(code_snippets or [])
|
|
code_vec = extract_contract_features(all_code).tolist()
|
|
|
|
combined = semantic_vec + code_vec
|
|
|
|
return EmbeddingResult(
|
|
vector=combined,
|
|
dims=len(combined),
|
|
model="crypto_multihead_v1",
|
|
head="scam_pattern",
|
|
metadata={
|
|
"pattern": pattern_name,
|
|
"severity": severity,
|
|
"indicators": indicators or [],
|
|
},
|
|
)
|
|
|
|
async def embed_contract(
|
|
self,
|
|
address: str,
|
|
code: str,
|
|
chain: str = "ethereum",
|
|
verified: bool = False,
|
|
) -> EmbeddingResult:
|
|
"""Full contract audit embedding."""
|
|
return await ContractCodeEmbedder.embed(
|
|
code,
|
|
semantic_embedder=self._openrouter,
|
|
)
|
|
|
|
async def embed_news(
|
|
self,
|
|
title: str,
|
|
content: str,
|
|
source: str = "",
|
|
tags: list[str] | None = None,
|
|
) -> EmbeddingResult:
|
|
"""News article embedding for market intel."""
|
|
text = f"TITLE: {title}\nSOURCE: {source}\nTAGS: {', '.join(tags or [])}\nCONTENT: {content[:3000]}"
|
|
vec = await self._semantic_embed_one(text, "semantic")
|
|
|
|
return EmbeddingResult(
|
|
vector=vec,
|
|
dims=len(vec),
|
|
model="openai/text-embedding-3-large",
|
|
head="news",
|
|
metadata={"title": title, "source": source, "tags": tags or []},
|
|
)
|
|
|
|
async def embed_query(self, query: str) -> list[float]:
|
|
"""Embed a search query for RAG retrieval."""
|
|
# Augment crypto-specific context
|
|
augmented = f"Cryptocurrency scam detection query: {query}"
|
|
return await self._semantic_embed_one(augmented, "semantic")
|
|
|
|
# ─── SIMILARITY & SEARCH ────────────────────────────────────
|
|
|
|
@staticmethod
|
|
def cosine_similarity(a: list[float], b: list[float]) -> float:
|
|
"""Cosine similarity between two vectors."""
|
|
a_arr = np.array(a, dtype=np.float32)
|
|
b_arr = np.array(b, dtype=np.float32)
|
|
dot = np.dot(a_arr, b_arr)
|
|
norm_a = np.linalg.norm(a_arr)
|
|
norm_b = np.linalg.norm(b_arr)
|
|
if norm_a == 0 or norm_b == 0:
|
|
return 0.0
|
|
return float(dot / (norm_a * norm_b))
|
|
|
|
@staticmethod
|
|
def weighted_similarity(
|
|
vec_a: list[float],
|
|
vec_b: list[float],
|
|
semantic_dim: int,
|
|
weights: tuple[float, float, float, float] = (0.4, 0.2, 0.2, 0.2),
|
|
) -> float:
|
|
"""
|
|
Weighted similarity with head-aware decomposition.
|
|
Weights: (semantic, code, behavioral, wallet)
|
|
Assumes vector layout: [semantic | code(128) | behavioral(64) | wallet(64)]
|
|
"""
|
|
if len(vec_a) != len(vec_b):
|
|
# Different dims — use common prefix
|
|
min_len = min(len(vec_a), len(vec_b))
|
|
return CryptoEmbedder.cosine_similarity(vec_a[:min_len], vec_b[:min_len])
|
|
|
|
code_start = semantic_dim
|
|
behav_start = code_start + 128
|
|
wallet_start = behav_start + 64
|
|
|
|
sem_sim = CryptoEmbedder.cosine_similarity(vec_a[:semantic_dim], vec_b[:semantic_dim])
|
|
code_sim = (
|
|
CryptoEmbedder.cosine_similarity(vec_a[code_start:behav_start], vec_b[code_start:behav_start])
|
|
if len(vec_a) > code_start
|
|
else 0.0
|
|
)
|
|
behav_sim = (
|
|
CryptoEmbedder.cosine_similarity(vec_a[behav_start:wallet_start], vec_b[behav_start:wallet_start])
|
|
if len(vec_a) > behav_start
|
|
else 0.0
|
|
)
|
|
wallet_sim = (
|
|
CryptoEmbedder.cosine_similarity(vec_a[wallet_start:], vec_b[wallet_start:])
|
|
if len(vec_a) > wallet_start
|
|
else 0.0
|
|
)
|
|
|
|
return weights[0] * sem_sim + weights[1] * code_sim + weights[2] * behav_sim + weights[3] * wallet_sim
|
|
|
|
# ─── VECTOR STORE INTEGRATION ───────────────────────────────
|
|
|
|
async def store(
|
|
self,
|
|
collection: str,
|
|
doc_id: str,
|
|
vector: list[float],
|
|
metadata: dict[str, Any] | None = None,
|
|
content: str = "",
|
|
) -> bool:
|
|
"""Store an embedding in Redis for fast retrieval."""
|
|
try:
|
|
r = await self._get_redis()
|
|
key = f"rag:{collection}:{doc_id}"
|
|
|
|
doc = {
|
|
"id": doc_id,
|
|
"vector": vector,
|
|
"metadata": metadata or {},
|
|
"content": content[:5000],
|
|
"stored_at": datetime.now(UTC).isoformat(),
|
|
}
|
|
await r.setex(key, 86400 * 30, json.dumps(doc)) # 30 day TTL
|
|
|
|
# Add to collection index
|
|
await r.sadd(f"rag:idx:{collection}", doc_id)
|
|
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to store embedding: {e}")
|
|
return False
|
|
|
|
async def search(
|
|
self,
|
|
query: str,
|
|
collection: str = "scam_patterns",
|
|
limit: int = 10,
|
|
min_similarity: float = 0.6,
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Search for similar documents using cosine similarity.
|
|
Compares query embedding against stored doc vectors (uses common prefix).
|
|
"""
|
|
query_vec = await self.embed_query(query)
|
|
q_dim = len(query_vec)
|
|
|
|
try:
|
|
r = await self._get_redis()
|
|
doc_ids = await r.smembers(f"rag:idx:{collection}")
|
|
|
|
if not doc_ids:
|
|
return []
|
|
|
|
# Batch fetch all documents
|
|
keys = [f"rag:{collection}:{did}" for did in doc_ids]
|
|
pipe = r.pipeline()
|
|
for k in keys:
|
|
pipe.get(k)
|
|
results = await pipe.execute()
|
|
|
|
# Compute similarities — compare only common prefix dimensions
|
|
scored = []
|
|
for _i, data in enumerate(results):
|
|
if not data:
|
|
continue
|
|
try:
|
|
doc = json.loads(data)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
doc_vec = doc.get("vector", [])
|
|
# Handle JSON-string vectors (from hash re-embed)
|
|
if isinstance(doc_vec, str):
|
|
try:
|
|
doc_vec = json.loads(doc_vec)
|
|
except (json.JSONDecodeError, TypeError):
|
|
continue
|
|
if not doc_vec or not isinstance(doc_vec, list):
|
|
continue
|
|
|
|
# Compare only the semantic portion (query has semantic only, doc may be multi-head)
|
|
compare_len = min(q_dim, len(doc_vec))
|
|
sim = self.cosine_similarity(query_vec[:compare_len], doc_vec[:compare_len])
|
|
if sim >= min_similarity:
|
|
scored.append(
|
|
{
|
|
"id": doc["id"],
|
|
"similarity": round(sim, 4),
|
|
"metadata": doc.get("metadata", {}),
|
|
"content": doc.get("content", "")[:500],
|
|
}
|
|
)
|
|
|
|
scored.sort(key=lambda x: x["similarity"], reverse=True)
|
|
return scored[:limit]
|
|
|
|
except Exception as e:
|
|
logger.error(f"Search failed: {e}")
|
|
return []
|
|
|
|
@property
|
|
def stats(self) -> dict[str, Any]:
|
|
return {
|
|
"initialized": self._initialized,
|
|
"providers": {
|
|
"local_bge": self._local is not None,
|
|
"openrouter": self._openrouter is not None,
|
|
"huggingface": self._huggingface is not None,
|
|
},
|
|
"primary": "local/bge-small-en-v1.5 (384d, $0/mo)",
|
|
"cache": {
|
|
"hits": self._cache_hits,
|
|
"misses": self._cache_misses,
|
|
"ratio": round(self._cache_hits / max(1, self._cache_hits + self._cache_misses), 3),
|
|
},
|
|
"dimensions": DIMS,
|
|
"collections": COLLECTIONS,
|
|
}
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# SEED DATA: Known scam pattern templates
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
KNOWN_SCAM_PATTERNS = [
|
|
{
|
|
"name": "Honeypot — Sell Disabled",
|
|
"description": "Token where only the creator can sell. Buyers are trapped. Implemented via maxSellAmount=0, tradingEnabled=false, or blacklist of all non-owner addresses.",
|
|
"indicators": [
|
|
"maxSellAmount=0",
|
|
"tradingEnabled=false",
|
|
"onlyOwner can transfer",
|
|
"blacklist all addresses",
|
|
],
|
|
"severity": "critical",
|
|
"code_snippets": [
|
|
"maxSellAmount = 0",
|
|
"tradingEnabled = false",
|
|
"require(_isExcludedFromFees[msg.sender] || tradingEnabled)",
|
|
"function enableTrading() external onlyOwner",
|
|
],
|
|
},
|
|
{
|
|
"name": "Unlimited Mint / Rug Pull",
|
|
"description": "Owner can mint unlimited tokens, diluting holders to zero before draining liquidity.",
|
|
"indicators": [
|
|
"unrestricted mint()",
|
|
"onlyOwner mint",
|
|
"no supply cap",
|
|
"mint + removeLiquidity",
|
|
],
|
|
"severity": "critical",
|
|
"code_snippets": [
|
|
"function mint(address to, uint256 amount) external onlyOwner",
|
|
"mint(msg.sender, 1000000 * 10**decimals())",
|
|
],
|
|
},
|
|
{
|
|
"name": "Hidden Fee Manipulation",
|
|
"description": "Owner can set fees to 100% after launch, making sells impossible or stealing all transfers.",
|
|
"indicators": [
|
|
"setTax(100)",
|
|
"dynamic fee > 50%",
|
|
"setFee external onlyOwner",
|
|
"no max fee cap",
|
|
],
|
|
"severity": "high",
|
|
"code_snippets": [
|
|
"function setFee(uint256 _fee) external onlyOwner",
|
|
"taxFee = 99",
|
|
"liquidityFee = 99",
|
|
],
|
|
},
|
|
{
|
|
"name": "Liquidity Drain Backdoor",
|
|
"description": "Hidden function allowing owner to remove all liquidity. Often disguised as 'manualSwap' or 'rescueToken'.",
|
|
"indicators": [
|
|
"manualSwap",
|
|
"removeLiquidity",
|
|
"drain",
|
|
"sweep",
|
|
"rescueToken with onlyOwner",
|
|
],
|
|
"severity": "critical",
|
|
"code_snippets": [
|
|
"function manualSwap() external onlyOwner",
|
|
"function removeAllLiquidity() external onlyOwner",
|
|
"uniswapV2Router.removeLiquidity",
|
|
],
|
|
},
|
|
{
|
|
"name": "Proxy Upgrade Trap",
|
|
"description": "Upgradeable proxy where admin can change implementation to a malicious contract post-launch.",
|
|
"indicators": [
|
|
"upgradeTo",
|
|
"upgradeToAndCall",
|
|
"UUPS",
|
|
"TransparentUpgradeableProxy",
|
|
"changeAdmin",
|
|
],
|
|
"severity": "high",
|
|
"code_snippets": [
|
|
"function upgradeTo(address newImplementation) external onlyOwner",
|
|
"function upgradeToAndCall(address newImplementation, bytes memory data) external onlyOwner",
|
|
],
|
|
},
|
|
{
|
|
"name": "Bundle/Sniper Launch",
|
|
"description": "Creator uses bundling to acquire large supply at launch before others can buy, then dumps.",
|
|
"indicators": [
|
|
"multi-wallet buy at block 0",
|
|
"same funder",
|
|
"bundle detection",
|
|
"60%+ supply in 3 wallets",
|
|
],
|
|
"severity": "high",
|
|
"code_snippets": [],
|
|
},
|
|
{
|
|
"name": "Honeypot — Max Tx Limit",
|
|
"description": "Token with maxTxAmount set so low that only tiny amounts can be sold, trapping larger holders.",
|
|
"indicators": [
|
|
"maxTxAmount < 0.1% supply",
|
|
"maxSellAmount micro",
|
|
"anti-whale as antibot disguise",
|
|
],
|
|
"severity": "high",
|
|
"code_snippets": [
|
|
"maxTxAmount = 1000",
|
|
"maxSellAmount = 500",
|
|
"maxWalletAmount = 2000",
|
|
],
|
|
},
|
|
{
|
|
"name": "Fake Renounce / Ownership Hide",
|
|
"description": "Claims ownership renounced but keeps control via secondary address, proxy, or timelock trick.",
|
|
"indicators": [
|
|
"fake renounce",
|
|
"ownership not actually 0xdead",
|
|
"hidden owner",
|
|
"renounceOwnership not called",
|
|
],
|
|
"severity": "high",
|
|
"code_snippets": [],
|
|
},
|
|
{
|
|
"name": "Sandwich Bot Tax Farm",
|
|
"description": "Token with high taxes that benefits from MEV sandwich bots. Creator runs bots to extract from traders.",
|
|
"indicators": ["tax > 10%", "sandwich pattern", "same wallet MEV + deployer"],
|
|
"severity": "medium",
|
|
"code_snippets": [],
|
|
},
|
|
{
|
|
"name": "Wallet Drainer / Approval Scam",
|
|
"description": "Contract tricks users into approving token spending, then drains their wallets via transferFrom.",
|
|
"indicators": [
|
|
"unlimited approval request",
|
|
"transferFrom in unverified contract",
|
|
"approveAndCall",
|
|
"permit exploit",
|
|
],
|
|
"severity": "critical",
|
|
"code_snippets": [
|
|
"IERC20(token).transferFrom(victim, attacker, balance)",
|
|
"function claim(address token) external",
|
|
"permit(",
|
|
],
|
|
},
|
|
]
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
# SINGLETON
|
|
# ══════════════════════════════════════════════════════════════════════
|
|
|
|
_embedder: CryptoEmbedder | None = None
|
|
|
|
|
|
async def get_embedder() -> CryptoEmbedder:
|
|
global _embedder
|
|
if _embedder is None:
|
|
_embedder = CryptoEmbedder()
|
|
await _embedder.initialize()
|
|
return _embedder
|