Хочу поделиться полезной наработкой для тех, кто разрабатывает торговых роботов - шаблоном кода для асинхронной торговой стратегии (live и offline), этот код можно достаточно просто перестраивать под любое API провайдера данных по инструментам.

Сам код опубликовал на Github Тинькофф: https://github.com/Tinkoff/invest-python/blob/main/examples/wiseplat_live_strategy_print_ohlcv.py

Сразу приведу ссылку на этот же мною переработанный код, который использует данные из открытого доступа предоставляемые провайдером - биржей MOEX (даже не нужно получать токен) по торговым инструментам (правда с задержкой в 15 минут) https://github.com/WISEPLAT/Hackathon-Finam-NN-Trade-Robot/blob/master/7_live_strategy.py

Структура кода торговой стратегии

Обычно код для торговой стратегии состоит из следующих блоков:

• Блок запуска кода

• Блок подключения по API к провайдеру данных, в цикле которого по каждому инструменту запускается отдельный экземпляр торговой стратегии

• Каждый запущенный экземпляр стратегии имеет:

  • Блок входа (однократное выполнение кода), обычно используется для инициализации переменных и т.д.

  • Блок бесконечного цикла, в котором уже и выполняется логика торговой стратегии

    • например проверка, что рынок открыт - или ожидание его открытия

    • получение исторических и живых данных по инструменту

    • обработка полученных данных торговой логикой

Теперь давайте посмотри поближе на этот код.

Блок запуска кода

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 к провайдеру данных

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)

Блок входа (однократное выполнение кода)

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()

Блок бесконечного цикла (выполнение логики торговой стратегии)

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)

Доступные выше по ссылкам примеры кода для асинхронной торговой стратегии постарался хорошо задокументировать.

Надеюсь, что это окажется вам полезным +1 и поможет вам сэкономить ваше время!