Bridge¶
You will learn how to use Roq’s gateway bridges.
Conda¶
Install the “roq-deribit” package
$ conda install -y -c https://roq-trading.com/conda/unstable roq-deribit
All Roq’s gateways are distributed with various 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 by FIX clients.roq-deribit-bridge→ bridge gateway used by flatbuffers/protobuf/SBE clients (any programming languagE).
Note
The remaining binaries have other purposes not relevant for this blog post.
Note
Previous blog posts describe how to configure and deploy the gateways.
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.7
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¶
The following demonstrates how to encode a Roq message using protobuf
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 only create “roq.codec.Encoder” once.
Note
You can choose any of the available codec types (PROTOBUF, FLATBUFFERS or SBE).
Since this is Python, speed isn’t really important.
We chose Protobuf for the example simply because it’s easy to auto-generate the Protobuf schema and verify that Roq’s Python binding does the right thing.
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'
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}
Example¶
We will incrementally show the steps to build a small reactive application subscribing market data.
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 therefore list all the accounts and symbols that you are going to use.
This is design choice to reduce latency jitter during live trading. The design implies that the relatively heavy download operation can only happen when a client connects.
However, the “account” and “symbol” fields are regex which you can use to conveniently subscribe “many”.
This is a generic helper function used 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
Websocket is message oriented and we don’t need to care about the consumed length, it will always match the received message.
We can use Python’s type dispatch to handle specific messages.
This is the generic handler
@singledispatchmethod
async def _handle(self, value):
print(f"UNHANDLED: {value}")
And this is a specific handler for the HandshakeAck message
@_handle.register
async def _(self, handshake_ack: roq.HandshakeAck):
print(handshake_ack)
Note
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.
We should record this 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}")
The Ready event is important because it signals that the download phase has completed and that
the following updates are live
@_handle.register
async def _(self, ready: roq.Ready):
print(ready)
self._ready = True
print("*** READY ***")
The MarketByPriceUpdate event is used to communicate snapshot or incremental updates
@_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}")
A dictionary having the pair of exchange and symbol as key is used for caching order book objects.
We simply have to apply the received update event to maintain the correct order book state.
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.