Разработка системы алертов по smart money movements

Разработка системы алертов по smart money movements

Smart money — термин для обозначения крупных профессиональных участников рынка: хедж-фондов, маркет-мейкеров, инсайдеров. Идея мониторинга их движений проста: если крупные игроки начинают накапливать или распродавать позиции, это полезный сигнал для обычных трейдеров. Проблема — идентификация этих участников в псевдоанонимном блокчейне.

Что считается "smart money"

On-chain сигналы:

• Адреса кошельков, ассоциированных с известными фондами (a16z, Paradigm, Multicoin Capital) — публично известны из инвестиционных объявлений
• Аккумуляция малоизвестных токенов за 1–2 недели до мейджорных объявлений (листинг, партнёрство)
• Активность крупных LP-позиций в DeFi протоколах

On-exchange сигналы:

• Крупные limit orders в стакане
• Необычная активность в опционных контрактах (large OI buildup, unusual options flow)
• Funding rate + OI расхождение (smart money накапливает против розничного тренда)

Архитектура системы

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):
        """Загружаем базу известных кошельков из нескольких источников"""
        # 1. Ручная база данных (фонды, известные трейдеры)
        manual = await self.load_manual_database()

        # 2. Nansen Smart Money API (коммерческий сервис)
        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):
        """Мониторинг транзакций Ethereum в реальном времени"""
        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  # Обычная транзакция, не интересна

        # Классифицируем действие
        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 интеграция

Nansen — коммерческий сервис с крупнейшей базой labeled кошельков:

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]:
        """Получить ярлыки кошелька (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:
        """Анализ держателей токена: кто накапливает, кто продаёт"""
        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 smart money по токену за период"""
        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-активность smart money

Мониторинг торговой активности в 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"]

        # Обогащаем данными о кошельках
        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

Паттерны накопления

Детектор накопления: адрес последовательно покупает токен небольшими порциями:

class AccumulationDetector:
    async def detect_accumulation(
        self,
        address: str,
        token: str,
        days: int = 14
    ) -> AccumulationPattern:
        transfers = await self.get_token_transfers(address, token, days)

        # Только входящие
        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",
            )

Форматирование и доставка алертов

def format_smart_money_alert(event: SmartMoneyEvent) -> str:
    labels = ", ".join(event.wallet_labels) or "Smart Money"

    action_map = {
        "BUY": "🟢 накапливает",
        "SELL": "🔴 продаёт",
        "DEPOSIT_TO_EXCHANGE": "📤 вносит на биржу",
        "WITHDRAW_FROM_EXCHANGE": "📥 выводит с биржи",
    }
    action = action_map.get(event.action, event.action)

    return f"""🧠 Smart Money Alert

{labels} {action} {event.token}

Сумма: ${event.usd_value:,.0f}
{f"Всего за 7д: ${event.rolling_7d_usd:,.0f}" if event.rolling_7d_usd else ""}
Кошелёк: {event.address[:6]}...{event.address[-4:]}

🔗 {event.explorer_url}"""

Система smart money мониторинга — не серебряная пуля. Smart money ошибается, инсайдерская активность не всегда предсказывает рост. Это один из многих сигналов, который нужно анализировать в контексте других данных. Но при правильной реализации он даёт информационное преимущество, недоступное пользователям без подобных инструментов.