Трейдер ведёт канал с сигналами в 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 (sub-account) | 50-200 мс | Высокая | Средняя |
| Telegram webhook | 200-500 мс | Средняя | Низкая |
| On-chain мониторинг | 1-3 секунды | Низкая (зависит от сети) | Высокая |
Через 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.
Как рассчитывается размер копируемой позиции?
Мы реализуем три режима масштабирования:
| Режим | Описание | Когда использовать |
|---|---|---|
| Пропорциональный | Копирует тот же % капитала, что и у трейдера | Когда капитал соизмерим |
| Фиксированный | Копирует указанный объём (не более 10% от баланса follower) | Когда нужен лимит на сделку |
| Fixed risk | Ограничивает риск на сделку % от баланса | При агрессивном трейдинге |
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"
Процесс разработки бота под ключ
- Анализ требований — определяем источники сигналов, режимы копирования, риск-параметры.
- Проектирование архитектуры — выбираем стек (Python/Go, биржевые API, брокер сообщений).
- Реализация — пишем ядро бота, парсеры, риск-менеджер.
- Тестирование — на исторических данных и реальных сигналах (backtest + paper trade).
- Деплой — разворачиваем на VPS с мониторингом (Grafana, алерты).
- Обучение и документация — передаём исходники, инструкцию, настраиваем поддержку.
Базовая версия с одним источником сигналов занимает от 5 до 10 рабочих дней. Для полноценной системы с несколькими источниками и продвинутым риск-менеджментом — от 15 до 25 дней.
Готовы обсудить ваш проект? Свяжитесь с нами для бесплатной консультации. Получите демо-версию бота на ваших сигналах.
Что входит в работу
- Настройка источников сигналов (API, Telegram, ончейн).
- Реализация алгоритмов копирования (пропорциональный, фиксированный, fixed risk).
- Интеграция риск-менеджмента (drawdown, лимиты, стоп-лоссы).
- Тестирование на исторических данных и paper trade.
- Развёртывание на сервере с мониторингом.
- Документация и обучение.
- Поддержка в течение месяца после запуска.
Мы гарантируем прозрачность на каждом этапе: все исходники остаются у вас, нет скрытых комиссий. Закажите разработку бота по индивидуальному ТЗ — мы предложим решение под вашу стратегию.