Real-Time Financial Exchange Order Book

In this example we are going to walk through how you can maintain a limit order book in real-time with very little extra infrastructure with Bytewax.

Skill Level	Time to Complete	Level
Intermediate Python programming, asynchronous programming	Approx. 25 Min	Intermediate

Your Takeaway

At the end of this tutorial you will understand how to use Bytewax to analyze financial exchange data. You will learn to establish connections to a WebSocket for real-time data, use Bytewax’s operators to efficiently manage an order book, and apply analytical techniques to assess trading opportunities based on the dynamics of buy and sell orders.

Resources

bytewax/bytewax/docs/tutorials/orderbook-guide/orderbook_dataflow.py

Objectives

In this example we are going to walk through how you can maintain a limit order book in real-time with very little extra infrastructure with Bytewax.

We are going to:

• Connect to Coinbase via WebSockets for live order book updates.

• Initialize order books with current snapshots for major cryptocurrencies.

• Update order books in real-time with market changes.

• Utilize advanced data structures for efficient order book management.

• Process live data with Bytewax to maintain and summarize order books.

• Filter updates for significant market movements based on spread.

Concepts

To start off, we are going to diverge into some concepts around markets, exchanges and orders.

Order Book

A Limit Order Book, or just Order Book is a record of all limit orders that are made. A limit order is an order to buy (bid) or sell (ask) an asset for a given price. This could facilitate the exchange of dollars for shares or, as in our case, they could be orders to exchange crypto currencies. On exchanges, the limit order book is constantly changing as orders are placed every fraction of a second. The order book can give a trader insight into the market, whether they are looking to determine liquidity, to create liquidity, design a trading algorithm or maybe determine when bad actors are trying to manipulate the market.

Bid and Ask

In the order book, the ask price is the lowest price that a seller is willing to sell at and the bid price is the highest price that a buyer is willing to buy at. A limit order is different than a market order in that the limit order can be placed with generally 4 dimensions, the direction (buy/sell), the size, the price and the duration (time to expire). A market order, in comparison, has 2 dimensions, the direction and the size. It is up to the exchange to fill a market order and it is filled via what is available in the order book.

Level 2 Data

An exchange will generally offer a few different tiers of information that traders can subscribe to. These can be quite expensive for some exchanges, but luckily for us, most crypto exchanges provide access to the various levels for free! For maintaining our order book we are going to need at least level2 order information. This gives us granularity of each limit order so that we can adjust the book in real-time. Without the ability to maintain our own order book, the snapshot would get almost instantly out of date and we would not be able to act with perfect information. We would have to download the order book every millisecond, or faster, to stay up to date and since the order book can be quite large, this isn’t really feasible.

Alright, let’s get started!

Set up and imports

Before we begin, let’s import the necessary modules and set up the environment for building the dataflow.

Complete installation - we recommend using a virtual environment to manage your Python dependencies. You can install Bytewax using pip:

$ python -m venv venv
$ ./venv/bin/activate
(venv) $ pip install bytewax==0.21.0

Now, let’s import the required modules and set up the environment for building the dataflow.

dataflow.py

import json
from dataclasses import dataclass, field
from datetime import timedelta
from typing import Dict, List, Optional

import websockets
from bytewax import operators as op
from bytewax.connectors.stdio import StdOutSink
from bytewax.dataflow import Dataflow
from bytewax.inputs import FixedPartitionedSource, StatefulSourcePartition, batch_async

Websocket Input

Our goal is to build a scalable system that can monitor multiple cryptocurrency pairs across different workers in real time. By crafting an asynchronous function to connect to the Coinbase Pro WebSocket, we facilitate streaming of cryptocurrency data into our dataflow. This process involves the websockets Python library for WebSocket connections and bytewax for dataflow integration.

The function _ws_agen yields cryptocurrency data, paired with their identifiers (e.g., ["BTC-USD", "ETH-USD"]) by establishing a connection to the Coinbase Exchange WebSocket. It subscribes to the level2_batch feed for live order book updates, first sending a JSON subscription message and awaiting a confirmation response with ws.recv().

dataflow.py

async def _ws_agen(product_id):
    """Connect to websocket and yield messages as they arrive.

    This function is a generator that connects to the Coinbase websocket and
    yields messages as they arrive. It subscribes to the level2_batch channel for
    the given product_id.

    Args:
        product_id (_type_): The product_id to subscribe to.

    Yields:
        _type_: A tuple of the product_id and the message as a dictionary.
    """
    url = "wss://ws-feed.exchange.coinbase.com"
    async with websockets.connect(url) as websocket:
        msg = json.dumps(
            {
                "type": "subscribe",
                "product_ids": [product_id],
                "channels": ["level2_batch"],
            }
        )
        await websocket.send(msg)
        # The first msg is just a confirmation that we have subscribed.
        await websocket.recv()

        while True:
            msg = await websocket.recv()
            yield (product_id, json.loads(msg))

