Algorithmic Trading System

Abstract

This is a project I have been working on, since university. The stock market had always seemed like this really big game to me where thousands of people participate, and the prize is something more than pride, more than glory, it's something everyone wants... money.

Infrastructure Breakdown

The code interfaces with Interactive Brokers using their Python API. The IBgateway acts as a network gateway between my home servers and IB's servers. All micro-services including the gateway are hosted on a single PC in my basement. So everything in the architecture diagram is a logical separation until further compute is available. To manage the IB gateway, I use open source code called IBC (IB controller). It handles auto login, and timeout recovery but interacting with the GUI of the gateway. It's not perfect, but it adds some robustness to our most critical failure point.

My services can be broadly divided into producers and consumers. “Producers” connect directly to the gateway, and act as central handlers for market data and order management. These services include, the master market data client, the master order client, and the stock screener service. Generally any service that requires a direct connection to Interactive Brokers through the python API is classified this way. The data can then be multiplexed through a message queue to “consumer services” which perform calculations and route orders to the master order client. I acknowledge that it's not a true “consumer producer architecture” because in this case the consumers also send data back. It’s more aptly described as “Router-Dealer” or “Server-Worker” architecture.

This was the best approach due to how the IB API is designed. Based on the following constraints of the IB API, I have decided on my current architecture.

1. Orders are required to have a globally unique order ID, so it is better to have a single entity handle it.
2. It is also inefficient to request duplicated market data from multiple clients. Instead of each strategy requesting market data individually, and crowding the gateway, it is better to have a single client request the data, and multiplex it through my own message queue
3. IB gateway limits the maximum number of concurrent client connections to 10.

Designing for Failure on a budget

In a professional software development setting, we are constantly thinking of how things could fail, both at a software level, and a hardware level. We have failovers, graceful restarts, state restoration and many other solutions. In a personal project setting I believe we have diminishing returns on redundancy. How effective is having docker auto rebuilds when everything is hosted on my 12 year old PC? How useful is Grafana when my internet connection dies and I can no longer access my brokerage account. Therefore, instead of considering so many possible failure cases, I just code assuming my power could go out at any moment. This means offloading as much calculation as possible to IB’s servers. For example I would never open a position without a stoploss order attached.

System Visibility

It's critically important to see what your systems are doing. When managing multiple services there are a variety of tools for container orchestration. I have seen Hashicorp Nomad, Airflow and Kubernetes used in a professional environment. To view the logs/ performance of containers, Grafana Loki and Prometheus seem to be the standard. At the start, I didn’t bother with complex monitoring. I had only set up critical alerts to ping me on discord. While writing this post, I took the chance to set up Grafana, Loki and Prometheus. I have to admit it was much easier than I thought. It only took a couple hours of back and forth with an LLM to have everything up and running. Everything works out of the box with dockerized services, so all you have to do is compose up and forget. The CPU memory usage and IO tracking really help you understand how much your services can handle, and where the bottlenecks are.

Stock Screener

The purpose of the stock screener service is to find “in Play” stocks in real time using IB’s built in market scanner. I configure it with different parameters to find stocks that are experiencing volatility.

The IB scanner is a subscription service that provides callbacks every minute. But to increase the polling rate, I continuously cancel and resubscribe after getting results. To determine if a stock is “in play” I consider the following criterion.

• Currently trading between $0.5 - $25
• Cumulative volume of at least 1 Million (measured from 4am)
• Current volume 3 times above average (spiking volume)
• Total float below 15M shares

The architecture of the scanner is fairly simple. We have a main thread and a callback thread. The main thread calls reqScannerSubscription() for each scan code, and then waits on an event for the scanner callbacks to run. Once all the tickers have been saved to a dictionary. The main thread continues. It filters out the stocks which do not meet the float requirement, repeated stocks and ETFs. We then request news, and live market snapshots for those securities and push that to the PUB-SUB message queue.

def scannerData(self, reqId: int, rank: int, contractDetails: ContractDetails, ... ):
    contract = contractDetails.contract
    
    scan_appearance = ScanAppearance(
        scan_code=self.scanner_id_to_code[reqId],
        time_added=datetime.now(),
        time_exited=None            
    )

    with self.lock:
        if contract.symbol not in self.current_scan:
            self.current_scan[contract.symbol] = ScannedStock(
                contract=contract,
                scan_appearances=[scan_appearance]
            )
        else:
            self.current_scan[contract.symbol].scan_appearances.append(scan_appearance)
            
