Розробка бота для копіювання угод

Розробка бота для копіювання сділок

Бот копіювання сділок — це автоматизований інструмент, який відслідковує сділки обраного трейдера (джерела) та відтворює їх на вашому аккаунті. На відміну від повноцінної системи копітрейдингу на біржі, цей бот працює автономно: отримує сигнали через API, Telegram, webhooks або напрямку моніторить публічні ончейн-гаманці.

Джерела сигналів

1. Exchange API (той же аккаунт / sub-account)

Якщо джерело торгує на тій же біржі, найнадійніший варіант — polling його позицій через API:

class ExchangeCopyTrader:
    def __init__(self, source_api_key: str, follower_api_key: str, exchange: str):
        self.source_client = ExchangeClient(source_api_key)
        self.follower_client = ExchangeClient(follower_api_key)
        self.tracked_positions: dict = {}

    async def sync_positions(self):
        """Синхронізуємо позиції кожні N секунд"""
        source_positions = await self.source_client.get_open_positions()
        follower_positions = await self.follower_client.get_open_positions()

        source_map = {p.symbol: p for p in source_positions}
        follower_map = {p.symbol: p for p in follower_positions}

        # Нові позиції у джерела — відкриваємо у follower
        for symbol, pos in source_map.items():
            if symbol not in follower_map:
                await self.open_copied_position(pos)

        # Позиції закриті у джерела — закриваємо у follower
        for symbol in follower_map:
            if symbol not in source_map:
                await self.close_copied_position(symbol)

        # Змінився розмір позиції — коригуємо
        for symbol in source_map:
            if symbol in follower_map:
                source_size = source_map[symbol].size
                follower_size = follower_map[symbol].size
                scaled_size = source_size * self.config.scale_factor
                if abs(follower_size - scaled_size) / scaled_size > 0.05:
                    await self.adjust_position(symbol, scaled_size)

2. Webhook / Telegram сигнали

Багато трейдерів публікують сигнали в Telegram каналах. Бот парсить сообщення:

class TelegramSignalParser:
    # Паттерн: "BUY BTCUSDT @ 50000, SL: 48000, TP: 55000"
    SIGNAL_PATTERN = r'(BUY|SELL)\s+(\w+)\s+@\s+([\d.]+)(?:.*SL:\s*([\d.]+))?(?:.*TP:\s*([\d.]+))?'

    def parse_message(self, text: str) -> TradeSignal | None:
        import re
        match = re.search(self.SIGNAL_PATTERN, text, re.IGNORECASE)
        if not match:
            return None

        return TradeSignal(
            action=match.group(1).upper(),
            symbol=match.group(2).upper(),
            entry_price=float(match.group(3)),
            stop_loss=float(match.group(4)) if match.group(4) else None,
            take_profit=float(match.group(5)) if match.group(5) else None,
            source='telegram'
        )

3. On-chain гаманець моніторинг (DeFi)

Для DeFi: моніторинг ончейн транзакцій відомого гаманця (whale tracking):

class OnChainCopyTrader:
    def __init__(self, target_wallet: str, dex_router: str):
        self.target = target_wallet
        self.router = dex_router

    async def monitor_swaps(self):
        """Слухаємо Transfer та Swap события з гаманця target"""
        filter_params = {
            'fromBlock': 'latest',
            'address': self.router,
            'topics': [
                UNISWAP_SWAP_EVENT_TOPIC,
                None,  # from
                self.w3.keccak(text=self.target)
            ]
        }

        async for log in self.w3.eth.subscribe('logs', filter_params):
            swap = self.decode_swap_log(log)
            if swap and swap.amount_in_usd > self.config.min_copy_usd:
                await self.copy_swap(swap)

Масштабування позицій

def calculate_copy_size(
    source_trade: Trade,
    source_balance: float,
    follower_balance: float,
    mode: str = 'proportional',
    multiplier: float = 1.0
) -> float:
    if mode == 'proportional':
        # Копіюємо той же % капіталу
        source_percent = source_trade.size_usd / source_balance
        return follower_balance * source_percent * multiplier

    elif mode == 'fixed':
        return min(source_trade.size_usd * multiplier, follower_balance * 0.1)

    elif mode == 'fixed_risk':
        # Фіксований ризик на сділку (% від баланса)
        if source_trade.stop_loss:
            risk_percent = abs(source_trade.entry - source_trade.stop_loss) / source_trade.entry
            max_loss = follower_balance * (self.config.risk_per_trade / 100)
            return max_loss / risk_percent
        return follower_balance * 0.02  # default 2%

Затримка та Latency

Latency між сигналом та виконанням критична: поки ваш бот обробляє сигнал, ціна рухається:

class LatencyMonitor:
    def __init__(self):
        self.latencies = []

    async def execute_with_tracking(self, signal: TradeSignal) -> Execution:
        t0 = time.perf_counter()

        # Виконуємо ордер
        order = await self.exchange.place_market_order(
            symbol=signal.symbol,
            side=signal.action.lower(),
            amount=self.calculate_size(signal)
        )

        t1 = time.perf_counter()
        latency_ms = (t1 - t0) * 1000

        self.latencies.append(latency_ms)
        logger.info(f"Copy latency: {latency_ms:.1f}ms, fill: {order.fill_price}")

        # Алерт якщо занадто повільно
        if latency_ms > 500:
            await self.alert(f"High latency: {latency_ms:.0f}ms for {signal.symbol}")

        return order

Типові latency для різних джерел:

• Exchange sub-account API → ~50-200ms
• Telegram webhook → ~200-500ms
• On-chain моніторинг → ~1000-3000ms (залежить від мережі)

Риск-менеджмент для копіючого бота

class CopyBotRiskManager:
    def can_copy(self, signal: TradeSignal, account_state: AccountState) -> tuple[bool, str]:
        # Загальний drawdown
        if account_state.drawdown_percent > self.config.max_drawdown:
            return False, "max_drawdown_exceeded"

        # Кількість одночасних позицій
        if len(account_state.open_positions) >= self.config.max_positions:
            return False, "max_positions_reached"

        # Уже є позиція по цьому символу
        if signal.symbol in account_state.open_positions:
            return False, "symbol_already_open"

        # Мінімальний баланс для відкриття
        required = self.calculate_required_margin(signal)
        if account_state.free_balance < required * 1.1:  # 10% буфер
            return False, "insufficient_balance"

        return True, "ok"

Copy бот — відличний інструмент для трейдерів, які довіряють конкретному сигналізатору або хочуть автоматизувати слідування стратегії. Ключові ризики: latency погіршує виконання, джерело може змінити стратегію, бот може пропустити сигнал при відключенні. Обов'язково: логування всіх операцій та Telegram алерти про проблеми.