Розробка системи алертів за 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, і т.д.)"""
        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 помиляється, інсайдерська активність не завжди передбачає ріст. Це один з багатьох сигналів, який потрібно аналізувати в контексті інших даних. Але при правильній реалізації він дає інформаційну перевагу, недоступну користувачам без подібних інструментів.