Smart Money 追蹤器

本教程介紹如何構建一個 Smart Money 追蹤系統，實時監控聰明錢地址的交易行為，獲取有價值的市場訊號。

預計時間：45 分鐘
難度等級：⭐⭐⭐ 中級

目標

構建完整的 Smart Money 追蹤系統： 功能清單：

✅ 獲取 Smart Money 列表
✅ 訂閱 SM 交易實時流
✅ 交易解析和模式識別
✅ 訊號通知

方法論回顧

在開始之前，建議先閱讀 Smart Money 方法論，瞭解：

Smart Money 的定義和型別
評分模型（Smart Score）
訊號強度解讀
使用注意事項

Step 1：獲取 SM 列表

API 獲取 Smart Money 列表

# sm_client.py
import requests
from config import CHAINSTREAM_ACCESS_TOKEN, CHAINSTREAM_BASE_URL

class SmartMoneyClient:
    def __init__(self):
        self.headers = {'Authorization': f'Bearer {CHAINSTREAM_ACCESS_TOKEN}'}
    
    def get_smart_money_list(self, chain: str = 'ethereum', limit: int = 100) -> list:
        """获取 Smart Money 列表"""
        url = f"{CHAINSTREAM_BASE_URL}/discovery/smart-money"
        
        params = {
            'chain': chain,
            'limit': limit,
            'min_score': 70,  # 最低 Smart Score
            'sort': 'score_desc'
        }
        
        response = requests.get(url, headers=self.headers, params=params)
        return response.json().get('wallets', [])
        
        # 返回示例:
        # [
        #   {
        #     "address": "0xabc...",
        #     "smart_score": 92,
        #     "tags": ["whale", "defi_expert"],
        #     "stats": {
        #       "roi_90d": "45.2%",
        #       "win_rate": "72%"
        #     }
        #   },
        #   ...
        # ]
    
    def get_wallet_activities(self, address: str, limit: int = 20) -> list:
        """获取钱包最近交易"""
        url = f"{CHAINSTREAM_BASE_URL}/wallets/{address}/activities"
        
        params = {'limit': limit}
        response = requests.get(url, headers=self.headers, params=params)
        return response.json().get('activities', [])

使用預設列表

# ChainStream 提供预设的 SM 列表
PRESET_LISTS = {
    'ethereum_top_100': 'eth_top_traders',
    'solana_smart_money': 'sol_smart_money',
    'base_early_adopters': 'base_early'
}

def get_preset_list(list_id: str) -> list:
    url = f"{CHAINSTREAM_BASE_URL}/discovery/lists/{list_id}"
    response = requests.get(url, headers=headers)
    return response.json().get('wallets', [])

Step 2：訂閱交易

WebSocket 訂閱

本教程中的 WebSocket 示例使用簡化的虛擬碼來說明概念。實際的 WebSocket 訂閱格式和可用頻道，請參閱 WebSocket API 參考文件。

# tracker.py
import asyncio
import json
import websockets
from config import CHAINSTREAM_ACCESS_TOKEN

class SmartMoneyTracker:
    def __init__(self, wallets: list):
        self.wallets = wallets
        self.ws_url = 'wss://realtime-dex.chainstream.io/connection/websocket'
        self.ws = None
    
    async def connect(self):
        """连接 WebSocket"""
        # 通过 URL 参数传递 token
        ws_url = f"{self.ws_url}?token={CHAINSTREAM_ACCESS_TOKEN}"
        self.ws = await websockets.connect(ws_url)
        print("✅ WebSocket 连接成功")
    
    async def subscribe(self):
        """订阅 Smart Money 交易"""
        # 方式1：订阅特定地址
        for wallet in self.wallets[:50]:  # 限制订阅数量
            msg = {
                'type': 'subscribe',
                'channel': 'wallet_activity',
                'params': {
                    'address': wallet['address'],
                    'events': ['swap', 'transfer']
                }
            }
            await self.ws.send(json.dumps(msg))
        
        # 方式2：订阅 Smart Money 聚合频道（推荐）
        msg = {
            'type': 'subscribe',
            'channel': 'smart_money',
            'params': {
                'chains': ['ethereum', 'arbitrum', 'base'],
                'min_score': 70,
                'min_value_usd': 10000
            }
        }
        await self.ws.send(json.dumps(msg))
        
        print("📡 已订阅 Smart Money 交易流")
    
    async def listen(self, callback):
        """监听交易事件"""
        async for message in self.ws:
            data = json.loads(message)
            
            if data.get('type') == 'smart_money_activity':
                await callback(data)
    
    async def run(self, callback):
        """运行追踪器"""
        while True:
            try:
                await self.connect()
                await self.subscribe()
                await self.listen(callback)
            except websockets.ConnectionClosed:
                print("⚠️ 连接断开，5秒后重连...")
                await asyncio.sleep(5)

