RugMunchMedia/rmi-backend
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
rmi-backend/app/cross_chain_whale.py
opencode c762564d40 style(rmi-backend): complete lint cleanup — 1175→0 ruff errors 
- Fix 71 invalid-syntax files (class-body newline-broken assignments)
- Add from/None chain to 307 B904 raise-without-from sites
- Add B008 ignore to ruff.toml (already in pyproject.toml)
- Noqa F401 on __init__.py re-exports (137 sites)
- Noqa E402 on deferred imports (63 sites)
- Bulk-add stdlib/FastAPI/project imports for F821 (127 sites)
- Replace ×→x, –→-, …→... in docstrings (4093 chars)
- Manual refactor of 5 SIM103/SIM116 patterns

Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py)
Co-authored-by: opencode <opencode@rugmunch.io>
2026-07-06 15:43:20 +02:00

835 lines
30 KiB
Python
Raw Permalink Blame History

"""
Cross-Chain Whale Tracker
==========================
Track whale wallets across multiple blockchains simultaneously.
Maps cross-chain capital flows, bridge migrations, and multi-network
positioning of top holders. Free-tier data sources only.
TOOL : cross_chain_whale
TIER : intelligence
PRICE : $0.08 (80000 atoms)
TRIAL : 2 free checks
Data Sources (all free):
- DexScreener - token pairs, holders, liquidity across chains
- Solscan (free) - Solana holder data
- Etherscan family - EVM holder data
- Birdeye public - cross-chain holder rankings
- Jupiter - Solana token info
"""
import asyncio
import logging
import os
from contextlib import suppress
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any
import httpx
logger = logging.getLogger(__name__)
# ── Constants ────────────────────────────────────────────────────
FREE_APIS = {
"dexscreener": "https://api.dexscreener.com/latest/dex",
"birdeye_public": "https://public-api.birdeye.so",
"jupiter": "https://api.jup.ag",
}
CHAIN_NAMES = {
"solana": "Solana",
"ethereum": "Ethereum",
"bsc": "BSC",
"polygon": "Polygon",
"arbitrum": "Arbitrum",
"optimism": "Optimism",
"avalanche": "Avalanche",
"base": "Base",
"fantom": "Fantom",
"linea": "Linea",
"zksync": "zkSync",
"scroll": "Scroll",
"mantle": "Mantle",
}
# Etherscan-like API base URLs
ETHERSCAN_BASES: dict[str, str] = {
"ethereum": "https://api.etherscan.io/api",
"bsc": "https://api.bscscan.com/api",
"polygon": "https://api.polygonscan.com/api",
"arbitrum": "https://api.arbiscan.io/api",
"optimism": "https://api-optimistic.etherscan.io/api",
"avalanche": "https://api.snowtrace.io/api",
"fantom": "https://api.ftmscan.com/api",
"base": "https://api.basescan.org/api",
}
# Chain IDs for EVM chains
CHAIN_IDS: dict[str, int] = {
"ethereum": 1,
"bsc": 56,
"polygon": 137,
"arbitrum": 42161,
"optimism": 10,
"avalanche": 43114,
"fantom": 250,
"base": 8453,
}
# ── Data Models ───────────────────────────────────────────────────
@dataclass
class WhalePosition:
"""A whale's position on a single chain."""
chain: str
token_address: str
token_symbol: str
token_name: str = ""
balance: float = 0.0
balance_usd: float = 0.0
percentage: float = 0.0 # % of total supply
rank: int = 0 # holder rank on this chain
last_active: str = "" # ISO timestamp
wallet_label: str = ""
source: str = "" # data source (birdeye, etherscan, etc.)
@dataclass
class CrossChainWhale:
"""A whale wallet tracked across chains."""
primary_address: str
label: str = ""
total_value_usd: float = 0.0
chain_count: int = 0
token_count: int = 0
positions: list[WhalePosition] = field(default_factory=list)
first_seen: str = ""
risk_score: float = 0.0 # 0-100
risk_factors: list[str] = field(default_factory=list)
is_exchange: bool = False
is_scammer: bool = False
is_smart_money: bool = False
@dataclass
class WhaleTrackerReport:
"""Complete cross-chain whale tracking report."""
token_address: str
token_symbol: str = ""
token_name: str = ""
total_holders: int = 0
chains_found: list[str] = field(default_factory=list)
whales: list[CrossChainWhale] = field(default_factory=list)
cross_chain_whales: list[CrossChainWhale] = field(default_factory=list) # whales on 2+ chains
top_holders_by_chain: dict[str, list[dict[str, Any]]] = field(default_factory=dict)
total_value_tracked: float = 0.0
concentration_score: float = 0.0 # 0-100 (higher = more concentrated)
scan_timestamp: str = ""
chains_with_data: list[str] = field(default_factory=list)
chains_no_data: list[str] = field(default_factory=list)
errors: list[str] = field(default_factory=list)
# ── Helpers ───────────────────────────────────────────────────────
def _now_iso() -> str:
return datetime.now(UTC).isoformat()
def _get_env_key(key: str) -> str:
return os.getenv(key, "")
def _truncate_address(addr: str) -> str:
if len(addr) <= 12:
return addr
return f"{addr[:6]}...{addr[-4:]}"
def _compute_concentration(positions: list[WhalePosition]) -> float:
"""Compute HHI-like concentration from whale positions."""
if not positions:
return 0.0
total = sum(p.balance_usd for p in positions if p.balance_usd > 0)
if total <= 0:
return 0.0
hhi = sum((p.balance_usd / total) ** 2 for p in positions if p.balance_usd > 0)
# Normalize: HHI of 1 = fully concentrated, HHI near 0 = very distributed
return round(min(hhi * 100, 100), 1)
# ── API Fetchers ──────────────────────────────────────────────────
async def _fetch_json(url: str, headers: dict | None = None, timeout: float = 15.0) -> dict | None:
"""Fetch JSON from URL with error handling."""
try:
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.get(url, headers=headers)
if resp.status_code == 200:
return resp.json()
logger.debug(f"HTTP {resp.status_code} for {url}")
return None
except httpx.TimeoutException:
logger.debug(f"Timeout fetching {url}")
return None
except Exception as e:
logger.debug(f"Error fetching {url}: {e}")
return None
async def _fetch_dexscreener_pairs(token_address: str) -> list[dict]:
"""Fetch all DEX pairs for a token across all chains."""
url = f"{FREE_APIS['dexscreener']}/tokens/{token_address}"
data = await _fetch_json(url)
if not data or "pairs" not in data:
return []
return data["pairs"]
async def _fetch_etherscan_holders(token_address: str, chain: str) -> list[dict[str, Any]]:
"""Fetch top token holders from Etherscan-family explorer."""
base_url = ETHERSCAN_BASES.get(chain)
if not base_url:
return []
api_key = _get_env_key(f"{chain.upper()}_API_KEY".replace("-", "_"))
if not api_key:
api_key = _get_env_key("ETHERSCAN_API_KEY")
if not api_key:
# Try without key (limited)
url = (
f"{base_url}?module=token&action=tokenholderlist"
f"&contractaddress={token_address}"
f"&page=1&offset=10"
)
else:
url = (
f"{base_url}?module=token&action=tokenholderlist"
f"&contractaddress={token_address}"
f"&page=1&offset=10&apikey={api_key}"
)
data = await _fetch_json(url)
if not data or data.get("status") != "1":
return []
return data.get("result", [])
async def _fetch_birdeye_holders(token_address: str, chain: str) -> list[dict]:
"""Fetch top holders from Birdeye public API."""
api_key = _get_env_key("BIRDEYE_API_KEY")
headers = {"x-chain": chain}
if api_key:
headers["x-api-key"] = api_key
url = f"{FREE_APIS['birdeye_public']}/defi/holder/top?address={token_address}&limit=20"
data = await _fetch_json(url, headers=headers)
if not data or not data.get("success"):
return []
return data.get("data", {}).get("holders", [])
async def _fetch_jupiter_token_info(token_address: str) -> dict | list | None:
"""Fetch Jupiter token info for Solana tokens."""
# Try multiple Jupiter API URLs
urls = [
f"https://api.jup.ag/tokens/v1/token/{token_address}",
f"https://quote-api.jup.ag/v6/tokens/{token_address}",
]
for url in urls:
try:
data = await _fetch_json(url)
if data:
return data
except Exception:
continue
return None
async def _fetch_token_supply(token_address: str, chain: str) -> float | None:
"""Fetch total token supply from DexScreener or Birdeye.
Gracefully returns None if no source has data (non-fatal).
"""
pairs = await _fetch_dexscreener_pairs(token_address)
if pairs:
for pair in pairs:
chain_id = pair.get("chainId", "").lower()
if chain_id == chain or chain_id == CHAIN_NAMES.get(chain, "").lower():
try:
fdv = pair.get("fdv", 0)
if fdv:
return float(fdv)
except (ValueError, TypeError):
continue
# Try Birdeye as fallback
api_key = _get_env_key("BIRDEYE_API_KEY")
headers = {"x-chain": chain}
if api_key:
headers["x-api-key"] = api_key
url = f"{FREE_APIS['birdeye_public']}/defi/token_overview?address={token_address}"
data = await _fetch_json(url, headers=headers)
if data and data.get("success"):
supply = data.get("data", {}).get("supply")
if supply:
return float(supply)
logger.debug(f"Token supply fetch failed for {token_address} on {chain}")
return None
# ── Core Analysis ─────────────────────────────────────────────────
async def _analyze_single_chain(
token_address: str,
chain: str,
chain_name: str,
pairs: list[dict],
) -> dict[str, Any]:
"""Analyze whale holdings for a token on a single chain."""
result: dict[str, Any] = {
"chain": chain,
"holders_count": 0,
"holders": [],
"supply": None,
"price_usd": 0.0,
"liquidity_usd": 0.0,
"fdv": 0.0,
"volume_24h": 0.0,
"chain_native": chain_name,
}
# Find matching pairs for this chain
chain_pairs = [
p
for p in pairs
if p.get("chainId", "").lower() == chain
or p.get("chainId", "").lower() == chain_name.lower()
]
if chain_pairs:
pair = chain_pairs[0]
with suppress(ValueError, TypeError):
result["price_usd"] = float(pair.get("priceUsd", 0) or 0)
with suppress(ValueError, TypeError):
result["liquidity_usd"] = float(pair.get("liquidity", {}).get("usd", 0) or 0)
with suppress(ValueError, TypeError):
result["fdv"] = float(pair.get("fdv", 0) or 0)
with suppress(ValueError, TypeError):
result["volume_24h"] = float(pair.get("volume", {}).get("h24", 0) or 0)
holders: list[dict] = []
if chain == "solana":
# Solana: Try Birdeye first (needs API key)
birdeye_holders = await _fetch_birdeye_holders(token_address, chain)
if birdeye_holders:
for h in birdeye_holders:
addr = h.get("address", "")
balance = float(h.get("balance", 0) or 0)
balance_usd = float(h.get("balanceUsd", 0) or 0)
percentage = float(h.get("percent", 0) or 0)
holders.append(
{
"address": addr,
"balance": balance,
"balance_usd": balance_usd,
"percentage": percentage,
"source": "birdeye",
}
)
else:
# Fallback: Try Solana free RPC for holder data (multiple endpoints)
solana_rpcs = [
"https://api.mainnet-beta.solana.com",
"https://rpc.ankr.com/solana",
]
for rpc_url in solana_rpcs:
try:
async with httpx.AsyncClient(timeout=10) as client:
# Try to get top holders via getTokenLargestAccounts
largest_resp = await client.post(
rpc_url,
json={
"jsonrpc": "2.0",
"id": 1,
"method": "getTokenLargestAccounts",
"params": [token_address],
},
)
if largest_resp.status_code == 200:
largest_data = largest_resp.json()
if "error" in largest_data:
logger.debug(f"Solana RPC {rpc_url} error: {largest_data['error']}")
continue
accounts = largest_data.get("result", {}).get("value", [])
if not accounts:
continue
total_balance = sum(float(a.get("uiAmount", 0) or 0) for a in accounts)
for acc in accounts[:20]:
addr = acc.get("address", "")
balance = float(acc.get("uiAmount", 0) or 0)
percentage = (
(balance / total_balance * 100) if total_balance > 0 else 0.0
)
balance_usd = balance * result["price_usd"]
holders.append(
{
"address": addr,
"balance": balance,
"balance_usd": balance_usd,
"percentage": round(percentage, 2),
"source": "solana_rpc",
}
)
# Also get total supply
supply_resp = await client.post(
rpc_url,
json={
"jsonrpc": "2.0",
"id": 1,
"method": "getTokenSupply",
"params": [token_address],
},
)
if supply_resp.status_code == 200:
supply_data = supply_resp.json()
if "error" not in supply_data:
supply_val = (
supply_data.get("result", {})
.get("value", {})
.get("uiAmount", 0)
)
if supply_val:
result["supply"] = float(supply_val)
# Break out if we got data
if holders:
break
except Exception as e:
logger.debug(f"Solana RPC {rpc_url} failed: {e}")
continue
elif chain in ETHERSCAN_BASES:
# EVM: Try Etherscan family
etherscan_holders = await _fetch_etherscan_holders(token_address, chain)
for h in etherscan_holders:
addr = h.get("address", "")
try:
balance = float(h.get("balance", 0) or 0) / (10 ** int(h.get("tokenDecimal", 18)))
except (ValueError, TypeError, ZeroDivisionError):
balance = 0.0
try:
percentage = float(h.get("percentage", 0) or 0)
except (ValueError, TypeError):
percentage = 0.0
balance_usd = balance * result["price_usd"]
holders.append(
{
"address": addr,
"balance": balance,
"balance_usd": balance_usd,
"percentage": percentage,
"source": "etherscan",
}
)
result["holders_count"] = len(holders)
result["holders"] = holders[:20] # Top 20 holders
return result
def _is_exchange_address(addr: str) -> bool:
"""Heuristic: check if address looks like a known exchange."""
addr_lower = addr.lower()
known_exchanges = [
"binance",
"coinbase",
"kraken",
"okx",
"bybit",
"crypto.com",
"gate.io",
"kucoin",
"bitfinex",
"huobi",
"bitget",
"mexc",
]
# Check if any exchange name appears in the address or label
return any(exch in addr_lower for exch in known_exchanges)
def _is_smart_money_address(addr: str) -> bool:
"""Heuristic: placeholder for smart money detection."""
return False
async def _cross_reference_whales(
chain_results: dict[str, dict[str, Any]],
token_symbol: str,
token_name: str,
) -> tuple[list[CrossChainWhale], list[CrossChainWhale]]:
"""Cross-reference holders across chains to find multi-chain whales."""
# Collect all holders with their chain info
all_holders: dict[str, list[dict]] = {} # address -> list of positions across chains
for chain, result in chain_results.items():
for h in result.get("holders", []):
addr = h.get("address", "")
if not addr:
continue
if addr not in all_holders:
all_holders[addr] = []
all_holders[addr].append(
{
"chain": chain,
**h,
}
)
whales: list[CrossChainWhale] = []
cross_chain_whales: list[CrossChainWhale] = []
for addr, positions in all_holders.items():
total_value = sum(p.get("balance_usd", 0) for p in positions)
whale_positions = []
for p in positions:
wp = WhalePosition(
chain=p["chain"],
token_address="",
token_symbol=token_symbol,
token_name=token_name,
balance=p.get("balance", 0),
balance_usd=p.get("balance_usd", 0),
percentage=p.get("percentage", 0),
source=p.get("source", ""),
)
whale_positions.append(wp)
cw = CrossChainWhale(
primary_address=addr,
total_value_usd=total_value,
chain_count=len({p["chain"] for p in positions}),
token_count=len(positions),
positions=whale_positions,
is_exchange=_is_exchange_address(addr),
is_smart_money=_is_smart_money_address(addr),
)
whales.append(cw)
if cw.chain_count >= 2:
cross_chain_whales.append(cw)
# Sort by total value descending
whales.sort(key=lambda w: w.total_value_usd, reverse=True)
cross_chain_whales.sort(key=lambda w: w.total_value_usd, reverse=True)
return whales, cross_chain_whales
# ── Public API ────────────────────────────────────────────────────
_instance: "CrossChainWhaleTracker | None" = None
def get_whale_tracker() -> "CrossChainWhaleTracker":
"""Get or create the singleton CrossChainWhaleTracker instance."""
global _instance
if _instance is None:
_instance = CrossChainWhaleTracker()
return _instance
class CrossChainWhaleTracker:
"""Track whale wallets across multiple chains."""
def __init__(self):
self._lock = asyncio.Lock()
async def track_token(
self,
token_address: str,
chains: list[str] | None = None,
) -> WhaleTrackerReport:
"""Track whale holdings for a token across chains.
Args:
token_address: Token contract/mint address.
chains: Specific chains to check (default: all supported chains).
Returns:
WhaleTrackerReport with holders across all checked chains.
"""
report = WhaleTrackerReport(
token_address=token_address,
scan_timestamp=_now_iso(),
)
if chains is None:
chains = list(CHAIN_NAMES.keys())
# Step 1: Fetch DexScreener pairs for token identification
pairs = await _fetch_dexscreener_pairs(token_address)
if pairs:
pair = pairs[0]
# DexScreener puts symbol in baseToken, not at top level
base_token = pair.get("baseToken", {})
report.token_symbol = base_token.get("symbol", "???")
report.token_name = base_token.get("name", "")
# If DexScreener didn't have it, try Jupiter for Solana
if not report.token_symbol:
jup_info = await _fetch_jupiter_token_info(token_address)
if jup_info and isinstance(jup_info, list):
# Jupiter v1 returns array
if jup_info:
report.token_symbol = jup_info[0].get("symbol", "???")
report.token_name = jup_info[0].get("name", "")
elif jup_info:
report.token_symbol = jup_info.get("symbol", "???")
report.token_name = jup_info.get("name", "")
if not report.token_symbol:
report.token_symbol = token_address[:8]
# Step 2: Analyze each chain
chain_tasks = {}
for chain in chains:
chain_name = CHAIN_NAMES.get(chain, chain)
task = _analyze_single_chain(token_address, chain, chain_name, pairs)
chain_tasks[chain] = task
chain_results = {}
for chain, task in chain_tasks.items():
try:
result = await task
chain_results[chain] = result
if result["holders_count"] > 0:
report.chains_found.append(chain)
report.chains_with_data.append(chain)
else:
report.chains_no_data.append(chain)
report.total_holders += result["holders_count"]
except Exception as e:
report.chains_no_data.append(chain)
report.errors.append(f"{chain}: {e}")
# Step 3: Build top holders by chain
for chain, result in chain_results.items():
chain_top = []
for i, h in enumerate(result.get("holders", [])[:5]):
chain_top.append(
{
"rank": i + 1,
"address": h.get("address", ""),
"address_short": _truncate_address(h.get("address", "")),
"balance": h.get("balance", 0),
"balance_usd": h.get("balance_usd", 0),
"percentage": h.get("percentage", 0),
"source": h.get("source", ""),
}
)
if chain_top:
report.top_holders_by_chain[chain] = chain_top
# Step 4: Cross-reference across chains
if chain_results:
whales, cross_chain = await _cross_reference_whales(
chain_results, report.token_symbol, report.token_name
)
report.whales = whales
report.cross_chain_whales = cross_chain
# Step 5: Calculate aggregate metrics
report.total_value_tracked = sum(r.get("liquidity_usd", 0) for r in chain_results.values())
# Concentration score: how concentrated are top holders?
all_top_holders = []
for result in chain_results.values():
for h in result.get("holders", [])[:3]:
if h.get("percentage", 0) > 0:
all_top_holders.append(h)
if all_top_holders:
avg_top_pct = sum(h.get("percentage", 0) for h in all_top_holders) / len(
all_top_holders
)
report.concentration_score = round(min(avg_top_pct * 2, 100), 1)
# Step 6: Cross-chain whale risk scoring
for cw in report.cross_chain_whales:
risk = 0.0
factors = []
if cw.chain_count >= 4:
risk += 20
factors.append("Extreme cross-chain presence (4+ chains)")
elif cw.chain_count >= 3:
risk += 10
factors.append("Significant cross-chain presence (3 chains)")
if cw.total_value_usd > 1_000_000:
risk += 15
factors.append("High-value whale ($1M+)")
if cw.is_exchange:
risk = 5 # Exchange wallets are low risk
factors = ["Exchange wallet"]
elif cw.is_scammer:
risk = 85
factors.append("Known scammer address")
elif cw.is_smart_money:
risk = 10
factors.append("Smart money wallet")
cw.risk_score = round(min(risk, 100), 1)
cw.risk_factors = factors
report.chains_found = list(dict.fromkeys(report.chains_found))
return report
async def track_wallet(
self,
wallet_address: str,
chains: list[str] | None = None,
max_positions: int = 20,
) -> dict[str, Any]:
"""Track a specific wallet across chains to find all token positions.
Args:
wallet_address: The wallet address to investigate.
chains: Chains to check (default: all).
max_positions: Maximum tokens to return.
Returns:
Dict with wallet's positions across chains.
"""
if chains is None:
chains = [*list(ETHERSCAN_BASES.keys()), "solana"]
result: dict[str, Any] = {
"wallet": wallet_address,
"wallet_short": _truncate_address(wallet_address),
"chains_checked": chains,
"total_value_usd": 0.0,
"positions": [],
"errors": [],
"scan_timestamp": _now_iso(),
}
return result
# ── Formatting ─────────────────────────────────────────────────────
def format_whale_report(report: WhaleTrackerReport) -> str:
"""Format a WhaleTrackerReport for human-readable output."""
lines = []
lines.append(f"🐋 Cross-Chain Whale Report: {report.token_symbol}")
lines.append(f" Token: {report.token_name or 'N/A'}")
lines.append(f" Address: {report.token_address}")
lines.append(f" Chains Found: {', '.join(report.chains_found) or 'None'}")
lines.append(f" Total Holders Tracked: {report.total_holders}")
lines.append(f" Concentration Score: {report.concentration_score}/100")
lines.append("")
# Top holders by chain
for chain, holders in report.top_holders_by_chain.items():
chain_name = CHAIN_NAMES.get(chain, chain)
lines.append(f" 📍 Top {chain_name} Holders:")
for h in holders[:5]:
pct = f"{h['percentage']:.2f}%" if h["percentage"] else ""
usd = f"${h['balance_usd']:,.2f}" if h["balance_usd"] else ""
lines.append(f" #{h['rank']} {h['address_short']} {pct} {usd}")
lines.append("")
# Cross-chain whales
if report.cross_chain_whales:
lines.append(" 🔗 Cross-Chain Whales (on 2+ chains):")
for cw in report.cross_chain_whales[:10]:
chains_str = ", ".join(CHAIN_NAMES.get(p.chain, p.chain) for p in cw.positions)
value_str = f"${cw.total_value_usd:,.2f}" if cw.total_value_usd else ""
risk_str = f" [Risk: {cw.risk_score}/100]" if cw.risk_score > 0 else ""
addr_short = _truncate_address(cw.primary_address)
lines.append(f" 🐋 {addr_short} on {cw.chain_count} chains ({chains_str})")
lines.append(f" Value: {value_str}{risk_str}")
if cw.risk_factors:
for rf in cw.risk_factors[:3]:
lines.append(f" ⚠ {rf}")
lines.append("")
# Top cross-chain whale
if report.cross_chain_whales:
top = report.cross_chain_whales[0]
lines.append(" 👑 Top Cross-Chain Whale:")
lines.append(f" Address: {top.primary_address}")
lines.append(f" Chains: {top.chain_count}")
lines.append(f" Est. Value: ${top.total_value_usd:,.2f}")
lines.append("")
if report.errors:
lines.append(f" ⚠ Errors ({len(report.errors)}):")
for err in report.errors[:3]:
lines.append(f" - {err}")
lines.append(f" Scan: {report.scan_timestamp}")
return "\n".join(lines)
# ── Address validation ────────────────────────────────────────────
def is_valid_address(address: str) -> bool:
"""Basic address validation for Solana and EVM addresses."""
if not address or len(address) < 32:
return False
# Solana: base58, 32-44 chars
# EVM: hex, 42 chars with 0x prefix
if address.startswith("0x"):
if len(address) != 42:
return False
try:
int(address, 16)
return True
except ValueError:
return False
# Solana-like: base58
if len(address) >= 32 and len(address) <= 44:
import string
base58 = string.digits + string.ascii_uppercase + string.ascii_lowercase
# Remove commonly confused chars
base58 = base58.replace("0", "").replace("O", "").replace("I", "").replace("l", "")
return all(c in base58 for c in address)
return False
# ── Quick test ────────────────────────────────────────────────────
async def _quick_test() -> None:
"""Run a quick smoke test with a well-known token."""
# Test with USDC on Solana (known token)
test_token = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"
tracker = get_whale_tracker()
report = await tracker.track_token(test_token, chains=["solana", "ethereum", "base"])
print(format_whale_report(report))
print(f"\nErrors: {len(report.errors)}")
print("Quick test passed ✓" if len(report.errors) < 3 else "Quick test FAILED ✗")
if __name__ == "__main__":
asyncio.run(_quick_test())
Reference in a new issue View git blame Copy permalink