Bridge

You will learn how to use Roq’s gateway bridge together with Roq’s python module and Python’s asyncio.

Conda

Install the roq-deribit package

$ conda install -y -c https://roq-trading.com/conda/unstable roq-deribit

All Roq’s gateways are packaged with various special-purpose binaries, e.g.

$ (cd $CONDA_PREFIX/bin; ls -1 roq-deribit*)

roq-deribit
roq-deribit-benchmark
roq-deribit-bridge
roq-deribit-dump
roq-deribit-filter
roq-deribit-fix-bridge
roq-deribit-strategy

In particular

• roq-deribit → classic gateway used by C++ clients.

• roq-deribit-fix-bridge → bridge gateway used with FIX clients.

• roq-deribit-bridge → bridge gateway used with any client communicating using native socket programming and auto-generated Flatbuffers, Protobuf or SBE interface code.

Note

The remaining binaries are not relevant for this blog post.

Note

Previous blog posts describe how to configure and deploy the gateways. The only major difference, when using the *-bridge solutions, is that the --client_listen_address command-line flag is the TCP port used for incoming websocket connections.

Python

Install the roq-python package

$ conda install -y -c https://roq-trading.com/conda/unstable roq-python

Checking

$ python -c 'import roq; print(roq.__version__)'

1.1.8

This library extends Python with various useful utilities.

In the following we will demonstrate

• The roq.codec module, in particular roq.codec.protobuf.

• The roq.market module, in particular how to maintain a L2 order book.

• Using Python’s asyncio, websocket and singledispatch to build a small reactive application.

Encode

This is a small example showing how to use Roq’s module to encode a Protobuf message

encoder = roq.codec.Encoder(roq.codec.Type.PROTOBUF)

message_info = roq.MessageInfo()

create_order = roq.CreateOrder(
     account=ACCOUNT,
     order_id=1,
     exchange=EXCHANGE,
     symbol=SYMBOL,
     side=roq.Side.BUY,
     order_type=roq.OrderType.LIMIT,
     quantity=1.0,
     price=50000.0,
)

message = encoder.encode(message_info, create_order)

print(message)

Note

In a real-world application, you should make sure to create the roq.codec.Encoder object only once.

Here we chose PROTOBUF because it has wide adoption.

However, it’s is not the fastest for encoding / decoding.

When speed is important, you may consider using (in order of speed):

• SBE (fastest)

• FLATBUFFERS (fast)

• PROTOBUF (slow)

The output may look something like this

b'\n\x08\x12\x00\x1a\x00"\x00*\x00\x8a\xf7\x02:\n\x02A1\x10\x01\x1a\x07deribit"\rBTC-PERPETUAL(\x01P\x02q\x00\x00\x00\x00\x00\x00\xf0?y\x00\x00\x00\x00\x00j\xe8@\x98\x01\x00\xa2\x01\x00'

You can take this output and verify using your own Protobuf auto-generated code that it is indeed a valid on-the-wire message.

Decode

Now that we have a protobuf encoded message, we can easily decode it

decoder = roq.codec.Decoder(roq.codec.Type.PROTOBUF)

length, message_info, value = decoder.decode(message)

# note!
#   length is only non-zero for a sufficiently large byte-buffer
#   this helps if the IP layer fragments messages:
#   - we can append to a byte buffer when we're unable to decode a message (length == 0)
#   - we can remove (length) from the front of a byte buffer when we successfully decodes a message

if length > 0:
    print(value)

The output may look something like this

{account="A1", order_id=1, exchange="deribit", symbol="BTC-PERPETUAL", side=BUY, position_effect=UNDEFINED, margin_mode=UNDEFINED, quantity_type=UNDEFINED, max_show_quantity=nan, order_type=LIMIT, time_in_force=UNDEFINED, execution_instructions=, request_template="", quantity=1, price=50000, stop_price=nan, leverage=nan, routing_id="", strategy_id=0, release_time_utc=0ns}

Note

In a real-world application, you should make sure to create the roq.codec.Decoder object only once.

Example

We will incrementally show the steps to build a small reactive application subscribing market data.

Note

You can find the full source code here.

We need the websockets package

$ conda install -y websockets

The dependencies

import asyncio

from functools import singledispatchmethod

import websockets

import roq

For convenience, we will define some global variables

URI = "ws://localhost:1234"

CODEC = roq.codec.Type.PROTOBUF

LOGIN = "trader"
PASSWORD = "secret"

ACCOUNT = "A1"
EXCHANGE = "deribit"
SYMBOL = "BTC-PERPETUAL"

Let us define a class where we can manage state

