A trader runs a signal channel in Telegram, but you can't manually open trades fast enough? A copy trading bot solves this: it automatically tracks the trader's positions and replicates them on your account. Unlike off-the-shelf solutions, our development allows flexible configuration of sources, risk management, and scaling algorithms. Execution latency is critical: a 200 ms difference can eat 5% of profit. Our bots use low-level APIs and WebSocket for minimal latency. We also implement logging and monitoring via Tenderly for on-chain trades.
One of our clients, managing a $2M portfolio, was suffering 0.3% slippage when copying via Telegram. We migrated him to direct exchange API and reduced latency from 400 ms to 80 ms, saving about $600 per month in slippage.
Why latency is critical
Latency between signal and execution directly impacts profitability. The longer the signal processing, the more likely the price moves against you. In high-frequency copying, even 100 ms can mean the difference between profit and loss. We use WebSocket and asynchronous calls to speed things up.
Which signal sources does the bot use?
The choice of source determines system latency and reliability. Let's compare three main options:
| Source | Typical latency | Reliability | Integration complexity |
|---|---|---|---|
| Exchange API (sub-account) | 50–200 ms | High | Medium |
| Telegram webhook | 200–500 ms | Medium | Low |
| On-chain monitoring | 1–3 seconds | Low (depends on network) | High |
Via Exchange API
If the source trades on the same exchange, polling its positions via API is the most reliable option:
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):
"""Sync positions every N seconds"""
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}
# New positions on source — open on follower
for symbol, pos in source_map.items():
if symbol not in follower_map:
await self.open_copied_position(pos)
# Positions closed on source — close on follower
for symbol in follower_map:
if symbol not in source_map:
await self.close_copied_position(symbol)
# Position size changed — adjust
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)
Via Telegram webhook
Many traders publish signals in Telegram channels. The bot parses messages:
class TelegramSignalParser:
# Pattern to parse: "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 wallet monitoring
For DeFi: monitoring on-chain transactions of a known wallet (whale tracking). We listen to Swap events via WebSocket and filter transactions with amount > $1000. This allows copying large wallet trades on Uniswap. Implementation uses web3.py and log subscription.
How is the copied position size calculated?
We implement three scaling modes:
| Mode | Description | When to use |
|---|---|---|
| Proportional | Copies the same % of capital as the trader | When capital is comparable |
| Fixed | Copies a specified volume (max 10% of follower balance) | When a per-trade limit is needed |
| Fixed risk | Limits risk per trade as % of balance | For aggressive trading |
def calculate_copy_size(
source_trade: Trade,
source_balance: float,
follower_balance: float,
mode: str = 'proportional',
multiplier: float = 1.0
) -> float:
if mode == 'proportional':
# Copy same % of capital
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':
# Fixed risk per trade (% of balance)
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%
How to minimize latency?
Latency between signal and execution is critical: while your bot processes the signal, the price moves. Copying via API is 2–3 times faster than Telegram signals.
class LatencyMonitor:
def __init__(self):
self.latencies = []
async def execute_with_tracking(self, signal: TradeSignal) -> Execution:
t0 = time.perf_counter()
# Execute order
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}")
# Alert if too slow
if latency_ms > 500:
await self.alert(f"High latency: {latency_ms:.0f}ms for {signal.symbol}")
return order
Typical latencies for different sources:
- Exchange sub-account API → ~50–200ms
- Telegram webhook → ~200–500ms
- On-chain monitoring → ~1000–3000ms (depends on network)
What does risk management include for a copy bot?
class CopyBotRiskManager:
def can_copy(self, signal: TradeSignal, account_state: AccountState) -> tuple[bool, str]:
# Overall drawdown
if account_state.drawdown_percent > self.config.max_drawdown:
return False, "max_drawdown_exceeded"
# Number of simultaneous positions
if len(account_state.open_positions) >= self.config.max_positions:
return False, "max_positions_reached"
# Already have a position in this symbol
if signal.symbol in account_state.open_positions:
return False, "symbol_already_open"
# Minimum balance to open
required = self.calculate_required_margin(signal)
if account_state.free_balance < required * 1.1: # 10% buffer
return False, "insufficient_balance"
return True, "ok"
Turnkey bot development process
- Requirements analysis — define signal sources, copying modes, risk parameters.
- Architecture design — choose tech stack (Python/Go, exchange APIs, message broker).
- Implementation — write bot core, parsers, risk manager.
- Testing — on historical data and real signals (backtest + paper trade).
- Deployment — deploy on VPS with monitoring (Grafana, alerts).
- Training and documentation — hand over source code, manual, set up support.
A basic version with one signal source takes 5–10 business days. A full system with multiple sources and advanced risk management takes 15–25 days.
Ready to discuss your project? Contact us for a free consultation. Get a demo version of the bot on your signals.
What's included
- Configuration of signal sources (API, Telegram, on-chain).
- Implementation of copying algorithms (proportional, fixed, fixed risk).
- Integration of risk management (drawdown, limits, stop-losses).
- Testing on historical data and paper trade.
- Server deployment with monitoring.
- Documentation and training.
- Support for one month after launch.
We guarantee transparency at every step: all source code stays with you, no hidden fees. Order a custom bot development — we'll propose a solution tailored to your strategy.