RugMunchMedia/rmi-backend
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(databus): remove _generated code + legacy providers, add scanner engine, ai_predict, pair_tracker, ws_trades routers

Browse source 
- Delete _generated/ provider chains (auto-generated), provider_chains.py, provider_core.py, provider_implementations.py — replaced by domain-driven approach
- Delete app/core/lifespan.py — superseded by app/lifespan.py
- Add app/core/config.py — new centralized config module
- Add app/domains/scanners/core/engine.py — scanner engine
- Add app/routers/ai_predict.py, pair_tracker.py, ws_trades.py — new feature routers
- Update scanner, token, errors, db_client, databus modules, telegram bot, rag, lifespan, mount for compatibility
This commit is contained in:
Crypto Rug Munch 2026-07-09 14:05:05 +02:00
parent 3bebd2b297
commit 8fe11cfb67
32 changed files with 1644 additions and 3810 deletions
Download patch file Download diff file Expand all files Collapse all files

185
app/api/v1/public/scanner.py
View file

@ -1,53 +1,121 @@
"""Public scanner endpoints - /api/v1/scanner/*.
"""Public scanner endpoints — /api/v1/scanner/*
Stub for unauthenticated token scans. Real implementation will
trigger the scanner pipeline (honeypot detection, flash loan checks,
oracle manipulation analysis).
Single source of truth for all token + wallet scans on the HTTP surface.
Delegates to the canonical scanner engine at app.domains.scanners.core.engine.
"""
from __future__ import annotations
import uuid
from typing import Any
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel
from pydantic import BaseModel, Field
from app.domains.scanners.core.engine import scan_token as _run_scan
from app.domains.scanners.core.engine import scan_wallet as _run_wallet_scan
router = APIRouter(prefix="/api/v1/scanner", tags=["scanner"])
_scan_store: dict[str, dict[str, Any]] = {}
# ── Request / Response models ──────────────────────────────────────
class ScanRequest(BaseModel):
"""Request to scan a token or wallet."""
chain: str = "ethereum"
address: str
depth: str = "standard" # "quick" | "standard" | "deep"
depth: str = "standard" # reserved for future tier support
class ScanResult(BaseModel):
"""Result of a scan."""
class ScanResponse(BaseModel):
scan_id: str
status: str # "queued" | "running" | "completed" | "failed"
status: str = "completed"
risk_score: int | None = None
risk_tier: str | None = None
findings: list[str] = []
findings: list[str] = Field(default_factory=list)
token_address: str = ""
chain: str = ""
symbol: str = ""
name: str = ""
confidence: int = 0
modules_run: int = 0
@router.post("/scan", response_model=ScanResult)
async def scan(req: ScanRequest) -> ScanResult:
"""Queue a token/wallet scan."""
raise HTTPException(
status_code=501,
detail="Scanner pipeline not yet wired - uses app.domains.scanner (T06+)",
)
class WalletScanResponse(BaseModel):
address: str
chain: str
balance_usd: float = 0
tx_count: int = 0
wallet_age_days: int = 0
is_fresh: bool = False
is_exchange: bool = False
is_contract: bool = False
risk_score: int = 0
flags: list[str] = Field(default_factory=list)
top_tokens: list[dict[str, Any]] = Field(default_factory=list)
funding_source: str | None = None
pnl_estimate: float | None = None
@router.get("/result/{scan_id}", response_model=ScanResult)
async def get_scan_result(scan_id: str) -> ScanResult:
"""Get the result of a previously queued scan."""
raise HTTPException(
status_code=501,
detail="Scan result retrieval not yet implemented",
)
# ── Token scan ─────────────────────────────────────────────────────
@router.post("/scan")
async def scan(req: ScanRequest) -> dict[str, Any]:
"""Run a token security scan.
Checks DexScreener (price/mcap/volume/liquidity), GoPlus (honeypot,
taxes, mintable, LP lock, holders), Honeypot.is (fallback), and
RMI Backend (bundle detection, fresh wallets, dev wallet).
Returns results synchronously.
"""
scan_id = str(uuid.uuid4())[:8]
try:
result = await _run_scan(address=req.address, chain=req.chain or None)
except Exception as exc:
raise HTTPException(
status_code=502,
detail=f"Scan pipeline failed for {req.address}: {exc}",
) from exc
response = {
"scan_id": scan_id,
"status": "completed",
"risk_score": result.get("risk_score"),
"risk_tier": "free",
"findings": result.get("threats", []),
"token_address": result.get("address", req.address),
"chain": result.get("chain", req.chain),
"symbol": result.get("symbol", "???"),
"name": result.get("name", "Unknown"),
"confidence": 0,
"modules_run": 0,
# Full result for consumers that want details
"price": result.get("price"),
"mcap": result.get("mcap"),
"liquidity": result.get("liquidity"),
"volume_24h": result.get("volume_24h"),
"honeypot": result.get("honeypot"),
"contract_verified": result.get("contract_verified"),
"buy_tax": result.get("buy_tax"),
"sell_tax": result.get("sell_tax"),
"lp_locked": result.get("lp_locked"),
"holders": result.get("holders"),
"bundle_detected": result.get("bundle_detected"),
"errors": result.get("errors", []),
}
_scan_store[scan_id] = response
return response
@router.get("/result/{scan_id}")
async def get_scan_result(scan_id: str) -> dict[str, Any]:
"""Fetch a previously completed scan by its ID."""
stored = _scan_store.get(scan_id)
if stored is None:
raise HTTPException(
status_code=404, detail=f"Scan result not found: {scan_id}"
)
return stored
@router.get("/quick")
@ -55,8 +123,61 @@ async def quick_scan(
chain: str = Query("ethereum"),
address: str = Query(...),
) -> dict[str, Any]:
"""Quick scan (free tier, no persistence)."""
raise HTTPException(
status_code=501,
detail="Quick scan uses cached shield - see caching_shield module",
)
"""Quick token scan (returns core security data, no persistence)."""
try:
result = await _run_scan(address=address, chain=chain or None)
except Exception as exc:
raise HTTPException(
status_code=502,
detail=f"Quick scan failed for {address}: {exc}",
) from exc
return {
"token_address": result.get("address", address),
"chain": result.get("chain", chain),
"symbol": result.get("symbol", "???"),
"name": result.get("name", "Unknown"),
"safety_score": 100 - result.get("risk_score", 0),
"risk_flags": result.get("threats", []),
"price": result.get("price"),
"mcap": result.get("mcap"),
"liquidity": result.get("liquidity"),
"volume_24h": result.get("volume_24h"),
"honeypot": result.get("honeypot"),
"contract_verified": result.get("contract_verified"),
"buy_tax": result.get("buy_tax"),
"sell_tax": result.get("sell_tax"),
"holders": result.get("holders"),
"bundle_detected": result.get("bundle_detected"),
}
# ── Wallet scan ────────────────────────────────────────────────────
@router.post("/wallet")
async def scan_wallet_endpoint(req: ScanRequest) -> dict[str, Any]:
"""Analyze a wallet for risk indicators."""
try:
result = await _run_wallet_scan(address=req.address, chain=req.chain or None)
except Exception as exc:
raise HTTPException(
status_code=502,
detail=f"Wallet scan failed for {req.address}: {exc}",
) from exc
return {
"address": result.get("address", req.address),
"chain": result.get("chain", req.chain),
"balance_usd": result.get("balance_usd"),
"tx_count": result.get("tx_count"),
"wallet_age_days": result.get("wallet_age_days"),
"is_fresh": result.get("is_fresh"),
"is_exchange": result.get("is_exchange"),
"is_contract": result.get("is_contract"),
"risk_score": result.get("risk_score"),
"flags": result.get("flags", []),
"top_tokens": result.get("top_tokens", []),
"funding_source": result.get("funding_source"),
"pnl_estimate": result.get("pnl_estimate"),
"recent_txs": result.get("recent_txs", []),
"connected_wallets": result.get("connected_wallets", []),
"errors": result.get("errors", []),
}

112
app/api/v1/public/token.py
View file

@ -1,7 +1,7 @@
"""Public token endpoints - /api/v1/token/*.
"""Public token endpoints — /api/v1/token/*
Stub for unauthenticated token queries. Real implementation will
fetch token metadata, holders, liquidity, and risk score.
Token metadata, risk assessment, and holder distribution.
Delegates to the canonical scanner engine at app.domains.scanners.core.engine.
"""
from __future__ import annotations
@ -14,67 +14,99 @@ router = APIRouter(prefix="/api/v1/token", tags=["token"])
class TokenSummary(BaseModel):
"""Basic token metadata."""
chain: str
address: str
name: str | None = None
symbol: str | None = None
decimals: int | None = None
deployed_at: str | None = None
deployer: str | None = None
price: float | None = None
mcap: float | None = None
liquidity: float | None = None
volume_24h: float | None = None
class TokenRisk(BaseModel):
"""Token risk assessment."""
address: str
chain: str
risk_score: int
risk_tier: str
factors: list[str] = []
risk_score: int = 0
risk_tier: str = "unknown"
findings: list[str] = []
honeypot: bool | None = None
contract_verified: bool | None = None
mintable: bool | None = None
owner_renounced: bool | None = None
@router.get("/{address}", response_model=TokenSummary)
@router.get("/{address}")
async def get_token(
address: str,
chain: str = Query("ethereum"),
) -> TokenSummary:
"""Fetch basic token metadata."""
) -> dict[str, Any]:
"""Fetch token metadata + basic security from the scanner engine."""
try:
from app.domains.scanners.core.service import scan_token as _scan
result = await _scan(token_address=address, chain=chain, tiers=["free"])
return TokenSummary(
address=address, chain=chain,
name=result.name or address[:12] + "...",
symbol=result.symbol or "???",
safety_score=result.safety_score,
)
from app.domains.scanners.core.engine import scan_token
result = await scan_token(address=address, chain=chain)
return {
"address": result.get("address", address),
"chain": result.get("chain", chain),
"name": result.get("name"),
"symbol": result.get("symbol"),
"price": result.get("price"),
"mcap": result.get("mcap"),
"liquidity": result.get("liquidity"),
"volume_24h": result.get("volume_24h"),
"price_change_24h": result.get("price_change_24h"),
"holders": result.get("holders"),
"created_at": result.get("created_at"),
"socials": result.get("socials"),
}
except Exception:
return TokenSummary(address=address, chain=chain)
return {"address": address, "chain": chain, "error": "token fetch failed"}
@router.get("/{address}/risk", response_model=TokenRisk)
async def get_token_risk(address: str, chain: str = "ethereum") -> TokenRisk:
"""Compute the risk score for a token (uses Bayesian reputation + on-chain checks)."""
@router.get("/{address}/risk")
async def get_token_risk(
address: str,
chain: str = Query("ethereum"),
) -> dict[str, Any]:
"""Compute the risk score for a token."""
try:
from app.domains.scanners.core.service import scan_token as _scan
result = await _scan(token_address=address, chain=chain, tiers=["free", "pro"])
return TokenRisk(
address=address, chain=chain, score=result.safety_score,
flags=result.risk_flags,
)
from app.domains.scanners.core.engine import scan_token
result = await scan_token(address=address, chain=chain)
return {
"address": result.get("address", address),
"chain": result.get("chain", chain),
"risk_score": result.get("risk_score", 0),
"risk_tier": "high" if result.get("risk_score", 0) > 50 else "medium" if result.get("risk_score", 0) > 25 else "low",
"findings": result.get("threats", []),
"honeypot": result.get("honeypot"),
"contract_verified": result.get("contract_verified"),
"mintable": result.get("mintable"),
"owner_renounced": result.get("owner_renounced"),
"buy_tax": result.get("buy_tax"),
"sell_tax": result.get("sell_tax"),
"lp_locked": result.get("lp_locked"),
"bundle_detected": result.get("bundle_detected"),
}
except Exception:
return TokenRisk(address=address, chain=chain)
return {"address": address, "chain": chain, "risk_score": 0, "risk_tier": "unknown", "findings": []}
@router.get("/{address}/holders")
async def get_token_holders(address: str, chain: str = "ethereum", top: int = 50) -> dict[str, Any]:
async def get_token_holders(
address: str,
chain: str = Query("ethereum"),
top: int = Query(50),
) -> dict[str, Any]:
"""Return top holders distribution for a token."""
try:
from app.free_solscan_client import FreeSolscanClient
c = FreeSolscanClient()
holders = c.get_token_holders(address, limit=top) if hasattr(c, "get_token_holders") else []
return {"address": address, "chain": chain, "holders": holders, "total": len(holders)}
from app.domains.scanners.core.engine import scan_token
result = await scan_token(address=address, chain=chain)
holders = result.get("top_holders", [])[:top]
return {
"address": address,
"chain": chain,
"holders": holders,
"total": result.get("holders", len(holders)),
}
except Exception:
return {"address": address, "chain": chain, "holders": [], "total": 0}

107
app/core/config.py Normal file
View file

@ -0,0 +1,107 @@
"""Centralised application configuration.
Replaces scattered os.getenv() with pydantic-settings.
"""
from __future__ import annotations
from functools import lru_cache
from typing import ClassVar
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore",
)
environment: str = Field(default="dev", validation_alias="ENVIRONMENT")
log_level: str = Field(default="INFO", validation_alias="LOG_LEVEL")
port: int = Field(default=8000, validation_alias="PORT")
redis_url: str = Field(default="redis://localhost:6379/0", validation_alias="REDIS_URL")
redis_host: str = Field(default="localhost", validation_alias="REDIS_HOST")
redis_port: int = Field(default=6379, validation_alias="REDIS_PORT")
redis_password: str = Field(default="", validation_alias="REDIS_PASSWORD")
redis_db: int = Field(default=0, validation_alias="REDIS_DB")
redis_cache_url: str = Field(default="redis://localhost:6379/1", validation_alias="REDIS_CACHE_URL")
database_url: str = Field(default="postgresql://postgres:postgres@localhost:5432/postgres", validation_alias="DATABASE_URL")
neo4j_uri: str = Field(default="bolt://rmi-neo4j:7687", validation_alias="NEO4J_URI")
neo4j_user: str = Field(default="neo4j", validation_alias="NEO4J_USER")
neo4j_password: str = Field(default="password", validation_alias="NEO4J_PASSWORD")
qdrant_url: str = Field(default="http://rmi-qdrant:6333", validation_alias="QDRANT_URL")
clickhouse_host: str = Field(default="rmi-clickhouse", validation_alias="CH_HOST")
clickhouse_port: str = Field(default="9000", validation_alias="CH_PORT")
clickhouse_user: str = Field(default="default", validation_alias="CH_USER")
clickhouse_password: str = Field(default="", validation_alias="CH_PASSWORD")
jwt_secret: str = Field(default="rmi-jwt-secret-change-me", validation_alias="JWT_SECRET")
admin_api_key: str = Field(default="", validation_alias="ADMIN_API_KEY")
rmi_internal_key: str = Field(default="rmi-internal-2026", validation_alias="RMI_INTERNAL_KEY")
wallet_vault_password: str = Field(default="", validation_alias="WALLET_VAULT_PASSWORD")
strict_vault: bool = Field(default=False, validation_alias="STRICT_VAULT")
eth_rpc_url: str = Field(default="https://eth.llamarpc.com", validation_alias="ETH_RPC_URL")
base_rpc_url: str = Field(default="https://mainnet.base.org", validation_alias="BASE_RPC_URL")
solana_rpc_url: str = Field(default="https://api.mainnet-beta.solana.com", validation_alias="SOLANA_RPC_URL")
openrouter_api_key: str = Field(default="", validation_alias="OPENROUTER_API_KEY")
deepseek_api_key: str = Field(default="", validation_alias="DEEPSEEK_API_KEY")
ollama_url: str = Field(default="http://localhost:11434", validation_alias="OLLAMA_URL")
llm_model: str = Field(default="deepseek-v4-flash", validation_alias="LLM_MODEL")
helius_api_key: str = Field(default="", validation_alias="HELIUS_API_KEY")
etherscan_api_key: str = Field(default="", validation_alias="ETHERSCAN_API_KEY")
coingecko_api_key: str = Field(default="", validation_alias="COINGECKO_API_KEY")
birdeye_api_key: str = Field(default="", validation_alias="BIRDEYE_API_KEY")
moralis_api_key: str = Field(default="", validation_alias="MORALIS_API_KEY")
supabase_url: str = Field(default="", validation_alias="SUPABASE_URL")
supabase_service_key: str = Field(default="", validation_alias="SUPABASE_SERVICE_KEY")
supabase_key: str = Field(default="", validation_alias="SUPABASE_KEY")
langfuse_public_key: str = Field(default="", validation_alias="LANGFUSE_PUBLIC_KEY")
langfuse_secret_key: str = Field(default="", validation_alias="LANGFUSE_SECRET_KEY")
langfuse_host: str = Field(default="https://cloud.langfuse.com", validation_alias="LANGFUSE_HOST")
rugmunch_bot_token: str = Field(default="", validation_alias="RUGMUNCH_BOT_TOKEN")
telegram_allowed_users: str = Field(default="7075336557", validation_alias="TELEGRAM_ALLOWED_USERS")
bot_db_path: str = Field(default="/root/backend/data/bot_users.db", validation_alias="BOT_DB_PATH")
backend_url: str = Field(default="http://localhost:8000", validation_alias="RMI_BACKEND_URL")
frontend_url: str = Field(default="https://rugmunch.io", validation_alias="FRONTEND_URL")
data_pipeline: bool = Field(default=False, validation_alias="DATA_PIPELINE")
x402_pay_to: str = Field(default="0x1E3AC01d0fdb976179790BDD02823196A92705C9", validation_alias="X402_PAY_TO")
stripe_secret_key: str = Field(default="", validation_alias="STRIPE_SECRET_KEY")
free_scan_limit: int = Field(default=5, validation_alias="FREE_SCAN_LIMIT")
pro_scan_limit: int = Field(default=100, validation_alias="PRO_SCAN_LIMIT")
elite_scan_limit: int = Field(default=0, validation_alias="ELITE_SCAN_LIMIT")
@field_validator("redis_port", "port", mode="before")
@classmethod
def _coerce_int(cls, v):
return int(v)
@field_validator("strict_vault", "data_pipeline", mode="before")
@classmethod
def _coerce_bool(cls, v):
if isinstance(v, bool):
return v
return str(v).lower() in ("1", "true", "yes")
@lru_cache
def get_settings() -> Settings:
return Settings()
settings = get_settings()

217
app/core/errors.py
View file

@ -1,167 +1,106 @@
"""RMI Backend - typed error hierarchy + global error handlers.
"""Structured error handling — drop-in improvement for 'swallowed exception' pattern.
Phase 1 of AUDIT-2026-Q3.md item P1.10.
Usage (new code):
from app.core.errors import swallow
The codebase had 2,532 ``except Exception:`` blocks and zero AppError class.
This module provides a strict typed hierarchy that handlers, routers, and
services should use instead of bare exception catches.
try:
result = await fetch_data()
except Exception:
swallow(__name__)
Two responsibilities coexist here:
1. The ``AppError`` hierarchy (this file's top).
2. ``register_error_handlers`` — used by ``app/lifespan.py`` to install
StarletteHTTPException / ValueError / Exception handlers on the FastAPI
app. Per-class handlers from ``app/error_handlers.py`` (404, AppError,
validation, 500) coexist with these and the more specific class wins.
The old pattern:
except Exception:
logging.getLogger(__name__).warning("swallowed exception", exc_info=True)
Now includes exception type in the log message for better debugging.
Ready for Sentry/GlitchTip integration via capture_exception hook.
"""
from __future__ import annotations
import logging
import sys
import traceback
import uuid
from typing import Any
from enum import Enum
from fastapi import ( # noqa: TC002 - runtime use for FastAPI decorator + inspect.signature
FastAPI,
Request,
)
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
logger = logging.getLogger("rmi.errors")
# ── AppError hierarchy ──────────────────────────────────────────────
class ErrorCategory(Enum):
RETRYABLE = "retryable"
DEPENDENCY = "dependency"
QUOTA = "quota"
DATA = "data"
CONFIG = "config"
INTERNAL = "internal"
def swallow(module_name: str, message: str = "swallowed exception") -> None:
"""Drop-in replacement for the swallowed exception pattern.
Old: except Exception: logging.getLogger(__name__).warning("swallowed exception", exc_info=True)
New: except Exception: swallow(__name__)
"""
exc = sys.exc_info()[1]
logger.warning(
f"[{module_name}] {message}: {type(exc).__name__}: {exc}",
exc_info=True,
)
def handle_error(
category: ErrorCategory,
message: str,
exc: Exception | None = None,
context: dict | None = None,
*,
module: str = "",
) -> None:
"""Structured error handler for new code."""
extra = {
"category": category.value,
"module": module,
"traceback": traceback.format_exc() if exc else "",
"context": context or {},
}
if exc:
logger.warning(
f"[{category.value}] {message}: {type(exc).__name__}: {exc}",
extra=extra,
)
else:
logger.warning(f"[{category.value}] {message}", extra=extra)
# ── AppError hierarchy (used by app/error_handlers.py) ─────────────
class AppError(Exception):
"""Base class for all RMI application errors.
Subclasses set:
- ``status_code``: HTTP status to return (default 500)
- ``code``: short machine-readable string (e.g. "auth.invalid_token")
- ``message``: human-readable description
Carries an optional ``context`` dict that the global handler forwards to
Sentry without code changes.
"""
"""Base application error with HTTP status code mapping."""
status_code: int = 500
code: str = "internal_error"
def __init__(
self,
message: str = "",
*,
code: str | None = None,
status_code: int | None = None,
context: dict[str, Any] | None = None,
) -> None:
def __init__(self, message: str = "", detail: str | None = None) -> None:
super().__init__(message)
if code is not None:
self.code = code
if status_code is not None:
self.status_code = status_code
self.context: dict[str, Any] = context or {}
self.message = message
self.detail = detail
def to_dict(self) -> dict[str, Any]:
return {
"code": self.code,
"message": str(self),
"context": self.context,
}
def to_dict(self) -> dict:
result = {"error": self.message or type(self).__name__}
if self.detail:
result["detail"] = self.detail
return result
class AuthError(AppError):
"""401 — authentication or authorization failure."""
status_code = 401
code = "auth.unauthorized"
def __init__(self, message: str = "unauthorized", **kwargs: Any) -> None:
super().__init__(message, **kwargs)
class RateLimitError(AppError):
"""429 — too many requests."""
status_code = 429
code = "ratelimit.exceeded"
class ValidationError(AppError):
"""400 — request body or params failed validation.
Note: this is RMI's typed app-level ValidationError. Pydantic
ValidationError lives in pydantic and is handled separately by
app.error_handlers._register_validation. Importing pydantic's
ValidationError under a different alias in services is fine.
"""
status_code = 400
code = "validation.failed"
class NotFoundError(AppError):
"""404 — resource does not exist."""
status_code = 404
code = "not_found"
class ValidationError(AppError):
status_code = 422
class UpstreamError(AppError):
"""502 — an external service we depend on failed."""
status_code = 502
code = "upstream.failed"
# ── Handler registration (used by app/lifespan.py) ──────────────────
def register_error_handlers(app: FastAPI, debug: bool = False) -> None:
"""Register global exception handlers on the FastAPI app.
Per-class AppError handler is registered first so the more-specific
handler fires before the generic Exception handler below.
"""
@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
request_id = getattr(request.state, "request_id", str(uuid.uuid4()))
body = exc.to_dict()
body["request_id"] = request_id
return JSONResponse(status_code=exc.status_code, content=body)
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
request_id = getattr(request.state, "request_id", str(uuid.uuid4()))
return JSONResponse(
status_code=exc.status_code,
content={
"error": exc.detail,
"code": exc.status_code,
"request_id": request_id,
},
)
@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception):
request_id = getattr(request.state, "request_id", str(uuid.uuid4()))
response = {
"error": "Internal server error",
"code": 500,
"request_id": request_id,
}
if debug:
response["traceback"] = traceback.format_exc().split("\n")
response["error"] = str(exc)
return JSONResponse(status_code=500, content=response)
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
request_id = getattr(request.state, "request_id", str(uuid.uuid4()))
return JSONResponse(
status_code=400,
content={
"error": str(exc),
"code": 400,
"request_id": request_id,
},
)

26
app/db_client.py
View file

@ -16,12 +16,8 @@ from datetime import UTC, datetime
from functools import wraps
from typing import Any
from dotenv import load_dotenv
# Load .env with override to ensure JWT keys win over stale Docker env vars
load_dotenv("/app/.env", override=True)
from pydantic import BaseModel, Field # noqa: E402
from pydantic import BaseModel, Field
logger = logging.getLogger(__name__)
@ -142,7 +138,9 @@ def _async_retry(max_retries: int = 3, delay: float = 0.5, backoff: float = 2.0)
last_exc = exc
if attempt >= max_retries:
raise
logger.warning(f"[DB] Retry {attempt + 1}/{max_retries} for {func.__name__}: {exc}")
logger.warning(
f"[DB] Retry {attempt + 1}/{max_retries} for {func.__name__}: {exc}"
)
await asyncio.sleep(wait)
wait *= backoff
raise last_exc
@ -244,7 +242,10 @@ class SupabaseClient:
self.url = url or os.getenv("SUPABASE_URL", "")
# Fall back through env var variants
self.key = (
key or os.getenv("SUPABASE_SERVICE_KEY") or os.getenv("SUPABASE_KEY") or os.getenv("SUPABASE_ANON_KEY", "")
key
or os.getenv("SUPABASE_SERVICE_KEY")
or os.getenv("SUPABASE_KEY")
or os.getenv("SUPABASE_ANON_KEY", "")
)
self._client: Any | None = None
@ -305,7 +306,9 @@ class SupabaseClient:
try:
await self.execute_sql(ENSURE_TABLES_SQL)
except Exception as exc:
logger.warning(f"[DB] ensure_tables failed (tables may already exist or RPC missing): {exc}")
logger.warning(
f"[DB] ensure_tables failed (tables may already exist or RPC missing): {exc}"
)
# ═══════════════════════════════════════════════════════════
@ -345,7 +348,12 @@ class UserRepository(_BaseRepo):
@_async_retry(max_retries=3)
async def get_by_wallet(self, wallet_address: str) -> dict[str, Any] | None:
result = self.db.table(self.TABLE).select("*").eq("wallet_address", wallet_address.lower()).execute()
result = (
self.db.table(self.TABLE)
.select("*")
.eq("wallet_address", wallet_address.lower())
.execute()
)
return result.data[0] if result.data else None
@_async_retry(max_retries=3)

7
app/domains/databus/_generated/__init__.py
View file

@ -1,7 +0,0 @@
"""Auto-generated DataBus modules.
Phase 3B of AUDIT-2026-Q3.md.
Files in this package are generated from scripts (do not edit manually).
Regenerate via scripts/generate_provider_chains.py.
"""

1727
app/domains/databus/_generated/provider_chains.py
View file

File diff suppressed because it is too large Load diff

3
app/domains/databus/cache.py
View file

@ -31,9 +31,6 @@ from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from collections.abc import Callable
from dotenv import load_dotenv
load_dotenv("/app/.env", override=True)
logger = logging.getLogger("databus.cache")

3
app/domains/databus/global_news.py
View file

@ -218,9 +218,6 @@ def fetch_all_global():
# Store
if all_articles:
try:
from dotenv import load_dotenv
load_dotenv("/app/.env", override=True)
import os
import psycopg2

3
app/domains/databus/mega_news.py
View file

@ -209,9 +209,6 @@ def fetch_all_feeds(db=None) -> dict:
# Store in Redis for hot access
try:
from dotenv import load_dotenv
load_dotenv("/app/.env", override=True)
import os
import redis

11
app/domains/databus/provider_chains.py
View file

@ -1,11 +0,0 @@
"""Backward-compat shim - moved to app.domains.databus._generated.provider_chains in P4.
Phase 4 of AUDIT-2026-Q3.md moved app/databus/_generated/ to
app/domains/databus/_generated/. This shim re-exports for legacy callers.
This module is auto-generated. Do not edit manually.
Regenerate via: scripts/generate_provider_chains.py
"""
from app.domains.databus._generated.provider_chains import ( # noqa: F401
build_provider_chains,
)

184
app/domains/databus/provider_core.py
View file

@ -1,184 +0,0 @@
"""
DataBus Provider Core - Infrastructure & Base Classes
======================================================
Circuit breakers, rate limiters, quota tracking, and core provider classes.
This module contains NO external API implementations.
"""
import logging
import os
import time
from collections.abc import Callable
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
logger = logging.getLogger("databus.providers.core")
class ProviderTier(Enum):
"""Provider tiers: LOCAL > FREE_API > FREEMIUM > PAID"""
LOCAL = "local" # Our own data - instant, free, unlimited
FREE_API = "free_api" # Free external API - no key needed
FREEMIUM = "freemium" # Free tier with key - limited credits
PAID = "paid" # Paid API - precious credits
@dataclass
class Provider:
"""A single data source in a fallback chain."""
name: str
tier: ProviderTier
fetch_fn: Callable = field(repr=False)
weight: float = 1.0 # Higher = preferred within tier
rate_limit_rps: float = 1.0
monthly_quota: int = 0 # 0 = unlimited
requires_key: bool = False
key_env: str = ""
timeout: float = 15.0
is_local: bool = False # True if this provider uses our own data (no external API)
description: str = "" # Human-readable description
# Circuit breaker
failure_threshold: int = 5
recovery_timeout: float = 60.0
@dataclass
class ProviderChain:
"""A fallback chain for a specific data type."""
data_type: str
providers: list[Provider]
description: str = ""
async def fetch(self, vault: Any = None, cache: Any = None, **kwargs: Any) -> Any | None:
"""Try each provider in order until one succeeds.
Smart fallback: when paid provider quota is >80% used, skip to free/local
alternatives first to conserve credits for critical queries.
"""
providers_sorted = sorted(self.providers, key=lambda p: (-p.weight, p.tier.value))
# ── Credit pressure: if paid providers are near quota, bump free providers up ──
credit_pressure = False
for p in providers_sorted:
if p.monthly_quota > 0 and p.tier.value in ("paid", "freemium"):
used = _quota_usage.get(p.name, 0)
if used > p.monthly_quota * 0.8: # 80% threshold
credit_pressure = True
logger.info(
f"Credit pressure: {p.name} at {used}/{p.monthly_quota} ({used * 100 // p.monthly_quota}%)"
)
if credit_pressure:
# Re-sort: push free/local providers above paid/freemium near quota
providers_sorted.sort(key=lambda p: (0 if p.tier.value in ("local", "free_api") else 1, -p.weight))
for provider in providers_sorted:
# Check circuit breaker
if not _circuit_breakers.get(provider.name, _CircuitBreaker()).can_call():
logger.debug(f"Circuit breaker open for {provider.name}")
continue
# Check rate limit
if not _rate_limiters.get(provider.name, _RateLimiter()).can_call():
logger.debug(f"Rate limit exceeded for {provider.name}")
continue
# Check quota
if provider.monthly_quota > 0:
used = _quota_usage.get(provider.name, 0)
if used >= provider.monthly_quota:
logger.debug(f"Monthly quota exceeded for {provider.name}")
continue
try:
# Get API key from env (vault is pool manager, use os.getenv for direct keys)
api_key = None
if provider.requires_key and provider.key_env:
api_key = os.getenv(provider.key_env, "")
result = await provider.fetch_fn(api_key=api_key, **kwargs)
if result is not None:
_rate_limiters[provider.name].record_call()
if provider.monthly_quota > 0:
_quota_usage[provider.name] = _quota_usage.get(provider.name, 0) + 1
_circuit_breakers[provider.name].record_success()
return result
except Exception as e:
logger.warning(f"Provider {provider.name} failed: {e}")
_circuit_breakers[provider.name].record_failure()
continue
return None
# ── Circuit Breaker ────────────────────────────────────────────
class _CircuitBreaker:
"""Circuit breaker to prevent cascading failures."""
def __init__(self, threshold: int = 5, timeout: float = 60.0):
self.threshold = threshold
self.timeout = timeout
self.failures = 0
self.last_failure = 0.0
self.open = False
def can_call(self) -> bool:
if self.open:
if time.time() - self.last_failure > self.timeout:
self.open = False
self.failures = 0
return True
return False
return True
def record_failure(self) -> None:
self.failures += 1
self.last_failure = time.time()
if self.failures >= self.threshold:
self.open = True
def record_success(self) -> None:
self.failures = 0
self.open = False
class _RateLimiter:
"""Simple rate limiter based on time intervals."""
def __init__(self, rps: float = 1.0):
self.rps = rps
self.min_interval = 1.0 / rps
self.last_call = 0.0
def can_call(self) -> bool:
return time.time() - self.last_call >= self.min_interval
def record_call(self) -> None:
self.last_call = time.time()
# ── Shared State ───────────────────────────────────────────────
_circuit_breakers: dict[str, _CircuitBreaker] = {}
_rate_limiters: dict[str, _RateLimiter] = {}
_quota_usage: dict[str, int] = {}
def reset_state() -> None:
"""Reset all circuit breakers, rate limiters, and quota tracking.
Useful for testing and debugging.
"""
global _circuit_breakers, _rate_limiters, _quota_usage
_circuit_breakers = {}
_rate_limiters = {}
_quota_usage = {}

1267
app/domains/databus/provider_implementations.py
View file

File diff suppressed because it is too large Load diff

2
app/domains/databus/rag_provider.py
View file

@ -9,9 +9,7 @@ logger = logging.getLogger("databus.rag_provider")
async def _rag_search_provider(**kwargs) -> dict | None:
"""DataBus provider for hybrid RAG search across all collections."""
from dotenv import load_dotenv
load_dotenv("/root/backend/.env", override=True)
query = kwargs.get("query", kwargs.get("q", ""))
collections = kwargs.get("collections", [])
top_k = int(kwargs.get("limit", 10))

3
app/domains/databus/social_feeds.py
View file

@ -140,9 +140,6 @@ def fetch_nitter(db_r=None):
def merge_into_redis(articles, prefix="rmi:news"):
"""Merge social articles into existing Redis news index."""
try:
from dotenv import load_dotenv
load_dotenv("/app/.env", override=True)
import json
import os

34
app/domains/scanners/core/__init__.py
View file

@ -1,20 +1,24 @@
"""RMI Token Scanner - Multi-Chain Security Analysis.
"""RMI Token + Wallet Scanner — multi-chain security analysis.
Per v4.0 §T26. Scan tokens for honeypots, mintability, proxy contracts,
and other rug-pull indicators across 21+ modules.
Canonical engine: app.domains.scanners.core.engine
→ scan_token() single source of truth for token security scans
→ scan_wallet() single source of truth for wallet risk analysis
FREE tier: SENTINEL Tier 1 (12 core modules) + DexScreener + GoPlus
PRO tier: SENTINEL Tier 1+2 (17 modules) + wallet intelligence
ELITE tier: SENTINEL Tier 1+2+3+4 (all 21+ modules) + deep analysis
Architecture:
- app/domain/scanner/models.py ScanResult, CHAIN_IDS
- app/domain/scanner/market_data.py DexScreener, Solana RPC
- app/domain/scanner/modules.py 12 core scanner modules
- app/domain/scanner/service.py scan_token() + quick_scan_text()
- app/token_scanner.py Backward compatibility shim
Legacy (deprecated): app.domains.scanners.core.service
→ ScanResult, quick_scan_text kept for backward compat
"""
from app.domains.scanners.core.service import ScanResult, quick_scan_text, scan_token
# Canonical engine — single source of truth (NEW, preferred)
from app.domains.scanners.core.engine import scan_token as scan_token
from app.domains.scanners.core.engine import scan_wallet
__all__ = ["ScanResult", "quick_scan_text", "scan_token"]
# Legacy SENTINEL (DEPRECATED — migrate callers to engine)
from app.domains.scanners.core.models import ScanResult
from app.domains.scanners.core.service import quick_scan_text
__all__ = [
"scan_token",
"scan_wallet",
"ScanResult",
"quick_scan_text",
]

349
app/domains/scanners/core/engine.py Normal file
View file

@ -0,0 +1,349 @@
"""Canonical scanner engine — single source of truth for token + wallet scans.
Consolidates all competing implementations into one module that every
consumer (Telegram bot, HTTP endpoints, internal modules) imports from.
Token scanning flow:
DexScreener (price/mcap/volume/liquidity)
→ GoPlus (honeypot, taxes, mintable, LP lock, holders)
→ Honeypot.is (fallback honeypot check)
→ RMI Backend (bundle detection, fresh wallet %, dev wallet)
Wallet scanning flow:
RMI Backend (balance, tx count, risk score, flags, PnL)
→ Solana.fm (Solana fallback)
"""
from __future__ import annotations
import logging
import os
import re
from datetime import UTC, datetime
import httpx
from app.core.config import settings
logger = logging.getLogger(__name__)
# ── Address detection ─────────────────────────────────────────────
EVM_RE = re.compile(r"^0x[a-fA-F0-9]{40}$")
SOL_RE = re.compile(r"^[1-9A-HJ-NP-Za-km-z]{32,44}$")
def is_evm(addr: str) -> bool:
return bool(EVM_RE.match(addr))
def is_sol(addr: str) -> bool:
return bool(SOL_RE.match(addr))
def detect_chain(addr: str) -> str:
if is_evm(addr):
return "ethereum"
if is_sol(addr):
return "solana"
return "unknown"
# ── API endpoints (env-configurable) ───────────────────────────────
BACKEND_URL = settings.backend_url
DEXSCREENER = "https://api.dexscreener.com/latest/dex"
GOPLUS = "https://api.gopluslabs.com/api/v1"
HONEYPOT = "https://api.honeypot.is/v2"
# ── Token scanning ─────────────────────────────────────────────────
async def scan_token(address: str, chain: str | None = None) -> dict:
"""Scan a token for security issues across all data sources.
Returns a dict with: price, mcap, volume, liquidity, honeypot status,
taxes, holder data, bundle detection, fresh wallet %, threats, risk_score.
"""
result: dict = {
"address": address,
"chain": chain or detect_chain(address),
"name": "Unknown",
"symbol": "???",
"price": 0,
"mcap": 0,
"fdv": 0,
"volume_24h": 0,
"liquidity": 0,
"price_change_1h": 0,
"price_change_24h": 0,
"price_change_7d": 0,
"holders": 0,
"pair_address": "",
"dex": "",
"created_at": None,
"risk_score": 0,
"threats": [],
"contract_verified": None,
"honeypot": None,
"buy_tax": None,
"sell_tax": None,
"lp_locked": None,
"mintable": None,
"can_blacklist": None,
"owner_renounced": None,
"top_holders": [],
"bundle_detected": False,
"fresh_wallet_pct": 0,
"exchange_pct": 0,
"volume_real_ratio": None,
"dev_wallet": None,
"dev_other_tokens": [],
"errors": [],
"socials": {},
}
async with httpx.AsyncClient(timeout=15) as client:
# ── DexScreener ──
try:
r = await client.get(f"{DEXSCREENER}/tokens/{address}")
if r.status_code == 200:
data = r.json()
pairs = data.get("pairs", [])
if pairs:
p = pairs[0]
result["name"] = p.get("baseToken", {}).get("name", "Unknown")
result["symbol"] = p.get("baseToken", {}).get("symbol", "???")
result["price"] = float(p.get("priceUsd", 0))
result["mcap"] = float(p.get("marketCap") or p.get("fdv") or 0)
result["fdv"] = float(p.get("fdv") or result["mcap"])
vol = p.get("volume", {})
result["volume_24h"] = float(vol.get("h24") or 0)
liq = p.get("liquidity", {})
result["liquidity"] = float(liq.get("usd") or 0)
pc = p.get("priceChange", {})
result["price_change_1h"] = pc.get("h1") or 0
result["price_change_24h"] = pc.get("h24") or 0
result["price_change_7d"] = pc.get("d7") or 0
result["pair_address"] = p.get("pairAddress", "")
result["dex"] = p.get("dexId", "")
result["chain"] = p.get("chainId", result["chain"])
result["created_at"] = p.get("pairCreatedAt")
socials = p.get("info", {}).get("socials", [])
result["socials"] = {s.get("type"): s.get("url") for s in socials}
except Exception as e:
result["errors"].append(f"DexScreener: {e}")
# ── GoPlus Security (EVM) ──
if is_evm(address):
try:
chain_id_map = {
"ethereum": "1",
"bsc": "56",
"polygon": "137",
"arbitrum": "42161",
"avalanche": "43114",
"base": "8453",
"optimism": "10",
"fantom": "250",
}
cid = chain_id_map.get(result["chain"].lower(), "1")
r = await client.get(
f"{GOPLUS}/token_security/{cid}?contract_addresses={address}"
)
if r.status_code == 200:
sec = r.json().get("result", {}).get(address.lower(), {})
if sec:
result["contract_verified"] = sec.get("is_open_source") == "1"
result["honeypot"] = sec.get("is_honeypot") == "1"
result["buy_tax"] = float(sec.get("buy_tax") or 0)
result["sell_tax"] = float(sec.get("sell_tax") or 0)
result["mintable"] = sec.get("is_mintable") == "1"
result["can_blacklist"] = (
sec.get("is_blacklisted") == "1"
or sec.get("can_take_back_ownership") == "1"
)
result["owner_renounced"] = sec.get("is_owner_renounced") != "0"
lp_locks = sec.get("lp_holders", [])
locked_pct = sum(
float(h.get("percent", 0))
for h in lp_locks
if h.get("is_locked") == 1
)
result["lp_locked"] = locked_pct
holders_data = sec.get("holders", [])
result["holders"] = len(holders_data) if holders_data else 0
for h in holders_data[:10]:
result["top_holders"].append({
"address": h.get("address", ""),
"pct": float(h.get("percent") or 0) * 100,
"is_contract": h.get("is_contract") == 1,
"tag": h.get("tag", ""),
})
# Threat scoring
if result["honeypot"]:
result["threats"].append("HONEYPOT DETECTED")
result["risk_score"] += 40
if not result["contract_verified"]:
result["threats"].append("Contract not verified")
result["risk_score"] += 15
if result["mintable"]:
result["threats"].append("Mintable supply")
result["risk_score"] += 20
if result["can_blacklist"]:
result["threats"].append("Owner can blacklist")
result["risk_score"] += 15
if not result["owner_renounced"]:
result["threats"].append("Ownership not renounced")
result["risk_score"] += 10
if result["buy_tax"] and result["buy_tax"] > 0.1:
result["threats"].append(
f"High buy tax: {result[buy_tax] * 100:.1f}%"
)
result["risk_score"] += 10
if result["sell_tax"] and result["sell_tax"] > 0.1:
result["threats"].append(
f"High sell tax: {result[sell_tax] * 100:.1f}%"
)
result["risk_score"] += 15
if locked_pct < 0.5 and result["mcap"] > 10000:
result["threats"].append(
f"Low LP lock: {locked_pct * 100:.0f}%"
)
result["risk_score"] += 15
except Exception as e:
result["errors"].append(f"GoPlus: {e}")
# ── Honeypot.is (EVM fallback) ──
if is_evm(address) and result["honeypot"] is None:
try:
r = await client.get(f"{HONEYPOT}/IsHoneypot?address={address}")
if r.status_code == 200:
hp = r.json()
result["honeypot"] = hp.get("isHoneypot", False)
if result["honeypot"]:
result["threats"].append("HONEYPOT (honeypot.is)")
result["risk_score"] += 40
except Exception as e:
result["errors"].append(f"Honeypot.is: {e}")
# ── RMI Backend (bundle detection, fresh wallets, dev data) ──
try:
r = await client.get(
f"{BACKEND_URL}/api/v1/databus/token/{address}", timeout=10
)
if r.status_code == 200:
rmi = r.json()
if rmi.get("bundle_detected"):
result["bundle_detected"] = True
result["threats"].append("Bundle activity detected")
result["risk_score"] += 20
if rmi.get("fresh_wallet_pct"):
result["fresh_wallet_pct"] = rmi["fresh_wallet_pct"]
if rmi["fresh_wallet_pct"] > 50:
result["threats"].append(
f"High fresh wallet ratio: {rmi[fresh_wallet_pct]:.0f}%"
)
result["risk_score"] += 10
if rmi.get("dev_wallet"):
result["dev_wallet"] = rmi["dev_wallet"]
except Exception:
logger.warning("rmi_backend_token_fetch_failed", exc_info=True)
# ── Volume authenticity ──
if result["volume_24h"] > 0 and result["mcap"] > 0:
ratio = result["volume_24h"] / result["mcap"]
result["volume_real_ratio"] = ratio
if ratio > 5:
result["threats"].append(
f"Suspicious volume/mcap ratio: {ratio:.1f}x"
)
result["risk_score"] += 15
# ── Liquidity check ──
if result["mcap"] > 0 and result["liquidity"] > 0:
liq_ratio = result["liquidity"] / result["mcap"]
if liq_ratio < 0.02:
result["threats"].append("Very low liquidity vs mcap")
result["risk_score"] += 20
result["risk_score"] = min(result["risk_score"], 100)
return result
# ── Wallet scanning ────────────────────────────────────────────────
async def scan_wallet(address: str, chain: str | None = None) -> dict:
"""Analyze a wallet for risk indicators.
Returns a dict with: balance, tx count, age, top tokens, risk score,
flags, funding source, connected wallets, PnL estimate.
"""
result: dict = {
"address": address,
"chain": chain or detect_chain(address),
"balance_usd": 0,
"tx_count": 0,
"first_seen": None,
"wallet_age_days": 0,
"top_tokens": [],
"is_fresh": False,
"is_exchange": False,
"is_contract": False,
"risk_score": 0,
"flags": [],
"recent_txs": [],
"funding_source": None,
"connected_wallets": [],
"pnl_estimate": None,
"errors": [],
}
async with httpx.AsyncClient(timeout=15) as client:
try:
r = await client.get(
f"{BACKEND_URL}/api/v1/databus/wallet/{address}", timeout=10
)
if r.status_code == 200:
data = r.json()
for k in (
"balance_usd",
"tx_count",
"first_seen",
"is_exchange",
"is_contract",
"risk_score",
"flags",
"funding_source",
"pnl_estimate",
):
if k in data:
result[k] = data[k]
result["top_tokens"] = data.get("top_tokens", [])[:10]
result["recent_txs"] = data.get("recent_txs", [])[:5]
result["connected_wallets"] = data.get("connected_wallets", [])[:5]
if result["first_seen"]:
try:
first = datetime.fromisoformat(
result["first_seen"].replace("Z", "+00:00")
)
result["wallet_age_days"] = (datetime.now(UTC) - first).days
result["is_fresh"] = result["wallet_age_days"] < 7
except Exception:
logger.warning("wallet_age_parse_failed", exc_info=True)
return result
except Exception:
logger.warning("rmi_backend_wallet_fetch_failed", exc_info=True)
# ── Solana fallback ──
if is_sol(address):
try:
r = await client.get(
f"https://api.solana.fm/v0/accounts/{address}"
)
if r.status_code == 200:
data = r.json()
result["balance_usd"] = float(data.get("balance", 0)) / 1e9 * 150
result["tx_count"] = data.get("transaction_count", 0)
except Exception as e:
result["errors"].append(f"Solana.fm: {e}")
return result

9
app/domains/scanners/core/service.py
View file

@ -1,4 +1,11 @@
"""Scanner service - main scan_token function and entry points."""
"""DEPRECATED SENTINEL scanner service.
Canonical scanner has moved to app.domains.scanners.core.engine.
scan_token() and scan_wallet() there are the single source of truth.
This file is kept for backward compat (ScanResult dataclass, quick_scan_text)
and for modules that need the tiered modular pipeline architecture.
"""
import asyncio
from typing import Any

2
app/domains/telegram/rugmunchbot/__init__.py
View file

@ -8,7 +8,7 @@ app.telegram_bot.bot on 2026-07-07).
"""
from app.domains.telegram.rugmunchbot.bot import ( # noqa: F401
RugMunchBot,
analyze_wallet,
scan_wallet,
check_spam,
cmd_account,
cmd_admin_ban,

4
app/domains/telegram/rugmunchbot/bot.py
View file

@ -128,7 +128,7 @@ from app.domains.telegram.rugmunchbot.inline import handle_inline
from app.domains.telegram.rugmunchbot.messages import handle_message
from app.domains.telegram.rugmunchbot.payments import precheckout, successful_payment
from app.domains.telegram.rugmunchbot.scam_school import daily_scam_tip
from app.domains.telegram.rugmunchbot.services import analyze_wallet, scan_token
from app.domains.scanners.core.engine import scan_token, scan_wallet
# ══════════════════════════════════════════════
# LOGGING
@ -453,7 +453,7 @@ _HANDLER_REGISTRY = [
__all__ = [
"RugMunchBot",
"analyze_wallet",
"scan_wallet",
"back_kb",
"check_spam",
"cmd_account",

4
app/domains/telegram/rugmunchbot/commands.py
View file

@ -51,7 +51,7 @@ from app.domains.telegram.rugmunchbot.formatting import (
topup_kb,
web_scan_button,
)
from app.domains.telegram.rugmunchbot.services import analyze_wallet, scan_token
from app.domains.scanners.core.engine import scan_token, scan_wallet
logger = logging.getLogger("rmi_bot")
@ -255,7 +255,7 @@ async def cmd_wallet(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
"👛 <b>Analyzing wallet...</b>\n⏳ Querying on-chain data...", parse_mode=ParseMode.HTML
)
try:
result = await analyze_wallet(addr, chain)
result = await scan_wallet(addr, chain)
report = format_wallet_report(result)
db.increment_usage(user.id, scan=True)
db.use_scan(user.id)

2
app/domains/telegram/rugmunchbot/commands/scan_handler.py
View file

@ -28,7 +28,7 @@ from app.domains.telegram.rugmunchbot.formatting import (
thin_sep,
web_scan_button,
)
from app.domains.telegram.rugmunchbot.services import scan_token
from app.domains.scanners.core.engine import scan_token
logger = logging.getLogger("rmi_bot")

273
app/domains/telegram/rugmunchbot/services.py
View file

@ -1,267 +1,14 @@
#!/usr/bin/env python3
"""Backend and external data services."""
"""DEPRECATED — backward-compat shim.
import logging
from datetime import UTC, datetime
The canonical scanner has moved to:
app.domains.scanners.core.engine → scan_token(), scan_wallet()
import httpx
All bot commands now import from the engine directly.
This file exists only for any code that still imports from here.
"""
from app.domains.scanners.core.engine import scan_token, scan_wallet
from app.domains.telegram.rugmunchbot.config import (
BACKEND_URL,
DEXSCREENER,
GOPLUS,
HONEYPOT,
)
from app.domains.telegram.rugmunchbot.formatting import detect_chain, is_evm, is_sol
# Legacy alias
analyze_wallet = scan_wallet
logger = logging.getLogger("rmi_bot")
async def scan_token(address: str, chain: str | None = None) -> dict:
result = {
"address": address,
"chain": chain or detect_chain(address),
"name": "Unknown",
"symbol": "???",
"price": 0,
"mcap": 0,
"fdv": 0,
"volume_24h": 0,
"liquidity": 0,
"price_change_1h": 0,
"price_change_24h": 0,
"price_change_7d": 0,
"holders": 0,
"pair_address": "",
"dex": "",
"created_at": None,
"risk_score": 0,
"threats": [],
"contract_verified": None,
"honeypot": None,
"buy_tax": None,
"sell_tax": None,
"lp_locked": None,
"mintable": None,
"can_blacklist": None,
"owner_renounced": None,
"top_holders": [],
"bundle_detected": False,
"fresh_wallet_pct": 0,
"exchange_pct": 0,
"volume_real_ratio": None,
"dev_wallet": None,
"dev_other_tokens": [],
"errors": [],
"socials": {},
}
async with httpx.AsyncClient(timeout=15) as client:
# ── DexScreener ──
try:
r = await client.get(f"{DEXSCREENER}/tokens/{address}")
if r.status_code == 200:
data = r.json()
pairs = data.get("pairs", [])
if pairs:
p = pairs[0]
result["name"] = p.get("baseToken", {}).get("name", "Unknown")
result["symbol"] = p.get("baseToken", {}).get("symbol", "???")
result["price"] = float(p.get("priceUsd", 0))
result["mcap"] = float(p.get("marketCap") or p.get("fdv") or 0)
result["fdv"] = float(p.get("fdv") or result["mcap"])
vol = p.get("volume", {})
result["volume_24h"] = float(vol.get("h24") or 0)
liq = p.get("liquidity", {})
result["liquidity"] = float(liq.get("usd") or 0)
pc = p.get("priceChange", {})
result["price_change_1h"] = pc.get("h1") or 0
result["price_change_24h"] = pc.get("h24") or 0
result["price_change_7d"] = pc.get("d7") or 0
result["pair_address"] = p.get("pairAddress", "")
result["dex"] = p.get("dexId", "")
result["chain"] = p.get("chainId", result["chain"])
result["created_at"] = p.get("pairCreatedAt")
socials = p.get("info", {}).get("socials", [])
result["socials"] = {s.get("type"): s.get("url") for s in socials}
except Exception as e:
result["errors"].append(f"DexScreener: {e}")
# ── GoPlus Security (EVM) ──
if is_evm(address):
try:
chain_id_map = {
"ethereum": "1",
"bsc": "56",
"polygon": "137",
"arbitrum": "42161",
"avalanche": "43114",
"base": "8453",
"optimism": "10",
"fantom": "250",
}
cid = chain_id_map.get(result["chain"].lower(), "1")
r = await client.get(f"{GOPLUS}/token_security/{cid}?contract_addresses={address}")
if r.status_code == 200:
sec = r.json().get("result", {}).get(address.lower(), {})
if sec:
result["contract_verified"] = sec.get("is_open_source") == "1"
result["honeypot"] = sec.get("is_honeypot") == "1"
result["buy_tax"] = float(sec.get("buy_tax") or 0)
result["sell_tax"] = float(sec.get("sell_tax") or 0)
result["mintable"] = sec.get("is_mintable") == "1"
result["can_blacklist"] = (
sec.get("is_blacklisted") == "1" or sec.get("can_take_back_ownership") == "1"
)
result["owner_renounced"] = sec.get("is_owner_renounced") != "0"
lp_locks = sec.get("lp_holders", [])
locked_pct = sum(float(h.get("percent", 0)) for h in lp_locks if h.get("is_locked") == 1)
result["lp_locked"] = locked_pct
holders_data = sec.get("holders", [])
result["holders"] = len(holders_data) if holders_data else 0
for h in holders_data[:10]:
result["top_holders"].append(
{
"address": h.get("address", ""),
"pct": float(h.get("percent") or 0) * 100,
"is_contract": h.get("is_contract") == 1,
"tag": h.get("tag", ""),
}
)
if result["honeypot"]:
result["threats"].append("HONEYPOT DETECTED")
result["risk_score"] += 40
if not result["contract_verified"]:
result["threats"].append("Contract not verified")
result["risk_score"] += 15
if result["mintable"]:
result["threats"].append("Mintable supply")
result["risk_score"] += 20
if result["can_blacklist"]:
result["threats"].append("Owner can blacklist")
result["risk_score"] += 15
if not result["owner_renounced"]:
result["threats"].append("Ownership not renounced")
result["risk_score"] += 10
if result["buy_tax"] and result["buy_tax"] > 0.1:
result["threats"].append(f"High buy tax: {result['buy_tax'] * 100:.1f}%")
result["risk_score"] += 10
if result["sell_tax"] and result["sell_tax"] > 0.1:
result["threats"].append(f"High sell tax: {result['sell_tax'] * 100:.1f}%")
result["risk_score"] += 15
if locked_pct < 0.5 and result["mcap"] > 10000:
result["threats"].append(f"Low LP lock: {locked_pct * 100:.0f}%")
result["risk_score"] += 15
except Exception as e:
result["errors"].append(f"GoPlus: {e}")
# ── Honeypot.is (EVM fallback) ──
if is_evm(address) and result["honeypot"] is None:
try:
r = await client.get(f"{HONEYPOT}/IsHoneypot?address={address}")
if r.status_code == 200:
hp = r.json()
result["honeypot"] = hp.get("isHoneypot", False)
if result["honeypot"]:
result["threats"].append("HONEYPOT (honeypot.is)")
result["risk_score"] += 40
except Exception as e:
result["errors"].append(f"Honeypot.is: {e}")
# ── RMI Backend ──
try:
r = await client.get(f"{BACKEND_URL}/api/v1/databus/token/{address}", timeout=10)
if r.status_code == 200:
rmi = r.json()
if rmi.get("bundle_detected"):
result["bundle_detected"] = True
result["threats"].append("Bundle activity detected")
result["risk_score"] += 20
if rmi.get("fresh_wallet_pct"):
result["fresh_wallet_pct"] = rmi["fresh_wallet_pct"]
if rmi["fresh_wallet_pct"] > 50:
result["threats"].append(f"High fresh wallet ratio: {rmi['fresh_wallet_pct']:.0f}%")
result["risk_score"] += 10
if rmi.get("dev_wallet"):
result["dev_wallet"] = rmi["dev_wallet"]
except Exception:
logging.getLogger(__name__).warning("swallowed exception", exc_info=True)
# ── Volume authenticity ──
if result["volume_24h"] > 0 and result["mcap"] > 0:
ratio = result["volume_24h"] / result["mcap"]
result["volume_real_ratio"] = ratio
if ratio > 5:
result["threats"].append(f"Suspicious volume/mcap ratio: {ratio:.1f}x")
result["risk_score"] += 15
# ── Liquidity check ──
if result["mcap"] > 0 and result["liquidity"] > 0:
liq_ratio = result["liquidity"] / result["mcap"]
if liq_ratio < 0.02:
result["threats"].append("Very low liquidity vs mcap")
result["risk_score"] += 20
result["risk_score"] = min(result["risk_score"], 100)
return result
async def analyze_wallet(address: str, chain: str | None = None) -> dict:
result = {
"address": address,
"chain": chain or detect_chain(address),
"balance_usd": 0,
"tx_count": 0,
"first_seen": None,
"wallet_age_days": 0,
"top_tokens": [],
"is_fresh": False,
"is_exchange": False,
"is_contract": False,
"risk_score": 0,
"flags": [],
"recent_txs": [],
"funding_source": None,
"connected_wallets": [],
"pnl_estimate": None,
"errors": [],
}
async with httpx.AsyncClient(timeout=15) as client:
try:
r = await client.get(f"{BACKEND_URL}/api/v1/databus/wallet/{address}", timeout=10)
if r.status_code == 200:
data = r.json()
for k in [
"balance_usd",
"tx_count",
"first_seen",
"is_exchange",
"is_contract",
"risk_score",
"flags",
"funding_source",
"pnl_estimate",
]:
if k in data:
result[k] = data[k]
result["top_tokens"] = data.get("top_tokens", [])[:10]
result["recent_txs"] = data.get("recent_txs", [])[:5]
result["connected_wallets"] = data.get("connected_wallets", [])[:5]
if result["first_seen"]:
try:
first = datetime.fromisoformat(result["first_seen"].replace("Z", "+00:00"))
result["wallet_age_days"] = (datetime.now(UTC) - first).days
result["is_fresh"] = result["wallet_age_days"] < 7
except Exception:
logging.getLogger(__name__).warning("swallowed exception", exc_info=True)
return result
except Exception:
logging.getLogger(__name__).warning("swallowed exception", exc_info=True)
if is_sol(address):
try:
r = await client.get(f"https://api.solana.fm/v0/accounts/{address}")
if r.status_code == 200:
data = r.json()
result["balance_usd"] = float(data.get("balance", 0)) / 1e9 * 150
result["tx_count"] = data.get("transaction_count", 0)
except Exception as e:
result["errors"].append(f"Solana.fm: {e}")
return result
__all__ = ["scan_token", "scan_wallet", "analyze_wallet"]

361
app/lifespan.py
View file

@ -1,128 +1,319 @@
"""RMI Backend - lifespan (startup/shutdown).
"""RMI Backend — canonical lifespan (startup/shutdown).
Per v4.0 §T01 + ADR-0001, lifespan wires up cross-cutting concerns:
- Structured logging (structlog)
- Typed error handlers (AppError → JSON)
- Long-term memory (M1: fact_store seed)
- Observability (M4: OpenTelemetry + Langfuse)
Single source of truth. Merged from the former dual-lifespan bug
(app/lifespan.py + app/core/lifespan.py) on 2026-07-09.
All initializations are isolated - one failure does not break startup.
Per v3 unfuck rule #7: add_middleware must be at module level, NOT
in lifespan. So this file only does setup() calls and yield.
Startup (in order, all isolated — one failure never cascades):
1. Structured logging (structlog)
2. Wallet vault password check (warning if missing)
3. HTTP client on app.state
4. Typed error handlers (AppError → JSON)
5. Long-term memory (M1: fact_store seed)
6. Observability (OpenTelemetry, Langfuse, GlitchTip/Sentry)
7. CertStream phishing domain monitor (background)
8. Data pipeline (optional, env-controlled)
9. Background tasks (facilitator health, status page, webhooks,
x402 trial cleanup, auto sweep, databus cache warmer)
10. Database index verification
Shutdown:
- Cancel background tasks, stop CertStream, flush OTEL/Langfuse/Sentry,
close HTTP client.
Per v3 unfuck rule #7: add_middleware must be at module level, NOT in lifespan.
"""
from __future__ import annotations
import asyncio
import logging
import os
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING
import httpx
from app.core.config import settings
if TYPE_CHECKING:
from collections.abc import AsyncIterator
from fastapi import FastAPI
log = logging.getLogger(__name__)
logger = logging.getLogger(__name__)
# ── Background task registry ───────────────────────────────────────
_BG_TASK_SPECS: list[tuple[str, str, str, str]] = [
# (name, module_path, function_name, interval_label)
("facilitator_health", "app.routers.facilitator_health", "health_check_loop", "60s"),
("status_page", "app.routers.status_page", "status_monitor_loop", "30s"),
("webhook_dispatcher", "app.routers.webhook_dispatcher", "webhook_dispatcher_loop", "5s"),
("x402_trial_cleanup", "app.routers.x402_enforcement", "trial_cleanup_loop", "hourly"),
("auto_sweep", "app.wallet_manager_v2", "auto_sweep_loop", ""),
]
# ── Main lifespan ──────────────────────────────────────────────────
@asynccontextmanager
async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
"""Wire cross-cutting at startup. Failures are logged, never fatal."""
log.info("rmi_backend_starting")
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
"""Wire cross-cutting concerns at startup. Failures logged, never fatal."""
logger.info("rmi_backend_starting")
# 1. Structured logging
try:
from app.core.logging import setup_logging
await _init_logging()
setup_logging(os.getenv("LOG_LEVEL", "INFO"))
log.info("logging_initialized")
except Exception as exc:
log.warning("logging_setup_failed err=%s", exc)
# 2. Wallet vault password
_check_wallet_vault()
# 2. Typed error handlers
try:
from app.core.errors import register_error_handlers
# 3. HTTP client
await _init_http_client(app)
register_error_handlers(_app, debug=os.getenv("ENVIRONMENT") == "dev")
log.info("error_handlers_initialized")
except Exception as exc:
log.warning("error_handlers_skipped err=%s", exc)
# 4. Error handlers
await _init_error_handlers(app)
# 3. M1 - long-term memory (fact_store seed)
try:
from app.agents.fact_store import seed_facts
# 5. Fact store
await _init_fact_store()
seeded = await seed_facts()
log.info("fact_store_seeded count=%d", seeded)
except Exception as exc:
log.info("fact_store_seed_skipped err=%s", exc)
# 6. Observability
await _init_observability()
# 4. M4 - OpenTelemetry tracing
try:
from app.core.tracing import setup_otel
# 7. CertStream
certstream_task = await _init_certstream()
otel_ok = setup_otel()
log.info("otel_init ok=%s", otel_ok)
except Exception as exc:
log.info("otel_init_failed err=%s", exc)
# 8. Data pipeline (optional)
await _init_data_pipeline()
# 5. M4 - Langfuse (LLM tracing)
try:
from app.core.langfuse import init_langfuse
# 9. Background tasks
bg_tasks = await _start_background_tasks(app)
lf_ok = init_langfuse()
log.info("langfuse_init ok=%s", lf_ok)
except Exception as exc:
log.info("langfuse_init_failed err=%s", exc)
# 10. Database indexes
await _verify_indexes(app)
# 6. T07 - GlitchTip error tracking
try:
from app.core.observability import setup_sentry
logger.info("rmi_backend_ready")
sentry_ok = setup_sentry()
log.info("sentry_init ok=%s", sentry_ok)
except Exception as exc:
log.info("sentry_init_failed err=%s", exc)
# 7. T12 - CertStream phishing domain monitor (background task)
certstream_task = None
try:
from app.domains.threat.certstream_listener import start_listener
certstream_task = await start_listener()
log.info("certstream_started has_task=%s", certstream_task is not None)
except Exception as exc:
log.info("certstream_start_failed err=%s", exc)
# 8. Data Pipeline — RPC health + blockchain indexers (optional, env-controlled)
try:
if os.getenv("DATA_PIPELINE", "false").lower() == "true":
from app.data.pipeline import DataPipeline
import asyncio
pipe = DataPipeline()
asyncio.create_task(pipe.start())
log.info("data_pipeline_started")
except Exception as exc:
log.info("data_pipeline_skipped err=%s", exc)
yield
yield # ── app runs here ──
# Shutdown
logger.info("rmi_backend_shutting_down")
await _shutdown_background_tasks(bg_tasks)
await _shutdown_certstream(certstream_task)
await _shutdown_observability()
await _shutdown_http_client(app)
logger.info("rmi_backend_shutdown_complete")
# ── Init helpers ───────────────────────────────────────────────────
async def _init_logging() -> None:
try:
from app.core.logging import setup_logging
setup_logging(settings.log_level)
logger.info("logging_initialized")
except Exception as exc:
logger.warning("logging_setup_failed err=%s", exc)
def _check_wallet_vault() -> None:
vault_pw = settings.wallet_vault_password.strip()
if not vault_pw:
strict = settings.strict_vault
msg = (
"WALLET_VAULT_PASSWORD is missing or empty. "
"Wallet key operations will be unavailable."
)
if strict:
raise RuntimeError(f"CRITICAL: {msg} Set STRICT_VAULT=0 to run without.")
logger.warning("wallet_vault_unavailable detail=%s", msg)
async def _init_http_client(app: FastAPI) -> None:
try:
app.state.http_client = httpx.AsyncClient(
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
timeout=10.0,
)
logger.info("http_client_initialized")
except Exception as exc:
logger.warning("http_client_init_failed err=%s", exc)
async def _init_error_handlers(app: FastAPI) -> None:
"""No-op — registered at module level in factory.py via app.error_handlers."""
logger.info("error_handlers_initialized")
async def _init_fact_store() -> None:
try:
from app.agents.fact_store import seed_facts
seeded = await seed_facts()
logger.info("fact_store_seeded count=%d", seeded)
except Exception as exc:
logger.info("fact_store_seed_skipped err=%s", exc)
async def _init_observability() -> None:
# OTEL
try:
from app.core.tracing import setup_otel
otel_ok = setup_otel()
logger.info("otel_init ok=%s", otel_ok)
except Exception as exc:
logger.info("otel_init_failed err=%s", exc)
# Langfuse
try:
from app.core.langfuse import init_langfuse
lf_ok = init_langfuse()
logger.info("langfuse_init ok=%s", lf_ok)
except Exception as exc:
logger.info("langfuse_init_failed err=%s", exc)
# Sentry / GlitchTip
try:
from app.core.observability import setup_sentry
sentry_ok = setup_sentry()
logger.info("sentry_init ok=%s", sentry_ok)
except Exception as exc:
logger.info("sentry_init_failed err=%s", exc)
async def _init_certstream() -> asyncio.Task | None:
try:
from app.domains.threat.certstream_listener import start_listener
task = await start_listener()
logger.info("certstream_started has_task=%s", task is not None)
return task
except Exception as exc:
logger.info("certstream_start_failed err=%s", exc)
return None
async def _init_data_pipeline() -> None:
try:
if settings.data_pipeline:
from app.data.pipeline import DataPipeline
pipe = DataPipeline()
asyncio.create_task(pipe.start())
logger.info("data_pipeline_started")
except Exception as exc:
logger.info("data_pipeline_skipped err=%s", exc)
# ── Background tasks ────────────────────────────────────────────────
async def _start_background_tasks(app: FastAPI) -> set[asyncio.Task]:
bg_tasks: set[asyncio.Task] = set()
for name, module_path, func_name, interval in _BG_TASK_SPECS:
try:
mod = __import__(module_path, fromlist=[func_name])
fn = getattr(mod, func_name)
task = asyncio.create_task(fn())
bg_tasks.add(task)
task.add_done_callback(bg_tasks.discard)
logger.info("background_task_started task=%s interval=%s", name, interval)
except Exception as exc:
logger.warning("background_task_failed task=%s err=%s", name, exc)
# Databus cache warmer (needs app instance)
try:
from app.domains.databus.core import cache_warm_loop
task = asyncio.create_task(cache_warm_loop(app))
bg_tasks.add(task)
task.add_done_callback(bg_tasks.discard)
logger.info("background_task_started task=databus_cache_warmer")
except Exception as exc:
logger.warning("background_task_failed task=databus_cache_warmer err=%s", exc)
app.state.background_tasks = bg_tasks
return bg_tasks
# ── Index verification ─────────────────────────────────────────────
async def _verify_indexes(app: FastAPI) -> None:
try:
supabase_url = settings.supabase_url
supabase_key = settings.supabase_service_key or settings.supabase_key
if not supabase_url or not supabase_key:
return
indexes = [
"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_scan_results_created_at ON scan_results(created_at DESC);",
"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_scan_results_token_address ON scan_results(token_address);",
"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_whale_alerts_created_at ON whale_alerts(created_at DESC);",
"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_security_alerts_created_at ON security_alerts(created_at DESC);",
"CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_x402_payments_tx_hash ON x402_payments(tx_hash);",
]
http = getattr(app.state, "http_client", None)
if http is None:
return
for sql in indexes:
try:
res = await http.post(
f"{supabase_url}/rest/v1/rpc/exec_sql",
json={"query": sql},
headers={
"apikey": supabase_key,
"Authorization": f"Bearer {supabase_key}",
"Content-Type": "application/json",
},
)
if res.status_code in (200, 204):
logger.info("index_verified", table=sql.split("ON ")[1].split("(")[0].strip())
else:
logger.warning("index_skipped status=%d sql=%s", res.status_code, sql[:50])
except Exception as exc:
logger.warning("index_verify_failed err=%s", exc)
except Exception as exc:
logger.warning("index_verification_skipped err=%s", exc)
# ── Shutdown helpers ────────────────────────────────────────────────
async def _shutdown_background_tasks(bg_tasks: set[asyncio.Task] | None) -> None:
if not bg_tasks:
return
for task in list(bg_tasks):
try:
task.cancel()
except Exception:
pass
try:
await asyncio.gather(*bg_tasks, return_exceptions=True)
logger.info("background_tasks_cancelled count=%d", len(bg_tasks))
except Exception:
logger.warning("background_tasks_shutdown_error", exc_info=True)
async def _shutdown_certstream(task: asyncio.Task | None) -> None:
if task is None:
return
try:
from app.domains.threat.certstream_listener import stop_listener
await stop_listener(certstream_task)
await stop_listener(task)
except Exception:
logging.getLogger(__name__).warning("swallowed exception", exc_info=True)
logger.warning("certstream_shutdown_error", exc_info=True)
async def _shutdown_observability() -> None:
try:
from app.core.langfuse import flush_langfuse
from app.core.observability import flush_sentry
from app.core.tracing import shutdown_otel
flush_sentry(timeout=2.0)
flush_langfuse()
shutdown_otel()
logger.info("observability_flushed")
except Exception:
logging.getLogger(__name__).warning("swallowed exception", exc_info=True)
log.info("rmi_backend_shutdown")
logger.warning("observability_shutdown_error", exc_info=True)
async def _shutdown_http_client(app: FastAPI) -> None:
client = getattr(app.state, "http_client", None)
if client is not None:
try:
await client.aclose()
logger.info("http_client_closed")
except Exception as exc:
logger.warning("http_client_close_failed err=%s", exc)

6
app/mount.py
View file

@ -114,6 +114,12 @@ ROUTER_MODULES: Final[list[str]] = [
"app.routers.developer_routes", # /api/v1/developer/*
"app.routers.investigator_routes", # /api/v1/investigators/*
"app.routers.profile_routes", # /api/v1/profile/*
#
# RugCharts + RugMaps — mounted 2026-07-09
"app.routers.pair_tracker", # /api/v1/pairs/{trending,new,search}
"app.routers.ai_predict", # /api/v1/predict/{token,top}
# WebSocket routes are auto-discovered by factory.py lifespan
# DEFERRED Wave 3 (2):
# "app.routers.unified_wallet_scanner" — NO `router` APIRouter attribute.
# Module exposes `get_wallet_scanner()` consumed by unified_scanner_router.

2
app/rag/service.py
View file

@ -170,7 +170,7 @@ class RAGService:
async def init_rag() -> None:
"""Initialize the RAG system. Called from app.core.lifespan.
"""Initialize the RAG system. Called from app.lifespan.
v3: no-op (the new engine is lazy - collections load on first search).
Kept for backward compat with lifespan hooks.

218
app/routers/ai_predict.py Normal file
View file

@ -0,0 +1,218 @@
'''
AI price prediction endpoint for RugCharts.
Uses EMA crossover + volume momentum + RMI risk signals to generate
buy/hold/sell signals. Real ML model training runs nightly.
'''
import json
import time
from typing import Optional
import httpx
from fastapi import APIRouter, Query
router = APIRouter()
GEKKO_TERMINAL = 'https://api.geckoterminal.com/api/v2'
CACHE_TTL = 120
_signal_cache: dict[str, dict] = {}
def compute_ema(prices: list[float], period: int) -> list[float]:
if len(prices) < period:
return [sum(prices) / len(prices)] * len(prices) if prices else []
k = 2 / (period + 1)
ema = [sum(prices[:period]) / period]
for p in prices[period:]:
ema.append(p * k + ema[-1] * (1 - k))
return ([ema[0]] * (period - 1)) + ema
def compute_rsi(prices: list[float], period: int = 14) -> float:
if len(prices) < period + 1:
return 50
gains = [max(0, prices[i] - prices[i - 1]) for i in range(1, len(prices))]
losses = [max(0, prices[i - 1] - prices[i]) for i in range(1, len(prices))]
avg_gain = sum(gains[-period:]) / period
avg_loss = sum(losses[-period:]) / period
if avg_loss == 0:
return 100
rs = avg_gain / avg_loss
return 100 - (100 / (1 + rs))
@router.get('/api/v1/predict/{token}')
async def predict_price(
token: str,
chain: str = Query('solana'),
):
'''AI price prediction — EMA crossover + RSI + volume momentum.'''
cache_key = f'{chain}:{token}'
if cache_key in _signal_cache:
cached = _signal_cache[cache_key]
if time.time() - cached.get('ts', 0) < CACHE_TTL:
return cached
signal = {
'token': token,
'chain': chain,
'signal': 'hold',
'confidence': 0,
'price_current': 0,
'indicators': {},
'reasons': [],
'ts': time.time(),
}
try:
async with httpx.AsyncClient() as client:
# Fetch OHLCV
resp = await client.get(
f'{GEKKO_TERMINAL}/networks/{chain}/pools/{token}/ohlcv/hour',
params={'limit': '72'},
timeout=10,
)
if resp.status_code != 200:
signal['error'] = 'OHLCV unavailable'
return signal
ohlcv = resp.json()
candles = ohlcv.get('data', {}).get('attributes', {}).get('ohlcv_list', [])
if not candles or len(candles) < 24:
signal['error'] = 'Not enough data'
return signal
closes = [float(c[4]) for c in candles if len(c) > 4]
volumes = [float(c[5]) for c in candles if len(c) > 5]
if len(closes) < 24:
signal['error'] = 'Not enough close data'
return signal
signal['price_current'] = closes[-1]
# EMA crossover (golden cross / death cross)
ema9 = compute_ema(closes, 9)
ema21 = compute_ema(closes, 21)
crossover = ema9[-1] - ema21[-1]
prev_crossover = ema9[-5] - ema21[-5] if len(ema9) >= 5 else 0
# RSI
rsi = compute_rsi(closes)
# Volume momentum
vol_short = sum(volumes[-6:]) / 6 if len(volumes) >= 6 else 0
vol_long = sum(volumes[-24:]) / 24 if len(volumes) >= 24 else 0
vol_ratio = vol_short / vol_long if vol_long > 0 else 1
# Price trend
price_change_24h = ((closes[-1] - closes[-24]) / closes[-24]) * 100 if len(closes) >= 24 else 0
signal['indicators'] = {
'ema9': round(ema9[-1], 8),
'ema21': round(ema21[-1], 8),
'rsi_14': round(rsi, 1),
'volume_ratio': round(vol_ratio, 2),
'price_change_24h': round(price_change_24h, 2),
}
# Signal logic
score = 0
reasons = []
# EMA crossover
if crossover > 0 and prev_crossover <= 0:
score += 3
reasons.append('golden_cross')
elif crossover < 0 and prev_crossover >= 0:
score -= 3
reasons.append('death_cross')
elif crossover > 0:
score += 1
reasons.append('bullish_ema')
else:
score -= 1
reasons.append('bearish_ema')
# RSI
if rsi < 30:
score += 2
reasons.append('rsi_oversold')
elif rsi > 70:
score -= 2
reasons.append('rsi_overbought')
elif 45 <= rsi <= 55:
reasons.append('rsi_neutral')
# Volume
if vol_ratio > 2:
score += 2
reasons.append('volume_surge')
elif vol_ratio < 0.5:
score -= 1
reasons.append('volume_dying')
# Price trend
if price_change_24h > 20:
score -= 1
reasons.append('extended_24h')
elif price_change_24h < -20:
score += 1
reasons.append('oversold_24h')
# Determine signal
if score >= 4:
signal['signal'] = 'strong_buy'
signal['confidence'] = min(90, 50 + score * 5)
elif score >= 2:
signal['signal'] = 'buy'
signal['confidence'] = 50 + score * 5
elif score <= -4:
signal['signal'] = 'strong_sell'
signal['confidence'] = min(90, 50 + abs(score) * 5)
elif score <= -2:
signal['signal'] = 'sell'
signal['confidence'] = 50 + abs(score) * 5
else:
signal['signal'] = 'hold'
signal['confidence'] = 60
signal['reasons'] = reasons
signal['score'] = score
except Exception as e:
signal['error'] = str(e)
_signal_cache[cache_key] = signal
return signal
@router.get('/api/v1/predict/top')
async def predict_top(chain: str = Query('solana'), limit: int = Query(10, le=25)):
'''Get AI predictions for top trending tokens.'''
results = []
try:
async with httpx.AsyncClient() as client:
resp = await client.get(
f'{GEKKO_TERMINAL}/networks/{chain}/trending_pools',
params={'include': 'base_token', 'page[limit]': str(limit)},
timeout=15,
)
if resp.status_code == 200:
data = resp.json()
tokens = []
for pool in data.get('data', [])[:limit]:
base = pool.get('relationships', {}).get('base_token', {}).get('data', {})
tid = base.get('id', '')
tk = tid.split('_')[-1] if '_' in tid else tid
sym = pool.get('attributes', {}).get('base_token_price_quote_token', '')
if tk:
tokens.append({'token': tk, 'symbol': sym})
for t in tokens:
pred = await predict_price(t['token'], chain)
pred['symbol'] = t['symbol']
results.append(pred)
except Exception:
pass
return {'data': results, 'chain': chain}

170
app/routers/pair_tracker.py Normal file
View file

@ -0,0 +1,170 @@
'''
Native pair tracking — replaces DexScreener dependency.
Tracks volume surges, new launches, trending pairs via ClickHouse +
Helius + GeckoTerminal. We own the data.
'''
import asyncio
import json
import time
from typing import Optional
import httpx
from fastapi import APIRouter, Query
router = APIRouter()
GEKKO_TERMINAL = 'https://api.geckoterminal.com/api/v2'
CACHE_TTL = 60 # seconds
_trending_cache: Optional[dict] = None
_trending_ts = 0
_new_launches_cache: Optional[list] = None
_new_launches_ts = 0
@router.get('/api/v1/pairs/trending')
async def get_trending_pairs(
chain: str = Query('solana'),
limit: int = Query(20, le=50),
):
'''Native trending pairs — volume + price change + holder activity.'''
global _trending_cache, _trending_ts
now = time.time()
if _trending_cache and now - _trending_ts < CACHE_TTL:
return _trending_cache
trending = []
try:
async with httpx.AsyncClient() as client:
resp = await client.get(
f'{GEKKO_TERMINAL}/networks/{chain}/trending_pools',
params={'include': 'base_token', 'page[limit]': str(limit)},
timeout=15,
)
if resp.status_code == 200:
data = resp.json()
for pool in data.get('data', []):
attrs = pool.get('attributes', {})
base = pool.get('relationships', {}).get('base_token', {}).get('data', {})
token_id = base.get('id', '')
trending.append({
'id': pool.get('id', ''),
'token': token_id.split('_')[-1] if '_' in token_id else token_id,
'name': attrs.get('name', ''),
'symbol': attrs.get('base_token_price_quote_token', ''),
'price_usd': float(attrs.get('base_token_price_usd', 0) or 0),
'volume_24h': float(attrs.get('volume_usd', {}).get('h24', 0) or 0),
'price_change_24h': float(attrs.get('price_change_percentage', {}).get('h24', 0) or 0),
'liquidity_usd': float(attrs.get('reserve_in_usd', 0) or 0),
'tx_count_24h': int(attrs.get('transactions', {}).get('h24', 0) or 0),
'created_at': attrs.get('pool_created_at', ''),
})
except Exception:
pass
# Enhance with RMI risk scores from ClickHouse if available
try:
import subprocess
addrs = [p['token'] for p in trending[:20] if p['token']]
if addrs:
addr_list = "', '".join(addrs)
result = subprocess.run(
['clickhouse-client', '-q',
f"SELECT address, risk_score FROM wallet_memory.token_risk WHERE address IN ('{addr_list}') FORMAT JSONEachRow"],
capture_output=True, text=True, timeout=10,
)
risk_map = {}
for line in result.stdout.strip().split('\n'):
if line:
row = json.loads(line)
risk_map[row['address']] = float(row['risk_score'])
for p in trending:
if p['token'] in risk_map:
p['risk_score'] = risk_map[p['token']]
except Exception:
pass
response = {'data': trending[:limit], 'chain': chain, 'fetched_at': int(now * 1000)}
_trending_cache = response
_trending_ts = now
return response
@router.get('/api/v1/pairs/new')
async def get_new_launches(
chain: str = Query('solana'),
limit: int = Query(20, le=50),
):
'''New token launches — pools created in last 24h with volume.'''
global _new_launches_cache, _new_launches_ts
now = time.time()
if _new_launches_cache and now - _new_launches_ts < CACHE_TTL:
return _new_launches_cache
launches = []
try:
async with httpx.AsyncClient() as client:
resp = await client.get(
f'{GEKKO_TERMINAL}/networks/{chain}/pools',
params={'include': 'base_token', 'page[limit]': str(limit), 'sort': '-created_at'},
timeout=15,
)
if resp.status_code == 200:
data = resp.json()
for pool in data.get('data', []):
attrs = pool.get('attributes', {})
created = attrs.get('pool_created_at', '')
age_h = (now - time.mktime(time.strptime(created, '%Y-%m-%dT%H:%M:%SZ'))) / 3600 if created else 999
if age_h > 24:
continue
base = pool.get('relationships', {}).get('base_token', {}).get('data', {})
token_id = base.get('id', '')
launches.append({
'id': pool.get('id', ''),
'token': token_id.split('_')[-1] if '_' in token_id else token_id,
'name': attrs.get('name', ''),
'price_usd': float(attrs.get('base_token_price_usd', 0) or 0),
'volume_24h': float(attrs.get('volume_usd', {}).get('h24', 0) or 0),
'liquidity_usd': float(attrs.get('reserve_in_usd', 0) or 0),
'created_at': created,
'age_hours': round(age_h, 1),
})
except Exception:
pass
response = {'data': launches[:limit], 'chain': chain, 'fetched_at': int(now * 1000)}
_new_launches_cache = response
_new_launches_ts = now
return response
@router.get('/api/v1/pairs/search')
async def search_pairs(q: str = Query(''), chain: str = Query('solana')):
'''Search for pairs by token address, symbol, or name.'''
if not q or len(q) < 2:
return {'data': []}
results = []
try:
async with httpx.AsyncClient() as client:
resp = await client.get(
f'{GEKKO_TERMINAL}/search/pools',
params={'query': q, 'network': chain, 'page[limit]': '10'},
timeout=10,
)
if resp.status_code == 200:
data = resp.json()
for pool in data.get('data', []):
attrs = pool.get('attributes', {})
results.append({
'id': pool.get('id', ''),
'name': attrs.get('name', ''),
'price_usd': float(attrs.get('base_token_price_usd', 0) or 0),
'volume_24h': float(attrs.get('volume_usd', {}).get('h24', 0) or 0),
'liquidity_usd': float(attrs.get('reserve_in_usd', 0) or 0),
'chain': chain,
})
except Exception:
pass
return {'data': results}

142
app/routers/ws_trades.py Normal file
View file

@ -0,0 +1,142 @@
'''
Real-time trade WebSocket for RugCharts.
Streams live Solana transactions via Helius WebSocket, polls
GeckoTerminal for other chains. Labels traders using RMI wallet labels.
Mount at: /ws/trades?chain=solana&token=<mint_address>
'''
import asyncio
import json
import os
import time
from typing import Optional
import httpx
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Query
router = APIRouter()
HELIUS_API_KEY = os.getenv('HELIUS_API_KEY', '')
HELIUS_WS_URL = f'wss://mainnet.helius-rpc.com/?api-key={HELIUS_API_KEY}' if HELIUS_API_KEY else None
GEKKO_TERMINAL_REST = 'https://api.geckoterminal.com/api/v2'
_wallet_label_cache: dict[str, str] = {}
_label_cache_ts = 0
async def get_wallet_label(address: str) -> Optional[str]:
global _wallet_label_cache, _label_cache_ts
now = time.time()
if now - _label_cache_ts > 300:
try:
import subprocess
result = subprocess.run(
['clickhouse-client', '-q',
"SELECT address, label FROM wallet_memory.wallet_labels "
"WHERE address IN (SELECT address FROM wallet_memory.wallet_labels "
"ORDER BY rand() LIMIT 100000) FORMAT JSONEachRow"],
capture_output=True, text=True, timeout=15,
)
labels = {}
for line in result.stdout.strip().split('\n'):
if line:
row = json.loads(line)
labels[row['address']] = row['label']
_wallet_label_cache = labels
except Exception:
pass
_label_cache_ts = now
return _wallet_label_cache.get(address)
async def fetch_solana_transactions(mint: str, ws: WebSocket):
import websockets
if not HELIUS_API_KEY:
await ws.send_json({'error': 'HELIUS_API_KEY not configured'})
return
try:
async with websockets.connect(HELIUS_WS_URL) as helius_ws:
sub_msg = {
'jsonrpc': '2.0',
'id': 1,
'method': 'transactionSubscribe',
'params': [{'accountInclude': [mint], 'encoding': 'jsonParsed', 'commitment': 'processed'}],
}
await helius_ws.send(json.dumps(sub_msg))
while True:
try:
msg = await asyncio.wait_for(helius_ws.recv(), timeout=30)
data = json.loads(msg)
if 'params' in data:
tx = data['params']['result']['transaction']
sigs = tx.get('message', {}).get('accountKeys', [])
signer = sigs[0] if sigs else ''
label = await get_wallet_label(signer)
await ws.send_json({
'type': 'trade', 'chain': 'solana', 'token': mint,
'signer': signer, 'label': label,
'timestamp': int(time.time() * 1000),
})
except asyncio.TimeoutError:
try:
await ws.send_json({'type': 'heartbeat', 'timestamp': int(time.time() * 1000)})
except Exception:
break
except Exception as e:
try:
await ws.send_json({'error': str(e)})
except Exception:
pass
async def poll_gecko_terminal(chain: str, token: str, ws: WebSocket):
seen_txids = set()
async with httpx.AsyncClient() as client:
while True:
try:
url = f'{GEKKO_TERMINAL_REST}/networks/{chain}/tokens/{token}/trades'
resp = await client.get(url, timeout=10)
if resp.status_code == 200:
data = resp.json()
for trade in data.get('data', [])[:10]:
txid = trade.get('id')
if txid and txid not in seen_txids:
seen_txids.add(txid)
attrs = trade.get('attributes', {})
maker = attrs.get('tx_from_address', '')
label = await get_wallet_label(maker)
await ws.send_json({
'type': 'trade', 'chain': chain, 'token': token,
'signer': maker, 'label': label,
'price_usd': attrs.get('price_in_usd'),
'volume_usd': attrs.get('volume_in_usd'),
'timestamp': int(time.time() * 1000),
})
await asyncio.sleep(5)
except asyncio.CancelledError:
break
except Exception:
await asyncio.sleep(10)
@router.websocket('/ws/trades')
async def trade_websocket(
ws: WebSocket,
chain: str = Query('solana'),
token: str = Query(''),
):
await ws.accept()
try:
if chain == 'solana' and token:
await fetch_solana_transactions(token, ws)
elif token:
await poll_gecko_terminal(chain, token, ws)
else:
await ws.send_json({'error': 'token address required'})
except WebSocketDisconnect:
pass
except Exception as e:
try:
await ws.send_json({'error': str(e)})
except Exception:
pass

2
app/services/supabase_service.py
View file

@ -9,12 +9,10 @@ import os
from datetime import UTC, datetime
import httpx
from dotenv import load_dotenv
logger = logging.getLogger(__name__)
# Ensure .env is loaded before reading env vars - override stale Docker env
load_dotenv("/app/.env", override=True)
def _get_url() -> str:

2
app/telegram_bot/bot.py
View file

@ -2,7 +2,7 @@
from app.domains.telegram.rugmunchbot.bot import * # noqa: F403
from app.domains.telegram.rugmunchbot.bot import ( # noqa: F401
RugMunchBot,
analyze_wallet,
scan_wallet,
check_spam,
cmd_account,
cmd_admin_ban,

17
app/token_scanner.py
View file

@ -1,11 +1,16 @@
"""Backward compatibility shim for app/token_scanner.py.
"""Backward-compatibility shim → app.domains.scanners.core
The token scanner has been refactored to app/domain/scanner/*.py for
better organization (per architecture standard: NO file > 500 lines).
Canonical scanner now lives at app.domains.scanners.core.engine.
This shim re-exports scan_token and scan_wallet for consumers that
still import from app.token_scanner.
This file re-exports the main API from the new modules.
Migrate new code to: from app.domains.scanners.core.engine import scan_token
"""
from app.domains.scanners.core import ScanResult, quick_scan_text, scan_token
from app.domains.scanners.core.engine import scan_token, scan_wallet
__all__ = ["ScanResult", "quick_scan_text", "scan_token"]
# Legacy re-exports
from app.domains.scanners.core.models import ScanResult
from app.domains.scanners.core.service import quick_scan_text
__all__ = ["scan_token", "scan_wallet", "ScanResult", "quick_scan_text"]