RugMunchMedia/rmi-backend
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
rmi-backend/app/defi_protocol_auditor.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

844 lines
33 KiB
Python
Raw Permalink Blame History

"""
DeFi Protocol Safety Auditor
=============================
Aggregates TVL, social sentiment, contract audit status, deployer reputation,
and on-chain risk indicators to produce a comprehensive safety score for any
DeFi protocol. Uses DeFiLlama for TVL/protocol data, social intel for sentiment,
and on-chain analysis for risk signals.
Output: SafetyReport with per-category scores, risk flags, and an AI-generated
executive summary powered by DeepSeek/MiniMax.
Competitive differentiator:
- Nansen tracks wallet labels but doesn't rate protocol safety
- DeBank shows portfolio but not risk assessment
- CertiK audits cost $$$ and cover only specific contracts
- We combine ALL signals into one free-to-check score
Usage:
from app.defi_protocol_auditor import DefiProtocolAuditor
auditor = DefiProtocolAuditor()
report = await auditor.audit("aave")
print(report.summary())
"""
import asyncio
import logging
import os
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
from typing import Any
logger = logging.getLogger(__name__)
# ═══════════════════════════════════════════════════════════════
# Enums & Types
# ═══════════════════════════════════════════════════════════════
class AuditStatus(Enum):
VERIFIED = "verified" # Audited by reputable firm, no critical issues
AUDITED = "audited" # Audited, some minor issues
PARTIAL = "partial" # Partially audited (some contracts only)
UNKNOWN = "unknown" # No audit information found
NO_AUDIT = "no_audit" # Confirmed no audit
SUSPICIOUS = "suspicious" # Audited by unknown/untrustworthy firm
@property
def score(self) -> float:
return {
"verified": 1.0,
"audited": 0.8,
"partial": 0.5,
"unknown": 0.3,
"no_audit": 0.1,
"suspicious": 0.0,
}[self.value]
class LiquidityStatus(Enum):
LOCKED_LONG = "locked_long" # Locked for 1+ years
LOCKED = "locked" # Locked for 3-12 months
LOCKED_SHORT = "locked_short" # Locked for < 3 months
UNKNOWN = "unknown"
UNLOCKED = "unlocked" # No lock detected
REMOVED = "removed" # Liquidity removed (exit scam)
@property
def score(self) -> float:
return {
"locked_long": 1.0,
"locked": 0.8,
"locked_short": 0.5,
"unknown": 0.3,
"unlocked": 0.1,
"removed": 0.0,
}[self.value]
class RiskLevel(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
UNKNOWN = "unknown"
@dataclass
class SafetyCategory:
"""Score for one category of the protocol assessment."""
name: str
score: float # 0.0 (worst) to 1.0 (best)
weight: float # Contribution weight to overall score
details: str # Human-readable explanation
flags: list[str] = field(default_factory=list)
@property
def weighted(self) -> float:
return self.score * self.weight
@dataclass
class SafetyReport:
"""Complete protocol safety assessment."""
protocol_name: str
slug: str
chain: str
timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
# Category scores
audit_score: SafetyCategory = field(default_factory=lambda: SafetyCategory("Audit", 0.0, 0.30, ""))
tvl_score: SafetyCategory = field(default_factory=lambda: SafetyCategory("TVL & Activity", 0.0, 0.20, ""))
social_score: SafetyCategory = field(default_factory=lambda: SafetyCategory("Social Sentiment", 0.0, 0.15, ""))
deployer_score: SafetyCategory = field(default_factory=lambda: SafetyCategory("Deployer Reputation", 0.0, 0.20, ""))
liquidity_score: SafetyCategory = field(default_factory=lambda: SafetyCategory("Liquidity Status", 0.0, 0.15, ""))
# Global flags
red_flags: list[str] = field(default_factory=list)
warnings: list[str] = field(default_factory=list)
positives: list[str] = field(default_factory=list)
# AI summary
ai_summary: str = ""
def overall_score(self) -> float:
"""Weighted average of all category scores."""
total_weight = sum(c.weight for c in self.categories())
if total_weight == 0:
return 0.0
return sum(c.weighted for c in self.categories()) / total_weight
def risk_level(self) -> RiskLevel:
score = self.overall_score()
if score >= 0.8:
return RiskLevel.LOW
elif score >= 0.6:
return RiskLevel.MEDIUM
elif score >= 0.3:
return RiskLevel.HIGH
return RiskLevel.CRITICAL
def categories(self) -> list[SafetyCategory]:
return [
self.audit_score,
self.tvl_score,
self.social_score,
self.deployer_score,
self.liquidity_score,
]
def summary(self, detailed: bool = False) -> str:
"""Quick human-readable summary."""
emoji = {"low": "✅", "medium": "⚠️", "high": "🔴", "critical": "🚨", "unknown": "❓"}
level = self.risk_level()
icon = emoji.get(level.value, "❓")
lines = [
f"{icon} **{self.protocol_name}** (on {self.chain})",
f" Overall Safety: {self.overall_score():.0%} - **{level.value.upper()}** risk",
f" Red Flags: {len(self.red_flags)} | Warnings: {len(self.warnings)} | Positives: {len(self.positives)}",
"",
]
if detailed:
for cat in self.categories():
bar = "▓" * int(cat.score * 10) + "░" * (10 - int(cat.score * 10))
lines.append(f" {cat.name:20s} [{bar}] {cat.score:.0%}")
if cat.details:
lines.append(f" → {cat.details}")
for flag in cat.flags:
lines.append(f" ⚠ {flag}")
lines.append("")
if self.red_flags:
lines.append("🚨 RED FLAGS:")
for f in self.red_flags:
lines.append(f" • {f}")
if self.warnings:
lines.append("⚠️ WARNINGS:")
for w in self.warnings:
lines.append(f" • {w}")
if self.positives:
lines.append("✅ POSITIVES:")
for p in self.positives:
lines.append(f" • {p}")
if self.ai_summary:
lines.append("")
lines.append("🤖 AI Assessment:")
lines.append(f" {self.ai_summary}")
return "\n".join(lines)
def to_dict(self) -> dict:
return {
"protocol": self.protocol_name,
"chain": self.chain,
"timestamp": self.timestamp,
"overall_score": round(self.overall_score(), 4),
"risk_level": self.risk_level().value,
"categories": {
c.name.lower().replace(" ", "_"): {
"score": c.score,
"weight": c.weight,
"details": c.details,
"flags": c.flags,
}
for c in self.categories()
},
"red_flags": self.red_flags,
"warnings": self.warnings,
"positives": self.positives,
"ai_summary": self.ai_summary,
}
# ═══════════════════════════════════════════════════════════════
# Known Audit Firms (reputable)
# ═══════════════════════════════════════════════════════════════
REPUTABLE_AUDITORS = {
"trailofbits",
"trail of bits",
"consensys",
"diligence",
"openzeppelin",
"certik",
"slowmist",
"peckshield",
"quantstamp",
"hacken",
"halborn",
"veridise",
"immunefi",
"code4rena",
"audithero",
"sherlock",
"cyfrin",
"codehawks",
"salus",
"chainsecurity",
}
# ═══════════════════════════════════════════════════════════════
# DefiLlama integration
# ═══════════════════════════════════════════════════════════════
DEFILLAMA_API = "https://api.llama.fi"
DEFILLAMA_API_TIMEOUT = 10
async def _fetch_json(url: str, params: dict[str, Any] | None = None) -> Any:
"""Simple HTTP fetch with timeout."""
import httpx
try:
async with httpx.AsyncClient(timeout=DEFILLAMA_API_TIMEOUT) as client:
resp = await client.get(url, params=params)
if resp.status_code == 200:
return resp.json()
logger.warning(f"HTTP {resp.status_code} from {url}")
return None
except Exception as e:
logger.warning(f"Fetch failed for {url}: {e}")
return None
async def _fetch_defillama_protocol(slug: str) -> dict[str, Any] | None:
"""Fetch protocol data from DeFiLlama."""
data = await _fetch_json(f"{DEFILLAMA_API}/protocol/{slug}")
if data is None:
data = await _fetch_json(f"{DEFILLAMA_API}/protocol/{slug.lower()}")
return data
async def _fetch_defillama_tvl(slug: str) -> list | None:
"""Fetch TVL history from DeFiLlama."""
return await _fetch_json(f"{DEFILLAMA_API}/protocol/{slug}")
async def _search_defillama(query: str) -> list[dict] | None:
"""Search DeFiLlama for a protocol."""
data = await _fetch_json(f"{DEFILLAMA_API}/search", {"q": query})
return data # Returns list of matches
# ═══════════════════════════════════════════════════════════════
# Social Sentiment (uses existing social intel)
# ═══════════════════════════════════════════════════════════════
async def _get_social_sentiment(protocol: str, chain: str) -> dict:
"""Check social sentiment for a protocol using available signals."""
signals = {
"mentions_24h": 0,
"positive_ratio": 0.5,
"trending": False,
"rug_mentions": 0,
"sources": ["coingecko", "twitter_intel"],
}
# CoinGecko trending check
api_key = os.getenv("COINGECKO_API_KEY", "")
try:
import httpx
async with httpx.AsyncClient(timeout=8) as client:
headers = {"x-cg-demo-api-key": api_key} if api_key else {}
resp = await client.get("https://api.coingecko.com/api/v3/search/trending", headers=headers)
if resp.status_code == 200:
trending = resp.json().get("coins", [])
for coin in trending:
item = coin.get("item", {})
name = (item.get("name", "") or "").lower()
symbol = (item.get("symbol", "") or "").lower()
if protocol.lower() in name or protocol.lower() in symbol:
signals["trending"] = True
signals["mentions_24h"] = item.get("market_cap_rank", 0) * 100
break
except Exception as e:
logger.debug(f"Coingecko trending check failed: {e}")
return signals
# ═══════════════════════════════════════════════════════════════
# AI Summary Generator (uses DeepSeek/MiniMax)
# ═══════════════════════════════════════════════════════════════
async def _generate_ai_summary(report: SafetyReport) -> str:
"""Generate an AI-powered executive summary of the safety report."""
try:
prompt = (
f"Protocol: {report.protocol_name} on {report.chain}\n"
f"Overall Safety Score: {report.overall_score():.0%}\n"
f"Risk Level: {report.risk_level().value.upper()}\n\n"
f"Category Scores:\n"
)
for cat in report.categories():
prompt += f" - {cat.name}: {cat.score:.0%} - {cat.details}\n"
if report.red_flags:
prompt += "\nRed Flags:\n" + "\n".join(f" ⛔ {f}" for f in report.red_flags)
if report.warnings:
prompt += "\nWarnings:\n" + "\n".join(f" ⚠ {w}" for w in report.warnings)
if report.positives:
prompt += "\nPositives:\n" + "\n".join(f" ✅ {p}" for p in report.positives)
prompt += (
"\n\nWrite a 2-3 sentence executive assessment of this DeFi protocol's safety. "
"Be direct, factual, and actionable. Include: (1) is it safe to interact with, "
"(2) what are the main risks, (3) recommended precautions."
)
# Try DeepSeek API first (configured locally), fall back to inline assessment
deepseek_key = os.getenv("DEEPSEEK_API_KEY", "")
if deepseek_key:
import httpx
body = {
"model": "deepseek-chat",
"messages": [
{
"role": "system",
"content": "You are a DeFi security analyst. Assess protocol safety concisely.",
},
{"role": "user", "content": prompt},
],
"max_tokens": 300,
"temperature": 0.3,
}
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.post(
"https://api.deepseek.com/v1/chat/completions",
headers={
"Authorization": f"Bearer {deepseek_key}",
"Content-Type": "application/json",
},
json=body,
)
if resp.status_code == 200:
data = resp.json()
return data["choices"][0]["message"]["content"].strip()
# Fallback: rule-based summary
score = report.overall_score()
if score >= 0.8:
return (
f"{report.protocol_name} appears to be a well-established protocol with "
f"strong safety signals. Audited by reputable firms, healthy TVL, and "
f"positive community sentiment. Standard security precautions recommended."
)
elif score >= 0.6:
return (
f"{report.protocol_name} has moderate safety indicators. While major risks "
f"are not evident, users should conduct their own research and verify "
f"the specific contracts they interact with. Monitor for changes."
)
elif score >= 0.3:
return (
f"⚠️ CAUTION: {report.protocol_name} shows significant risk indicators. "
f"Consider avoiding until outstanding audit and liquidity concerns are "
f"resolved. If you must interact, use minimal funds."
)
else:
return (
f"🚨 WARNING: {report.protocol_name} has CRITICAL risk factors. "
f"Multiple red flags detected. Strongly advise against depositing funds. "
f"This protocol may be an active scam or exit risk."
)
except Exception as e:
logger.warning(f"AI summary generation failed: {e}")
return "AI summary unavailable. Please review category scores manually."
# ═══════════════════════════════════════════════════════════════
# Main Auditor Class
# ═══════════════════════════════════════════════════════════════
class DefiProtocolAuditor:
"""Comprehensive DeFi protocol safety auditor."""
def __init__(self):
self._cache: dict[str, tuple[float, SafetyReport]] = {}
self._cache_ttl = 300 # 5 minutes
async def audit(self, protocol_slug: str, chain: str = "ethereum") -> SafetyReport:
"""Run full safety audit on a DeFi protocol.
Args:
protocol_slug: Protocol name/slug (e.g., "aave", "uniswap", "pancakeswap")
chain: Primary chain the protocol operates on
Returns:
SafetyReport with all category scores and AI summary
"""
cache_key = f"{protocol_slug}:{chain}"
now = time.time()
# Check cache
if cache_key in self._cache:
cached_time, cached_report = self._cache[cache_key]
if now - cached_time < self._cache_ttl:
return cached_report
report = SafetyReport(
protocol_name=protocol_slug.capitalize(),
slug=protocol_slug,
chain=chain,
)
# Phase 1: Fetch data from all sources concurrently
defillama_data, tvl_data, social_data = await asyncio.gather(
_fetch_defillama_protocol(protocol_slug),
_fetch_defillama_tvl(protocol_slug),
_get_social_sentiment(protocol_slug, chain),
return_exceptions=True,
)
if isinstance(defillama_data, Exception):
defillama_data = None
logger.warning(f"DeFiLlama lookup failed for {protocol_slug}: {defillama_data}")
if isinstance(tvl_data, Exception):
tvl_data = None
if isinstance(social_data, Exception):
social_data = {
"mentions_24h": 0,
"positive_ratio": 0.5,
"trending": False,
"rug_mentions": 0,
}
# Phase 2: Score each category
dl_data: dict | None = defillama_data if isinstance(defillama_data, dict) else None
tvl_data if isinstance(tvl_data, dict) else None
soc_data: dict = (
social_data
if isinstance(social_data, dict)
else {"mentions_24h": 0, "positive_ratio": 0.5, "trending": False, "rug_mentions": 0}
)
self._score_audit(report, dl_data)
self._score_tvl(report, dl_data)
self._score_social(report, soc_data)
self._score_deployer(report, dl_data)
self._score_liquidity(report, dl_data)
# Phase 3: Generate global flags
self._generate_flags(report)
# Phase 4: AI summary
report.ai_summary = await _generate_ai_summary(report)
# Cache
self._cache[cache_key] = (now, report)
return report
def _score_audit(self, report: SafetyReport, data: dict | None):
"""Score based on audit status from DeFiLlama."""
if not data:
report.audit_score.score = 0.3
report.audit_score.details = "No audit data available from public sources."
report.audit_score.flags.append("Unable to verify audit status")
return
audits = data.get("audits", [])
if not audits:
# Check if there are other signals
open_source = data.get("openSource", False)
report.audit_score.score = 0.5 if open_source else 0.3
report.audit_score.details = "Audit status unknown" + (
" (open source - better transparency)" if open_source else ""
)
return
# Analyze audits
max_score = 0.0
audit_details = []
for audit in audits:
if isinstance(audit, str):
# Some DeFiLlama entries store audit links as strings
audit_details.append(f"📄 Audit on file ({audit[:40]})")
max_score = max(max_score, 0.5)
continue
auditor = (audit.get("auditor", "") or "").lower().strip()
audit.get("link", "") or ""
date = audit.get("date", "") or ""
if any(firm in auditor for firm in REPUTABLE_AUDITORS):
max_score = max(max_score, 0.9)
audit_details.append(f"✅ Audited by {auditor} ({date})")
elif auditor:
max_score = max(max_score, 0.3)
audit_details.append(f"⚠️ Audited by {auditor} ({date}) - unknown firm")
else:
audit_details.append("📄 Audit on file")
report.audit_score.score = max_score
report.audit_score.details = "; ".join(audit_details[:3])
# Check if we have *any* reputable audits
if audit_details and not any(
any(firm in ad.lower() for firm in REPUTABLE_AUDITORS) for ad in str(audit_details).split(";")
):
report.audit_score.flags.append("No audits from well-known firms - verify independently")
def _score_tvl(self, report: SafetyReport, data: dict | None):
"""Score based on TVL data."""
if not data:
report.tvl_score.score = 0.3
report.tvl_score.details = "No TVL data available."
return
tvl = data.get("tvl", [])
current_tvl = data.get("currentChainTvls", {})
current_tvl.get(report.chain, 0) or 0
# Total TVL across all chains
total_tvl = sum(v for v in current_tvl.values() if isinstance(v, (int, float)))
# TVL stability check
tvl_stable = True
if len(tvl) > 7:
# Check last week TVL trend
try:
recent = [x.get("totalLiquidityUSD", 0) for x in tvl[-7:] if isinstance(x, dict)]
if recent and len(recent) >= 2:
change = (recent[-1] - recent[0]) / (recent[0] or 1)
if abs(change) > 0.5:
tvl_stable = False
except (KeyError, IndexError, TypeError):
tvl_stable = True
# Scoring: higher TVL = more "too big to fail" safety
if total_tvl > 1_000_000_000: # $1B+
report.tvl_score.score = 0.95
report.tvl_score.details = f"Very high TVL (${total_tvl:,.0f}) - strong market confidence"
elif total_tvl > 100_000_000: # $100M+
report.tvl_score.score = 0.85
report.tvl_score.details = f"High TVL (${total_tvl:,.0f}) - healthy protocol"
elif total_tvl > 10_000_000: # $10M+
report.tvl_score.score = 0.70
report.tvl_score.details = f"Moderate TVL (${total_tvl:,.0f}) - established"
elif total_tvl > 1_000_000: # $1M+
report.tvl_score.score = 0.50
report.tvl_score.details = f"Low TVL (${total_tvl:,.0f}) - higher risk"
elif total_tvl > 100_000: # $100K+
report.tvl_score.score = 0.30
report.tvl_score.details = f"Very low TVL (${total_tvl:,.0f}) - high risk"
else:
report.tvl_score.score = 0.10
report.tvl_score.details = "Minimal TVL - possible ghost protocol"
if not tvl_stable:
report.tvl_score.flags.append("TVL fluctuated >50% in the last week - potential exit or attack")
# Age check
start_date = data.get("listedAt", 0)
if start_date:
import datetime as dt
try:
age_days = (datetime.now(UTC) - dt.datetime.fromtimestamp(start_date / 1000, tz=UTC)).days
if age_days < 30:
report.tvl_score.flags.append(f"Protocol is very new ({age_days} days old)")
elif age_days < 90:
report.tvl_score.flags.append(f"Recently launched ({age_days} days old)")
except Exception:
pass
def _score_social(self, report: SafetyReport, social: dict):
"""Score based on social sentiment signals."""
mentions = social.get("mentions_24h", 0)
positive_ratio = social.get("positive_ratio", 0.5)
trending = social.get("trending", False)
rug_mentions = social.get("rug_mentions", 0)
# Base score from positive ratio
score = positive_ratio
# Trending bonus
if trending:
score = min(1.0, score + 0.15)
report.social_score.flags.append("Currently trending on CoinGecko - high visibility")
# Rug mention penalty
if rug_mentions > 5:
score = max(0.0, score - 0.2)
report.social_score.flags.append(f"Multiple 'rug' mentions detected ({rug_mentions} in 24h)")
# Very few mentions is a warning (unless it's a very new protocol)
if mentions < 10 and not trending:
report.social_score.flags.append("Very low social engagement - limited community visibility")
report.social_score.score = score
report.social_score.details = f"Social mentions: {mentions} in 24h | Positive ratio: {positive_ratio:.0%}" + (
" | Currently trending 🔥" if trending else ""
)
def _score_deployer(self, report: SafetyReport, data: dict | None):
"""Score based on deployer/team reputation."""
if not data:
report.deployer_score.score = 0.3
report.deployer_score.details = "No deployer information available."
return
name = data.get("name", "")
# Check for known safe protocols
BLUECHIP_PROTOCOLS = {
"aave",
"uniswap",
"curve",
"compound",
"makerdao",
"lido",
"pancakeswap",
"quickswap",
"sushiswap",
"balancer",
"yearn",
"convex",
"fraxlend",
"synthetix",
"instadapp",
"1inch",
"stargate",
"radiant",
"pendle",
"gmx",
"traderjoe",
"camelot",
"aerodrome",
"velodrome",
"spark",
"morpho",
}
name_lower = name.lower().strip()
if name_lower in BLUECHIP_PROTOCOLS or report.slug.lower() in BLUECHIP_PROTOCOLS:
report.deployer_score.score = 0.95
report.deployer_score.details = "Established blue-chip DeFi protocol"
report.deployer_score.flags.append(f"{report.protocol_name} is a well-known, battle-tested protocol")
return
# Check DeFiLlama metadata for reputation signals
chains = data.get("chains", [])
if isinstance(chains, list) and len(chains) >= 3:
# Multi-chain = more legit
report.deployer_score.score = 0.7
report.deployer_score.details = f"Deployed on {len(chains)} chains - moderate distribution"
elif isinstance(chains, list) and len(chains) >= 1:
report.deployer_score.score = 0.5
report.deployer_score.details = f"Deployed on {len(chains)} chain(s)"
else:
report.deployer_score.score = 0.4
report.deployer_score.details = "Single-chain protocol - limited track record"
# Fork detection
if data.get("forkedFrom"):
fork = data["forkedFrom"]
if isinstance(fork, str) and fork.lower() in BLUECHIP_PROTOCOLS:
report.deployer_score.score = min(1.0, report.deployer_score.score + 0.1)
report.deployer_score.details += f" (forked from {fork})"
elif isinstance(fork, str):
report.deployer_score.flags.append(f"Forked from {fork} - verify original is legit")
def _score_liquidity(self, report: SafetyReport, data: dict | None):
"""Score based on liquidity status."""
if not data:
report.liquidity_score.score = 0.3
report.liquidity_score.details = "No liquidity information available."
return
pools = data.get("pools", [])
if not pools:
report.liquidity_score.score = 0.4
report.liquidity_score.details = "No pool data on DeFiLlama"
return
# Check for locked liquidity signals
total_liquidity = 0
pool_count = len(pools)
for pool in pools:
if isinstance(pool, dict):
tvl_usd = pool.get("tvlUsd", 0) or 0
total_liquidity += tvl_usd
if total_liquidity > 0:
if total_liquidity > 10_000_000:
report.liquidity_score.score = 0.85
elif total_liquidity > 1_000_000:
report.liquidity_score.score = 0.70
elif total_liquidity > 100_000:
report.liquidity_score.score = 0.50
else:
report.liquidity_score.score = 0.30
report.liquidity_score.details = f"${total_liquidity:,.0f} total liquidity across {pool_count} pools"
else:
report.liquidity_score.score = 0.2
report.liquidity_score.details = "No liquidity detected in pools"
def _generate_flags(self, report: SafetyReport):
"""Generate cross-category red flags, warnings, and positives."""
# Red flags: any category critically low
for cat in report.categories():
if cat.score <= 0.15:
report.red_flags.append(f"Critical {cat.name} score ({cat.score:.0%}): {cat.details}")
# Red flag if overall score is critical
if report.overall_score() < 0.3:
report.red_flags.insert(0, "Protocol has CRITICAL risk profile - avoid if possible")
# Warnings
for cat in report.categories():
if 0.15 < cat.score <= 0.4:
report.warnings.append(f"Low {cat.name} score ({cat.score:.0%}): {cat.details[:60]}")
# Positives
for cat in report.categories():
if cat.score >= 0.8:
report.positives.append(f"Strong {cat.name}: {cat.details[:80]}")
def invalidate_cache(self, protocol_slug: str, chain: str = "ethereum"):
"""Force re-audit on next call."""
self._cache.pop(f"{protocol_slug}:{chain}", None)
def cache_stats(self) -> dict:
return {"cached_protocols": len(self._cache), "ttl_seconds": self._cache_ttl}
# ═══════════════════════════════════════════════════════════════
# Singleton for FastAPI integration
# ═══════════════════════════════════════════════════════════════
_defi_auditor: DefiProtocolAuditor | None = None
def get_auditor() -> DefiProtocolAuditor:
global _defi_auditor
if _defi_auditor is None:
_defi_auditor = DefiProtocolAuditor()
return _defi_auditor
# ═══════════════════════════════════════════════════════════════
# FastAPI Router
# ═══════════════════════════════════════════════════════════════
# Import and use in main.py or a router:
#
# from app.defi_protocol_auditor import get_auditor
# router = APIRouter(prefix="/api/v1/defi", tags=["defi"])
#
# @router.get("/audit/{protocol}")
# async def protocol_audit(protocol: str, chain: str = "ethereum"):
# auditor = get_auditor()
# report = await auditor.audit(protocol, chain=chain)
# return report.to_dict()
#
# ═══════════════════════════════════════════════════════════════
# Standalone Test
# ═══════════════════════════════════════════════════════════════
if __name__ == "__main__":
import asyncio
async def main():
auditor = DefiProtocolAuditor()
# Audit a well-known protocol
print("=" * 60)
print("Auditing: Uniswap (known safe)")
print("=" * 60)
report = await auditor.audit("uniswap")
print(report.summary(detailed=True))
print("\n" + "=" * 60)
print("Auditing: Aave")
print("=" * 60)
report2 = await auditor.audit("aave")
print(report2.summary(detailed=True))
print("\n" + "=" * 60)
print("Full JSON output for Uniswap:")
print("=" * 60)
import json
print(json.dumps(report.to_dict(), indent=2))
asyncio.run(main())
Reference in a new issue View git blame Copy permalink