External data along with trades

[OnlyC]

Hi,
There are times I need to subscribe for external data outside of nautilus, ex: open interest, BTC dominance, or even my own bias to the bot.
With live trading, I believe can do that by sending http requests every x minutes.
But with backtest, I have trades & external data ready in files. How can I feed those saved data along with trades, like the real flow of time. Ex:

1583020803145000: trade#1 & open_interest#1
1583020803155000: open_interest#2
1583020803276000: trade#2

TRADES.csv

timestamp,trade_id,price,quantity,buyer_maker
1583020803145000,ccc3c1fa-212c-e8b0-1706-9b9c4f3d5ecf,8531.5,2152,True
1583020803276000,b46c84da-6942-b404-ab94-18a9ad497fdd,8531.5,1,True

OPEN_INTEREST.csv

timestamp,open_interest
1583020803145000,88531.5
1583020803155000,75531.5

I tried clock.set_time_alert() but it run on real-time, therefore only allow alert_time > utc_now().

Thanks in advance!

[limx0]

Hey @OnlyC - it's actually much simpler than that - nautilus has first class support for "custom data".

Simply create a Data subclass to represent your data in your project:

class OpenInterest(Data):
    def __init__(self, value, ts_event=0, ts_init=0):
        super().__init__(ts_event, ts_init)
        self.value = value

Then in your backtest code - add a parsing function that creates these objects from your raw csv data:

def load_open_interest(filename):
    data = []
    for _, row in  pd.read_csv(filename).iterrows():
        data.append(
            OpenInterest(value=row['open_interest'], timestamp=row['timestamp'])
        )
    return data

Then, you can simply add this data to your backtest engine as "generic_data", giving the ClientId it comes from (for subscribing later).

engine = BacktestEngine()

open_interest_data = load_open_interest("OPEN_INTEREST.csv")
engine.add_generic_data(ClientId("Binance"), open_interest_data)

Then in your trading strategy - you simply subscribe to this data via the ClientId and DataType. Updates will come through the on_data method.

def on_start(self):
    self.subscribe_data(
        client_id=ClientId("Binance"), data_type=DataType(OpenInterest)
    )

def on_data(self, data: Data):
    if data.data_type == DataType(OpenInterest):
        # do something! 

The great thing about this approach is that if your live adapter emits this data in the same way (wrapped up as OpenInterest Data objects) your trading strategy does not need to change at all for live vs backtest.

Let me know if you have any troubles or other questions

[limx0]

You can also load your own trade ticks via engine.add_trade_ticks(instrument_id, trade_ticks) in a similar fashion (parsing your csv data into TradeTick objects).

Let us know if you have any issues

[OnlyC]

Base on @limx0 answer, I finally managed to done it. But, with a modification in next() function of BacktestDataProducer. I don't know if it's the error in the code or not. If I keep the original code, engine will push all on_data events at first, then all trade_ticks follow after, throw error because the clock already run to the end of Open Interest list.

Original:
` cdef int64_t next_timestamp_ns = INT64_MAX
cdef int choice = 0

    if self._next_quote_tick is not None:
        next_timestamp_ns = self._next_quote_tick.ts_init
        choice = 1

    if self._next_trade_tick is not None:
        if choice == 0 or self._next_trade_tick.ts_init <= next_timestamp_ns:
            choice = 2

    cdef Data next_data = None
    if self._next_data is not None:
        # This next_timestamp_ns is = INT64_MAX or self._next_quote_tick.ts_init, no connection with _next_trade_tick.
        if choice == 0 or self._next_data.ts_init <= next_timestamp_ns:
            next_data = self._next_data
            self._iterate_stream()
            return next_data

    if choice == 1:
        next_data = self._next_quote_tick
        self._iterate_quote_ticks()
    elif choice == 2:
        next_data = self._next_trade_tick
        self._iterate_trade_ticks()

    return next_data`

Modified: (added 1 line)
`cdef int64_t next_timestamp_ns = INT64_MAX
cdef int choice = 0

    if self._next_quote_tick is not None:
        next_timestamp_ns = self._next_quote_tick.ts_init
        choice = 1

    if self._next_trade_tick is not None:
        if choice == 0 or self._next_trade_tick.ts_init <= next_timestamp_ns:
            choice = 2
            next_timestamp_ns = self._next_trade_tick.ts_init
            """ I see that next_timestamp_ns variable being used to choose the lowest timestamp in data stream to send out first. 
            It does get update with _next_quote_tick above but not with _next_trade_tick. This line make sure it doesn't get overtaken by _next_data below. """

    cdef Data next_data = None
    if self._next_data is not None:
        if choice == 0 or self._next_data.ts_init <= next_timestamp_ns:
            next_data = self._next_data
            self._iterate_stream()
            return next_data

    if choice == 1:
        next_data = self._next_quote_tick
        self._iterate_quote_ticks()
    elif choice == 2:
        next_data = self._next_trade_tick
        self._iterate_trade_ticks()

    return next_data`

[cjdsellers]

You're correct, that line is necessary for the logic of the method to be correct. Great find! ➕

I've done a pass over this whole block and refactored, updated next_timestamp_ns to next_ts_init to align with a variable naming change. Also the if choice == 0 comparison is probably partially why I missed that line you added - due to the asymmetry with the lines above. The reason for the choice comparison was probably for performance optimization when timestamps weren't represented by int64_t. Now it seems unnecessary to keep checking choice so I removed those also.

Unit tests all pass, and this will be in my next push to develop.

        # Determine next data element
        cdef int64_t next_ts_init = INT64_MAX
        cdef int choice = 0

        if self._next_quote_tick is not None:
            next_ts_init = self._next_quote_tick.ts_init
            choice = 1

        if self._next_trade_tick is not None:
            if self._next_trade_tick.ts_init <= next_ts_init:
                next_ts_init = self._next_trade_tick.ts_init
                choice = 2

        cdef Data next_data = None
        if self._next_data is not None:
            if self._next_data.ts_init <= next_ts_init:
                next_data = self._next_data
                self._iterate_stream()
                return next_data

        if choice == 1:
            next_data = self._next_quote_tick
            self._iterate_quote_ticks()
        elif choice == 2:
            next_data = self._next_trade_tick
            self._iterate_trade_ticks()

        return next_data

Many thanks!