或使用 Webhook

# 创建 Webhook 订阅 Smart Money 活动
webhook_config = {
    'url': 'https://your-server.com/sm-webhook',
    'events': ['smartmoney.buy', 'smartmoney.sell', 'smartmoney.new_token'],
    'filters': {
        'min_score': 70,
        'min_value_usd': 10000,
        'chains': ['ethereum', 'arbitrum']
    }
}

Step 3：分析與通知

交易解析

# analyzer.py
from dataclasses import dataclass
from typing import Optional

@dataclass
class SmartMoneySignal:
    wallet_address: str
    wallet_score: int
    wallet_tags: list
    action: str  # buy, sell
    token_symbol: str
    token_address: str
    amount_usd: float
    price: float
    signal_strength: str  # strong, medium, weak
    timestamp: str

class SignalAnalyzer:
    
    def analyze(self, event: dict) -> Optional[SmartMoneySignal]:
        """分析交易事件，生成信号"""
        wallet = event.get('wallet', {})
        transaction = event.get('transaction', {})
        
        # 计算信号强度
        strength = self._calc_signal_strength(wallet, transaction)
        
        if strength == 'ignore':
            return None
        
        return SmartMoneySignal(
            wallet_address=wallet['address'],
            wallet_score=wallet.get('smart_score', 0),
            wallet_tags=wallet.get('tags', []),
            action=transaction.get('type', 'unknown'),
            token_symbol=transaction.get('token', {}).get('symbol', 'Unknown'),
            token_address=transaction.get('token', {}).get('address', ''),
            amount_usd=float(transaction.get('value_usd', 0)),
            price=float(transaction.get('price', 0)),
            signal_strength=strength,
            timestamp=event.get('timestamp', '')
        )
    
    def _calc_signal_strength(self, wallet: dict, tx: dict) -> str:
        """计算信号强度"""
        score = wallet.get('smart_score', 0)
        value_usd = float(tx.get('value_usd', 0))
        is_new_token = tx.get('is_first_buy', False)
        
        # 强信号条件
        if score >= 85 and value_usd >= 100000:
            return 'strong'
        if is_new_token and score >= 80:
            return 'strong'
        
        # 中等信号
        if score >= 70 and value_usd >= 50000:
            return 'medium'
        
        # 弱信号
        if score >= 60 and value_usd >= 10000:
            return 'weak'
        
        return 'ignore'

通知模組

# notifier.py
import aiohttp
from config import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID

class Notifier:
    
    async def send_signal(self, signal):
        """发送信号通知"""
        
        # 信号强度 emoji
        strength_emoji = {
            'strong': '🔴',
            'medium': '🟡',
            'weak': '🟢'
        }
        
        # 动作 emoji
        action_emoji = '📈' if signal.action == 'buy' else '📉'
        
        message = f"""
{strength_emoji.get(signal.signal_strength, '⚪')} Smart Money 信号

{action_emoji} **{signal.action.upper()}** {signal.token_symbol}

👤 地址: `{signal.wallet_address[:10]}...`
⭐ Smart Score: {signal.wallet_score}
🏷️ 标签: {', '.join(signal.wallet_tags)}

💰 金额: ${signal.amount_usd:,.0f}
💵 价格: ${signal.price:.6f}

⏰ {signal.timestamp}
        """
        
        await self._send_telegram(message)
    
    async def _send_telegram(self, message: str):
        """发送 Telegram 消息"""
        url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
        
        async with aiohttp.ClientSession() as session:
            await session.post(url, json={
                'chat_id': TELEGRAM_CHAT_ID,
                'text': message,
                'parse_mode': 'Markdown'
            })

