> ## Documentation Index
> Fetch the complete documentation index at: https://docs.chainstream.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Smart Money 追蹤器

> 追蹤 Smart Money 地址，獲取交易訊號

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

<Info>
  **預計時間**：45 分鐘\
  **難度等級**：⭐⭐⭐ 中級
</Info>

***

## 目標

構建完整的 Smart Money 追蹤系統：

```mermaid theme={null}
flowchart TD
    A[获取 SM 列表] --> B[订阅交易 WebSocket]
    B --> C[分析通知]
```

**功能清單**：

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

***

## 方法論回顧

在開始之前，建議先閱讀 [Smart Money 方法論](/zh-Hant/docs/data-products/smart-money)，瞭解：

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

***

## Step 1：獲取 SM 列表

### API 獲取 Smart Money 列表

```python theme={null}
# 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', [])
```

### 使用預設列表

```python theme={null}
# 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 訂閱

<Info>
  本教程中的 WebSocket 示例使用簡化的虛擬碼來說明概念。實際的 WebSocket 訂閱格式和可用頻道，請參閱 [WebSocket API 參考文件](/zh-Hant/api-reference/endpoint/websocket/api)。
</Info>

```python theme={null}
# 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

```python theme={null}
# 创建 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：分析與通知

### 交易解析

```python theme={null}
# 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'
```

### 通知模組

```python theme={null}
# 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'
            })
```

***

## 完整程式碼

```python theme={null}
# 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())
```

***

## 應用場景

### 跟單參考

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

### 市場情緒指標

```python theme={null}
# 计算 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'
    }
```

***

## 注意事項

<Warning>
  **重要提示**：

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

***

## 相關文件

<CardGroup cols={2}>
  <Card title="Smart Money 方法論" icon="brain" href="/zh-Hant/docs/data-products/smart-money">
    理解 SM 評分模型
  </Card>

  <Card title="Webhook 基礎" icon="webhook" href="/zh-Hant/docs/recipes/webhook-fundamentals">
    使用 Webhook 替代 WebSocket
  </Card>
</CardGroup>