To efficiently process and manage this data, we implement the CoinbasePartition class, extending Bytewax’s StatefulSourcePartition. This enables us to obtain the current orderbook at the beginning of the stream when we subscribe.

Within CoinbasePartition, the _ws_agen function is used for data fetching using batch_async which allows us to collect a batch of incoming data for up to 0.5 seconds or after receiving 100 messages, optimizing data processing and state management. This structure ensures an efficient, scalable, and fault-tolerant system for real-time cryptocurrency market monitoring.

In this section we defined the key building blocks to enable asynchronous WebSocket connections and efficient data processing. Before we can establish a dataflow to maintain the order book, we need to define the data classes - this will enable a structured approach to data processing and management. Let’s take a look at this in the next section.

dataflow.py

class CoinbasePartition(StatefulSourcePartition):
    """Process messages from the Coinbase websocket as they arrive.

    This class is a partition that connects to the Coinbase websocket and
    yields messages as they arrive. It subscribes to the level2_batch channel for
    the given product_id.

    For more information on StatefulSourcePartition, see the documentation:
    https://docs.bytewax.io/stable/api/bytewax/bytewax.inputs.html#bytewax.inputs.StatefulSourcePartition

    Args:
        StatefulSourcePartition : The base class for a partition.
    """

    def __init__(self, product_id):
        """Initializes the partition with the given product_id.

        Args:
            product_id (str): The product_id to subscribe to.
        """
        agen = _ws_agen(product_id)
        self._batcher = batch_async(agen, timedelta(seconds=0.5), 100)

    def next_batch(self):
        """This function returns the next batch of messages from the websocket.

        Returns:
            _type_: A tuple of the product_id and the message as a dictionary.
        """
        return next(self._batcher)

    def snapshot(self):
        """This function returns a snapshot of the partition.

        Returns:
            _type_: _description_
        """
        return None

Defining data classes

Through the Python dataclasses library we can define the structure of the data we are working with. As part of this approach we define three dataclasses:

• CoinbaseSource: Serves as a source for partitioning data based on cryptocurrency product IDs. It is crucial for organizing and distributing the data flow across multiple workers, facilitating parallel processing of cryptocurrency pairs.

dataflow.py

@dataclass
class CoinbaseSource(FixedPartitionedSource):
    """Connect to the Coinbase websocket and yield messages as they arrive.

    This class is a source that connects to the Coinbase websocket and
    yields messages as they arrive. It subscribes to the level2_batch channel for
    the given product_ids.

    Note that the methods associated with this class are required parts
    of the FixedPartitionedSource class.

    For more information on FixedPartitionedSource, see the documentation:
    https://docs.bytewax.io/stable/api/bytewax/bytewax.inputs.html#bytewax.inputs.FixedPartitionedSourceß

    Args:
        FixedPartitionedSource: The base class for a source.
    """

    product_ids: List[str]

    def list_parts(self):
        """This function returns the partitions for the source.

        Returns:
            List (str): The list of product_ids.
        """
        return self.product_ids

    def build_part(self, step_id, for_key, _resume_state):
        """This function builds a partition for the given product_id.

        Args:
            step_id (_type_): The step_id of the input operator.
            for_key (_type_): Which partition to build.
                                Will always be one of the keys returned
                                by list_parts on this worker.
            _resume_state (_type_): State data containing where in
                                    the input stream this partition should
                                    be begin reading during this execution.

        Returns:
            CoinbasePartition: The partition for the given product_id.
        """
        return CoinbasePartition(for_key)

• OrderBookSummary: Summarizes the state of an order book at a point in time, encapsulating the bid and ask prices, sizes, and the spread. This class is immutable (frozen=True), ensuring that each instance is a snapshot that cannot be altered, which is essential for accurate historical analysis and decision-making.

dataflow.py

@dataclass(frozen=True)
class OrderBookSummary:
    """Represents a summary of the order book state."""

    bid_price: float
    bid_size: float
    ask_price: float
    ask_size: float
    spread: float

• OrderBookState: Maintains the current state of the order book, including all bids and asks. It allows for dynamic updates as new market data arrives, keeping track of the best bid and ask prices and their respective sizes.

dataflow.py