完整程式碼

# main.py
import asyncio
from sm_client import SmartMoneyClient
from tracker import SmartMoneyTracker
from analyzer import SignalAnalyzer
from notifier import Notifier

async def main():
    # 初始化
    client = SmartMoneyClient()
    analyzer = SignalAnalyzer()
    notifier = Notifier()
    
    # 获取 Smart Money 列表
    print("📋 获取 Smart Money 列表...")
    wallets = client.get_smart_money_list(chain='ethereum', limit=100)
    print(f"✅ 获取到 {len(wallets)} 个 Smart Money 地址")
    
    # 定义回调处理
    async def on_activity(event):
        signal = analyzer.analyze(event)
        
        if signal:
            print(f"🎯 {signal.signal_strength.upper()} | {signal.action} {signal.token_symbol} | ${signal.amount_usd:,.0f}")
            await notifier.send_signal(signal)
    
    # 启动追踪
    tracker = SmartMoneyTracker(wallets)
    print("🚀 启动 Smart Money 追踪器...")
    await tracker.run(on_activity)

if __name__ == '__main__':
    asyncio.run(main())

應用場景

跟單參考

# 筛选强信号用于参考
def filter_strong_signals(signals: list) -> list:
    return [
        s for s in signals 
        if s.signal_strength == 'strong' 
        and s.action == 'buy'
    ]

市場情緒指標

# 计算 SM 买卖比
def calc_sentiment(signals: list, hours: int = 24) -> dict:
    recent = filter_recent(signals, hours)
    
    buys = sum(1 for s in recent if s.action == 'buy')
    sells = sum(1 for s in recent if s.action == 'sell')
    
    ratio = buys / max(sells, 1)
    
    return {
        'buy_count': buys,
        'sell_count': sells,
        'ratio': ratio,
        'sentiment': 'bullish' if ratio > 1.5 else 'bearish' if ratio < 0.7 else 'neutral'
    }

注意事項

重要提示：

不要盲目跟單 — SM 訊號僅供參考
注意延遲 — 鏈上確認 + 分析存在延遲
分散關注 — 多個 SM 共振訊號更有價值
獨立研究 — 結合基本面做判斷

Smart Money 方法論

理解 SM 評分模型

Webhook 基礎

使用 Webhook 替代 WebSocket

資料產品

接入方式

AI Agents

合規 (KYT/KYA)

教學

實戰範例

平台

生態

參考

Smart Money 追蹤器

目標

方法論回顧

Step 1：獲取 SM 列表

API 獲取 Smart Money 列表

使用預設列表

Step 2：訂閱交易

WebSocket 訂閱

或使用 Webhook

Step 3：分析與通知

交易解析

通知模組

完整程式碼

應用場景

跟單參考

市場情緒指標

注意事項

相關文件

Smart Money 方法論

Webhook 基礎

​目標

​方法論回顧

​Step 1：獲取 SM 列表

​API 獲取 Smart Money 列表

​使用預設列表

​Step 2：訂閱交易

​WebSocket 訂閱

​或使用 Webhook

​Step 3：分析與通知

​交易解析

​通知模組

​完整程式碼

​應用場景

​跟單參考

​市場情緒指標

​注意事項

​相關文件

Smart Money 方法論

Webhook 基礎

目標

方法論回顧

Step 1：獲取 SM 列表

API 獲取 Smart Money 列表

使用預設列表

Step 2：訂閱交易

WebSocket 訂閱

或使用 Webhook

Step 3：分析與通知

交易解析

通知模組

完整程式碼

應用場景

跟單參考

市場情緒指標

注意事項

相關文件