Implementing a Simple Trading Bot Framework in Python

Building a trading bot from scratch teaches you more about markets than any course. Here's a clean, extensible framework.

Design Principles

1. Separation of concerns — strategy logic, execution, and risk management are separate modules
2. Event-driven — react to market data, don't poll
3. Paper trading first — always simulate before going live
4. Logging everything — you can't debug what you can't see

Core Architecture

from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

class Side(Enum):
    BUY = 'buy'
    SELL = 'sell'

@dataclass
class Signal:
    symbol: str
    side: Side
    strength: float  # 0 to 1
    timestamp: datetime
    metadata: dict = None

@dataclass
class Order:
    symbol: str
    side: Side
    quantity: float
    order_type: str  # 'market', 'limit'
    price: float = None

class Strategy(ABC):
    @abstractmethod
    def on_bar(self, symbol: str, bar: dict) -> Signal | None:
        pass

    @abstractmethod
    def on_fill(self, order: Order, fill_price: float):
        pass

Strategy Implementation

class MovingAverageCross(Strategy):
    def __init__(self, fast_period=10, slow_period=30):
        self.fast = fast_period
        self.slow = slow_period
        self.prices = {}

    def on_bar(self, symbol, bar):
        if symbol not in self.prices:
            self.prices[symbol] = []

        self.prices[symbol].append(bar['close'])
        prices = self.prices[symbol]

        if len(prices) < self.slow:
            return None

        fast_ma = sum(prices[-self.fast:]) / self.fast
        slow_ma = sum(prices[-self.slow:]) / self.slow
        prev_fast = sum(prices[-self.fast-1:-1]) / self.fast
        prev_slow = sum(prices[-self.slow-1:-1]) / self.slow

        # Crossover detection
        if prev_fast <= prev_slow and fast_ma > slow_ma:
            return Signal(symbol, Side.BUY, 0.8, datetime.now())
        elif prev_fast >= prev_slow and fast_ma < slow_ma:
            return Signal(symbol, Side.SELL, 0.8, datetime.now())

        return None

    def on_fill(self, order, fill_price):
        logger.info(f"Filled: {order.side.value} {order.symbol} @ {fill_price}")

Risk Manager

class RiskManager:
    def __init__(self, max_risk_per_trade=0.01, max_daily_loss=0.03,
                 max_positions=5):
        self.max_risk = max_risk_per_trade
        self.max_daily_loss = max_daily_loss
        self.max_positions = max_positions
        self.daily_pnl = 0
        self.open_positions = {}

    def check_signal(self, signal, account_balance):
        # Daily loss limit
        if self.daily_pnl <= -account_balance * self.max_daily_loss:
            logger.warning("Daily loss limit reached")
            return None

        # Max positions
        if len(self.open_positions) >= self.max_positions:
            if signal.symbol not in self.open_positions:
                logger.warning("Max positions reached")
                return None

        # Calculate position size
        risk_amount = account_balance * self.max_risk
        return risk_amount

    def update_pnl(self, pnl):
        self.daily_pnl += pnl

    def reset_daily(self):
        self.daily_pnl = 0

Paper Trading Engine

class PaperTrader:
    def __init__(self, initial_balance=100000):
        self.balance = initial_balance
        self.positions = {}
        self.trade_history = []

    def execute(self, order, current_price):
        # Simulate slippage
        slippage = current_price * 0.0001
        fill_price = current_price + slippage if order.side == Side.BUY else current_price - slippage

        if order.side == Side.BUY:
            self.positions[order.symbol] = {
                'side': 'long',
                'entry': fill_price,
                'quantity': order.quantity
            }
        else:
            if order.symbol in self.positions:
                entry = self.positions[order.symbol]['entry']
                pnl = (fill_price - entry) * order.quantity
                self.balance += pnl
                self.trade_history.append({
                    'symbol': order.symbol,
                    'pnl': pnl,
                    'entry': entry,
                    'exit': fill_price
                })
                del self.positions[order.symbol]

        return fill_price

Putting It Together

def run_bot(strategy, risk_manager, paper_trader, data_feed):
    for bar in data_feed:
        signal = strategy.on_bar(bar['symbol'], bar)

        if signal:
            risk_amount = risk_manager.check_signal(signal, paper_trader.balance)
            if risk_amount:
                quantity = risk_amount / bar['close']
                order = Order(signal.symbol, signal.side, quantity, 'market')
                fill = paper_trader.execute(order, bar['close'])
                strategy.on_fill(order, fill)

    # Print results
    print(f"Final balance: ${paper_trader.balance:,.2f}")
    print(f"Total trades: {len(paper_trader.trade_history)}")

Important Considerations

• Always paper trade first — never deploy untested code with real money
• Handle disconnections — network failures will happen
• Log everything — timestamps, signals, orders, fills, errors
• Monitor resource usage — memory leaks kill bots over time

This framework is a starting point. Production bots need error handling, reconnection logic, and monitoring. Whether you trade your own bot or use it to practice before taking a firm evaluation, solid engineering matters. More on choosing the right trading environment at propfirmkey.com.

---

Have you built a trading bot? What framework did you use?