Custom Strategy Trading Bot Development

Custom Strategy Trading Bot Development

A custom trading bot is the implementation of a specific trading idea as an automated algorithm. Unlike ready-made DCA or grid bots, a custom strategy implies unique entry/exit logic, specific conditions, and parameters. Let's explore how to build from idea to production.

Trading Strategy Anatomy

Strategy Components

Any trading strategy consists of:

1. Signal generation: when to enter position (opening conditions)
2. Position sizing: how much to invest
3. Exit logic: when and how to close position
4. Risk management: stoploss, max losses, drawdown limits

class TradingStrategy:
    """Base interface for custom strategy"""

    def generate_signal(self, market_data: MarketData) -> Signal:
        """Returns BUY, SELL or HOLD"""
        raise NotImplementedError

    def calculate_position_size(
        self,
        signal: Signal,
        account_balance: Decimal,
        current_price: Decimal
    ) -> Decimal:
        """Position size in base currency"""
        raise NotImplementedError

    def should_exit(
        self,
        position: Position,
        market_data: MarketData
    ) -> ExitReason | None:
        """Returns exit reason or None"""
        raise NotImplementedError

Example: Breakout with Volume Confirmation

Strategy: buy on local maximum breakout with volume confirmation:

class BreakoutStrategy(TradingStrategy):
    def __init__(self, lookback: int = 20, volume_multiplier: float = 1.5):
        self.lookback = lookback
        self.volume_multiplier = volume_multiplier

    def generate_signal(self, data: MarketData) -> Signal:
        closes = data.close[-self.lookback:]
        volumes = data.volume[-self.lookback:]

        resistance = max(closes[:-1])  # max without last candle
        current_close = closes[-1]
        current_volume = volumes[-1]
        avg_volume = sum(volumes[:-1]) / len(volumes[:-1])

        # Breakout condition: price broke resistance + volume above average
        if (current_close > resistance and
                current_volume > avg_volume * self.volume_multiplier):
            return Signal.BUY

        # Downside breakout — short
        support = min(closes[:-1])
        if (current_close < support and
                current_volume > avg_volume * self.volume_multiplier):
            return Signal.SELL

        return Signal.HOLD

Position Sizing: Kelly Criterion and Fixed Fraction

Position size is one of the most important strategy elements. Too large → one loss wipes out account. Too small → low returns.

Fixed Fraction (% of capital): simplest approach:

def fixed_fraction_sizing(balance: Decimal, risk_percent: float = 2.0) -> Decimal:
    return balance * Decimal(str(risk_percent / 100))

Kelly Criterion: optimizes capital growth:

Kelly % = (Win Rate × Avg Win / Avg Loss - (1 - Win Rate)) / (Avg Win / Avg Loss)

def kelly_sizing(win_rate: float, avg_win: float, avg_loss: float) -> float:
    profit_ratio = avg_win / avg_loss
    kelly = (win_rate * profit_ratio - (1 - win_rate)) / profit_ratio
    # Use half Kelly to reduce risk (Half-Kelly)
    return max(0, kelly * 0.5)

In practice, use Half-Kelly or Quarter-Kelly: full Kelly formula is aggressive and requires accurate historical data.

Event-driven Bot Architecture

Main Loop

class TradingBot:
    def __init__(self, strategy: TradingStrategy, exchange_client, risk_manager):
        self.strategy = strategy
        self.exchange = exchange_client
        self.risk = risk_manager
        self.position: Position | None = None

    async def run(self):
        async for candle in self.exchange.subscribe_candles(self.symbol, '1h'):
            await self.on_candle(candle)

    async def on_candle(self, candle: Candle):
        # Update market data
        self.market_data.append(candle)

        # Check current position
        if self.position:
            exit_reason = self.strategy.should_exit(self.position, self.market_data)
            if exit_reason:
                await self.close_position(exit_reason)
                return

        # Check for new signals
        if not self.position:
            signal = self.strategy.generate_signal(self.market_data)
            if signal != Signal.HOLD:
                await self.open_position(signal)

    async def open_position(self, signal: Signal):
        # Risk checks
        if not self.risk.can_open_position():
            logger.info("Risk manager blocked new position")
            return

        balance = await self.exchange.get_balance()
        price = self.market_data.last_close
        size = self.strategy.calculate_position_size(signal, balance, price)

        order = await self.exchange.place_order(
            side='buy' if signal == Signal.BUY else 'sell',
            size=size,
            order_type='market'
        )

        self.position = Position(
            entry_price=order.fill_price,
            size=size,
            side=signal,
            opened_at=datetime.utcnow()
        )
        logger.info(f"Opened {signal} position: {size} @ {order.fill_price}")

Risk Management Module

class RiskManager:
    def __init__(self, config: RiskConfig):
        self.max_daily_loss = config.max_daily_loss_percent  # % of balance
        self.max_drawdown = config.max_drawdown_percent
        self.max_consecutive_losses = config.max_consecutive_losses

        self.daily_pnl = Decimal(0)
        self.peak_balance = None
        self.consecutive_losses = 0

    def can_open_position(self) -> bool:
        # Daily loss limit
        if self.daily_pnl < -self.max_daily_loss:
            logger.warning("Daily loss limit reached")
            return False

        # Maximum drawdown
        if self.peak_balance:
            drawdown = (self.peak_balance - self.current_balance) / self.peak_balance
            if drawdown > self.max_drawdown:
                logger.warning(f"Max drawdown {drawdown:.1%} exceeded")
                return False

        # Consecutive losses
        if self.consecutive_losses >= self.max_consecutive_losses:
            logger.warning(f"{self.consecutive_losses} consecutive losses")
            return False

        return True

Parameterization and Optimization

A good custom strategy shouldn't be hardcoded — all key parameters go to config and are optimized via backtesting:

# strategy_config.yaml
strategy:
  type: breakout
  lookback_period: 20
  volume_multiplier: 1.5
  entry_delay_candles: 1  # wait for candle close

risk:
  position_size_percent: 2.0
  stop_loss_percent: 2.5
  take_profit_percent: 5.0
  max_daily_loss_percent: 6.0
  max_drawdown_percent: 15.0
  max_consecutive_losses: 5

execution:
  exchange: binance
  symbol: BTCUSDT
  timeframe: 1h
  order_type: limit  # or market
  max_slippage_percent: 0.1

Grid search on parameters over historical data shows strategy robustness. A strategy that only works at lookback=20 and no other values — likely overfitted to history.

Logging and Monitoring

Production bot must have:

• Structured logging of every action with timestamp
• Telegram/email alerts on critical events (drawdown stop, API errors)
• Dashboard with current P&L, open positions, trade count
• Auto-restart on failures (supervisord or systemd)
• Heartbeat monitoring (Uptime Robot or Grafana alerting)

A bot without monitoring is a black box that can lose money unnoticed for weeks.