Custom Copy Trading Bot Development – API, Telegram, DeFi

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:

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:

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

1. Requirements analysis — define signal sources, copying modes, risk parameters.
2. Architecture design — choose tech stack (Python/Go, exchange APIs, message broker).
3. Implementation — write bot core, parsers, risk manager.
4. Testing — on historical data and real signals (backtest + paper trade).
5. Deployment — deploy on VPS with monitoring (Grafana, alerts).
6. 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.