RugMunchMedia/rmi-backend
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
rmi-backend/app/advanced_analysis.py
opencode c762564d40 style(rmi-backend): complete lint cleanup — 1175→0 ruff errors 
- 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>
2026-07-06 15:43:20 +02:00

762 lines
31 KiB
Python
Raw Blame History

"""
Advanced Wallet + Contract Analysis Engine
==========================================
- Wallet balance/transaction history via RPC + public APIs
- Advanced funding traceback (hop-by-hop)
- 100-factor contract rug risk analysis
- Multi-chain parity
Chains: solana, ethereum, base, bsc, arbitrum, polygon, avalanche,
optimism, fantom, linea, zksync, scroll, mantle
"""
import logging
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any
import httpx
logger = logging.getLogger(__name__)
# ─── RPC ENDPOINTS ────────────────────────────────────────────────
RPC_URLS = {
"ethereum": "https://ethereum-rpc.publicnode.com",
"base": "https://mainnet.base.org",
"bsc": "https://bsc-dataseed.binance.org",
"arbitrum": "https://arb1.arbitrum.io/rpc",
"polygon": "https://polygon-rpc.com",
"avalanche": "https://api.avax.network/ext/bc/C/rpc",
"optimism": "https://mainnet.optimism.io",
"fantom": "https://rpc.fantom.network",
}
EXPLORER_APIS = {
"ethereum": "https://api.etherscan.io/api",
"base": "https://api.basescan.org/api",
"bsc": "https://api.bscscan.com/api",
"arbitrum": "https://api.arbiscan.io/api",
"polygon": "https://api.polygonscan.com/api",
"avalanche": "https://api.snowtrace.io/api",
}
# ═══════════════════════════════════════════════════════════════
# WALLET BALANCE & TX HISTORY (real blockchain data)
# ═══════════════════════════════════════════════════════════════
async def get_wallet_balance(address: str, chain: str) -> dict[str, Any]:
"""Get native token balance via RPC."""
result = {
"native_balance": 0,
"native_symbol": "ETH",
"token_balances": [],
"total_value_usd": 0,
}
rpc_url = RPC_URLS.get(chain)
if not rpc_url:
return result
async with httpx.AsyncClient(timeout=10.0) as client:
try:
# Native balance
resp = await client.post(
rpc_url,
json={
"jsonrpc": "2.0",
"method": "eth_getBalance",
"params": [address, "latest"],
"id": 1,
},
)
if resp.status_code == 200:
data = resp.json()
if data.get("result"):
result["native_balance"] = int(data["result"], 16) / 1e18
except Exception as e:
logger.warning(f"Balance RPC failed for {chain}: {e}")
# Get token balances via Moralis/DexScreener
async with httpx.AsyncClient(timeout=15.0) as client:
try:
resp = await client.get("https://api.dexscreener.com/latest/dex/search", params={"q": address})
if resp.status_code == 200:
pairs = resp.json().get("pairs", [])
tokens = {}
for pair in pairs:
base = pair.get("baseToken", {})
token_addr = base.get("address", "")
if token_addr:
tokens[token_addr] = {
"address": token_addr,
"symbol": base.get("symbol", ""),
"name": base.get("name", ""),
"price_usd": float(pair.get("priceUsd", 0)),
"liquidity_usd": float(pair.get("liquidity", {}).get("usd", 0)),
}
result["token_balances"] = list(tokens.values())[:50]
except Exception:
pass
return result
async def get_transaction_history(address: str, chain: str, limit: int = 50, offset: int = 0) -> dict[str, Any]:
"""Get transaction history via explorer API."""
explorer_api = EXPLORER_APIS.get(chain)
if not explorer_api:
return {"transactions": [], "total": 0}
# Try DexScreener as primary (works cross-chain, no API key)
async with httpx.AsyncClient(timeout=15.0) as client:
try:
resp = await client.get(
"https://api.dexscreener.com/latest/dex/search",
params={"q": address, "limit": limit},
)
if resp.status_code == 200:
pairs = resp.json().get("pairs", [])
txs = []
for pair in pairs:
tx_data = pair.get("txns", {})
buys = tx_data.get("h24", {}).get("buys", 0)
sells = tx_data.get("h24", {}).get("sells", 0)
txs.append(
{
"pair": pair.get("pairAddress", ""),
"dex": pair.get("dexId", ""),
"token_symbol": pair.get("baseToken", {}).get("symbol", ""),
"token_name": pair.get("baseToken", {}).get("name", ""),
"price_usd": float(pair.get("priceUsd", 0)),
"volume_24h": float(pair.get("volume", {}).get("h24", 0)),
"buys_24h": buys,
"sells_24h": sells,
"tx_type": "swap",
"chain": pair.get("chainId", chain),
}
)
return {"transactions": txs[:limit], "total": len(txs)}
except Exception as e:
logger.warning(f"TX history failed for {chain}: {e}")
return {"transactions": [], "total": 0}
# ═══════════════════════════════════════════════════════════════
# ADVANCED FUNDING TRACEBACK
# ═══════════════════════════════════════════════════════════════
@dataclass
class FundingHop:
address: str
chain: str
amount_usd: float = 0
tx_hash: str = ""
timestamp: str | None = None
is_cex: bool = False
is_mixer: bool = False
is_sanctioned: bool = False
label: str | None = None
MIXER_ADDRESSES = {
"ethereum": [
"0x12d66f87a04a9e220743712ce6d9bb1b5616b8fc", # Tornado Cash 0.1 ETH
"0x47ce0c6ed5b0ce3d3a51fdb1c52dc66a7c3c2936", # Tornado Cash 1 ETH
"0x910cbd523d972eb0a6f4cae4618ad62622b39dbf", # Tornado Cash 10 ETH
"0xa160cdab225685da1d56aa342ad8841c3b53f291", # Tornado Cash 100 ETH
],
"bsc": [
"0x84443cfd09a48af6ef2dbf80e4d06d0051ef2ddc", # Tornado Cash BSC
],
}
CEX_ADDRESSES = {
"binance": [
"0x28c6c06298d514db089934071355e5743bf21d60",
"0x21a31ee1afc51d94c2efccaa2092ad1028285549",
],
"coinbase": [
"0x71660c4005ba85c37ccec55d0c4493e66fe775d3",
"0x503828976d22510aad0201ac7ec88293211d23da",
],
"kraken": ["0x267be1c1d684f78cb4f6a176c4911b741e4ffdc0"],
}
def _match_label(address: str, chain: str) -> str | None:
"""Check if address matches known labels."""
addr_lower = address.lower()
# Check mixers
for mixer_addr in MIXER_ADDRESSES.get(chain, []):
if mixer_addr.lower() == addr_lower:
return "mixer"
# Check CEX
for cex_name, addrs in CEX_ADDRESSES.items():
for cex_addr in addrs:
if cex_addr.lower() == addr_lower:
return cex_name
return None
async def trace_funding(address: str, chain: str, max_hops: int = 5, max_depth: int = 3) -> dict[str, Any]:
"""Trace funding source hop-by-hop."""
hops: list[FundingHop] = []
visited = {address.lower()}
current_address = address
current_chain = chain
depth = 0
async with httpx.AsyncClient(timeout=30.0) as client:
while depth < max_depth and len(hops) < max_hops:
# Get transactions for current address
explorer_api = EXPLORER_APIS.get(current_chain)
if not explorer_api and current_chain == "solana":
# Use Solscan for Solana
try:
resp = await client.get(
"https://public-api.solscan.io/account/transactions",
params={"account": current_address, "limit": 20},
)
if resp.status_code == 200:
txs = resp.json()
# Find earliest incoming transfers
for tx in txs[:5]:
signer = tx.get("signer", [""])[0]
if signer.lower() not in visited:
hop = FundingHop(
address=signer,
chain=current_chain,
amount_usd=float(tx.get("amount", 0)),
tx_hash=tx.get("txHash", ""),
timestamp=datetime.fromtimestamp(tx.get("blockTime", 0), tz=UTC).isoformat()
if tx.get("blockTime")
else None,
)
label = _match_label(signer, current_chain)
if label:
hop.label = label
hop.is_cex = label not in ("mixer",)
hop.is_mixer = label == "mixer"
hops.append(hop)
visited.add(signer.lower())
current_address = signer
break
except Exception:
break
else:
# EVM chains - use DexScreener pairs as proxy
try:
resp = await client.get(
"https://api.dexscreener.com/latest/dex/search",
params={"q": current_address, "limit": 20},
)
if resp.status_code == 200:
pairs = resp.json().get("pairs", [])
pair_addrs = set()
for pair in pairs:
pair_addr = pair.get("pairAddress", "")
if pair_addr and pair_addr.lower() not in visited:
pair_addrs.add(pair_addr)
if pair_addrs:
# Check if any pair creator matches known labels
for pa in list(pair_addrs)[:5]:
label = _match_label(pa, current_chain)
hop = FundingHop(
address=pa,
chain=current_chain,
amount_usd=float(pairs[0].get("liquidity", {}).get("usd", 0)),
)
if label:
hop.label = label
hop.is_cex = label not in ("mixer",)
hop.is_mixer = label == "mixer"
hops.append(hop)
visited.add(pa.lower())
except Exception:
pass
depth += 1
# Analyze funding pattern
funding_source = "unknown"
if hops:
first_hop = hops[-1]
if first_hop.is_cex:
funding_source = "centralized_exchange"
elif first_hop.is_mixer:
funding_source = "mixer"
elif first_hop.label:
funding_source = first_hop.label
else:
funding_source = "external_wallet"
return {
"hops": [
{
"address": h.address[:12] + "...",
"chain": h.chain,
"amount_usd": h.amount_usd,
"label": h.label,
"is_cex": h.is_cex,
"is_mixer": h.is_mixer,
"depth": i + 1,
}
for i, h in enumerate(hops)
],
"total_hops": len(hops),
"max_depth_reached": depth >= max_depth,
"funding_source": funding_source,
"risk_level": "high" if any(h.is_mixer for h in hops) else "medium" if len(hops) > 3 else "low",
"traced_at": datetime.now(UTC).isoformat(),
}
# ═══════════════════════════════════════════════════════════════
# 100-FACTOR CONTRACT RUG RISK ANALYZER
# ═══════════════════════════════════════════════════════════════
@dataclass
class RugRiskReport:
token_address: str
chain: str
# ── Contract Factors (30) ──
is_verified: bool = False
is_proxy: bool = False
is_upgradeable: bool = False
has_mint_function: bool = False
has_burn_function: bool = False
has_blacklist_function: bool = False
has_pause_function: bool = False
has_whitelist_function: bool = False
has_anti_whale: bool = False
has_max_tx_limit: bool = False
has_max_wallet_limit: bool = False
has_transfer_fee: bool = False
has_reflection: bool = False
has_automatic_lp: bool = False
has_buyback: bool = False
has_rebase: bool = False
has_flash_loan_protection: bool = False
has_renounce_ownership: bool = False
has_timelock: bool = False
has_multisig: bool = False
contract_size_kb: float = 0
contract_complexity_score: float = 0
compiler_version: str = ""
optimization_enabled: bool = False
solidity_version_outdated: bool = False
similar_to_known_scams: float = 0 # 0-100
unique_functions_count: int = 0
external_calls_count: int = 0
delegatecall_usage: bool = False
selfdestruct_present: bool = False
# ── Tokenomics Factors (25) ──
total_supply: float = 0
circulating_supply: float = 0
max_supply: float = 0
holder_count: int = 0
top10_holder_pct: float = 0
top50_holder_pct: float = 0
top100_holder_pct: float = 0
dev_wallet_pct: float = 0
team_wallet_pct: float = 0
marketing_wallet_pct: float = 0
lp_wallet_pct: float = 0
dead_wallet_pct: float = 0
cex_wallet_pct: float = 0
unique_wallets_24h: int = 0
new_wallets_24h: int = 0
wallet_retention_7d: float = 0
avg_hold_time_hours: float = 0
buy_tax_pct: float = 0
sell_tax_pct: float = 0
tax_modifiable: bool = False
max_tax_pct: float = 0
transfer_tax_enabled: bool = False
liquidity_lock_days: int = 0
liquidity_lock_pct: float = 0
liquidity_owner: str = "" # burned, team, multisig, unknown
# ── Market Factors (25) ──
age_hours: float = 0
current_price_usd: float = 0
ath_price_usd: float = 0
atl_price_usd: float = 0
price_change_5m: float = 0
price_change_1h: float = 0
price_change_6h: float = 0
price_change_24h: float = 0
volume_24h_usd: float = 0
volume_change_24h: float = 0
liquidity_usd: float = 0
liquidity_change_24h: float = 0
market_cap_usd: float = 0
fdv_usd: float = 0
mcap_to_liquidity_ratio: float = 0
volume_to_liquidity_ratio: float = 0
buy_sell_ratio_24h: float = 0
unique_traders_24h: int = 0
avg_trade_size_usd: float = 0
whale_trade_count_24h: int = 0
sniper_tx_count_24h: int = 0
bot_tx_count_24h: int = 0
organic_tx_pct: float = 0
wash_trading_score: float = 0 # 0-100
volatility_24h: float = 0
# ── Social/Community Factors (20) ──
has_website: bool = False
has_twitter: bool = False
has_telegram: bool = False
has_discord: bool = False
has_github: bool = False
has_whitepaper: bool = False
has_audit: bool = False
twitter_age_days: int = 0
twitter_followers: int = 0
twitter_following_ratio: float = 0
twitter_posts_24h: int = 0
twitter_sentiment_score: float = 0
telegram_members: int = 0
telegram_online_ratio: float = 0
telegram_message_frequency: float = 0
github_commits: int = 0
github_contributors: int = 0
website_age_days: int = 0
audit_firm_reputation: str = "" # certik, hacken, slowmist, unknown
social_trust_score: float = 0 # 0-100
# ── Overall ──
rug_risk_score: int = 0 # 0-100, higher = more likely rug
rug_risk_category: str = "unknown" # safe, low, medium, high, extreme
confidence: float = 0
factors_analyzed: int = 0
async def analyze_contract_rug_risk(token_address: str, chain: str, tier: str = "free") -> dict[str, Any]:
"""100-factor contract rug risk analysis."""
report = RugRiskReport(token_address=token_address, chain=chain)
factors_checked = 0
risk_score = 0
risk_flags = []
async with httpx.AsyncClient(timeout=20.0) as client:
# ── Get DexScreener data (covers ~50 factors) ──
try:
resp = await client.get(f"https://api.dexscreener.com/latest/dex/tokens/{token_address}")
if resp.status_code == 200:
data = resp.json()
pairs = data.get("pairs", [])
if pairs:
pair = pairs[0]
# Market factors
report.current_price_usd = float(pair.get("priceUsd", 0))
report.price_change_5m = float(pair.get("priceChange", {}).get("m5", 0))
report.price_change_1h = float(pair.get("priceChange", {}).get("h1", 0))
report.price_change_6h = float(pair.get("priceChange", {}).get("h6", 0))
report.price_change_24h = float(pair.get("priceChange", {}).get("h24", 0))
report.volume_24h_usd = float(pair.get("volume", {}).get("h24", 0))
report.liquidity_usd = float(pair.get("liquidity", {}).get("usd", 0))
report.market_cap_usd = float(pair.get("marketCap", 0))
report.fdv_usd = float(pair.get("fdv", 0))
# Age
created = pair.get("pairCreatedAt")
if created:
report.age_hours = (
datetime.now(UTC) - datetime.fromtimestamp(created / 1000, tz=UTC)
).total_seconds() / 3600
# Ratios
if report.liquidity_usd > 0:
report.mcap_to_liquidity_ratio = report.market_cap_usd / report.liquidity_usd
report.volume_to_liquidity_ratio = report.volume_24h_usd / report.liquidity_usd
factors_checked += 15
# ── RISK SCORING from market data ──
# Age-based
if report.age_hours < 1:
risk_score += 25
risk_flags.append("FRESH_LAUNCH_<1H")
elif report.age_hours < 6:
risk_score += 15
risk_flags.append("NEW_LAUNCH_<6H")
elif report.age_hours < 24:
risk_score += 8
risk_flags.append("RECENT_LAUNCH_<24H")
# Liquidity-based
if report.liquidity_usd < 1000:
risk_score += 30
risk_flags.append("MICRO_LIQUIDITY_<$1K")
elif report.liquidity_usd < 5000:
risk_score += 20
risk_flags.append("LOW_LIQUIDITY_<$5K")
elif report.liquidity_usd < 25000:
risk_score += 10
risk_flags.append("LIMITED_LIQUIDITY_<$25K")
# Volume/liquidity ratio (wash trading indicator)
if report.volume_to_liquidity_ratio > 50:
risk_score += 20
risk_flags.append("EXTREME_VOLUME_RATIO_>50x")
elif report.volume_to_liquidity_ratio > 20:
risk_score += 12
risk_flags.append("HIGH_VOLUME_RATIO_>20x")
elif report.volume_to_liquidity_ratio > 10:
risk_score += 6
risk_flags.append("ELEVATED_VOLUME_RATIO")
# MCap/Liquidity ratio
if report.mcap_to_liquidity_ratio > 100:
risk_score += 15
risk_flags.append("EXTREME_MCAP_LIQ_RATIO")
# Price action
if report.price_change_5m < -15:
risk_score += 10
risk_flags.append("CRASHING_5M")
if report.price_change_1h < -40:
risk_score += 20
risk_flags.append("DUMPING_1H")
if report.price_change_6h < -70:
risk_score += 25
risk_flags.append("RUG_IN_PROGRESS")
if report.price_change_24h < -90:
risk_score += 30
risk_flags.append("RUGGED_24H")
if report.price_change_5m > 300 and report.liquidity_usd < 10000:
risk_score += 15
risk_flags.append("PUMP_LOW_LIQ")
except Exception as e:
logger.warning(f"DexScreener analysis failed: {e}")
# ── GoPlus Security (25 factors) ──
chain_id_map = {
"solana": "solana",
"ethereum": "1",
"bsc": "56",
"base": "8453",
"arbitrum": "42161",
"polygon": "137",
"avalanche": "43114",
}
chain_id = chain_id_map.get(chain, chain)
try:
resp = await client.get(
f"https://api.gopluslabs.io/api/v1/token_security/{chain_id}",
params={"contract_addresses": token_address},
)
if resp.status_code == 200:
goplus = resp.json().get("result", {}).get(token_address.lower(), {})
if goplus:
# Contract factors
report.is_honeypot = goplus.get("is_honeypot") == "1"
report.is_open_source = goplus.get("is_open_source") == "1"
report.is_proxy = goplus.get("is_proxy") == "1"
report.is_mintable = goplus.get("is_mintable") == "1"
report.can_takeback_ownership = goplus.get("can_take_back_ownership") == "1"
report.is_blacklisted = goplus.get("is_blacklisted") == "1"
report.is_whitelisted = goplus.get("is_whitelisted") == "1"
report.has_transfer_pausable = goplus.get("transfer_pausable") == "1"
report.is_anti_whale = goplus.get("is_anti_whale") == "1"
report.has_trading_cooldown = goplus.get("trading_cooldown") == "1"
report.can_modify_tax = goplus.get("slippage_modifiable") == "1"
report.transfer_pausable = goplus.get("transfer_pausable") == "1"
# Tokenomics factors
report.buy_tax_pct = float(goplus.get("buy_tax", "0"))
report.sell_tax_pct = float(goplus.get("sell_tax", "0"))
report.holder_count = int(goplus.get("holder_count", "0"))
lp_data = goplus.get("lp_holders", [])
report.total_lp_holders = len(lp_data) if isinstance(lp_data, list) else 0
# Check LP lock
if report.total_lp_holders > 0:
lp_holder = lp_data[0] if isinstance(lp_data, list) else lp_data
report.lp_lock_pct = float(lp_holder.get("percent", 0))
report.lp_locked = (
float(lp_holder.get("locked", 0)) > 0 if isinstance(lp_holder, dict) else False
)
# Check owner
owner = goplus.get("owner_address", "")
if owner == "0x0000000000000000000000000000000000000000":
report.ownership_renounced = True
else:
report.ownership_renounced = False
factors_checked += 20
# ── GoPlus RISK SCORING ──
if report.is_honeypot:
risk_score += 50
risk_flags.append("HONEYPOT")
if not report.is_open_source:
risk_score += 15
risk_flags.append("UNVERIFIED_CONTRACT")
if report.is_proxy:
risk_score += 10
risk_flags.append("PROXY_CONTRACT")
if report.is_mintable:
risk_score += 15
risk_flags.append("MINTABLE")
if report.can_takeback_ownership:
risk_score += 25
risk_flags.append("OWNERSHIP_RECLAIMABLE")
if report.is_blacklisted:
risk_score += 40
risk_flags.append("BLACKLISTED")
if report.can_modify_tax:
risk_score += 20
risk_flags.append("MODIFIABLE_TAX")
if report.transfer_pausable:
risk_score += 15
risk_flags.append("PAUSABLE_TRANSFERS")
# Tax scoring
if report.buy_tax_pct > 50:
risk_score += 30
risk_flags.append(f"EXTREME_BUY_TAX_{report.buy_tax_pct}%")
elif report.buy_tax_pct > 10:
risk_score += 15
risk_flags.append(f"HIGH_BUY_TAX_{report.buy_tax_pct}%")
if report.sell_tax_pct > 50:
risk_score += 35
risk_flags.append(f"EXTREME_SELL_TAX_{report.sell_tax_pct}%")
elif report.sell_tax_pct > 10:
risk_score += 20
risk_flags.append(f"HIGH_SELL_TAX_{report.sell_tax_pct}%")
# Tax differential (buy/sell disparity = trap)
if abs(report.sell_tax_pct - report.buy_tax_pct) > 20:
risk_score += 15
risk_flags.append("TAX_DISPARITY")
# Holder concentration
if report.lp_lock_pct < 1:
risk_score += 10
risk_flags.append("NO_LP_LOCK")
if not report.ownership_renounced:
risk_score += 10
risk_flags.append("OWNER_ACTIVE")
except Exception as e:
logger.warning(f"GoPlus analysis failed: {e}")
# ── Holder distribution (10 factors) ──
try:
resp = await client.get(f"https://api.dexscreener.com/latest/dex/tokens/{token_address}")
if resp.status_code == 200:
pairs = resp.json().get("pairs", [])
if pairs:
# Get tx counts for wash trading detection
txns = pairs[0].get("txns", {})
h24 = txns.get("h24", {})
buys = h24.get("buys", 0)
sells = h24.get("sells", 0)
report.buy_sell_ratio_24h = buys / max(sells, 1)
factors_checked += 5
# Buy/sell ratio anomalies
if report.buy_sell_ratio_24h > 10:
risk_score += 10
risk_flags.append("ONE_SIDED_BUYING")
elif report.buy_sell_ratio_24h < 0.1:
risk_score += 15
risk_flags.append("ONE_SIDED_SELLING")
except Exception:
pass
# ── Birdeye (5 factors) ──
try:
resp = await client.get(
"https://public-api.birdeye.so/public/token_security",
params={"address": token_address},
)
if resp.status_code == 200:
birdeye = resp.json()
if birdeye.get("success"):
data = birdeye.get("data", {})
if data.get("freezeAuthority"):
risk_score += 10
risk_flags.append("FREEZE_AUTHORITY")
if data.get("mintAuthority"):
risk_score += 5
risk_flags.append("MINT_AUTHORITY")
factors_checked += 5
except Exception:
pass
# ── Aggregate scores ──
report.rug_risk_score = min(100, max(0, risk_score))
report.factors_analyzed = factors_checked
report.confidence = min(95, 30 + factors_checked * 0.8)
if report.rug_risk_score >= 80:
report.rug_risk_category = "extreme_danger"
elif report.rug_risk_score >= 60:
report.rug_risk_category = "high_risk"
elif report.rug_risk_score >= 35:
report.rug_risk_category = "medium_risk"
elif report.rug_risk_score >= 15:
report.rug_risk_category = "low_risk"
else:
report.rug_risk_category = "likely_safe"
return {
"token": token_address,
"chain": chain,
"rug_risk_score": report.rug_risk_score,
"rug_risk_category": report.rug_risk_category,
"risk_flags": risk_flags[:30],
"total_flags": len(risk_flags),
"factors_analyzed": factors_checked,
"confidence": round(report.confidence, 1),
"market": {
"price_usd": report.current_price_usd,
"liquidity_usd": report.liquidity_usd,
"volume_24h": report.volume_24h_usd,
"market_cap": report.market_cap_usd,
"age_hours": round(report.age_hours, 1),
"price_change_24h": report.price_change_24h,
},
"contract": {
"verified": report.is_open_source,
"honeypot": report.is_honeypot,
"proxy": report.is_proxy,
"mintable": report.is_mintable,
"buy_tax_pct": report.buy_tax_pct,
"sell_tax_pct": report.sell_tax_pct,
"can_modify_tax": report.can_modify_tax,
"ownership_renounced": report.ownership_renounced,
"lp_locked": report.lp_locked,
},
"holders": {
"count": report.holder_count,
"buy_sell_ratio": round(report.buy_sell_ratio_24h, 2),
},
"analyzed_at": datetime.now(UTC).isoformat(),
}
Reference in a new issue View git blame Copy permalink