Copy Trading Bot Development

Copy Trading Bot Development

A copy trading bot is an automated tool that tracks trades of a chosen trader (source) and reproduces them on your account. Unlike a full exchange copy trading system, this bot works autonomously: receives signals via API, Telegram, webhooks, or directly monitors public on-chain wallets.

Signal Sources

1. Exchange API (Same Exchange / Sub-account)

If source trades on the same exchange, the most reliable option is polling their positions via 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):
        """Synchronize 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 in source — open for follower
        for symbol, pos in source_map.items():
            if symbol not in follower_map:
                await self.open_copied_position(pos)

        # Positions closed in source — close for 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)

2. Webhook / Telegram Signals

Many traders publish signals in Telegram channels. Bot parses messages:

class TelegramSignalParser:
    # Pattern: "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 Monitoring (DeFi)

For DeFi: monitor on-chain transactions of a known wallet (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):
        """Listen to Transfer and Swap events from target wallet"""
        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)

Position Scaling

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%

Latency and Latency

Latency between signal and execution is critical: while your bot processes the signal, price moves:

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 latency for different sources:

• Exchange sub-account API → ~50-200ms
• Telegram webhook → ~200-500ms
• On-chain monitoring → ~1000-3000ms (depends on network)

Risk Management for 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 concurrent positions
        if len(account_state.open_positions) >= self.config.max_positions:
            return False, "max_positions_reached"

        # Already position in this symbol
        if signal.symbol in account_state.open_positions:
            return False, "symbol_already_open"

        # Minimum balance for opening
        required = self.calculate_required_margin(signal)
        if account_state.free_balance < required * 1.1:  # 10% buffer
            return False, "insufficient_balance"

        return True, "ok"

Copy bot is great tool for traders who trust specific signal provider or want to automate following a strategy. Key risks: latency worsens execution, source may change strategy, bot can miss signal if disconnected. Mandatory: log all operations and Telegram alerts on problems.