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>
65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
"""NFT Wash Trading Detector - detects fake volume in NFT collections."""
|
|
|
|
import os
|
|
|
|
from fastapi import APIRouter, Query
|
|
|
|
router = APIRouter(prefix="/api/v1/nft-detector", tags=["nft-detector"])
|
|
|
|
BACKEND = os.getenv("BACKEND_URL", "http://localhost:8000")
|
|
RMI_KEY = os.getenv("RMI_INTERNAL_KEY", "rmi-internal-2026")
|
|
|
|
|
|
@router.get("/check/{collection_address}")
|
|
async def check_nft_wash(collection_address: str, chain: str = Query("ethereum")):
|
|
"""Check NFT collection for wash trading patterns."""
|
|
|
|
# Use DataBus volume authenticity engine with NFT-specific heuristics
|
|
score = 85 # Placeholder - real implementation queries DataBus
|
|
|
|
# NFT-specific wash trading patterns
|
|
patterns = []
|
|
patterns.append({"pattern": "Self-trades", "description": "Same wallet buying and selling", "detected": False})
|
|
patterns.append({"pattern": "Round-trip", "description": "A→B→A within minutes", "detected": False})
|
|
patterns.append({"pattern": "Bid stuffing", "description": "Fake bids to inflate floor price", "detected": False})
|
|
patterns.append(
|
|
{"pattern": "Wash pool", "description": "Group of wallets trading among themselves", "detected": False}
|
|
)
|
|
|
|
if score < 30:
|
|
risk = "CRITICAL"
|
|
emoji = "🔴"
|
|
elif score < 60:
|
|
risk = "HIGH"
|
|
emoji = "🟡"
|
|
elif score < 80:
|
|
risk = "MEDIUM"
|
|
emoji = "⚪"
|
|
else:
|
|
risk = "LOW"
|
|
emoji = "🟢"
|
|
|
|
return {
|
|
"collection": collection_address,
|
|
"chain": chain,
|
|
"authenticity_score": score,
|
|
"risk": risk,
|
|
"emoji": emoji,
|
|
"wash_patterns": patterns,
|
|
"recommendation": "Likely authentic trading"
|
|
if score >= 80
|
|
else "Suspicious activity detected - verify before buying",
|
|
}
|
|
|
|
|
|
@router.get("/floor-check/{collection_address}")
|
|
async def check_floor_manipulation(collection_address: str, chain: str = Query("ethereum")):
|
|
"""Check if NFT floor price is being manipulated."""
|
|
return {
|
|
"collection": collection_address,
|
|
"floor_price_eth": 0.0,
|
|
"bid_ask_spread_pct": 0,
|
|
"suspicious_bids_24h": 0,
|
|
"manipulation_risk": "LOW",
|
|
"note": "NFT data providers integration pending",
|
|
}
|