2
$\begingroup$

Background: I'm building a trading system for the crypto market with Python, and currently having problems on how to effectively save my real time orders/trades to disk, so that I could monitor more easily or do further analysis afterwards.

The current design is as follows:

For each asset I have a Strategy class, with a Position class bound to it. The strategy manages a websocket connection to the exchange, and when the strategy receives any order update, it will update its position. Simple code snippet is like:

class Position():
    def __init__():
        self.orders = {}
        ...

    def update(self, order_info):
        ... # process position
        self.orders[order_info['id']] = order_info # update 

class Strategy():
    def __init__():
        self.name = ''
        self.position = Position()

    def on_order_update(self, order_info):
        self.position.update(order_info)

Now, all orders are saved in memory in the position class of each strategy, and I want to also save them to disk (maybe a database?). What I have in mind is to simply write the order status to a database every time the strategy receives it. Therefore I would start a db connection when initializing the position class, and add few lines to write to database in its update method.

However, I'm wondering whether there are better ways to achieve this, and I'm mostly worried about the performance of the trading system. I'm mostly doing intraday trading and I would expect each strategy to have around 1 to 2 orders per second, and would trade on up to 100 assets across exchanges at the same time.

My approach mentioned above would then create and manage a lot of connections to the database (also need to deal with reconnections). Also the writing is executed in the same process of the strategy so it might stall the strategy if writing takes long time? Since I do not need the orders data to be instantaneously available (a delay of seconds is not really a problem), maybe it's a better idea to have another process to read the orders' info of all strategies and write them periodically to the database? Or cache the data in redis first and then persist to disk?

Question: What is a more effective approach for my senario?

Disclaimer: I have very limited knowledge on trading system engineering so any suggestions or pointers are more than welcome.

$\endgroup$
2
  • $\begingroup$ Have you looked into asyncio? $\endgroup$
    – Bob Jansen
    May 28, 2023 at 7:35
  • $\begingroup$ Yes, currently all methods of the Strategy class are async, but the Position class have only normal functions. I'm looking at aiomysql as well, though I'm not sure how much improvement it could yield. $\endgroup$ May 28, 2023 at 7:54

2 Answers 2

2
$\begingroup$

Some approaches I can think of.

  1. Let the hardware capture all incoming and outgoing traffic and filter out the orders and order updates. This has the advantage that it shouldn't come with any added latency and removes this concern from your strategy code but requires more hardware, software and operational work.

  2. Capture all incoming and outgoing packages and process them in a separate process. This doesn't require any changes to the hardware but still introduces more software which needs to be maintained.

The options above would correspond roughly to the NoopConsumer() approach in the code below.

  1. (Not tested) Share memory using mmap and append only the data you want share from your strategy process and read from another process to keep up to date.

  2. Write async as done below, it's probably faster if the overhead of writing is larger than of context switching which seems likely but needs to be experimented with.

  3. Have blocking writes, maybe if you can write really fast.

This code was built from scratch today and I'm not a regular asyncio user so it can probably be improved. I think it is a good starting point nonetheless.

"""
A simple experiment to time different ways to save orders

Start the servers with

> python main.py server 5

and start the experiment with

> python main.py client 5

Servers can be reused.

Example output:

Connecting to 5 exchanges with sync writer
Connecting to 8080
Connecting to 8081
Connecting to 8082
Connecting to 8083
Connecting to 8084
Took: 5.667398263s
Took: 5.693023433s
Took: 5.707320206s
Took: 5.716781134s
Took: 5.720934913s
Connecting to 5 exchanges with async writer
Connecting to 8080
Connecting to 8081
Connecting to 8082
Connecting to 8083
Connecting to 8084
Took: 5.684973048s
Took: 5.700082285s
Took: 5.709185018s
Took: 5.713764676s
Took: 5.721333215s
Connecting to 5 exchanges with NoopConsumer
Connecting to 8080
Connecting to 8081
Connecting to 8082
Connecting to 8083
Connecting to 8084
Took: 2.466461159s
Took: 2.553055324s
Took: 2.563424393s
Took: 2.563971055s
Took: 2.60299299s
Connecting to 5 exchanges with SqliteConsumer synced writes
Connecting to 8080
Connecting to 8081
Connecting to 8082
Connecting to 8083
Connecting to 8084
Took: 10.003863398s
Took: 10.026773816s
Took: 10.058823117s
Took: 10.08572986s
Took: 10.12419061s
Connecting to 5 exchanges with SqliteConsumer async writes
Connecting to 8080
Connecting to 8081
Connecting to 8082
Connecting to 8083
Connecting to 8084
Took: 9.547965043s
Took: 9.583894856s
Took: 9.603546447s
Took: 9.623569694s
Took: 9.648144184s
"""

