Holistic evaluation in PositionManager

[petmo]

From my understanding, the current dispatcher (both backtest and live) operates by sending individual ticks for each coin/ticker, with strategy/backtest processing these individually throughout (this is what's illustrated in the examples).

What would be the best approach for a setup where the position manager (or strategy?) waits to collect ticks for all tickers in scope, before making a decision. This would be required for e.g. holistic risk evaluation of all current and new positions, or prioritizing between buy/sell signals.

Is this supported in the current dispatcher setup, or do you need to integrate custom code in the position manager / strategy? What would the best of doing so?

For context, I've tried experimenting with position manager collecting signals and waiting with async until all expected ticks are collected, and then executing the holistic evaluation. However, this has issues especially in the backtest dispatcher, where signals are sent frequently via the async and not always in "expected" / chronological order.

Any ideas welcome!

[gbeced]

Would something like this work as a starting point ? Note that I didn't include the position manager part because that will depend on what you want to do with the information you collected.

# Bars can be downloaded using this command:
# python -m basana.external.bitstamp.tools.download_bars -c BTC/USD -p 1d -s 2021-01-01 -e 2021-12-31 \
# -o bitstamp_btcusd_day.csv

from decimal import Decimal
from typing import Dict, List, Optional
import asyncio
import itertools
import logging

from basana.external.bitstamp import csv
import basana as bs
import basana.backtesting.exchange as backtesting_exchange


# https://docs.python.org/3/library/itertools.html#itertools-recipes
def take(n, iterable):
    "Return first n items of the iterable as a list."
    return list(itertools.islice(iterable, n))


def all_equal(iterable, key=None):
    "Returns True if all the elements are equal to each other."
    return len(take(2, itertools.groupby(iterable, key))) <= 1


class Strategy(bs.TradingSignalSource):
    def __init__(self, dispatcher: bs.EventDispatcher, pairs: List[bs.Pair]):
        super().__init__(dispatcher)
        self._last_events: Dict[bs.Pair, Optional[bs.BarEvent]] = {
            pair: None for pair in pairs
        }

    async def on_bar_event(self, bar_event: bs.BarEvent):
        self._last_events[bar_event.bar.pair] = bar_event

        if all_equal(
            [bar_event.when for bar_event in self._last_events.values() if bar_event]
        ):
            logging.info("All bars available {}".format(
                [str(pair) for pair in self._last_events.keys()]
            ))
            # TODO: Push a signal ?


async def main():
    logging.basicConfig(level=logging.INFO, format="[%(asctime)s %(levelname)s] %(message)s")

    event_dispatcher = bs.backtesting_dispatcher()
    pairs = [
        bs.Pair("BTC", "USD"),
        bs.Pair("ETH", "USD"),
        bs.Pair("LTC", "USD"),
    ]
    exchange = backtesting_exchange.Exchange( event_dispatcher, initial_balances={"USD": Decimal(10000)})
    strategy = Strategy(event_dispatcher, pairs)

    for pair in pairs:
        exchange.set_pair_info(pair, bs.PairInfo(4, 2))
        exchange.subscribe_to_bar_events(pair, strategy.on_bar_event)
        # For the purpose of this example we'll use the same CSV for all symbols.
        exchange.add_bar_source(csv.BarSource(pair, "bitstamp_btcusd_day.csv", "1d"))

    # Run the backtest.
    await event_dispatcher.run()


if __name__ == "__main__":
    asyncio.run(main())

What I'm actually doing in a pairs trading strategy where I need to keep a window of the last X closing values for a couple pairs is to use a Pandas dataframe like this:

async def on_bar_event(self, bar_event: bs.BarEvent):
    self._df.at[bar_event.bar.datetime, bar_event.bar.pair.base_symbol] = float(bar_event.bar.close)
    self._df = self._df.iloc[-self._window_size:]
    if self._df_ready():
        # Analyze and maybe push a signal ?
        pass

def _df_ready(self):
    if len(self._df.columns) != 2:
        return False
    if len(self._df) != self._window_size:
        return False
    if self._df.isnull().values.any():
        return False
    return True