@dataclass
class OrderBookState:
    """Maintains the state of the order book."""

    bids: Dict[float, float] = field(default_factory=dict)
    asks: Dict[float, float] = field(default_factory=dict)
    bid_price: Optional[float] = None
    ask_price: Optional[float] = None

    def update(self, data):
        """Update the order book state with the given data.

        Args:
            data: The data to update the order book state with.
        """
        # Initialize bids and asks if they're empty
        if not self.bids:
            self.bids = {float(price): float(size) for price, size in data["bids"]}
            self.bid_price = max(self.bids.keys(), default=None)
        if not self.asks:
            self.asks = {float(price): float(size) for price, size in data["asks"]}
            self.ask_price = min(self.asks.keys(), default=None)

        # Process updates from the "changes" field in the data
        for change in data.get("changes", []):
            side, price_str, size_str = change
            price, size = float(price_str), float(size_str)

            target_dict = self.asks if side == "sell" else self.bids

            # If size is zero, remove the price level; otherwise,
            # update/add the price level
            if size == 0.0:
                target_dict.pop(price, None)
            else:
                target_dict[price] = size

            # After update, recalculate the best bid and ask prices
            if side == "sell":
                self.ask_price = min(self.asks.keys(), default=None)
            else:
                self.bid_price = max(self.bids.keys(), default=None)

    def spread(self) -> float:
        """Calculate the spread between the best bid and ask prices.

        Returns:
            float: The spread between the best bid and ask prices.
        """
        return self.ask_price - self.bid_price  # type: ignore

    def summarize(self):
        """Summarize the order book state.

        Returns:
            OrderBookSummary: A summary of the order book state.
        """
        return OrderBookSummary(
            bid_price=self.bid_price,
            bid_size=self.bids[self.bid_price],
            ask_price=self.ask_price,
            ask_size=self.asks[self.ask_price],
            spread=self.spread(),
        )

In this section, we have defined the data classes that will enable us to maintain the order book in real time. These classes will be used to structure the data flow and manage the state of the order book. Now that we have defined the data classes, we can proceed to construct the dataflow to maintain the order book.

Constructing The Dataflow

Before we get to the exciting part of our order book dataflow, we need to create our Dataflow object and prep the data. We’ll start with creating a Dataflow named ‘orderbook’. Once this is initialized, we can incorporate an input data source into the data flow. We can do this by using the bytewax module operator bytewax.operators which we’ve imported here by a shorter name, op. We will use the operator input, specify its step_id as “input”, pass the orderbook dataflow object along with the data source - in this case the source of data is the CoinbaseSource class we defined earlier initialized with the ids ["BTC-USD", "ETH-USD", "BTC-EUR", "ETH-EUR"].

dataflow.py

flow = Dataflow("orderbook")
inp = op.input(
    "input", flow, CoinbaseSource(["BTC-USD", "ETH-USD", "BTC-EUR", "ETH-EUR"])
)
# ('BTC-USD', {
#     'type': 'l2update',
#     'product_id': 'BTC-USD',
#     'changes': [['buy', '36905.39', '0.00334873']],
#     'time': '2022-05-05T17:25:09.072519Z',
# })

Now that we have input for our Dataflow, we will establish a dataflow pipeline for processing live cryptocurrency order book updates. We will focus on analysis and data filtration based on order book spreads. Our goal is for the pipeline to extract and highlight trading opportunities through the analysis of spreads. Let’s take a look at key components of the dataflow pipeline:

The mapper function updates and summarizes the order book state, initializing a new OrderBookState object if one does not exist, and applying new data updates. The result is a state-summary tuple with key metrics like the best bid and ask prices, sizes, and the spread. We can then use the stateful_map operator to apply the mapper function to each update received in our input batch.

dataflow.py

def mapper(state, value):
    """Update the state with the given value and return the state and a summary."""
    if state is None:
        state = OrderBookState()

    state.update(value)
    return (state, state.summarize())


stats = op.stateful_map("orderbook", inp, mapper)
# ('BTC-USD', (36905.39, 0.00334873, 36905.4, 1.6e-05, 0.010000000002037268))

The last step is to filter the summaries by spread percentage, with a focus on identifying significant spreads greater than 0.1% of the ask price - we will use this as a proxy for trading opportunities. We will define the function and use filter to apply the filter to summaries from the 'orderbook' dataflow.

dataflow.py

# # filter on 0.1% spread as a per
def just_large_spread(prod_summary):
    """Filter out products with a spread less than 0.1%."""
    product, summary = prod_summary
    return summary.spread / summary.ask_price > 0.0001


state = op.filter("big_spread", stats, just_large_spread)

To return the result of the dataflow, we can use output.

dataflow.py

op.output("out", stats, StdOutSink())

Executing the Dataflow

Now we can run our completed dataflow:

> python -m bytewax.run dataflow:flow

This will process real-time order book data for three cryptocurrency pairs: BTC-USD, ETH-EUR, and ETH-USD. Each summary provided detailed insights into the bid and ask sides of the market, including prices and sizes.

('BTC-EUR', OrderBookSummary(bid_price=60152.78, bid_size=0.0104, ask_price=60173.35, ask_size=0.02238611, spread=20.56999999999971))
('BTC-USD', OrderBookSummary(bid_price=65677.38, bid_size=0.05, ask_price=65368.71, ask_size=0.001663, spread=-308.67000000000553))
('ETH-EUR', OrderBookSummary(bid_price=3095.16, bid_size=0.20712451, ask_price=3079.9, ask_size=0.14696149, spread=-15.259999999999764))

Summary

That’s it! You’ve learned how to use websockets with Bytewax and how to leverage stateful map to maintain the state in a streaming application.