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

Трейдер ведёт канал с сигналами в Telegram, но вы не успеваете вручную открывать сделки? Бот для копирования сделок решает эту проблему: он автоматически отслеживает позиции трейдера и воспроизводит их на вашем аккаунте. В отличие от готовых решений, наша разработка позволяет гибко настраивать источники, риск-менеджмент и алгоритмы масштабирования. Задержка исполнения критична: разница в 200 мс может съесть 5% прибыли. Наши боты используют низкоуровневые API и WebSocket для минимальной latency. Также мы внедряем логирование и мониторинг через Tenderly для on-chain сделок.

Один из наших клиентов, управляющий портфелем на $2M, жаловался на проскальзывание 0.3% при копировании через Telegram. Мы перевели его на direct exchange API и снизили latency с 400 ms до 80 ms, что сэкономило около $600 в месяц на проскальзывании.

Почему задержка критична?

Latency между сигналом и исполнением напрямую влияет на доходность. Чем дольше обрабатывается сигнал, тем больше вероятность, что цена изменится не в вашу пользу. При высокочастотном копировании даже 100 ms могут означать разницу между профитом и убытком. Мы используем WebSocket и асинхронные вызовы для ускорения.

Какие источники сигналов использует бот?

Выбор источника определяет задержку и надёжность системы. Сравним три основных варианта:

Через Exchange API

Если источник торгует на той же бирже, самый надёжный вариант — 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) 

Через Telegram webhook

Многие трейдеры публикуют сигналы в 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' ) 

On-chain мониторинг кошельков

Для DeFi: мониторинг on-chain транзакций известного кошелька (whale tracking). Мы слушаем события Swap через WebSocket и фильтруем транзакции с суммой > $1000. Это позволяет копировать сделки крупных кошельков на Uniswap. Реализация осуществляется через web3.py и подписку на logs.

Как рассчитывается размер копируемой позиции?

Мы реализуем три режима масштабирования:

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 между сигналом и исполнением критична: пока ваш бот обрабатывает сигнал, цена движется. Копирование через API быстрее телеграм-сигналов в 2-3 раза.

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" 

Процесс разработки бота под ключ

1. Анализ требований — определяем источники сигналов, режимы копирования, риск-параметры.
2. Проектирование архитектуры — выбираем стек (Python/Go, биржевые API, брокер сообщений).
3. Реализация — пишем ядро бота, парсеры, риск-менеджер.
4. Тестирование — на исторических данных и реальных сигналах (backtest + paper trade).
5. Деплой — разворачиваем на VPS с мониторингом (Grafana, алерты).
6. Обучение и документация — передаём исходники, инструкцию, настраиваем поддержку.

Базовая версия с одним источником сигналов занимает от 5 до 10 рабочих дней. Для полноценной системы с несколькими источниками и продвинутым риск-менеджментом — от 15 до 25 дней.

Готовы обсудить ваш проект? Свяжитесь с нами для бесплатной консультации. Получите демо-версию бота на ваших сигналах.

Что входит в работу

• Настройка источников сигналов (API, Telegram, ончейн).
• Реализация алгоритмов копирования (пропорциональный, фиксированный, fixed risk).
• Интеграция риск-менеджмента (drawdown, лимиты, стоп-лоссы).
• Тестирование на исторических данных и paper trade.
• Развёртывание на сервере с мониторингом.
• Документация и обучение.
• Поддержка в течение месяца после запуска.

Мы гарантируем прозрачность на каждом этапе: все исходники остаются у вас, нет скрытых комиссий. Закажите разработку бота по индивидуальному ТЗ — мы предложим решение под вашу стратегию.