import asyncio
import sqlite3
import sys
import time
import numpy as np
from websockets.server import serve
from websockets import connect

DEBUG = False
# Server
hostname = "localhost"
base_port = 8080
average_wait_time = 2
num_orders = 1000
p = 0.5
speed = 1000
seed = 42

# Client
sync_overhead = 0.002
async_overhead = 0.002


np.random.seed(seed)


def print_settings():
    print(
        f"""Settings:
hostname:       {hostname}
base_port:      {base_port}
num_orders:     {num_orders}
p:              {p}
speed:          {speed}
seed:           {seed}"""
    )


class NoopConsumer:
    """Takes a message and does nothing"""

    def write(self, exchange_id, order_id):
        pass


class Consumer:
    """Consumes and writes a message and adds some overhead"""

    def __init__(self, overhead):
        self.overhead = overhead
        self.messages_received = []

    def write(self, exchange_id, order_id):
        self.messages_received.append(f"{exchange_id}|{order_id}")
        time.sleep(self.overhead)

    async def async_write(self, exchange_id, order_id):
        self.messages_received.append(f"{exchange_id}|{order_id}")
        time.sleep(self.overhead)


class SqliteConsumer:
    """Non-thread safe writer to a sqlite3 in 'test_db.sqlite3'"""

    def __init__(self):
        self.con = sqlite3.connect("test_db.sqlite3")
        self.cur = self.con.cursor()
        self.cur.execute("DROP TABLE IF EXISTS orders")
        self.cur.execute("CREATE TABLE orders(exchange_id int, order_id int)")

    def write(self, exchange_id, order_id):
        self.cur.execute(
            f"INSERT INTO orders (exchange_id, order_id) VALUES ({exchange_id}, {order_id})"
        )
        self.con.commit()

    async def async_write(self, exchange_id, order_id):
        self.cur.execute(
            f"INSERT INTO orders (exchange_id, order_id) VALUES ({exchange_id}, {order_id})"
        )
        self.con.commit()


class Exchange:
    """
    Bare functionality to model an exchange

    An exchange listens on hostname:port and sends exchange_id:order_id over a
    websocket at random intervals.
    """

    def __init__(self, exchange_id, hostname, port, average_wait_time):
        self.exchange_id = exchange_id
        self.hostname = hostname
        self.port = port
        self.wait_times = np.random.poisson(average_wait_time, num_orders)
        print(f"Creating {self.exchange_id} on {self.hostname}:{self.port}")
        print(f"Average wait time param: {average_wait_time}")
        print(f"Average wait time: {self.wait_times.sum() / (speed * num_orders)}")
        print(f"Total wait time: {self.wait_times.sum() / speed}")

    async def order_feed(self, websocket):
        async for message in websocket:
            if message == "start":
                start = time.time_ns()
                send_time = 0
                for i, wait_time in enumerate(self.wait_times):
                    await asyncio.sleep(wait_time / speed)
                    start_send = time.time_ns()
                    await websocket.send(f"{self.exchange_id}:{i}")
                    send_time += time.time_ns() - start_send
                await websocket.send("done")
                print(
                    f"Took {(time.time_ns() - start) / 1e9}s to send all orders on {self.exchange_id}"
                )
                print(f"Total send time: {send_time / 1e9}s")
                print_settings()

    async def run(self):
        async with serve(self.order_feed, self.hostname, self.port):
            await asyncio.Future()


class Client:
    """Connects to an exchange and records some results

    The feed is started with 'start' and stopped when the message 'done' is
    recieved. A coin flip is performed to decide whether the message is saved.
    """

    def __init__(self, port, writer, write_async):
        self.port = port
        self.writer = writer
        self.write_async = write_async

    async def run(self):
        print(f"Connecting to {self.port}")
        async with connect(f"ws://{hostname}:{self.port}") as websocket:
            await websocket.send("start")
            start = time.time_ns()
            async for message in websocket:
                if message == "done":
                    break
                if np.random.choice([True, False]):
                    exchange_id, order_id = message.split(":")
                    if self.write_async:
                        await self.writer.async_write(exchange_id, order_id)
                    else:
                        self.writer.write(exchange_id, order_id)
            print(f"Took: {(time.time_ns() - start) / 1e9}s")