def scannerDataEnd(self, reqId: int):
    print("ScannerDataEnd for", self.scanner_id_to_code[reqId])
    # Current scan only adds things that are not in our watchlist
    print("current scan", self.current_scan.keys())
    
    # Wake up main thread to merge callback data
    with self.lock:
        self.scans_pending -= 1
        
        print("scan finished", self.scanner_id_to_code[reqId])
        if self.scans_pending == 0:
            print("all scans done setting event")
            self.scanner_done_event.set()
            self.scans_pending = len(self.my_scan_codes)
            self.scan_to_process = self.current_scan.copy()
            self.current_scan.clear()

Processing Level 1 Market Data in IB

Originally I tried using reqMktData() for a live connection. It's the most obvious choice (just looking at the name) but at this point I think it's more of a legacy service, and has many problems. The design of reqMktData() is problematic. You are not getting actual ticks, rather you are sampling and aggregating ticks every 250ms. That in itself is fine, because approximating order flow is likely fine for my strategies. The problem arises in how the data is presented. Data is returned via two separate callbacks. TickPrice() and TickSize(). TickPrice() returns to you bid, ask, and last price traded, Ticksize() returns to you the bid, ask and last trade volume. The problem is that the order of these callbacks is not guaranteed. Therefore it is possible to incorrectly classify buy orders and sell orders especially with the highly volatile securities that I am trading.

Market Data Master

Purpose

class IBApp(EWrapper, EClient):
    def __init__(self):
        EClient.__init__(self, self)
        self.bars = {}  # reqId -> list of BarData, keyed for multiple simultaneous subs

    def historicalData(self, reqId: int, bar: BarData):
        # Initial backfill — called once per bar in the lookback window
        self.bars.setdefault(reqId, []).append(bar)

    def historicalDataUpdate(self, reqId: int, bar: BarData):
        # Live updates to the in-progress (or newly started) bar
        existing = self.bars[reqId]
        if existing and existing[-1].date == bar.date:
            existing[-1] = bar  # same bar, mutate in place
        else:
            existing.append(bar)  # bar rolled over, new one started

    def historicalDataEnd(self, reqId: int, start: str, end: str):
        print(f"Initial backfill complete for reqId {reqId}, now streaming live updates")

Market Proto

Market Client

Market Thread

Candle Store

Client Wrapper

Order Master

The order client is the central client which routes all orders to IB's servers. Using the message queue, independent strategies send their order requests to the order client, which tags each order with a unique ID, and sends it to IB.

Purpose

Reuse code, else we would have to implement order management for every strategy, as well as connect it to IB. Allows us to expand the number of strategies unbounded by the IB client connection limit.

...
while True:
    now = datetime.now().time()
    if now > time(20, 0): # 8:00 PM
        break

    try:
        sender, empty,  command_id, order_binary = order_socket.recv_multipart()
        order_msg = msg.TradeOrder()
        order_msg.ParseFromString(order_binary)

        logger.info(f"Message Received {order_msg.action} {order_msg.ticker}")

        # Send a single order
        if order_msg.classification == msg.OrderAction.NEW:
            client.send_order(order_msg,sender, command_id)

        elif order_msg.classification == msg.OrderAction.DELETE:
            client.cancel_existing_order(order_msg.order_id, sender, command_id)
            
        elif order_msg.classification == msg.OrderAction.MODIFY:
            client.send_order(order_msg,sender, command_id, modify=True)
        
        elif order_msg.classification == msg.OrderAction.GETID:
            start_id, end_id = client.get_order_id_slice(int(order_msg.qty))
            resp = msg.Ticket(
                order_id_start=start_id,
                order_id_end = end_id
            )
            order_socket.send_multipart([sender, b"", resp.SerializeToString()])

    except zmq.Again:
        # It just loops back up to check 'is_market_open' again
        continue

Order Proto

For the proto design, I just mimicked the IB Order object, allowing support for every order type that we might use. IB will just ignore the values of fields that are unrelated to the order you are sending. Currently supporting the following orders:

1. 1. Limit
2. 2. Market
3. 3. Stop/Market if touched
4. 4. Stop Limit/ Limit if touched
5. 5. Dynamic Stop

enum OrderAction {
    NEW = 0;
    MODIFY = 1;
    GETID = 2;
    DELETE = 3;
}

message TradeOrder {
  OrderAction classification = 1;
  string ticker = 2;
  string action = 3;
  float qty = 4;
  string order_type = 5;
  // Stop/StopLimit/Limit Orders
  optional double auxprice = 6;
  optional double lmt_price = 7;
  // To cancel or modify an order
  optional int32 order_id = 8;
  // Trailing Stop Orders
  optional float trailingPercent = 9;
  optional int32 trailStopPrice = 10;

}

