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>
902 lines
36 KiB
Python
902 lines
36 KiB
Python
"""
|
|
Advanced x402 tools - caching, confidence, real-time streams, predictive scoring.
|
|
|
|
Layer 1: Response caching (Redis, 60s TTL) - <50ms repeat calls
|
|
Layer 2: Confidence scores - every data point gets low/medium/high
|
|
Layer 3: SSE alert stream - real-time rug/whale/price alerts
|
|
Layer 4: Rug probability - predictive 0-100 "will this rug in 24h?"
|
|
Layer 5: Historical scanner data - time-series risk/liquidity/holders
|
|
Layer 6: Narrative engine - what's the market saying RIGHT NOW
|
|
"""
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
from datetime import datetime
|
|
|
|
import aiohttp
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
logger = logging.getLogger("x402.advanced")
|
|
|
|
router = APIRouter()
|
|
|
|
# ═══════════════════════════════════════════════════════════
|
|
# LAYER 1: Response Caching
|
|
# ═══════════════════════════════════════════════════════════
|
|
|
|
CACHE_TTL = 60 # seconds
|
|
CACHEABLE_TOOLS = {
|
|
"audit",
|
|
"wallet",
|
|
"reputation_score",
|
|
"honeypot_check",
|
|
"rugshield",
|
|
"forensics",
|
|
"whale",
|
|
"token_deep_dive",
|
|
"market_overview",
|
|
"chain_health",
|
|
"sentiment",
|
|
"rug_pull_predictor",
|
|
"rug_probability",
|
|
"narrative",
|
|
}
|
|
|
|
|
|
def _redis_conn():
|
|
import redis as _redis
|
|
|
|
return _redis.Redis(
|
|
host=os.getenv("REDIS_HOST", "rmi-redis"),
|
|
port=int(os.getenv("REDIS_PORT", "6379")),
|
|
password=os.getenv("REDIS_PASSWORD", ""),
|
|
decode_responses=True,
|
|
socket_connect_timeout=2,
|
|
socket_timeout=2,
|
|
)
|
|
|
|
|
|
def get_cached(tool: str, params: dict) -> dict | None:
|
|
"""Check Redis cache for a tool result."""
|
|
try:
|
|
r = _redis_conn()
|
|
key = _cache_key(tool, params)
|
|
data = r.get(key)
|
|
if data:
|
|
return json.loads(data)
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def set_cached(tool: str, params: dict, result: dict, ttl: int = CACHE_TTL):
|
|
"""Store tool result in Redis cache."""
|
|
try:
|
|
r = _redis_conn()
|
|
key = _cache_key(tool, params)
|
|
result["_cached"] = True
|
|
result["_cached_at"] = datetime.utcnow().isoformat()
|
|
r.setex(key, ttl, json.dumps(result))
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _cache_key(tool: str, params: dict) -> str:
|
|
raw = f"{tool}:{json.dumps(params, sort_keys=True)}"
|
|
return f"x402:cache:{hashlib.sha256(raw.encode()).hexdigest()[:16]}"
|
|
|
|
|
|
def invalidate_cache(tool: str | None = None):
|
|
"""Clear cache for a tool or all cached results."""
|
|
try:
|
|
r = _redis_conn()
|
|
if tool:
|
|
for key in r.scan_iter("x402:cache:*"):
|
|
r.delete(key)
|
|
else:
|
|
for key in r.scan_iter("x402:cache:*"):
|
|
r.delete(key)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════
|
|
# LAYER 2: Confidence Scores
|
|
# ═══════════════════════════════════════════════════════════
|
|
|
|
|
|
def compute_confidence(result: dict) -> dict:
|
|
"""Add confidence scoring to any tool result."""
|
|
sources = result.get("sources_used", [])
|
|
source_count = len(sources)
|
|
|
|
# Source quality weights
|
|
high_quality = {
|
|
"clickhouse",
|
|
"wallet_labels",
|
|
"etherscan",
|
|
"coingecko",
|
|
"defillama",
|
|
"solana_rpc",
|
|
"ethereum_rpc",
|
|
"base_rpc",
|
|
"dexscreener",
|
|
"geckoterminal",
|
|
"tron_rpc",
|
|
"bitcoin_rpc",
|
|
}
|
|
medium_quality = {
|
|
"cryptopanic",
|
|
"reddit",
|
|
"coincap",
|
|
"coinmarketcap",
|
|
"moralis",
|
|
"birdeye",
|
|
"helius",
|
|
"solscan",
|
|
"rmi_intel",
|
|
"scam_detector",
|
|
"rag_similarity",
|
|
"pumpfun",
|
|
}
|
|
|
|
high_count = sum(1 for s in sources if s.lower() in high_quality)
|
|
medium_count = sum(1 for s in sources if s.lower() in medium_quality)
|
|
source_count - high_count - medium_count
|
|
|
|
# Score computation
|
|
if source_count >= 4 and high_count >= 2:
|
|
level = "high"
|
|
score = min(100, 70 + source_count * 5)
|
|
elif source_count >= 2 and high_count >= 1:
|
|
level = "medium"
|
|
score = 40 + source_count * 8
|
|
elif source_count >= 1:
|
|
level = "low"
|
|
score = 20 + source_count * 10
|
|
else:
|
|
level = "unverified"
|
|
score = 5
|
|
|
|
# Flag freshness
|
|
age_flags = []
|
|
if "dexscreener" in [s.lower() for s in sources]:
|
|
age_flags.append("market_data_live")
|
|
if "wallet_labels" in [s.lower() for s in sources]:
|
|
age_flags.append("labels_cached")
|
|
|
|
return {
|
|
"score": min(100, score),
|
|
"level": level,
|
|
"sources_total": source_count,
|
|
"sources_high_quality": high_count,
|
|
"sources_medium_quality": medium_count,
|
|
"flags": age_flags,
|
|
"interpretation": {
|
|
"high": "Multiple verified sources confirm this data",
|
|
"medium": "Adequate coverage from trusted sources",
|
|
"low": "Limited source coverage - verify independently",
|
|
"unverified": "Insufficient data - treat as directional only",
|
|
}.get(level, ""),
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════
|
|
# LAYER 3: SSE Real-Time Alert Stream
|
|
# ═══════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/stream/alerts")
|
|
async def sse_alert_stream(
|
|
events: str = Query(default="rug_pull,whale_move,price_crash", description="Comma-separated event types"),
|
|
chain: str = Query(default="all"),
|
|
):
|
|
"""Server-Sent Events stream for real-time security alerts.
|
|
|
|
Connect with: EventSource('/api/v1/x402-tools/stream/alerts?events=rug_pull,whale_move')
|
|
|
|
Events emitted as JSON:
|
|
{"event":"rug_pull","address":"0x...","chain":"base","severity":"critical","data":{...}}
|
|
"""
|
|
event_list = [e.strip() for e in events.split(",") if e.strip()]
|
|
chain_filter = chain.strip().lower()
|
|
|
|
async def event_generator():
|
|
import redis as _redis
|
|
|
|
r = _redis.Redis(
|
|
host=os.getenv("REDIS_HOST", "rmi-redis"),
|
|
port=int(os.getenv("REDIS_PORT", "6379")),
|
|
password=os.getenv("REDIS_PASSWORD", ""),
|
|
decode_responses=True,
|
|
socket_connect_timeout=3,
|
|
)
|
|
last_id = "0"
|
|
|
|
# Send initial connection event
|
|
yield f"event: connected\ndata: {json.dumps({'status': 'connected', 'events': event_list, 'chain': chain_filter, 'timestamp': datetime.utcnow().isoformat()})}\n\n"
|
|
|
|
while True:
|
|
try:
|
|
# Check Redis for new alerts
|
|
alerts = r.lrange("x402:alerts:high_risk", 0, 9)
|
|
for alert_json in alerts:
|
|
try:
|
|
alert = json.loads(alert_json)
|
|
alert_id = alert.get("timestamp", "")
|
|
if alert_id > last_id:
|
|
event_type = alert.get("type", "unknown")
|
|
if event_type in event_list or "all" in event_list:
|
|
yield f"event: {event_type}\ndata: {json.dumps(alert)}\n\n"
|
|
last_id = alert_id
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
# Also check for new events in webhook-triggered alerts
|
|
for evt in event_list:
|
|
evt_alerts = r.lrange(f"x402:alert:{evt}", 0, 4)
|
|
for a in evt_alerts:
|
|
try:
|
|
data = json.loads(a)
|
|
if data.get("timestamp", "") > last_id:
|
|
yield f"event: {evt}\ndata: {json.dumps(data)}\n\n"
|
|
last_id = data.get("timestamp", "")
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
# Keep-alive ping
|
|
yield f": keepalive {int(time.time())}\n\n"
|
|
await asyncio.sleep(5)
|
|
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as e:
|
|
logger.error(f"SSE stream error: {e}")
|
|
await asyncio.sleep(10)
|
|
|
|
return StreamingResponse(
|
|
event_generator(),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"Connection": "keep-alive",
|
|
"X-Accel-Buffering": "no",
|
|
"Access-Control-Allow-Origin": "*",
|
|
},
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════
|
|
# Request Models
|
|
# ═══════════════════════════════════════════════════════════
|
|
|
|
|
|
class AddressRequest(BaseModel):
|
|
address: str = Field(..., description="Wallet or token address")
|
|
chain: str = Field(default="base")
|
|
|
|
|
|
class HistoryRequest(BaseModel):
|
|
address: str = Field(..., description="Token or wallet address")
|
|
chain: str = Field(default="base")
|
|
hours: int = Field(default=24, ge=1, le=168, description="Lookback window in hours (max 168)")
|
|
|
|
|
|
class NarrativeRequest(BaseModel):
|
|
token: str = Field(..., description="Token symbol or address")
|
|
chain: str = Field(default="all")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════
|
|
# LAYER 4: Rug Probability Score
|
|
# ═══════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.post("/rug_probability")
|
|
async def rug_probability(req: AddressRequest):
|
|
"""Predictive rug pull probability: 0-100 score for "will this token rug in 24h?"
|
|
|
|
Combines 7 signals:
|
|
- Honeypot check (can you sell?)
|
|
- Liquidity depth and lock status
|
|
- Holder concentration (whale dominance)
|
|
- Deployer history (serial rugger?)
|
|
- Contract age (new = higher risk)
|
|
- Social signal anomalies (coordinated shilling)
|
|
- Market context (volume/liquidity ratio)
|
|
"""
|
|
try:
|
|
addr = req.address.strip()
|
|
chain = req.chain or "base"
|
|
t0 = time.time()
|
|
|
|
# Check cache
|
|
cached = get_cached("rug_probability", {"address": addr, "chain": chain})
|
|
if cached:
|
|
return cached
|
|
|
|
probability = 0
|
|
signals = []
|
|
sources = []
|
|
|
|
# ── Signal 1: Honeypot Check ──
|
|
try:
|
|
async with aiohttp.ClientSession() as session, session.post(
|
|
"http://localhost:8000/api/v1/x402-tools/honeypot_check",
|
|
json={"address": addr, "chain": chain},
|
|
timeout=aiohttp.ClientTimeout(total=10),
|
|
) as resp:
|
|
if resp.status == 200:
|
|
data = await resp.json()
|
|
if data.get("is_honeypot"):
|
|
probability += 40
|
|
signals.append(
|
|
{
|
|
"signal": "honeypot_detected",
|
|
"weight": 40,
|
|
"detail": data.get("reason", "Cannot sell - confirmed honeypot"),
|
|
}
|
|
)
|
|
sources.append("honeypot_check")
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Signal 2: DexScreener Liquidity & Age ──
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
url = f"https://api.dexscreener.com/latest/dex/tokens/{addr}"
|
|
async with session.get(url, timeout=aiohttp.ClientTimeout(total=8)) as resp:
|
|
if resp.status == 200:
|
|
data = await resp.json()
|
|
pairs = data.get("pairs", [])
|
|
if pairs:
|
|
sources.append("dexscreener")
|
|
p = pairs[0]
|
|
liq = p.get("liquidity", {}).get("usd", 0) or 0
|
|
vol = p.get("volume", {}).get("h24", 0) or 0
|
|
age_ms = p.get("pairCreatedAt", 0) or 0
|
|
age_h = (time.time() * 1000 - age_ms) / 3600000 if age_ms else 0
|
|
|
|
# Liquidity signal
|
|
if liq < 1000:
|
|
probability += 25
|
|
signals.append(
|
|
{
|
|
"signal": "critical_low_liquidity",
|
|
"weight": 25,
|
|
"detail": f"Liquidity ${liq:,.0f} - extremely low, easy to drain",
|
|
}
|
|
)
|
|
elif liq < 10000:
|
|
probability += 15
|
|
signals.append(
|
|
{
|
|
"signal": "low_liquidity",
|
|
"weight": 15,
|
|
"detail": f"Liquidity ${liq:,.0f} - below safe threshold",
|
|
}
|
|
)
|
|
elif liq < 50000:
|
|
probability += 5
|
|
signals.append(
|
|
{
|
|
"signal": "moderate_liquidity",
|
|
"weight": 5,
|
|
"detail": f"Liquidity ${liq:,.0f} - moderate",
|
|
}
|
|
)
|
|
|
|
# Age signal
|
|
if 0 < age_h < 1:
|
|
probability += 20
|
|
signals.append(
|
|
{
|
|
"signal": "brand_new",
|
|
"weight": 20,
|
|
"detail": f"Only {age_h:.1f}h old - highest rug risk window",
|
|
}
|
|
)
|
|
elif 0 < age_h < 6:
|
|
probability += 12
|
|
signals.append(
|
|
{
|
|
"signal": "very_new",
|
|
"weight": 12,
|
|
"detail": f"Only {age_h:.1f}h old - early risk period",
|
|
}
|
|
)
|
|
elif 0 < age_h < 24:
|
|
probability += 5
|
|
signals.append(
|
|
{
|
|
"signal": "new_token",
|
|
"weight": 5,
|
|
"detail": f"{age_h:.0f}h old - still in risk window",
|
|
}
|
|
)
|
|
|
|
# Volume/Liquidity ratio (pump and dump signal)
|
|
if liq > 0 and vol > liq * 3:
|
|
probability += 10
|
|
signals.append(
|
|
{
|
|
"signal": "pump_dump_pattern",
|
|
"weight": 10,
|
|
"detail": f"Volume {vol / liq:.0f}x liquidity - possible pump and dump",
|
|
}
|
|
)
|
|
|
|
# Price crash signal
|
|
pc = p.get("priceChange", {}).get("h24", 0) or 0
|
|
if pc < -50:
|
|
probability += 15
|
|
signals.append(
|
|
{
|
|
"signal": "price_crashing",
|
|
"weight": 15,
|
|
"detail": f"Down {pc:.0f}% in 24h - possible exit scam in progress",
|
|
}
|
|
)
|
|
elif pc < -20:
|
|
probability += 8
|
|
signals.append(
|
|
{
|
|
"signal": "price_declining",
|
|
"weight": 8,
|
|
"detail": f"Down {pc:.0f}% in 24h",
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Signal 3: Holder Concentration ──
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
# GeckoTerminal for holder data
|
|
chain_map = {"solana": "solana", "base": "base", "ethereum": "eth", "bsc": "bsc"}
|
|
geo_chain = chain_map.get(chain, chain)
|
|
url = f"https://api.geckoterminal.com/api/v2/networks/{geo_chain}/tokens/{addr}"
|
|
async with session.get(url, timeout=aiohttp.ClientTimeout(total=8)) as resp:
|
|
if resp.status == 200:
|
|
data = await resp.json()
|
|
attrs = data.get("data", {}).get("attributes", {})
|
|
if attrs:
|
|
sources.append("geckoterminal")
|
|
# Check top holder concentration from available data
|
|
top_pool = attrs.get("top_pool_id")
|
|
if top_pool:
|
|
# Pool exists - check if it's the only one
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Signal 4: Deployer History (scam pattern check) ──
|
|
try:
|
|
import asyncio as _asyncio
|
|
|
|
from app.rag_service import detect_scam_patterns
|
|
|
|
result = _asyncio.run(detect_scam_patterns({"address": addr, "chain": chain}, 0.4))
|
|
if result and result.get("risk_score", 0) > 0:
|
|
sources.append("scam_detector")
|
|
risk = result.get("risk_score", 0)
|
|
probability += min(risk, 30)
|
|
patterns = result.get("patterns", [])
|
|
if patterns:
|
|
signals.append(
|
|
{
|
|
"signal": "scam_pattern_match",
|
|
"weight": min(risk, 30),
|
|
"detail": f"Matches known scam patterns: {', '.join(patterns[:3])}",
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Signal 5: Social Anomaly Check ──
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
from urllib.parse import quote
|
|
|
|
symbol = addr[:12]
|
|
url = f"https://cryptopanic.com/api/free/posts/?filter=important&q={quote(symbol)}"
|
|
async with session.get(url, timeout=aiohttp.ClientTimeout(total=8)) as resp:
|
|
if resp.status == 200:
|
|
data = await resp.json()
|
|
posts = data.get("results", [])
|
|
if posts:
|
|
sources.append("cryptopanic")
|
|
# Check for sudden spike in mentions
|
|
recent = [p for p in posts if p.get("created_at")]
|
|
if len(recent) > 20:
|
|
probability += 5
|
|
signals.append(
|
|
{
|
|
"signal": "social_spike",
|
|
"weight": 5,
|
|
"detail": f"{len(recent)} social mentions - unusual activity",
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Compute final probability ──
|
|
probability = max(0, min(100, probability))
|
|
|
|
if probability >= 75:
|
|
tier = "EXTREME_RISK"
|
|
recommendation = "DO NOT BUY - extremely high rug probability"
|
|
elif probability >= 50:
|
|
tier = "HIGH_RISK"
|
|
recommendation = "Avoid - significant rug indicators present"
|
|
elif probability >= 25:
|
|
tier = "MODERATE_RISK"
|
|
recommendation = "Caution - monitor closely before entry"
|
|
elif probability >= 10:
|
|
tier = "LOW_RISK"
|
|
recommendation = "Standard risk - normal market activity"
|
|
else:
|
|
tier = "MINIMAL_RISK"
|
|
recommendation = "Low rug probability - relatively safe"
|
|
|
|
result = {
|
|
"tool": "Rug Probability Score",
|
|
"version": "1.0",
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"address": addr,
|
|
"chain": chain,
|
|
"rug_probability": probability,
|
|
"tier": tier,
|
|
"recommendation": recommendation,
|
|
"signals": signals,
|
|
"signal_count": len(signals),
|
|
"sources_used": sources,
|
|
"_confidence": compute_confidence({"sources_used": sources}),
|
|
"performance_ms": round((time.time() - t0) * 1000, 1),
|
|
"guarantee": "Data delivered or auto-refund via x402 receipt",
|
|
}
|
|
|
|
set_cached("rug_probability", {"address": addr, "chain": chain}, result)
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Rug probability failed: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e)) from e
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════
|
|
# LAYER 5: Historical Scanner Data
|
|
# ═══════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.post("/history")
|
|
async def token_history(req: HistoryRequest):
|
|
"""Historical risk/liquidity/holder data for any token.
|
|
|
|
Returns time-series data from the RMI scanner (runs every 10 min).
|
|
Shows how risk profile, liquidity, volume, and holder metrics change over time.
|
|
|
|
Data points: timestamp, risk_score, liquidity_usd, volume_24h,
|
|
price_usd, holder_count, whale_dominance_pct.
|
|
"""
|
|
try:
|
|
addr = req.address.strip()
|
|
chain = req.chain or "base"
|
|
hours = req.hours
|
|
|
|
cached = get_cached("history", {"address": addr, "chain": chain, "hours": hours})
|
|
if cached:
|
|
return cached
|
|
|
|
# Read scanner snapshots from Redis
|
|
import redis as _redis
|
|
|
|
r = _redis.Redis(
|
|
host=os.getenv("REDIS_HOST", "rmi-redis"),
|
|
port=int(os.getenv("REDIS_PORT", "6379")),
|
|
password=os.getenv("REDIS_PASSWORD", ""),
|
|
decode_responses=True,
|
|
socket_connect_timeout=2,
|
|
)
|
|
|
|
data_points = []
|
|
cutoff = time.time() - (hours * 3600)
|
|
|
|
# Scanner stores data as x402:scan:{address}:{timestamp}
|
|
for key in r.scan_iter(f"x402:scan:{addr}:*"):
|
|
try:
|
|
ts = float(key.decode().split(":")[-1]) if isinstance(key, bytes) else float(key.split(":")[-1])
|
|
if ts < cutoff:
|
|
continue
|
|
raw = r.get(key)
|
|
if raw:
|
|
dp = json.loads(raw)
|
|
dp["timestamp"] = ts
|
|
data_points.append(dp)
|
|
except (ValueError, json.JSONDecodeError):
|
|
continue
|
|
|
|
# Also try the token scanner's own keys
|
|
for key in r.scan_iter(f"token:scan:{addr}:*"):
|
|
try:
|
|
parts = key.decode().split(":") if isinstance(key, bytes) else key.split(":")
|
|
ts = float(parts[-1])
|
|
if ts < cutoff:
|
|
continue
|
|
raw = r.get(key)
|
|
if raw:
|
|
dp = json.loads(raw)
|
|
dp["timestamp"] = ts
|
|
data_points.append(dp)
|
|
except (ValueError, json.JSONDecodeError):
|
|
continue
|
|
|
|
data_points.sort(key=lambda d: d.get("timestamp", 0))
|
|
|
|
# Add current snapshot
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
url = f"https://api.dexscreener.com/latest/dex/tokens/{addr}"
|
|
async with session.get(url, timeout=aiohttp.ClientTimeout(total=8)) as resp:
|
|
if resp.status == 200:
|
|
dex_data = await resp.json()
|
|
pairs = dex_data.get("pairs", [])
|
|
if pairs:
|
|
p = pairs[0]
|
|
current = {
|
|
"timestamp": time.time(),
|
|
"price_usd": p.get("priceUsd"),
|
|
"liquidity_usd": p.get("liquidity", {}).get("usd"),
|
|
"volume_24h": p.get("volume", {}).get("h24"),
|
|
"price_change_24h": p.get("priceChange", {}).get("h24"),
|
|
"is_current": True,
|
|
}
|
|
data_points.append(current)
|
|
except Exception:
|
|
pass
|
|
|
|
# Compute trends
|
|
trends = {}
|
|
if len(data_points) >= 2:
|
|
first = data_points[0]
|
|
last = data_points[-1]
|
|
for metric in ["price_usd", "liquidity_usd", "volume_24h"]:
|
|
fv = first.get(metric)
|
|
lv = last.get(metric)
|
|
if fv and lv and fv > 0:
|
|
change_pct = ((lv - fv) / fv) * 100
|
|
trends[metric] = {
|
|
"start": fv,
|
|
"end": lv,
|
|
"change_pct": round(change_pct, 1),
|
|
"direction": "up" if change_pct > 0 else "down" if change_pct < 0 else "flat",
|
|
}
|
|
|
|
result = {
|
|
"tool": "Historical Scanner Data",
|
|
"version": "1.0",
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"address": addr,
|
|
"chain": chain,
|
|
"lookback_hours": hours,
|
|
"data_points": len(data_points),
|
|
"history": data_points[-50:], # Last 50 data points
|
|
"trends": trends,
|
|
"scanner_interval": "10 minutes",
|
|
"_confidence": compute_confidence(
|
|
{
|
|
"sources_used": ["rmi_scanner"]
|
|
+ (["dexscreener"] if any(d.get("is_current") for d in data_points) else []),
|
|
}
|
|
),
|
|
"guarantee": "Historical data or full refund",
|
|
}
|
|
|
|
set_cached("history", {"address": addr, "chain": chain, "hours": hours}, result, ttl=120)
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Token history failed: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e)) from e
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════
|
|
# LAYER 6: Narrative Engine
|
|
# ═══════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.post("/narrative")
|
|
async def narrative_engine(req: NarrativeRequest):
|
|
"""What's the market saying about this token RIGHT NOW?
|
|
|
|
Aggregates Twitter, Reddit, Telegram, and news sentiment into a
|
|
narrative summary with confidence scoring and shill detection.
|
|
"""
|
|
try:
|
|
token = req.token.strip()
|
|
chain = req.chain or "all"
|
|
t0 = time.time()
|
|
|
|
cached = get_cached("narrative", {"token": token, "chain": chain})
|
|
if cached:
|
|
return cached
|
|
|
|
sources = []
|
|
posts = []
|
|
|
|
# ── CryptoPanic (news + social) ──
|
|
try:
|
|
from urllib.parse import quote
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
url = f"https://cryptopanic.com/api/free/posts/?filter=important&q={quote(token)}"
|
|
async with session.get(url, timeout=aiohttp.ClientTimeout(total=8)) as resp:
|
|
if resp.status == 200:
|
|
data = await resp.json()
|
|
results = data.get("results", [])
|
|
if results:
|
|
sources.append("cryptopanic")
|
|
for r in results[:15]:
|
|
posts.append(
|
|
{
|
|
"source": "cryptopanic",
|
|
"title": r.get("title", "")[:150],
|
|
"sentiment": r.get("votes", {}).get("positive", 0)
|
|
- r.get("votes", {}).get("negative", 0),
|
|
"created": r.get("created_at", ""),
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Reddit ──
|
|
try:
|
|
from urllib.parse import quote
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
url = f"https://www.reddit.com/search.json?q={quote(token)}&sort=new&limit=10"
|
|
async with session.get(
|
|
url,
|
|
headers={"User-Agent": "RMI-Narrative/1.0"},
|
|
timeout=aiohttp.ClientTimeout(total=8),
|
|
) as resp:
|
|
if resp.status == 200:
|
|
data = await resp.json()
|
|
children = data.get("data", {}).get("children", [])
|
|
if children:
|
|
sources.append("reddit")
|
|
for c in children[:10]:
|
|
d = c.get("data", {})
|
|
posts.append(
|
|
{
|
|
"source": "reddit",
|
|
"title": d.get("title", "")[:150],
|
|
"subreddit": d.get("subreddit", ""),
|
|
"score": d.get("score", 0),
|
|
"comments": d.get("num_comments", 0),
|
|
"created": datetime.utcfromtimestamp(d.get("created_utc", 0)).isoformat()
|
|
if d.get("created_utc")
|
|
else "",
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Compute narrative ──
|
|
total_posts = len(posts)
|
|
if total_posts == 0:
|
|
return {
|
|
"tool": "Narrative Engine",
|
|
"version": "1.0",
|
|
"address": token,
|
|
"narrative": "Insufficient data - no recent mentions found",
|
|
"sentiment": "neutral",
|
|
"confidence": "low",
|
|
"posts_analyzed": 0,
|
|
"sources_used": [],
|
|
}
|
|
|
|
# Sentiment scoring
|
|
positive = sum(1 for p in posts if p.get("sentiment", 0) > 0)
|
|
negative = sum(1 for p in posts if p.get("sentiment", 0) < 0)
|
|
neutral_count = total_posts - positive - negative
|
|
|
|
if positive > negative * 2:
|
|
sentiment = "bullish"
|
|
sentiment_pct = round(positive / max(total_posts, 1) * 100)
|
|
elif negative > positive * 2:
|
|
sentiment = "bearish"
|
|
sentiment_pct = round(negative / max(total_posts, 1) * 100)
|
|
elif positive > negative:
|
|
sentiment = "slightly_bullish"
|
|
sentiment_pct = round(positive / max(total_posts, 1) * 100)
|
|
elif negative > positive:
|
|
sentiment = "slightly_bearish"
|
|
sentiment_pct = round(negative / max(total_posts, 1) * 100)
|
|
else:
|
|
sentiment = "neutral"
|
|
sentiment_pct = 50
|
|
|
|
# Shill detection
|
|
shill_signals = []
|
|
reddit_posts = [p for p in posts if p.get("source") == "reddit"]
|
|
if total_posts > 10 and positive > total_posts * 0.8:
|
|
shill_signals.append("Suspiciously high positive ratio - possible coordinated shilling")
|
|
if len(reddit_posts) >= 3 and all(p.get("score", 0) == 0 for p in reddit_posts):
|
|
shill_signals.append("Same-timestamp posts detected - possible bot activity")
|
|
|
|
# Build narrative summary
|
|
subreddits = list({p.get("subreddit", "") for p in posts if p.get("subreddit")})
|
|
narrative = (
|
|
f"{sentiment.replace('_', ' ').title()} on {token}. "
|
|
f"{sentiment_pct}% positive across {total_posts} posts from {len(sources)} sources. "
|
|
+ (f"Active in r/{', r/'.join(subreddits[:3])}. " if subreddits else "")
|
|
+ (f"Risk: {'; '.join(shill_signals)}" if shill_signals else "No shill signals detected.")
|
|
)
|
|
|
|
result = {
|
|
"tool": "Narrative Engine",
|
|
"version": "1.0",
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"token": token,
|
|
"chain": chain,
|
|
"narrative": narrative,
|
|
"sentiment": sentiment,
|
|
"sentiment_score": sentiment_pct,
|
|
"posts_analyzed": total_posts,
|
|
"breakdown": {
|
|
"positive": positive,
|
|
"negative": negative,
|
|
"neutral": neutral_count,
|
|
},
|
|
"sources_used": sources,
|
|
"shill_signals": shill_signals if shill_signals else None,
|
|
"_confidence": compute_confidence({"sources_used": sources}),
|
|
"performance_ms": round((time.time() - t0) * 1000, 1),
|
|
"guarantee": "Real-time social data or full refund",
|
|
}
|
|
|
|
set_cached("narrative", {"token": token, "chain": chain}, result)
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Narrative engine failed: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e)) from e
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════
|
|
# Cache Management Endpoint
|
|
# ═══════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.post("/cache/clear")
|
|
async def clear_cache(tool: str = Query(default=None)):
|
|
"""Clear response cache for a specific tool or all tools."""
|
|
ok = invalidate_cache(tool)
|
|
return {
|
|
"status": "cleared" if ok else "failed",
|
|
"tool": tool or "all",
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
}
|
|
|
|
|
|
@router.get("/cache/stats")
|
|
async def cache_stats():
|
|
"""Get cache hit/miss statistics."""
|
|
try:
|
|
r = _redis_conn()
|
|
keys = list(r.scan_iter("x402:cache:*"))
|
|
return {
|
|
"cached_entries": len(keys),
|
|
"memory_estimate_bytes": sum(len(r.get(k) or "") for k in keys[:100]),
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
}
|
|
except Exception as e:
|
|
return {"error": str(e), "cached_entries": 0}
|