Разработка бота для копирования сделок

Разработка бота для копирования сделок

Бот копирования сделок — это автоматизированный инструмент, который отслеживает сделки выбранного трейдера (источника) и воспроизводит их на вашем аккаунте. В отличие от полноценной системы копитрейдинга на бирже, этот бот работает автономно: получает сигналы через 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 сигналы

Многие traders публикуют сигналы в 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 wallet мониторинг (DeFi)

Для DeFi: мониторинг on-chain транзакций известного кошелька (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 (зависит от network)

Риск-менеджмент для копирующего бота

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 ухудшает исполнение, источник может изменить стратегию, bot может пропустить сигнал при отключении. Обязательны логирование всех операций и Telegram алерты о проблемах.