Template for an asynchronous trading strategy

I want to share a useful experience for those who develop trading robots – a code template for asynchronous trading strategy (live And offline), this code can be easily rebuilt for any instrument data provider API.

The code itself was published on Github by Tinkoff: https://github.com/Tinkoff/invest-python/blob/main/examples/wiseplat_live_strategy_print_ohlcv.py

I will immediately give a link to the same revised code by me, which uses data from open access provided by the provider – the MOEX exchange (you don’t even need to receive a token) for trading instruments (though with a delay of 15 minutes) https://github.com/WISEPLAT/Hackathon-Finam-NN-Trade-Robot/blob/master/7_live_strategy.py

Structure of the trading strategy code

Typically, the code for a trading strategy consists of the following blocks:

• Code Run Block

• API connection block to the data provider, in the cycle of which a separate instance of the trading strategy is launched for each instrument

• Each running strategy instance has:

  • Entry block (once execution of code), usually used to initialize variables, etc.

  • Block of an infinite loop, in which the logic of the trading strategy is already executed

    • e.g. checking that the market is open – or waiting for it to open

    • obtaining historical and live data on the instrument

    • processing of received data by trading logic

Now let’s take a closer look at this code.

Code Run Block

if __name__ == "__main__":

    sber_figi = "BBG004730N88"
    vtbr_figi = "BBG004730ZJ9"
    portfolio = {sber_figi, vtbr_figi}
    timeframe = CandleInterval.CANDLE_INTERVAL_1_MIN
    days_back = 1
    check_interval = 10  # seconds to check interval for new completed candle

    loop = asyncio.get_event_loop()
    task = loop.create_task(
        run_strategy(
            portfolio=portfolio,
            timeframe=timeframe,
            days_back=days_back,
            check_interval=check_interval,
        )
    )
    loop.run_until_complete(task)

API connection block to the data provider

async def run_strategy(portfolio, timeframe, days_back, check_interval):
    """From this function we are starting
    strategy for every ticker from portfolio.
    """
    async with AsyncClient(token=TOKEN, app_name="TinkoffApp") as client:
        strategy_tasks = []
        for instrument in portfolio:
            strategy = LogOnlyCandlesStrategy(
                figi=instrument,
                timeframe=timeframe,
                days_back=days_back,
                check_interval=check_interval,
                client=client,
            )
            strategy_tasks.append(asyncio.create_task(strategy.start()))
        await asyncio.gather(*strategy_tasks)

Entry Block (Single Code Execution)

async def start(self):
    """Strategy starts from this function."""
    if self.account_id is None:
        try:
            self.account_id = (
                (await self.client.users.get_accounts()).accounts.pop().id
            )
        except AioRequestError as are:
            logger.error("Error taking account id. Stopping strategy. %s", are)
            return
    await self.main_cycle()

Infinite loop block (execution of trading strategy logic)

async def main_cycle(self):
    """Main cycle for live strategy."""
    while True:
        try:
            await self.ensure_market_open()
            await self.get_historical_data()

            # put your strategy code here for live
            # to generate signals for buying or selling tickers
            logger.debug(
                "- live mode: run some strategy code to buy or sell - figi=%s",
                self.figi,
            )

        except AioRequestError as are:
            logger.error("Client error %s", are)

        await asyncio.sleep(self.check_interval)

I tried to document the code examples for an asynchronous trading strategy available at the links above.

I hope you find this useful +1 and help you save your time!