class Strategy:
    """A strategy holds mulitple connections"""

    def __init__(self, clients):
        self.clients = clients

    async def run_all(self):
        async with asyncio.TaskGroup() as tg:
            for client in self.clients:
                tg.create_task(client.run())


if __name__ == "__main__":
    if len(sys.argv) < 3:
        print("Provide either 'server' or 'client' as argument and a count")
    else:
        arg = sys.argv[1]
        num_servers = int(sys.argv[2])

        if arg == "server":
            print_settings()

            async def run_servers(num_servers):
                async with asyncio.TaskGroup() as tg:
                    for i in range(num_servers):
                        exchange = Exchange(
                            i + 1, hostname, base_port + i, average_wait_time
                        )
                        tg.create_task(exchange.run())

            asyncio.run(run_servers(num_servers))
        elif arg == "client":
            print(f"Connecting to {num_servers} exchanges with sync writer")
            clients = [
                Client(base_port + i, Consumer(sync_overhead), False)
                for i in range(num_servers)
            ]
            asyncio.run(Strategy(clients).run_all())
            if DEBUG:
                print("\n".join(strategy.writer.messages_received))

            print(f"Connecting to {num_servers} exchanges with async writer")
            clients = [
                Client(base_port + i, Consumer(async_overhead), True)
                for i in range(num_servers)
            ]
            asyncio.run(Strategy(clients).run_all())
            if DEBUG:
                print("\n".join(strategy.writer.messages_received))

            print(f"Connecting to {num_servers} exchanges with NoopConsumer")
            clients = [
                Client(base_port + i, NoopConsumer(), False) for i in range(num_servers)
            ]
            asyncio.run(Strategy(clients).run_all())
            if DEBUG:
                print("\n".join(strategy.writer.messages_received))

            print(
                f"Connecting to {num_servers} exchanges with SqliteConsumer synced writes"
            )
            clients = [
                Client(base_port + i, SqliteConsumer(), False)
                for i in range(num_servers)
            ]
            asyncio.run(Strategy(clients).run_all())

            print(
                f"Connecting to {num_servers} exchanges with SqliteConsumer async writes"
            )
            clients = [
                Client(base_port + i, SqliteConsumer(), True)
                for i in range(num_servers)
            ]
            asyncio.run(Strategy(clients).run_all())

        else:
            print(f"Unknown arg '{arg}', exiting")
$\endgroup$
3
  • $\begingroup$ Thanks a lot for your suggestions, will do some experiments around them in the coming days. $\endgroup$ May 29, 2023 at 5:34
  • $\begingroup$ I've added an experiment with the builtin sqlite3 as well. $\endgroup$
    – Bob Jansen
    May 29, 2023 at 17:38
  • $\begingroup$ What this misses is a measurement of response times but that should be possible to add. $\endgroup$
    – Bob Jansen
    May 29, 2023 at 18:40
1
$\begingroup$

If I may suggest, QuestDB is a time-series database written exactly for these use cases, with many users on the financial industry. Ingesting data can be done using any postgresql compatible library but not really efficient for high rate. For higher frequency you can ingest data using the ILP protocol, which is available via a native Python client.

An example for ingestion:

from questdb.ingress import Sender

with Sender('localhost', 9009) as sender:
    sender.row(
        'trades',
        symbols={'symbol': 'ETH-USD', 'side': 'buy'},
        columns={'prices': 2615.54 'volume': 0.00044})
    sender.flush()

Queries can be executed directly using any postgresql compatible library, like psycopg or sqlalchemy. An example of a candlechart at 15m intervals could be

SELECT 
    timestamp,
    first(price) AS open,
    last(price) AS close,
    min(price),
    max(price),
    sum(amount) AS volume
FROM trades
WHERE symbol = 'BTC-USD' AND timestamp > dateadd('d', -1, now())
SAMPLE BY 15m ALIGN TO CALENDAR;

This example can be run at the public demo instance https://demo.questdb.io using the public trades dataset, which is updated with data from coinbase every second or so.

You can use QuestDB self hosted (apache 2.0 license) or as a managed service using the QuestDB Cloud (with a free tier of $200).

$\endgroup$

Not the answer you're looking for? Browse other questions tagged or ask your own question.