message TradeUpdate {
    // Unique identifiers
    int32 order_id = 1;
    string ticker = 2;
    // Status info
    string status = 3;
    // Order Info
    string order_type = 4;
    string action = 5;  
    // Quantity tracking
    double total_qty = 6;
    double filled_qty = 7;
    double remaining_qty = 8;   
    // Pricing
    double avg_fill_price = 9;
    double last_fill_price = 10; // Price of the most recent execution
    // Updates for limit and stop orders
    optional double auxprice = 11;
    optional double lmt_price = 12;   
    // Metadata
    string timestamp = 13;      // ISO format or Unix
    optional string error_message = 14;
}

Order Client

The purpose of the order client is for the client to be able to access order updates, and send orders in a synchronous manner. The client has three components.

• 1. The order thread, a separate thread used to receive and send messages into the message queue going to Order Master.
• 2. The order book, shared memory that tracks the states of orders as updates come from the Order Master.
• 3. The client wrapper, the user facing interface that clients can call to wait for an order to fill, check the status of the order or submit a new order

Order Thread

The client communicates with the order thread using a python Queue (which is thread safe). New requests get put into the outgoing messages queue, and a future is declared. The future tracks the acknowledgement from the Order Master that the order has been submitted if the future times out, then we raise an network error.

def send_command(self, payload: bytes, parse: Callable[[bytes], object],
                    context_msg: str, timeout_s: float = ACK_TIMEOUT_S):
    """
    Enqueue a command for the network thread to send, then block the *calling* thread
    (not the network thread) until its ack arrives or timeout_s elapses.
    Returns None on timeout, after logging and alerting -- never raises.
    """
    future: concurrent.futures.Future = concurrent.futures.Future()
    command_id = uuid.uuid4().hex

    with self._pending_lock:
        self._pending[command_id] = _PendingCommand(future=future, parse=parse, context_msg=context_msg)

    self._outgoing.put(_OutgoingCommand(command_id=command_id, payload=payload))

    try:
        return future.result(timeout=timeout_s)
    except concurrent.futures.TimeoutError:
        with self._pending_lock:
            self._pending.pop(command_id, None)
        self._logger.error()
        self._pn.send_notif()
        return None

Order Book

When responses from the message queue come in, the Order thread writes the updated state of the order to the orderbook. This is the shared object that both the Client Thread, the Order Thread touch.

class OrderBook:
    """
    Locked dict of order_id -> OrderRecord, plus a derived net-position view.

    Written to only by the network thread (via `apply_update`).
    Read from by strategy threads (via `get_status` / `wait_for_terminal` /
    `get_position` / `get_all_positions` / `get_open_order_ids`).
    """

    def __init__(self):
        self._lock = threading.Lock()
        self._orders: Dict[int, OrderRecord] = {}
        # Net open position by ticker, derived from fills seen via apply_update.
        # Positive = long, negative = short. Absent key == flat.
        self._positions: Dict[str, float] = {}

    def apply_update(self, update: msg.TradeUpdate) -> None:
        with self._lock:
            record = self._orders.setdefault(update.order_id, OrderRecord(order_id=update.order_id))
            if _is_duplicate(record.latest, update):
                return

Client wrapper

The client thread provides public facing functions to check the status of an order as well as submit, cancel and modify orders. Each function call returns a response object giving a status update from the server, if the response is None, it means the server did not process the request and an error is sent to me on discord.

class LiveOrderClient():
    def __init__(self, strategy_id: str, logger: logging.Logger, pn: PushNotification, host=IB_HOST):
        self.strategy_id = strategy_id
        self.logger = logger

        server_addr = f"tcp://{host}:{ORDER_SERVICE_PORT}"
        self.order_book = OrderBook()
        self._network = OrderNetworkThread(server_addr, self.order_book, logger, pn, strategy_id)
        self._network.start()

    # ---- commands: fast acks only, never block on a fill ----

    def placeOrder(self, order: msg.TradeOrder) -> Optional[msg.TradeUpdate]:
        self.logger.info(f"{order.ticker} | {order.action} {order.order_type} : {order.qty}", extra={"extra_fields": {"Order": "Requested", "Type": "Opening"}})

        resp = self._network.send_command(
                    payload=order.SerializeToString(),
                    parse=_parse_trade_update,
                    context_msg=f"placeOrder {order.ticker} {order.action} {order.qty}",
                )
        
        return resp

Admin Client & Order Validator

Emerging Stocks Momentum Strategy

Buy on Gap Reversal Strategy

Batch Data Collection Service

Trading Central Console

Database Design

Mongo DB

PostgreSQL

IB Controller

ZMQ