class Worker:

    def __init__(self):
        self._uri = f"{URI}?codec={CODEC.name}"
        self._encoder = roq.codec.Encoder(CODEC)
        self._decoder = roq.codec.Decoder(CODEC)
        self._ws = None
        self._ready = False
        self._max_order_id = 0
        self._mbp_cache = {}

Note

All the following code-snippets are methods belonging to this class.

This is our main async loop function

async def _main_loop(self):
    try:
        print(f"connecting to {self._uri}...")
        async with websockets.connect(self._uri) as self._ws:
            print("sending handshake...")
            await self._send_handshake()
            print("receiving...")
            async for msg in self._ws:
                length, message_info, value = self._decoder.decode(msg)
                print(type(value))
                await self._handle(value)
    except asyncio.CancelledError:
        pass
    self._ws = None

Note

We keep the function relatively simple. A real implementation should deal with disconnect, reconnect, etc.

When connected, the client sends the initial handshake using this specific function

async def _send_handshake(self):
    handshake = roq.Handshake(
        login=LOGIN,
        password=PASSWORD,
        accounts=[
            ACCOUNT,
        ],
        symbols=[
            roq.ExchangeSymbol(EXCHANGE, SYMBOL),
        ],
    )
    await self._send(handshake)

Important

Roq’s gateways implement “static” subscriptions. You must list all the accounts and symbols that you are going to use.

A static subscription design was chosen to reduce latency jitter during live trading: The relatively heavy download operation can only happen when a client connects. After the initial download you should only receive incremental updates as they arrive from the exchange.

However, note that the account and symbol fields can be regular expression (regex) which allows you to subscribe “many”.

The handshake function is using a generic helper function to encode any message and send it on the connected websocket

async def _send(self, value):
    message_info = roq.MessageInfo()
    msg = self._encoder.encode(message_info, value)
    await self._ws.send(msg)

After the handshake, the loop function awaits new messages received on the websocket, then decodes those messages

length, message_info, value = self._decoder.decode(msg)

The decode function returns a tuple including the length of the consumed message.

Note

The websocket protocol is message oriented and we don’t need to care about the consumed length, it will always match the received message.

This particular code uses Python’s type dispatch to handle specific messages

length, message_info, value = self._decoder.decode(msg)
print(type(value))
await self._handle(value)

This is the generic handler

@singledispatchmethod
async def _handle(self, value):
    print(f"UNHANDLED: {value}")

We simply print UNHANDLED if a type hasn’t been registered.

This is a specific handler for the HandshakeAck message

@_handle.register
async def _(self, handshake_ack: roq.HandshakeAck):
    print(handshake_ack)

We expect this function to be called if the gateway accepts our handshake.

When the gateway accepts a client, it will go through a “download” sequence followed by a “ready” event.

The DownloadEnd event is important because it communicates the maximum order_id which has been used so far.

We should record this as max_order_id if we later want to send an order request

@_handle.register
async def _(self, download_end: roq.DownloadEnd):
    print(download_end)
    if download_end.max_order_id > self._max_order_id:
        self._max_order_id = download_end.max_order_id
        print(f"max_order_id={self._max_order_id}")

Finally, the Ready event is important because it signals that the download phase has completed and that the following updates will now be “live”

@_handle.register
async def _(self, ready: roq.Ready):
    print(ready)
    self._ready = True
    print("*** READY ***")

To demonstrate market data, we can handle the MarketByPriceUpdate event.

It is used to communicate snapshot or incremental updates for the order book.

@_handle.register
async def _(self, market_by_price_update: roq.MarketByPriceUpdate):
print(market_by_price_update)
key = (market_by_price_update.exchange, market_by_price_update.symbol)
mbp = self._mbp_cache.get(key)
if mbp is None:
    mbp = roq.market.mbp.MarketByPrice(*key)
    self._mbp_cache[key] = mbp
mbp.apply(market_by_price_update)
depth = mbp.extract(2)  # extract top 2 layers
print(f"DEPTH: {depth}")

In this example we use a dictionary to cache order book objects.

The lookup-key is the pair consisting of exchange and symbol.

The only “work” we have to do to maintain a correct order book is to apply the received update events as shown in the example.

mbp.apply(market_by_price_update)

The order book object has various methods useful for querying or working with order books.

For this demo, we extract the top 2 layers from the book like this

depth = mbp.extract(2)  # extract top 2 layers
print(f"DEPTH: {depth}")

Summary

We have demonstrated the building blocks required to communicate with Roq’s gateways using Python’s asyncio, websockets and Protobuf encoders/decoders. (You can find the full source code under “Links” below.)

Using Roq’s python binding is entirely optional: We simply avoid having to deal with low-level Protobuf auto-generated code.

You could just as well use your own auto-generated Python interfaces based on the raw schema definitions from Roq.

Links

• roq-demo

Blogs

• Schema

• Gateway

References