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>
1138 lines
43 KiB
Python
1138 lines
43 KiB
Python
"""
|
|
Cross-Chain Portfolio Risk Aggregator
|
|
======================================
|
|
Aggregates a wallet's token holdings across ALL supported chains and produces
|
|
a unified risk profile - the free, comprehensive alternative to Nansen Portfolio.
|
|
|
|
How it works:
|
|
1. Scans EVM chains (Ethereum, Base, BSC, Arbitrum, Polygon, Optimism, etc.)
|
|
via web3.py for native balance + ERC20 holdings
|
|
2. Scans Solana via solana-py for SPL token holdings
|
|
3. Runs each token through the Unified Scanner risk engine
|
|
4. Aggregates into a PortfolioRiskProfile with:
|
|
- Chain-by-chain breakdown of total value, risk scores
|
|
- Per-token risk scoring with scam/honeypot flags
|
|
- Concentration risk (single token/chains exposure)
|
|
- Overall portfolio health score (0-100)
|
|
5. Generates human-readable risk report
|
|
|
|
Standalone usage:
|
|
python3 portfolio_risk_aggregator.py <wallet_address>
|
|
python3 portfolio_risk_aggregator.py <wallet_address> --chains ethereum,base,solana
|
|
|
|
API usage:
|
|
from app.portfolio_risk_aggregator import PortfolioRiskAggregator
|
|
agg = PortfolioRiskAggregator()
|
|
profile = await agg.analyze("0x...")
|
|
print(profile.report())
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import re
|
|
import sys
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import UTC, datetime
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ─── TRY IMPORTS (graceful fallback) ───────────────────────────────
|
|
|
|
try:
|
|
from web3 import HTTPProvider
|
|
from web3 import Web3 as _Web3
|
|
|
|
WEB3_AVAILABLE = True
|
|
except ImportError:
|
|
WEB3_AVAILABLE = False
|
|
_Web3 = None
|
|
HTTPProvider = None
|
|
|
|
try:
|
|
from solana.rpc.api import Client as SolanaClient
|
|
|
|
SOLANA_AVAILABLE = True
|
|
except ImportError:
|
|
SOLANA_AVAILABLE = False
|
|
SolanaClient = None
|
|
|
|
try:
|
|
import httpx
|
|
|
|
HTTPX_AVAILABLE = True
|
|
except ImportError:
|
|
HTTPX_AVAILABLE = False
|
|
|
|
# ─── CONSTANTS ─────────────────────────────────────────────────────
|
|
|
|
DEFAULT_TIMEOUT = 15 # seconds per chain scan
|
|
MAX_TOKENS_PER_WALLET = 100 # max tokens to analyze per wallet
|
|
RISK_WEIGHT_TOKENS = 0.6 # weight for token-level risk in portfolio score
|
|
RISK_WEIGHT_CONCENTRATION = 0.25
|
|
RISK_WEIGHT_CHAIN_DIVERSITY = 0.15
|
|
|
|
# ERC20 ABI (minimal - balanceOf + decimals + symbol)
|
|
ERC20_ABI = json.loads("""
|
|
[
|
|
{"constant":true,"inputs":[{"name":"_owner","type":"address"}],"name":"balanceOf","outputs":[{"name":"balance","type":"uint256"}],"type":"function"},
|
|
{"constant":true,"inputs":[],"name":"decimals","outputs":[{"name":"","type":"uint8"}],"type":"function"},
|
|
{"constant":true,"inputs":[],"name":"symbol","outputs":[{"name":"","type":"string"}],"type":"function"},
|
|
{"constant":true,"inputs":[],"name":"name","outputs":[{"name":"","type":"string"}],"type":"function"},
|
|
{"constant":true,"inputs":[],"name":"totalSupply","outputs":[{"name":"","type":"uint256"}],"type":"function"}
|
|
]
|
|
""")
|
|
|
|
# ─── CHAIN CONFIGURATIONS ───────────────────────────────────────────
|
|
|
|
# ─── ADDRESS VALIDATION ─────────────────────────────────────────────
|
|
|
|
EVM_ADDRESS_RE = re.compile(r"^0x[a-fA-F0-9]{40}$")
|
|
SOLANA_ADDRESS_RE = re.compile(r"^[1-9A-HJ-NP-Za-km-z]{32,44}$")
|
|
|
|
|
|
def validate_wallet_address(address: str) -> tuple[bool, str]:
|
|
"""Validate a wallet address. Returns (is_valid, chain_hint)."""
|
|
if EVM_ADDRESS_RE.match(address):
|
|
return True, "evm"
|
|
if SOLANA_ADDRESS_RE.match(address):
|
|
return True, "solana"
|
|
if not address.startswith("0x") and len(address) >= 32:
|
|
return True, "solana" # lenient solana check
|
|
if address.startswith("0x") and len(address) == 42:
|
|
return True, "evm" # lenient EVM check (allow mixed case)
|
|
return False, ""
|
|
|
|
|
|
CHAIN_CONFIGS: dict[str, dict[str, Any]] = {
|
|
"ethereum": {
|
|
"rpc": "https://ethereum-rpc.publicnode.com",
|
|
"chain_id": 1,
|
|
"native_symbol": "ETH",
|
|
"decimals": 18,
|
|
"explorer": "https://etherscan.io",
|
|
"type": "evm",
|
|
},
|
|
"base": {
|
|
"rpc": "https://base-rpc.publicnode.com",
|
|
"chain_id": 8453,
|
|
"native_symbol": "ETH",
|
|
"decimals": 18,
|
|
"explorer": "https://basescan.org",
|
|
"type": "evm",
|
|
},
|
|
"arbitrum": {
|
|
"rpc": "https://arbitrum-rpc.publicnode.com",
|
|
"chain_id": 42161,
|
|
"native_symbol": "ETH",
|
|
"decimals": 18,
|
|
"explorer": "https://arbiscan.io",
|
|
"type": "evm",
|
|
},
|
|
"optimism": {
|
|
"rpc": "https://optimism-rpc.publicnode.com",
|
|
"chain_id": 10,
|
|
"native_symbol": "ETH",
|
|
"decimals": 18,
|
|
"explorer": "https://optimistic.etherscan.io",
|
|
"type": "evm",
|
|
},
|
|
"polygon": {
|
|
"rpc": "https://polygon-rpc.publicnode.com",
|
|
"chain_id": 137,
|
|
"native_symbol": "MATIC",
|
|
"decimals": 18,
|
|
"explorer": "https://polygonscan.com",
|
|
"type": "evm",
|
|
"poa": True,
|
|
},
|
|
"bsc": {
|
|
"rpc": "https://bsc-rpc.publicnode.com",
|
|
"chain_id": 56,
|
|
"native_symbol": "BNB",
|
|
"decimals": 18,
|
|
"explorer": "https://bscscan.com",
|
|
"type": "evm",
|
|
"poa": True,
|
|
},
|
|
"avalanche": {
|
|
"rpc": "https://avalanche-rpc.publicnode.com",
|
|
"chain_id": 43114,
|
|
"native_symbol": "AVAX",
|
|
"decimals": 18,
|
|
"explorer": "https://snowtrace.io",
|
|
"type": "evm",
|
|
"poa": True,
|
|
},
|
|
"linea": {
|
|
"rpc": "https://linea-rpc.publicnode.com",
|
|
"chain_id": 59144,
|
|
"native_symbol": "ETH",
|
|
"decimals": 18,
|
|
"explorer": "https://lineascan.build",
|
|
"type": "evm",
|
|
},
|
|
"zksync": {
|
|
"rpc": "https://zksync-rpc.publicnode.com",
|
|
"chain_id": 324,
|
|
"native_symbol": "ETH",
|
|
"decimals": 18,
|
|
"explorer": "https://explorer.zksync.io",
|
|
"type": "evm",
|
|
},
|
|
"scroll": {
|
|
"rpc": "https://scroll-rpc.publicnode.com",
|
|
"chain_id": 534352,
|
|
"native_symbol": "ETH",
|
|
"decimals": 18,
|
|
"explorer": "https://scrollscan.com",
|
|
"type": "evm",
|
|
},
|
|
"mantle": {
|
|
"rpc": "https://mantle-rpc.publicnode.com",
|
|
"chain_id": 5000,
|
|
"native_symbol": "MNT",
|
|
"decimals": 18,
|
|
"explorer": "https://mantlescan.xyz",
|
|
"type": "evm",
|
|
},
|
|
"fantom": {
|
|
"rpc": "https://fantom-rpc.publicnode.com",
|
|
"chain_id": 250,
|
|
"native_symbol": "FTM",
|
|
"decimals": 18,
|
|
"explorer": "https://ftmscan.com",
|
|
"type": "evm",
|
|
"poa": True,
|
|
},
|
|
"solana": {
|
|
"rpc": "https://solana-rpc.publicnode.com",
|
|
"chain_id": "solana",
|
|
"native_symbol": "SOL",
|
|
"decimals": 9,
|
|
"explorer": "https://solscan.io",
|
|
"type": "solana",
|
|
},
|
|
}
|
|
|
|
# Common high-value ERC20 token addresses for estimation
|
|
COMMON_TOKENS: dict[str, dict[str, str]] = {
|
|
"ethereum": {
|
|
"USDC": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
|
|
"USDT": "0xdAC17F958D2ee523a2206206994597C13D831ec7",
|
|
"WBTC": "0x2260FAC5E5542a773Aa44fBCfeDf7C193bc2C599",
|
|
"LINK": "0x514910771AF9Ca656af840dff83E8264EcF986CA",
|
|
"UNI": "0x1f9840a85d5aF5bf1D1762F925BDADdC4201F984",
|
|
"DAI": "0x6B175474E89094C44Da98b954EedeAC495271d0F",
|
|
"PEPE": "0x6982508145454Ce325dDbE47a25d4ec3d2311933",
|
|
"SHIB": "0x95aD61b0a150d79219dCF64E1E6Cc01f0B64C4cE",
|
|
},
|
|
"base": {
|
|
"USDC": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913",
|
|
"AERO": "0x940181a94A35A4569E4529A3CDfB74e38FD98631",
|
|
},
|
|
"bsc": {
|
|
"USDT": "0x55d398326f99059fF775485246999027B3197955",
|
|
"CAKE": "0x0E09FaBB73Bd3Ade0a17ECC321fD13a19e81cE82",
|
|
"BNB": "0xbb4CdB9CBd36B01bD1cBaEBF2De08d9173bc095c",
|
|
},
|
|
}
|
|
|
|
# Common scam token patterns (address snippets to check)
|
|
KNOWN_SCAM_TOKENS: set[str] = set()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Data Models
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class RiskLevel(Enum):
|
|
SAFE = "safe"
|
|
LOW = "low"
|
|
MEDIUM = "medium"
|
|
HIGH = "high"
|
|
CRITICAL = "critical"
|
|
|
|
@classmethod
|
|
def from_score(cls, score: float) -> "RiskLevel":
|
|
if score >= 80:
|
|
return cls.SAFE
|
|
elif score >= 60:
|
|
return cls.LOW
|
|
elif score >= 40:
|
|
return cls.MEDIUM
|
|
elif score >= 20:
|
|
return cls.HIGH
|
|
return cls.CRITICAL
|
|
|
|
|
|
@dataclass
|
|
class TokenHolding:
|
|
"""A single token holding at a wallet, with risk data."""
|
|
|
|
chain: str
|
|
symbol: str
|
|
name: str
|
|
contract_address: str
|
|
balance_raw: int
|
|
balance_formatted: float
|
|
decimals: int
|
|
estimated_usd_value: float = 0.0
|
|
risk_score: float = 50.0 # 0=critical risk, 100=safe
|
|
risk_flags: list[str] = field(default_factory=list)
|
|
is_native: bool = False
|
|
pct_of_portfolio: float = 0.0
|
|
|
|
def risk_level(self) -> RiskLevel:
|
|
return RiskLevel.from_score(self.risk_score)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass
|
|
class ChainPortfolio:
|
|
"""Aggregated holdings for a single chain."""
|
|
|
|
chain: str
|
|
total_value_usd: float = 0.0
|
|
token_count: int = 0
|
|
holdings: list[TokenHolding] = field(default_factory=list)
|
|
avg_risk_score: float = 50.0
|
|
high_risk_count: int = 0
|
|
critical_risk_count: int = 0
|
|
scan_error: str | None = None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"chain": self.chain,
|
|
"total_value_usd": round(self.total_value_usd, 2),
|
|
"token_count": self.token_count,
|
|
"avg_risk_score": round(self.avg_risk_score, 1),
|
|
"high_risk_count": self.high_risk_count,
|
|
"critical_risk_count": self.critical_risk_count,
|
|
"holdings": [h.to_dict() for h in self.holdings],
|
|
"scan_error": self.scan_error,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class PortfolioRiskProfile:
|
|
"""Complete cross-chain portfolio risk assessment."""
|
|
|
|
wallet_address: str
|
|
chains_scanned: list[str]
|
|
total_value_usd: float = 0.0
|
|
total_tokens: int = 0
|
|
chain_count: int = 0
|
|
overall_health_score: float = 50.0
|
|
concentration_risk_pct: float = 0.0 # % in top token
|
|
chain_portfolios: dict[str, ChainPortfolio] = field(default_factory=dict)
|
|
top_risk_tokens: list[TokenHolding] = field(default_factory=list)
|
|
findings: list[str] = field(default_factory=list)
|
|
timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
|
|
scan_errors: list[str] = field(default_factory=list)
|
|
|
|
def overall_risk_level(self) -> RiskLevel:
|
|
return RiskLevel.from_score(self.overall_health_score)
|
|
|
|
def report(self, format: str = "text") -> str:
|
|
"""Generate a human-readable risk report."""
|
|
if format == "json":
|
|
return json.dumps(self.to_dict(), indent=2, default=str)
|
|
|
|
lines = []
|
|
lines.append("=" * 64)
|
|
total_val = f"${self.total_value_usd:,.2f}" if self.total_value_usd > 0 else "Unknown"
|
|
lines.append(
|
|
f" PORTFOLIO RISK REPORT - {self.wallet_address[:12]}...{self.wallet_address[-6:]}"
|
|
)
|
|
lines.append(
|
|
f" Health: {self.overall_health_score:.0f}/100 ({self.overall_risk_level().value.upper()})"
|
|
)
|
|
lines.append(
|
|
f" Value: {total_val} | Tokens: {self.total_tokens} | Chains: {self.chain_count}"
|
|
)
|
|
lines.append("=" * 64)
|
|
|
|
if self.findings:
|
|
lines.append("")
|
|
lines.append("🔍 KEY FINDINGS:")
|
|
for f in self.findings:
|
|
lines.append(f" • {f}")
|
|
|
|
lines.append("")
|
|
lines.append(f"{'CHAIN':<14} {'VALUE':<16} {'TOKENS':<8} {'RISK':<8} {'HIGH/CRIT':<10}")
|
|
lines.append("-" * 56)
|
|
for chain_name, cp in sorted(self.chain_portfolios.items()):
|
|
val = f"${cp.total_value_usd:,.2f}" if cp.total_value_usd else "?"
|
|
avg_risk = f"{cp.avg_risk_score:.0f}"
|
|
high_crit = f"{cp.high_risk_count}/{cp.critical_risk_count}"
|
|
error_tag = f" ⚠️ {cp.scan_error}" if cp.scan_error else ""
|
|
lines.append(
|
|
f"{chain_name:<14} {val:<16} {cp.token_count:<8} {avg_risk:<8} {high_crit:<10}{error_tag}"
|
|
)
|
|
|
|
if self.top_risk_tokens:
|
|
lines.append("")
|
|
lines.append("⚠️ HIGHEST RISK TOKENS:")
|
|
for t in self.top_risk_tokens[:10]:
|
|
val = f"${t.estimated_usd_value:,.2f}" if t.estimated_usd_value > 0 else "N/A"
|
|
flags = ", ".join(t.risk_flags[:3])
|
|
pct = f"{t.pct_of_portfolio:.1f}%"
|
|
lines.append(
|
|
f" • {t.symbol:>8} ({t.chain}) - Score: {t.risk_score:.0f}/100 - {val} - {pct}"
|
|
)
|
|
if flags:
|
|
lines.append(f" ⚑ {flags}")
|
|
|
|
lines.append("")
|
|
lines.append("=" * 64)
|
|
lines.append(f" Report generated: {self.timestamp[:19]}")
|
|
lines.append("=" * 64)
|
|
|
|
return "\n".join(lines)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"wallet_address": self.wallet_address,
|
|
"chains_scanned": self.chains_scanned,
|
|
"total_value_usd": round(self.total_value_usd, 2),
|
|
"total_tokens": self.total_tokens,
|
|
"chain_count": self.chain_count,
|
|
"overall_health_score": round(self.overall_health_score, 1),
|
|
"concentration_risk_pct": round(self.concentration_risk_pct, 1),
|
|
"overall_risk_level": self.overall_risk_level().value,
|
|
"findings": self.findings,
|
|
"chain_portfolios": {k: v.to_dict() for k, v in self.chain_portfolios.items()},
|
|
"top_risk_tokens": [t.to_dict() for t in self.top_risk_tokens[:10]],
|
|
"scan_errors": self.scan_errors,
|
|
"timestamp": self.timestamp,
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Chain Connectors
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class ChainConnector:
|
|
"""Abstract chain connector base."""
|
|
|
|
async def get_native_balance(self, address: str) -> float:
|
|
raise NotImplementedError
|
|
|
|
async def get_token_balances(self, address: str) -> list[dict[str, Any]]:
|
|
raise NotImplementedError
|
|
|
|
|
|
class EVMConnector(ChainConnector):
|
|
"""EVM chain connector using web3.py."""
|
|
|
|
def __init__(self, chain_name: str, config: dict[str, Any]):
|
|
self.chain_name = chain_name
|
|
self.config = config
|
|
self._w3: Any = None
|
|
|
|
@property
|
|
def w3(self) -> Any:
|
|
if self._w3 is None:
|
|
if not WEB3_AVAILABLE:
|
|
raise RuntimeError("web3.py not installed")
|
|
provider = HTTPProvider(self.config["rpc"], request_kwargs={"timeout": DEFAULT_TIMEOUT})
|
|
self._w3 = _Web3(provider)
|
|
if self.config.get("poa"):
|
|
try:
|
|
from web3.middleware import geth_poa_middleware
|
|
|
|
self._w3.middleware_onion.inject(geth_poa_middleware, layer=0)
|
|
except ImportError:
|
|
try:
|
|
from web3.middleware import ExtraDataToPOAMiddleware
|
|
|
|
self._w3.middleware_onion.inject(ExtraDataToPOAMiddleware, layer=0)
|
|
except ImportError:
|
|
pass
|
|
return self._w3
|
|
|
|
async def get_native_balance(self, address: str) -> float:
|
|
"""Get native token balance in human-readable form."""
|
|
try:
|
|
bal_wei = await asyncio.to_thread(self.w3.eth.get_balance, address)
|
|
decimals = self.config["decimals"]
|
|
return float(bal_wei) / (10**decimals)
|
|
except ConnectionError as e:
|
|
logger.warning(f"{self.chain_name} RPC connection error (balance): {e}")
|
|
return 0.0
|
|
except TimeoutError as e:
|
|
logger.warning(f"{self.chain_name} RPC timeout (balance): {e}")
|
|
return 0.0
|
|
except ValueError as e:
|
|
logger.warning(f"{self.chain_name} value error (balance): {e}")
|
|
return 0.0
|
|
except Exception as e:
|
|
logger.warning(
|
|
f"{self.chain_name} unexpected native balance error: {type(e).__name__}: {e}"
|
|
)
|
|
return 0.0
|
|
|
|
async def get_token_balances(self, address: str) -> list[dict[str, Any]]:
|
|
"""Get ERC20 token balances for an address.
|
|
|
|
Uses a two-pass approach:
|
|
1. Check common high-value tokens first
|
|
2. If enabled, scan for all transfers to discover held tokens
|
|
"""
|
|
results = []
|
|
checked = set()
|
|
|
|
# First pass: check common tokens
|
|
common = COMMON_TOKENS.get(self.chain_name, {})
|
|
for _symbol, token_addr in common.items():
|
|
try:
|
|
contract = self.w3.eth.contract(
|
|
address=_Web3.to_checksum_address(token_addr),
|
|
abi=ERC20_ABI,
|
|
)
|
|
bal = await asyncio.to_thread(contract.functions.balanceOf(address).call)
|
|
if bal > 0:
|
|
dec = await asyncio.to_thread(contract.functions.decimals.call)
|
|
sym = await asyncio.to_thread(contract.functions.symbol.call)
|
|
results.append(
|
|
{
|
|
"address": token_addr,
|
|
"symbol": sym,
|
|
"balance_raw": bal,
|
|
"balance_formatted": float(bal) / (10**dec),
|
|
"decimals": dec,
|
|
}
|
|
)
|
|
checked.add(token_addr.lower())
|
|
except Exception:
|
|
pass
|
|
|
|
# Second pass: check recent transfer events to discover tokens
|
|
try:
|
|
# Look at ERC20 Transfer events to the wallet (last 5000 blocks)
|
|
latest = await asyncio.to_thread(self.w3.eth.block_number)
|
|
from_block = max(latest - 5000, 0)
|
|
transfer_topic = self.w3.keccak(text="Transfer(address,address,uint256)").hex()
|
|
|
|
logs = await asyncio.to_thread(
|
|
self.w3.eth.get_logs,
|
|
{
|
|
"address": None, # across all contracts
|
|
"fromBlock": from_block,
|
|
"toBlock": latest,
|
|
"topics": [
|
|
transfer_topic,
|
|
None,
|
|
_Web3.to_checksum_address(address).hex()
|
|
if address.startswith("0x")
|
|
else "0x" + address.zfill(64),
|
|
],
|
|
},
|
|
)
|
|
# Limit to avoid huge sets
|
|
for log in logs[:200]:
|
|
token_addr = log["address"].lower()
|
|
if token_addr in checked:
|
|
continue
|
|
checked.add(token_addr)
|
|
try:
|
|
contract = self.w3.eth.contract(address=token_addr, abi=ERC20_ABI)
|
|
bal = await asyncio.to_thread(contract.functions.balanceOf(address).call)
|
|
if bal > 0:
|
|
dec = await asyncio.to_thread(contract.functions.decimals.call)
|
|
sym = await asyncio.to_thread(contract.functions.symbol.call)
|
|
results.append(
|
|
{
|
|
"address": log["address"],
|
|
"symbol": sym,
|
|
"balance_raw": bal,
|
|
"balance_formatted": float(bal) / (10**dec),
|
|
"decimals": dec,
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
if len(results) >= MAX_TOKENS_PER_WALLET:
|
|
break
|
|
except Exception as e:
|
|
logger.debug(f"{self.chain_name} event scan error: {e}")
|
|
|
|
return results
|
|
|
|
|
|
class SolanaConnector(ChainConnector):
|
|
"""Solana chain connector using solana-py."""
|
|
|
|
def __init__(self):
|
|
self._client: Any = None
|
|
|
|
@property
|
|
def client(self) -> Any:
|
|
if self._client is None:
|
|
if not SOLANA_AVAILABLE:
|
|
raise RuntimeError("solana-py not installed")
|
|
self._client = SolanaClient(CHAIN_CONFIGS["solana"]["rpc"])
|
|
return self._client
|
|
|
|
async def get_native_balance(self, address: str) -> float:
|
|
"""Get SOL balance."""
|
|
try:
|
|
from solana.rpc.api import Pubkey as SolPubkey
|
|
|
|
pubkey = SolPubkey.from_string(address)
|
|
resp = await asyncio.to_thread(self.client.get_balance, pubkey)
|
|
if resp and resp.get("result"):
|
|
lamports = resp["result"]["value"]
|
|
return lamports / 1e9
|
|
return 0.0
|
|
except ConnectionError as e:
|
|
logger.warning(f"solana RPC connection error (balance): {e}")
|
|
return 0.0
|
|
except TimeoutError as e:
|
|
logger.warning(f"solana RPC timeout (balance): {e}")
|
|
return 0.0
|
|
except ValueError as e:
|
|
logger.warning(
|
|
f"solana address validation error: {e} - invalid pubkey: {address[:12]}..."
|
|
)
|
|
return 0.0
|
|
except Exception as e:
|
|
logger.warning(f"solana unexpected native balance error: {type(e).__name__}: {e}")
|
|
return 0.0
|
|
|
|
async def get_token_balances(self, address: str) -> list[dict[str, Any]]:
|
|
"""Get SPL token balances."""
|
|
results = []
|
|
try:
|
|
from solana.rpc.api import Pubkey as SolPubkey
|
|
|
|
pubkey = SolPubkey.from_string(address)
|
|
resp = await asyncio.to_thread(
|
|
self.client.get_token_accounts_by_owner_json_parsed, pubkey
|
|
)
|
|
if not resp or not resp.get("result"):
|
|
return results
|
|
for item in resp["result"]["value"]:
|
|
try:
|
|
parsed = item["account"]["data"]["parsed"]["info"]
|
|
mint = parsed.get("mint", "")
|
|
token_amount = parsed.get("tokenAmount", {})
|
|
ui_amount = float(token_amount.get("uiAmount", 0) or 0)
|
|
if ui_amount <= 0:
|
|
continue
|
|
results.append(
|
|
{
|
|
"address": mint,
|
|
"symbol": token_amount.get("symbol", "SPL"),
|
|
"balance_raw": int(token_amount.get("amount", "0")),
|
|
"balance_formatted": ui_amount,
|
|
"decimals": token_amount.get("decimals", 0),
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
except Exception as e:
|
|
logger.warning(f"solana token balances error: {e}")
|
|
return results
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Token Risk Scanner
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TokenRiskScorer:
|
|
"""Scores a token's risk using heuristics + on-chain data.
|
|
|
|
In production, this calls the unified scanner. For standalone use,
|
|
it applies heuristic checks based on token metadata and chain data.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._scanner_available = False
|
|
self._import_error: str | None = None
|
|
self._try_import_scanner()
|
|
|
|
def _try_import_scanner(self):
|
|
"""Try to import the unified scanner for real risk scoring."""
|
|
try:
|
|
import importlib
|
|
|
|
self._scanner_available = importlib.util.find_spec("app.unified_scanner") is not None
|
|
except ImportError:
|
|
pass
|
|
|
|
async def score_token(self, chain: str, address: str, symbol: str) -> dict[str, Any]:
|
|
"""Score a token. Returns risk_score (0-100, higher = safer) and flags."""
|
|
if not address or address == "0x0000000000000000000000000000000000000000":
|
|
return {"risk_score": 100, "risk_flags": [], "reason": "Native token (zero address)"}
|
|
|
|
if self._scanner_available:
|
|
try:
|
|
from app.unified_scanner import scan_token as _scan_token
|
|
|
|
result = await _scan_token(chain, address)
|
|
if result:
|
|
score = result.get("safety_score", 50)
|
|
flags = result.get("risk_flags", [])
|
|
return {"risk_score": score, "risk_flags": flags, "reason": "Scanner"}
|
|
except Exception:
|
|
pass
|
|
|
|
# Heuristic fallback
|
|
return self._heuristic_score(chain, address, symbol)
|
|
|
|
def _heuristic_score(self, chain: str, address: str, symbol: str) -> dict[str, Any]:
|
|
"""Heuristic risk assessment when scanner unavailable."""
|
|
score = 50 # default: neutral
|
|
flags = []
|
|
addr_lower = address.lower()
|
|
|
|
# Check known scam tokens
|
|
if addr_lower in KNOWN_SCAM_TOKENS:
|
|
score -= 40
|
|
flags.append("KNOWN_SCAM_TOKEN")
|
|
|
|
# Very new token heuristic (check if address appears in known lists)
|
|
if not addr_lower.startswith("0x") or len(address) < 20:
|
|
score -= 20
|
|
flags.append("NON_STANDARD_ADDRESS")
|
|
|
|
# Check name/symbol heuristics
|
|
upper_sym = symbol.upper() if symbol else ""
|
|
scam_indicators = ["ELON", "MUSK", "SHIBA", "FLOCKI", "SQUID", "SAFEMOON"]
|
|
for ind in scam_indicators:
|
|
if ind in upper_sym:
|
|
score -= 15
|
|
flags.append(f"SCAM_INDICATOR_{ind}")
|
|
break
|
|
|
|
# Check if it's a well-known token
|
|
for chain_tokens in COMMON_TOKENS.values():
|
|
for _known_sym, known_addr in chain_tokens.items():
|
|
if known_addr.lower() == addr_lower:
|
|
score = 90
|
|
flags = []
|
|
return {"risk_score": score, "risk_flags": flags, "reason": "Known token"}
|
|
|
|
return {"risk_score": max(0, min(100, score)), "risk_flags": flags, "reason": "Heuristic"}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Price Oracle (Coingecko-based estimation)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class PriceOracle:
|
|
"""Simple price oracle using Coingecko public API.
|
|
|
|
Caches prices in-memory to avoid rate limits.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._cache: dict[str, float] = {
|
|
"ETH": 2800.0,
|
|
"BTC": 68000.0,
|
|
"SOL": 145.0,
|
|
"BNB": 580.0,
|
|
"MATIC": 0.55,
|
|
"AVAX": 32.0,
|
|
"MNT": 0.85,
|
|
"FTM": 0.68,
|
|
"USDC": 1.0,
|
|
"USDT": 1.0,
|
|
"DAI": 1.0,
|
|
"WBTC": 68000.0,
|
|
"LINK": 14.0,
|
|
"UNI": 7.5,
|
|
"PEPE": 0.000012,
|
|
"SHIB": 0.000025,
|
|
"AERO": 0.85,
|
|
"CAKE": 2.10,
|
|
}
|
|
self._cache_time: float = 0
|
|
self._cache_ttl = 300 # 5 minutes
|
|
|
|
async def get_price(self, symbol: str) -> float:
|
|
"""Get USD price for a token symbol. Uses fallbacks."""
|
|
upper = symbol.upper()
|
|
cached = self._cache.get(upper)
|
|
if cached is not None:
|
|
return cached
|
|
|
|
# Try async lookup via Coingecko
|
|
try:
|
|
if HTTPX_AVAILABLE:
|
|
async with httpx.AsyncClient(timeout=5) as client:
|
|
url = f"https://api.coingecko.com/api/v3/simple/price?ids={upper.lower()}&vs_currencies=usd"
|
|
resp = await client.get(url)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
price = data.get(upper.lower(), {}).get("usd")
|
|
if price:
|
|
self._cache[upper] = float(price)
|
|
return float(price)
|
|
except Exception:
|
|
pass
|
|
|
|
return 0.0 # Unknown price
|
|
|
|
def estimate_native_price(self, chain: str) -> float:
|
|
"""Get estimated native token price for a chain."""
|
|
symbol = CHAIN_CONFIGS[chain]["native_symbol"]
|
|
return self._cache.get(symbol.upper(), 0.0)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Portfolio Risk Aggregator
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class PortfolioRiskAggregator:
|
|
"""Main portfolio risk analysis engine."""
|
|
|
|
def __init__(self, chains: list[str] | None = None):
|
|
self.chains = chains or [
|
|
"ethereum",
|
|
"base",
|
|
"arbitrum",
|
|
"optimism",
|
|
"polygon",
|
|
"bsc",
|
|
"avalanche",
|
|
"fantom",
|
|
"solana",
|
|
]
|
|
self.price_oracle = PriceOracle()
|
|
self.token_scorer = TokenRiskScorer()
|
|
self._connectors: dict[str, ChainConnector] = {}
|
|
|
|
def _get_connector(self, chain: str) -> ChainConnector:
|
|
"""Get or create a chain connector."""
|
|
if chain not in self._connectors:
|
|
config = CHAIN_CONFIGS.get(chain)
|
|
if not config:
|
|
raise ValueError(f"Unsupported chain: {chain}")
|
|
if config["type"] == "evm":
|
|
self._connectors[chain] = EVMConnector(chain, config)
|
|
elif config["type"] == "solana":
|
|
self._connectors[chain] = SolanaConnector()
|
|
else:
|
|
raise ValueError(f"Unknown chain type for {chain}: {config['type']}")
|
|
return self._connectors[chain]
|
|
|
|
async def analyze(self, wallet_address: str) -> PortfolioRiskProfile:
|
|
"""Full cross-chain portfolio risk analysis."""
|
|
profile = PortfolioRiskProfile(
|
|
wallet_address=wallet_address,
|
|
chains_scanned=list(self.chains),
|
|
)
|
|
|
|
# Scan each chain in parallel
|
|
tasks = []
|
|
for chain in self.chains:
|
|
tasks.append(self._scan_chain(chain, wallet_address))
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
|
|
# Process results
|
|
all_tokens: list[TokenHolding] = []
|
|
total_value = 0.0
|
|
|
|
for chain, result in zip(self.chains, results, strict=False):
|
|
if isinstance(result, Exception):
|
|
profile.scan_errors.append(f"{chain}: {str(result)[:100]}")
|
|
profile.chain_portfolios[chain] = ChainPortfolio(
|
|
chain=chain, scan_error=str(result)[:100]
|
|
)
|
|
continue
|
|
|
|
cp = result # ChainPortfolio
|
|
profile.chain_portfolios[chain] = cp
|
|
total_value += cp.total_value_usd
|
|
all_tokens.extend(cp.holdings)
|
|
|
|
# Calculate percentages
|
|
for token in all_tokens:
|
|
if total_value > 0:
|
|
token.pct_of_portfolio = (token.estimated_usd_value / total_value) * 100
|
|
|
|
# Calculate concentration risk
|
|
if all_tokens:
|
|
sorted_by_value = sorted(all_tokens, key=lambda t: t.estimated_usd_value, reverse=True)
|
|
top_token_pct = sorted_by_value[0].pct_of_portfolio if sorted_by_value else 0
|
|
profile.concentration_risk_pct = top_token_pct
|
|
|
|
# Top risk tokens (sorted by risk score ascending = most dangerous first)
|
|
risk_sorted = sorted(
|
|
[t for t in all_tokens if t.risk_score < 60],
|
|
key=lambda t: t.risk_score,
|
|
)
|
|
profile.top_risk_tokens = risk_sorted
|
|
|
|
profile.total_value_usd = total_value
|
|
profile.total_tokens = len(all_tokens)
|
|
profile.chain_count = len(
|
|
[
|
|
c
|
|
for c in self.chains
|
|
if c in profile.chain_portfolios and profile.chain_portfolios[c].scan_error is None
|
|
]
|
|
)
|
|
|
|
# Calculate overall health score
|
|
profile.overall_health_score = self._calculate_health_score(profile)
|
|
|
|
# Generate findings
|
|
profile.findings = self._generate_findings(profile)
|
|
|
|
return profile
|
|
|
|
async def _scan_chain(self, chain: str, wallet: str) -> ChainPortfolio:
|
|
"""Scan a single chain for wallet holdings."""
|
|
cp = ChainPortfolio(chain=chain)
|
|
|
|
try:
|
|
connector = self._get_connector(chain)
|
|
except ValueError as e:
|
|
cp.scan_error = str(e)
|
|
return cp
|
|
|
|
# Get native balance
|
|
try:
|
|
native_balance = await connector.get_native_balance(wallet)
|
|
except Exception as e:
|
|
native_balance = 0.0
|
|
logger.debug(f"{chain} native balance error: {e}")
|
|
|
|
# Get token balances
|
|
try:
|
|
token_data = await connector.get_token_balances(wallet)
|
|
except Exception as e:
|
|
token_data = []
|
|
logger.debug(f"{chain} token balances error: {e}")
|
|
|
|
# Process native token
|
|
native_price = self.price_oracle.estimate_native_price(chain)
|
|
native_value = native_balance * native_price
|
|
if native_balance > 0:
|
|
native_addr = "0x0000000000000000000000000000000000000000"
|
|
native_sym = CHAIN_CONFIGS[chain]["native_symbol"]
|
|
native_token = TokenHolding(
|
|
chain=chain,
|
|
symbol=native_sym,
|
|
name=f"Native {native_sym}",
|
|
contract_address=native_addr,
|
|
balance_raw=0,
|
|
balance_formatted=native_balance,
|
|
decimals=CHAIN_CONFIGS[chain]["decimals"],
|
|
estimated_usd_value=native_value,
|
|
risk_score=90, # Native tokens are generally low risk
|
|
risk_flags=[],
|
|
is_native=True,
|
|
)
|
|
cp.holdings.append(native_token)
|
|
|
|
# Process ERC20/SPL tokens
|
|
for td in token_data[:MAX_TOKENS_PER_WALLET]:
|
|
try:
|
|
addr = td.get("address", "")
|
|
symbol = td.get("symbol", "UNK")
|
|
formatted = td.get("balance_formatted", 0)
|
|
|
|
if formatted <= 0:
|
|
continue
|
|
|
|
# Get price estimate
|
|
price = await self.price_oracle.get_price(symbol)
|
|
usd_value = formatted * price
|
|
|
|
# Get risk score
|
|
risk_result = await self.token_scorer.score_token(chain, addr, symbol)
|
|
|
|
token = TokenHolding(
|
|
chain=chain,
|
|
symbol=symbol,
|
|
name=td.get("symbol", symbol),
|
|
contract_address=addr,
|
|
balance_raw=td.get("balance_raw", 0),
|
|
balance_formatted=formatted,
|
|
decimals=td.get("decimals", 18),
|
|
estimated_usd_value=usd_value,
|
|
risk_score=risk_result["risk_score"],
|
|
risk_flags=risk_result.get("risk_flags", []),
|
|
is_native=False,
|
|
)
|
|
cp.holdings.append(token)
|
|
except Exception as e:
|
|
logger.debug(f"Error processing token on {chain}: {e}")
|
|
|
|
# Calculate chain-level stats
|
|
cp.token_count = len(cp.holdings)
|
|
if cp.holdings:
|
|
cp.total_value_usd = sum(h.estimated_usd_value for h in cp.holdings)
|
|
cp.avg_risk_score = sum(h.risk_score for h in cp.holdings) / len(cp.holdings)
|
|
cp.high_risk_count = len([h for h in cp.holdings if h.risk_score < 40])
|
|
cp.critical_risk_count = len([h for h in cp.holdings if h.risk_score < 20])
|
|
|
|
return cp
|
|
|
|
def _calculate_health_score(self, profile: PortfolioRiskProfile) -> float:
|
|
"""Calculate overall portfolio health score (0-100)."""
|
|
if not profile.chain_portfolios:
|
|
return 50.0
|
|
|
|
# Component 1: Average token risk (0-100) weighted 60%
|
|
token_scores = []
|
|
for cp in profile.chain_portfolios.values():
|
|
for h in cp.holdings:
|
|
token_scores.append(h.risk_score)
|
|
avg_token_score = sum(token_scores) / len(token_scores) if token_scores else 50
|
|
|
|
# Component 2: Concentration risk (100 - % in single token), weighted 25%
|
|
conc_score = 100 - profile.concentration_risk_pct
|
|
|
|
# Component 3: Chain diversity, weighted 15%
|
|
chains_with_tokens = sum(
|
|
1 for cp in profile.chain_portfolios.values() if cp.token_count > 0
|
|
)
|
|
diversity_score = min(100, (chains_with_tokens / max(1, len(self.chains))) * 100)
|
|
|
|
health = (
|
|
avg_token_score * RISK_WEIGHT_TOKENS
|
|
+ conc_score * RISK_WEIGHT_CONCENTRATION
|
|
+ diversity_score * RISK_WEIGHT_CHAIN_DIVERSITY
|
|
)
|
|
|
|
return max(0, min(100, health))
|
|
|
|
def _generate_findings(self, profile: PortfolioRiskProfile) -> list[str]:
|
|
"""Generate actionable findings from the risk profile."""
|
|
findings = []
|
|
|
|
# Critical risk tokens
|
|
critical = []
|
|
for cp in profile.chain_portfolios.values():
|
|
for h in cp.holdings:
|
|
if h.risk_score < 20:
|
|
critical.append(h)
|
|
|
|
if critical:
|
|
syms = ", ".join(f"{h.symbol} ({h.chain})" for h in critical[:5])
|
|
findings.append(f"⚠️ {len(critical)} CRITICAL risk token(s) detected: {syms}")
|
|
|
|
# High risk tokens
|
|
high = []
|
|
for cp in profile.chain_portfolios.values():
|
|
for h in cp.holdings:
|
|
if 20 <= h.risk_score < 40:
|
|
high.append(h)
|
|
if high:
|
|
findings.append(f"⚠️ {len(high)} HIGH risk token(s) - investigate withdrawals")
|
|
|
|
# Concentration risk
|
|
if profile.concentration_risk_pct > 50:
|
|
findings.append(
|
|
f"🔴 High concentration risk: {profile.concentration_risk_pct:.0f}% in single token"
|
|
)
|
|
elif profile.concentration_risk_pct > 30:
|
|
findings.append(
|
|
f"🟡 Moderate concentration: {profile.concentration_risk_pct:.0f}% in top holding"
|
|
)
|
|
|
|
# Low chain diversity
|
|
chains_used = sum(1 for cp in profile.chain_portfolios.values() if cp.token_count > 0)
|
|
if chains_used <= 1:
|
|
findings.append("🟡 Single-chain portfolio - consider diversifying across chains")
|
|
elif chains_used >= 5:
|
|
findings.append(f"✅ Good cross-chain diversity ({chains_used} chains)")
|
|
|
|
# Scam flags
|
|
scam_tokens = []
|
|
for cp in profile.chain_portfolios.values():
|
|
for h in cp.holdings:
|
|
if any("SCAM" in f or "HONEYPOT" in f for f in h.risk_flags):
|
|
scam_tokens.append(h)
|
|
if scam_tokens:
|
|
findings.append(f"🚨 {len(scam_tokens)} token(s) with scam indicators found!")
|
|
|
|
# High value in risky tokens
|
|
risky_value = 0.0
|
|
for cp in profile.chain_portfolios.values():
|
|
for h in cp.holdings:
|
|
if h.risk_score < 40:
|
|
risky_value += h.estimated_usd_value
|
|
if risky_value > 1000:
|
|
findings.append(f"💰 ${risky_value:,.0f} total value in high/critical risk tokens")
|
|
|
|
# Clean portfolio
|
|
if not critical and not high and profile.overall_health_score >= 80:
|
|
findings.append("✅ Portfolio looks clean - no high-risk tokens detected")
|
|
|
|
return findings
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# CLI Entry Point
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
async def main():
|
|
"""CLI entry point."""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description="Cross-Chain Portfolio Risk Aggregator",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
python3 portfolio_risk_aggregator.py 0x742d35Cc6634C0532925a3b844Bc454e4438f44e
|
|
python3 portfolio_risk_aggregator.py 0x... --chains ethereum,base
|
|
python3 portfolio_risk_aggregator.py 0x... --format json
|
|
python3 portfolio_risk_aggregator.py <solana_wallet>
|
|
""",
|
|
)
|
|
parser.add_argument("wallet", help="Wallet address to analyze (EVM or Solana)")
|
|
parser.add_argument("--chains", help="Comma-separated chain list (default: all)")
|
|
parser.add_argument(
|
|
"--format", choices=["text", "json"], default="text", help="Output format (default: text)"
|
|
)
|
|
parser.add_argument("--debug", action="store_true", help="Enable debug logging")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.debug:
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
else:
|
|
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
|
|
|
|
chains = args.chains.split(",") if args.chains else None
|
|
|
|
# Validate wallet address
|
|
is_valid, chain_hint = validate_wallet_address(args.wallet)
|
|
if not is_valid:
|
|
print(f"❌ Invalid wallet address: {args.wallet}")
|
|
print(" Expected: 42-char hex (0x...) for EVM, or 32-44 base58 for Solana")
|
|
sys.exit(1)
|
|
|
|
aggregator = PortfolioRiskAggregator(chains=chains)
|
|
print(f"🔍 Analyzing wallet: {args.wallet[:16]}...{args.wallet[-6:]}")
|
|
print(f" Chains: {', '.join(aggregator.chains)}")
|
|
if chain_hint == "solana":
|
|
print(" [i] Solana address detected -- scanning SPL tokens")
|
|
else:
|
|
print(" [i] EVM address detected -- scanning ERC20 tokens")
|
|
print()
|
|
|
|
profile = await aggregator.analyze(args.wallet)
|
|
print(profile.report(format=args.format))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|