Development of Smart Money Movements Alerts System
Smart money — a term for large professional market participants: hedge funds, market makers, insiders. The idea of monitoring their movements is simple: if large players start accumulating or liquidating positions, it's a useful signal for regular traders. The problem — identification of these participants in the pseudo-anonymous blockchain.
What Counts as "Smart Money"
On-chain signals:
- Wallet addresses associated with known funds (a16z, Paradigm, Multicoin Capital) — publicly known from investment announcements
- Accumulation of obscure tokens in 1–2 weeks before major announcements (listing, partnership)
- Activity of large LP positions in DeFi protocols
On-exchange signals:
- Large limit orders in order book
- Unusual activity in options contracts (large OI buildup, unusual options flow)
- Funding rate + OI divergence (smart money accumulates against retail trend)
System Architecture
class SmartMoneyTracker:
def __init__(self, address_db, chain_client, alert_engine):
self.known_wallets: dict[str, WalletProfile] = {} # address → profile
self.chain = chain_client
self.alerts = alert_engine
async def load_known_wallets(self):
"""Load database of known wallets from multiple sources"""
# 1. Manual database (funds, known traders)
manual = await self.load_manual_database()
# 2. Nansen Smart Money API (commercial service)
nansen = await self.load_nansen_labels()
# 3. Arkham Intelligence API
arkham = await self.load_arkham_labels()
self.known_wallets = {**manual, **nansen, **arkham}
async def monitor_ethereum(self):
"""Real-time monitoring of Ethereum transactions"""
async for block in self.chain.subscribe_blocks():
for tx in block.transactions:
await self.analyze_transaction(tx)
async def analyze_transaction(self, tx: EthTransaction):
from_profile = self.known_wallets.get(tx.from_address.lower())
to_profile = self.known_wallets.get(tx.to_address.lower())
if not from_profile and not to_profile:
return # Ordinary transaction, not interesting
# Classify action
if tx.to_address == UNISWAP_V3_ROUTER:
await self.analyze_dex_trade(tx, from_profile)
elif await self.is_token_transfer(tx):
await self.analyze_token_movement(tx, from_profile, to_profile)
elif await self.is_defi_interaction(tx):
await self.analyze_defi_position(tx, from_profile)
Nansen API Integration
Nansen is a commercial service with the largest database of labeled wallets:
class NansenClient:
BASE_URL = "https://api.nansen.ai/v1"
def __init__(self, api_key: str):
self.session = httpx.AsyncClient(
headers={"n-api-key": api_key}
)
async def get_wallet_labels(self, address: str) -> list[str]:
"""Get wallet labels (Smart Money, Exchange, Whale, etc.)"""
resp = await self.session.get(
f"{self.BASE_URL}/labels/address/{address}"
)
if resp.status_code == 404:
return []
data = resp.json()
return data.get("labels", [])
async def get_token_god_mode(self, token_address: str) -> dict:
"""Analyze token holders: who's accumulating, who's selling"""
resp = await self.session.get(
f"{self.BASE_URL}/token/godMode",
params={"token_address": token_address}
)
return resp.json()
async def get_smart_money_flows(
self,
token_address: str,
days: int = 7
) -> dict:
"""Net flow of smart money for token for period"""
resp = await self.session.get(
f"{self.BASE_URL}/token/smartMoney",
params={"token_address": token_address, "days": days}
)
data = resp.json()
return {
"net_flow_usd": data["netFlowUSD"],
"buyers": data["smartMoneyBuyers"],
"sellers": data["smartMoneySellers"],
"unique_wallets": data["uniqueWallets"],
}
DEX Activity Monitoring
Monitoring trading activity in Uniswap/Curve/Balancer:
class DEXActivityMonitor:
UNISWAP_V3_SUBGRAPH = "https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3"
async def get_recent_large_swaps(
self,
min_usd: float = 100_000,
hours: int = 24
) -> list[dict]:
query = """
query LargeSwaps($minUSD: String!, $since: Int!) {
swaps(
where: {amountUSD_gt: $minUSD, timestamp_gt: $since}
orderBy: amountUSD
orderDirection: desc
first: 100
) {
id
timestamp
token0 { symbol }
token1 { symbol }
amountUSD
origin
transaction { id }
}
}
"""
since = int((datetime.now() - timedelta(hours=hours)).timestamp())
resp = await self.graphql_client.query(self.UNISWAP_V3_SUBGRAPH, query, {
"minUSD": str(min_usd),
"since": since
})
swaps = resp["data"]["swaps"]
# Enrich with wallet data
enriched = []
for swap in swaps:
labels = await self.nansen.get_wallet_labels(swap["origin"])
if "Smart Money" in labels or "Fund" in labels:
enriched.append({**swap, "labels": labels})
return enriched
Accumulation Patterns
Accumulation detector: address sequentially buys a token in small portions:
class AccumulationDetector:
async def detect_accumulation(
self,
address: str,
token: str,
days: int = 14
) -> AccumulationPattern:
transfers = await self.get_token_transfers(address, token, days)
# Only incoming
incoming = [t for t in transfers if t.to_address == address]
if len(incoming) < 3:
return None
total_accumulated = sum(t.value for t in incoming)
avg_interval = self.avg_time_between(incoming)
is_consistent = self.is_consistent_buying(incoming)
if is_consistent and len(incoming) >= 5:
return AccumulationPattern(
address=address,
token=token,
transactions=len(incoming),
total_value_usd=total_accumulated,
avg_interval_hours=avg_interval,
start_date=incoming[0].timestamp,
strength="STRONG" if len(incoming) >= 10 else "MODERATE",
)
Alert Formatting and Delivery
def format_smart_money_alert(event: SmartMoneyEvent) -> str:
labels = ", ".join(event.wallet_labels) or "Smart Money"
action_map = {
"BUY": "🟢 accumulating",
"SELL": "🔴 selling",
"DEPOSIT_TO_EXCHANGE": "📤 depositing to exchange",
"WITHDRAW_FROM_EXCHANGE": "📥 withdrawing from exchange",
}
action = action_map.get(event.action, event.action)
return f"""🧠 Smart Money Alert
{labels} {action} {event.token}
Amount: ${event.usd_value:,.0f}
{f"Total for 7d: ${event.rolling_7d_usd:,.0f}" if event.rolling_7d_usd else ""}
Wallet: {event.address[:6]}...{event.address[-4:]}
🔗 {event.explorer_url}"""
Smart money monitoring system is not a silver bullet. Smart money makes mistakes, insider activity doesn't always predict growth. It's one of many signals that should be analyzed in context of other data. But when implemented correctly, it provides information advantage unavailable to users without such tools.