Торговый робот на Python для Tinkoff Invest API
Время прочтения: 21 мин.
507

    Предупреждение

    Эта информация не является индивидуальной инвестиционной рекомендацией. Решение о соответствии инвестиций вашим целям и рискам – ваша задача. Наш сайт не несет ответственности за убытки от инвестиций в упомянутые финансовые инструменты.

    Предоставленные данные о доходности не гарантируют будущие результаты. Не следует полагаться только на эту информацию при инвестиционных решениях. Ценные бумаги не рекламируются, а числовые показатели основаны на данных от третьих сторон, за которые наш сайт ответственности не несет.

    Возможности программы

    Этот скрипт на Python представляет собой автоматизированную систему для распределения акций в портфеле инвестора с использованием Tinkoff Invest API.

    • Интеграция с Tinkoff Invest API: использует API для получения информации об акциях, ценах, статусе торгов и выполнения ордеров. Реализована функциональность для фильтрации и выбора акций на основе заданных критериев, таких как определение биржи, валюта, страна риска, и другие параметры.
    • Поддержка работы в реальном и тестовом режиме: скрипт может работать как с реальным портфелем, так и с «песочницей» (тестовым режимом Tinkoff Invest API).
    • Конфигурация портфеля: позволяет задавать параметры портфеля, такие как начальный баланс, комиссия брокера, и доли акций в процентах.
    • Равномерное распределение акций: поддерживает опцию равномерного распределения акций в портфеле, включая возможность учитывать привилегированные акции. Разделение обыкновенных акций поровну с привилегированными помогает предотвратить превышение желаемой доли отдельного эмитента в портфеле.
    • Расчет ордеров: осуществляет расчёт количества лотов для каждой акции в портфеле с учётом текущего баланса и комиссии брокера. Учитывается размер лота для акции.
    • Выполнение ордеров: скрипт может автоматически размещать ордера на покупку или продажу акций в соответствии с расчетным распределением процентов.
    • Ребалансировка портфеля: при расчёте лотов робот сравнивает их количество с имеющимися лотами в портфеле, что позволяет выполнять автоматическую ребалансировку.
    • Валидация данных: скрипт проверяет корректность входных данных, включая проверку типов переменных и условий.
    • Безопасность и обработка ошибок: включает механизмы обработки исключений и ошибок для обеспечения надежной работы скрипта.
    • Логирование действий: включает подробное логирование для отслеживания действий скрипта.

    Программа подходит для автоматизации процесса управления инвестиционным портфелем, минимизируя ручное вмешательство и оптимизируя распределение активов в соответствии с заданными параметрами. Пользователи могут либо задать свое распределение акций в процентах, либо система может автоматически разделить их равномерно.

    Настройка рабочего пространства

    Установка IDE PyCharm

    • Установите PyCharm Community Edition из Магазина приложений Ubuntu Linux.
    • Скачайте с официального сайта и запустите из распакованного каталога: bin/pycharm.sh
      Достаточно бесплатной версии (внизу страницы).
    • Можно использовать пакетный менеджер Snap:
    sudo snap install pycharm-community --classic

    Установка необходимых библиотек

    Необходимо установить официальную библиотеку tinkoff-investments, которая предоставляет инструменты для работы с Tinkoff Invest API. Установка осуществляется через менеджер пакетов pip. В PyCharm это можно сделать через меню File -> Settings -> Project: <Ваш проект> -> Python Interpreter. Здесь вы можете добавить новую библиотеку, нажав на + и введя tinkoff-investments в поисковую строку.

    PyCharm — Tinkoff Invest API

    Настройка и авторизация

    Для взаимодействия с API требуется токен, который можно получить в личном кабинете Тинькофф Инвестиции. Из соображений безопасности, рекомендуется хранить токен в переменных окружения. Для этого создайте в проекте PyCharm новый файл Python File, например с именем bot. Нажмите на него правой клавишей мыши и выберите Modify Run Configuration... -> Environment Variables. Добавьте переменную окружения INVEST_TOKEN с вашим личным токеном.

    Не публикуйте токены в общедоступные репозитории!

    PyCharm — Tinkoff Invest API

    Чтобы правильно работала проверка орфографии в коде, перейдите в настройках в раздел Editor -> Natural Languages и загрузите русский язык.

    Создание песочницы

    В режиме Sandbox, или «Песочницы» удобно тестировать торговые стратегии, не выводя сделки на основной аккаунт. Для управления песочницей я использую следующий скрипт.

    # Description: Скрипт для создания аккаунта в песочнице Tinkoff Invest API.
    
    # Импорт необходимых библиотек.
    import logging
    import os
    from decimal import Decimal
    
    # Импорт классов и функций из библиотеки Tinkoff Invest API.
    from tinkoff.invest import MoneyValue
    from tinkoff.invest.sandbox.client import SandboxClient
    from tinkoff.invest.utils import decimal_to_quotation, quotation_to_decimal
    
    # Получение токена инвестиций из переменных окружения.
    TOKEN = os.environ["INVEST_TOKEN"]
    
    # Баланс пополнения в рублях.
    BALANCE = 5000000
    
    # Конфигурация модуля logging для вывода логов с определенным уровнем.
    logging.basicConfig(level=logging.DEBUG)
    # Создание экземпляра logging с именем текущего модуля.
    logger = logging.getLogger(__name__)
    
    
    def main():
        # Создание клиента песочницы с использованием токена инвестиций.
        with SandboxClient(TOKEN) as client:
            # Преобразование десятичных чисел.
            balance = decimal_to_quotation(Decimal(BALANCE))
    
            # Получение всех счетов в песочнице.
            sandbox_accounts = client.users.get_accounts()
    
            # Закрытие всех счетов в песочнице.
            for sandbox_account in sandbox_accounts.accounts:
                client.sandbox.close_sandbox_account(account_id=sandbox_account.id)
    
            # Открытие нового счета в песочнице.
            sandbox_account = client.sandbox.open_sandbox_account()
            # Получение ID нового счета.
            account_id = sandbox_account.account_id
    
            # Добавление рублей на новый счет.
            client.sandbox.sandbox_pay_in(
                account_id=account_id,
                amount=MoneyValue(units=balance.units, nano=balance.nano, currency="rub")
            )
    
            # Вывод баланса счета.
            positions = client.operations.get_positions(account_id=account_id).money[0]
            logger.info(f"Баланс: {float(quotation_to_decimal(positions))} рублей.")
    
    
    # Выполнение главной функции, если этот скрипт запущен напрямую.
    if __name__ == "__main__":
        main()
    

    Код программы

    Пример портфеля акций из состава индекса Мосбиржи (за исключением депозитарных расписок):

    Переключение в режим реальной торговли находится в параметре SANDBOX. Будьте внимательны!

    # Description: Скрипт для распределения акций по портфелю в соответствии с указанными процентами.
    
    # Импорт модуля для логирования.
    import logging
    # Импорт математической библиотеки.
    import math
    # Импорт модуля для работы с операционной системой.
    import os
    # Импорт модуля для работы с временем.
    import time
    # Импорт модуля для создания клиентских объектов XML-RPC.
    import xmlrpc.client
    # Импорт модуля для работы со словарями.
    from collections import defaultdict
    
    # Импорт модуля для доступа к метаданным установленных распределений.
    import pkg_resources
    # Импорт модуля для парсинга и сравнения строк версий.
    from packaging import version
    # Импорт классов и функций из библиотеки Tinkoff Invest API.
    from tinkoff.invest import OrderDirection, ShareType, RealExchange, OrderExecutionReportStatus
    from tinkoff.invest import OrderType, SecurityTradingStatus, Client
    from tinkoff.invest.constants import INVEST_GRPC_API_SANDBOX, INVEST_GRPC_API
    from tinkoff.invest.utils import quotation_to_decimal
    
    # Получение токена из переменных окружения.
    TOKEN = os.environ['INVEST_TOKEN']
    
    # Режим работы с песочницей (True) или реальным портфелем (False).
    SANDBOX = True
    
    # Учитывать все акции из портфеля, не указанные в STOCKS.
    TOTAL = False
    
    # Равномерное распределение акций.
    EQUAL = True
    
    # Привилегированные акции в равных долях с обыкновенными.
    PREF = True
    
    # Планируемый баланс портфеля.
    BALANCE = 0
    
    # Комиссия брокера в процентах.
    RATE = 0.03
    
    # Тикеры акций и их доли в портфеле в процентах (если EQUAL = False).
    STOCKS = dict(
        AFKS=1,
        AFLT=1,
        # AGRO=1,
        ALRS=1,
        CBOM=1,
        CHMF=1,
        ENPG=1,
        FEES=1,
        # FIVE=1,
        GAZP=1,
        # GLTR=1,
        GMKN=1,
        HYDR=1,
        IRAO=1,
        LKOH=1,
        MAGN=1,
        MGNT=1,
        MOEX=1,
        MTSS=1,
        NLMK=1,
        NVTK=1,
        # OZON=1,
        PHOR=1,
        PIKK=1,
        PLZL=1,
        # POLY=1,
        POSI=1,
        # QIWI=1,
        ROSN=1,
        RTKM=1,
        RUAL=1,
        SBER=1,
        SBERP=1,
        SELG=1,
        SGZH=1,
        SNGS=1,
        SNGSP=1,
        TATN=1,
        TATNP=1,
        TCSG=1,
        TRNFP=1,
        UPRO=1,
        VKCO=1,
        VTBR=1,
        # YNDX=1,
    )
    
    # Дальнейший код редактировать не нужно.
    
    # Настройка параметров логирования.
    logging.basicConfig(level=logging.INFO, format='%(levelname)s %(filename)s:%(lineno)d %(message)s')
    # Создание объекта логирования.
    logger = logging.getLogger(__name__)
    
    
    def validate_var(var_name, var_type, condition=None, error_msg=None):
        """
        Проверка, что переменная с заданным именем имеет ожидаемый тип и удовлетворяет условию.
    
        :param var_name: (str) Имя переменной, которую нужно проверить.
        :param var_type: (type) Ожидаемый тип переменной.
        :param condition: (function, optional) Функция, которая принимает переменную.
        :param error_msg: (str, optional) Сообщение об ошибке, если условие не удовлетворено.
        :return: None
        """
    
        # Получение глобальной переменной по ее названию.
        var = globals().get(var_name)
    
        # Проверка, что переменная имеет ожидаемый тип.
        if not isinstance(var, var_type):
            raise TypeError(
                f'Ожидался тип {var_type.__name__} для "{var_name}", получено {type(var).__name__}.'
            )
    
        # Проверка, что переменная удовлетворяет условию, если оно предоставлено.
        if condition and not condition(var):
            error_message = error_msg if error_msg else f'Неверное значение для "{var_name}": {var}'
            raise ValueError(error_message)
    
    
    def check_variables():
        """
        Проверка, что переменные окружения имеют правильные значения.
    
        :return: None
        """
    
        # Проверка каждой переменной окружения.
        validate_var('TOKEN', str, lambda x: x)
        validate_var('SANDBOX', bool)
        validate_var('EQUAL', bool)
        validate_var('PREF', bool)
        validate_var('BALANCE', int, lambda x: x >= 0)
        validate_var('RATE', float, lambda x: 0 <= x <= 1)
    
        # Проверка, что все значения в словаре STOCKS больше или равны нулю, и являются строками или числами.
        validate_var(
            'STOCKS', dict,
            lambda x: all(isinstance(k, str) and isinstance(v, (int, float)) and v >= 0 for k, v in x.items())
        )
    
        # Интеграция проверки EQUAL и суммы значений STOCKS.
        validate_var(
            'EQUAL', bool, lambda x: x or sum(STOCKS.values()) == 100,
            'Сумма значений STOCKS должна быть 100, когда EQUAL равно False'
        )
    
    
    def check_latest_version():
        """
        Проверяет наличие новых версий для указанных библиотек 'tinkoff' и 'tinkoff-investments'.
    
        :return: None
        """
    
        # Список пакетов для проверки.
        packages = ['tinkoff', 'tinkoff-investments']
        # Создание клиента для взаимодействия с PyPI через XML-RPC.
        client = xmlrpc.client.ServerProxy('https://pypi.org/pypi')
    
        # Список для хранения информации об обновлениях.
        updates = []
        # Перебор каждого пакета для проверки.
        for pkg in packages:
            try:
                # Получение текущей версии пакета.
                current_version = pkg_resources.get_distribution(pkg).version
                # Получение доступных версий пакета с PyPI.
                available_versions = client.package_releases(pkg)
                # Новая версия пакета.
                new_version = available_versions[0]
    
                # Проверка, есть ли новые версии и сравнение с текущей версией.
                if available_versions and version.parse(current_version) < version.parse(new_version):
                    # Добавление информации о новой версии в список обновлений.
                    updates.append(f'{pkg} ({current_version} -> {new_version})')
    
            except pkg_resources.DistributionNotFound:
                # Пропуск пакета, если он не установлен.
                continue
    
        # Проверка, были ли найдены обновления и генерация исключения с перечнем обновлений.
        if updates:
            raise Exception(f'Доступны новые версии библиотек: {updates}.')
    
    
    def format_number(number):
        """
        Форматирование числа для вывода в лог.
    
        :param number: (float) Число для форматирования.
        :return: (str) Форматированное число в виде строки.
        """
    
        # Преобразование входного числа в float.
        number = float(format(number, 'f'))
    
        # Возвращение числа в формате строки с разделителями для тысяч, если число целое.
        return f'{int(number) if number.is_integer() else number:,}'
    
    
    def get_account_id(client, account_number=0):
        """
        Получение идентификатора аккаунта пользователя.
    
        :param client: (Client) Клиент Тинькофф API.
        :param account_number: (int, optional) Номер аккаунта пользователя.
        :return: (str) Идентификатор аккаунта.
        """
    
        # Получение списка аккаунтов пользователя.
        accounts = client.users.get_accounts()
    
        # Возврат идентификатора аккаунта.
        return accounts.accounts[account_number].id
    
    
    def get_portfolio(client, account_id):
        """
        Получение информации о портфеле аккаунта.
    
        :param client: (Client) Клиент Тинькофф API.
        :param account_id: (str) Идентификатор аккаунта.
        :return: (Portfolio) Информация о портфеле.
        """
    
        # Возврат информации о портфеле для указанного аккаунта.
        return client.operations.get_positions(account_id=account_id)
    
    
    def is_valid_stock(stock):
        """
        Проверка, что акция удовлетворяет условиям трейдинга.
    
        :param stock: (Stock) Объект акции.
        :return: (bool) True, если акция удовлетворяет условиям.
        """
    
        # Условия для акций.
        conditions = [
            # Торги Квалифицированных Биржевых Рынков.
            stock.class_code == 'TQBR',
            # Акции, торгуемые в рублях.
            stock.currency == 'rub',
    
            # Акции, торгуемые в России.
            stock.country_of_risk == 'RU',
            # Признак внебиржевой торговли.
            not stock.otc_flag,
    
            # Доступность для покупки.
            stock.buy_available_flag,
            # Доступность для продажи.
            stock.sell_available_flag,
    
            # Доступность для торговли через API.
            stock.api_trade_available_flag,
            # Торгуемые на Московской бирже.
            stock.real_exchange == RealExchange.REAL_EXCHANGE_MOEX,
    
            # Признак блокировки.
            not stock.blocked_tca_flag,
            # Признак ликвидности.
            stock.liquidity_flag,
        ]
    
        # True, если акция удовлетворяет условиям.
        return all(conditions)
    
    
    def get_stocks_info(client, stocks):
        """
        Сбор информации об акциях.
    
        :param client: (Client) Клиент Тинькофф API.
        :param stocks: (dict) Словарь акций и их долей.
        :return: (dict) Словарь с данными об акциях.
        """
    
        # Получение информации об инструментах.
        instruments_service = client.instruments
        # Получение последних цен для всех акций.
        last_prices_response = client.market_data.get_last_prices()
        # Словарь последних цен для всех акций.
        last_prices = {price.figi: price.price for price in last_prices_response.last_prices}
    
        # Фильтрация акций по условиям трейдинга.
        tradable_stocks = list(filter(is_valid_stock, instruments_service.shares().instruments))
        # Набор тикеров из торгуемых акций.
        tradable_tickers = {item.ticker for item in tradable_stocks}
    
        # Проверка, все ли тикеры из stocks присутствуют в tradable_tickers.
        missing_tickers = stocks.keys() - tradable_tickers
        if missing_tickers:
            raise ValueError(f'Акции не удовлетворяют условиям трейдинга: {missing_tickers}.')
    
        # Словарь с информацией об акциях.
        stock_data = {
            item.ticker: {
                # Идентификатор акции.
                'figi': item.figi,
                # ISIN акции.
                'isin': item.isin,
                # Размер лота.
                'lot_size': item.lot,
                # Признак доступности для квалифицированных инвесторов.
                'qualified_investor': item.for_qual_investor_flag,
                # Тип акции.
                'share_type': item.share_type,
                # Цена акции.
                'price': float(quotation_to_decimal(last_prices.get(item.figi))),
            } for item in tradable_stocks if last_prices.get(item.figi)
        }
    
        # Проверка, что для всех акций есть цены.
        missing_prices = {ticker for ticker, data in stock_data.items() if not data['price']}
        if missing_prices:
            raise ValueError(f'Отсутствуют цены для тикеров: {missing_prices}.')
    
        # Словарь с данными об акциях.
        return stock_data
    
    
    def distribute_equally(stocks, preferential_distribution):
        """
        Равномерное распределение акций.
    
        :param stocks: (dict) Словарь акций и их долей.
        :param preferential_distribution: (bool) Использовать распределение с привилегированными акциями.
        :return: (dict) Словарь с равномерным распределением акций.
        """
    
        # Расчет среднего процента для каждой акции при наличии дополнительного распределения.
        if preferential_distribution:
            # Создание словаря для группировки акций по их первым четырем буквам.
            grouped_stocks = defaultdict(list)
    
            for stock, percent in stocks.items():
                grouped_stocks[stock[:4]].append(stock)
            # Расчет среднего процента для каждой группы.
            percent = 100 / max(len(grouped_stocks), 1)
    
            # Возврат словаря с равномерным распределением для каждой группы.
            return {stock: percent / max(len(group), 1) for group in grouped_stocks.values() for stock in group}
        else:
            # Расчет среднего процента для каждой акции при отсутствии дополнительного распределения.
            percent = 100 / max(len(stocks), 1)
    
            # Возврат словаря с равномерным распределением для каждой акции.
            return {stock: percent for stock in stocks}
    
    
    def percent_allocation(stocks_info, stocks, adjusted_balance):
        """
        Распределение акций по процентам.
    
        :param stocks_info: (dict) Информация об акциях.
        :param stocks: (dict) Словарь акций и их долей.
        :param adjusted_balance: (float) Баланс с учетом комиссии.
        :return: (dict) Словарь акций и их долей.
        """
    
        # Акции, стоимость которых превышает свою долю в портфеле.
        high_cost_stocks = {
            ticker: percent for ticker, percent in stocks.items()
            # Если стоимость акции превышает ее долю в портфеле.
            if stocks_info[ticker]['price'] * stocks_info[ticker]['lot_size'] > adjusted_balance * percent / 100
        }
    
        # Суммарный процент акций, стоимость которых превышает ее долю в портфеле.
        total_percent = sum(high_cost_stocks.values())
    
        # Распределение процентов по остальным акциям.
        if total_percent > 0:
            stocks = {
                # Расчет нового процента для акции.
                ticker: percent + ((percent / max(100 - total_percent, 1)) * total_percent)
                # Если акция не входит в список акций, стоимость которых превышает ее долю в портфеле.
                if ticker not in high_cost_stocks else percent for ticker, percent in stocks.items()
            }
    
        # Возврат словаря с распределенными процентами.
        return stocks
    
    
    def use_all_stocks(stocks_info, portfolio, stocks):
        """
        Учитывать все акции в портфеле, не указанные в STOCKS.
    
        :param stocks_info: (dict) Информация об акциях.
        :param portfolio: (Portfolio) Портфель пользователя.
        :param stocks: (dict) Словарь акций и их долей.
        :return: (dict) Обновленный словарь stocks с добавленными тикерами для продажи.
        """
    
        # Создание множества всех тикеров из портфеля, используя stocks_info для преобразования FIGI в тикеры.
        portfolio_tickers = {
            ticker for stock in portfolio.securities if stock.balance is not None
            for ticker, info in stocks_info.items() if info['figi'] == stock.figi
        }
    
        # Добавление недостающих тикеров из портфеля в словарь stocks со значением 0.
        stocks.update({ticker: 0 for ticker in portfolio_tickers if ticker not in stocks})
    
        # Обновленный словарь stocks.
        return stocks
    
    
    def calculate_lots(stock_info, adjusted_balance, percent, portfolio):
        """
        Расчет желаемого количества лотов для ордеров.
    
        :param stock_info: (dict) Информация об акции.
        :param adjusted_balance: (float) Баланс портфеля с учетом комиссии.
        :param percent: (float) Доля акции в процентах.
        :param portfolio: (Portfolio) Портфель пользователя.
        :return: (int) Количество лотов для выставления ордера.
        """
    
        # Получение FIGI, цены, и размера лота акции.
        figi, price, lot_size = stock_info['figi'], stock_info['price'], stock_info['lot_size']
    
        # Расчет текущего количества лотов в портфеле.
        current_lots = int(sum(
            stock.balance for stock in portfolio.securities if stock.figi == figi and stock.balance is not None
        ) / max(lot_size, 1))
    
        # Расчет желаемого количества лотов для ордера.
        desired_amount = adjusted_balance * percent / 100
        # Расчет необходимого количества средств для покупки акций.
        lots_needed = int(math.floor(desired_amount / max(price * lot_size, 1) - current_lots))
    
        # Корректировка количества лотов для предотвращения отрицательных продаж.
        return max(-current_lots, lots_needed)
    
    
    def lots_and_cost(portfolio, stocks_info, all_stocks, equal, pref, balance, rate, stocks):
        """
        Расчет количества лотов и суммы ордера.
    
        :param portfolio: (Portfolio) Портфель пользователя.
        :param stocks_info: (dict) Информация об акциях.
        :param all_stocks: (bool) Учитывать все акции в портфеле.
        :param equal: (bool) Равномерное распределение акций.
        :param pref: (bool) Привилегированные акции в равномерном распределении с обыкновенными.
        :param balance: (float) Планируемый баланс портфеля.
        :param rate: (float) Комиссия брокера.
        :param stocks: (dict) Словарь акций и их долей.
        :return: (dict) Словарь с информацией о количестве лотов.
        """
    
        # Баланс портфеля с учетом комиссии.
        adjusted_balance = balance - (balance * rate / 100)
    
        # Равномерное распределение акций по процентам с учетом привилегированных акций.
        stocks = distribute_equally(stocks, pref) if equal else stocks
        # Дополнительное распределение процентов от акций, стоимость которых превышает ее долю в портфеле.
        stocks = percent_allocation(stocks_info, stocks, adjusted_balance)
        # Учитывать все акции в портфеле, не указанные в STOCKS.
        stocks = use_all_stocks(stocks_info, portfolio, stocks) if all_stocks else stocks
    
        # Словарь с информацией о количестве лотов.
        lots_and_value = {}
        # Расчет количества лотов для каждой акции.
        for ticker, percent in stocks.items():
            # Получение цены, размера лота и FIGI акции.
            stock_info = stocks_info[ticker]
            # Расчет желаемого количества лотов.
            lots = calculate_lots(stock_info, adjusted_balance, percent, portfolio)
    
            # Добавление информации о количестве лотов в словарь.
            if lots != 0:
                # Расчет суммы ордера в удобном для чтения формате.
                order_value = lots * stock_info['lot_size'] * stock_info['price']
                # Добавление информации о количестве лотов и сумме ордера.
                lots_and_value[ticker] = {'lots': lots, 'order_value': order_value}
    
        # Сортировка по возрастанию суммы для приоритета продаж над покупками при ребалансировке портфеля.
        return dict(
            sorted(
                lots_and_value.items(), key=lambda value: value[1]['order_value']
            )
        )
    
    
    def check_balance(lots, portfolio, broker_rate):
        """
        Проверка достаточности баланса для выполнения ордеров.
    
        :param lots: (dict) Количество лотов и стоимость ордера.
        :param portfolio: (Portfolio) Портфель пользователя.
        :param broker_rate: (float) Комиссия брокера.
        :return: None
        """
    
        # Суммарная стоимость ордеров.
        total_order_value = sum(order['order_value'] for order in lots.values())
        # Баланс портфеля в рублях.
        portfolio_balance = next((money.units for money in portfolio.money if money.currency == 'rub'), 0)
        # Общая сумма с учетом комиссии.
        total_with_rate = total_order_value + (total_order_value * broker_rate / 100)
    
        # Проверка достаточности баланса.
        if total_with_rate > portfolio_balance:
            raise ValueError(f'Недостаточный баланс: {format_number(portfolio_balance)} RUB.')
    
    
    def check_orders_by_share_type(lots, stocks_info):
        """
        Проверка ордеров по типу акций.
    
        :param lots: (dict) Количество лотов и стоимость ордера.
        :param stocks_info: (dict) Информация об акциях.
        :return: None
        """
    
        # Допустимые типы акций.
        valid_types = {ShareType.SHARE_TYPE_COMMON, ShareType.SHARE_TYPE_PREFERRED}
        # Нахождение ордеров с недопустимыми типами акций или ISIN.
        invalid_orders = [
            ticker for ticker in lots if
            stocks_info[ticker]['share_type'] not in valid_types or not
            stocks_info[ticker]['isin'].startswith('RU')
        ]
    
        # Если найдены недопустимые типы акций, вызов исключения.
        if invalid_orders:
            raise ValueError(f'Недопустимые типы акций или ISIN: {invalid_orders}.')
    
    
    def check_for_qualified_investors(lots, stocks_info):
        """
        Проверка ордеров на соответствие требованиям для квалифицированных инвесторов.
    
        :param lots: (dict) Количество лотов и стоимость ордера.
        :param stocks_info: (dict) Информация об акциях.
        :return: None
        """
    
        # Нахождение ордеров, доступных только квалифицированным инвесторам.
        qualified_orders = [ticker for ticker in lots if stocks_info[ticker]['qualified_investor']]
    
        # Если найдены такие ордеры, вызов исключения.
        if qualified_orders:
            raise ValueError(f'Инструменты только для квалифицированных инвесторов: {qualified_orders}.')
    
    
    def check_trading_status(client, lots, stocks_info):
        """
        Проверка статуса торговли акциями.
    
        :param client: (Client) Клиент Тинькофф API.
        :param lots: (dict) Количество лотов и стоимость ордера.
        :param stocks_info: (dict) Информация об акциях.
        :return: None
        """
    
        # Получение статусов торговли акциями.
        trading_statuses = {
            status.figi: status for status in client.market_data.get_trading_statuses().trading_statuses
        }
    
        # Поиск акций, которые не доступны для торговли.
        non_tradable = [
            ticker for ticker in lots if any(
                (
                    # Проверка, что статус торговли не является нормальным.
                    trading_statuses[stocks_info[ticker]['figi']].trading_status !=
                    SecurityTradingStatus.SECURITY_TRADING_STATUS_NORMAL_TRADING,
                    # Проверка, что рыночные ордера не доступны.
                    not trading_statuses[stocks_info[ticker]['figi']].market_order_available_flag,
                )
            )
        ]
    
        # Если найдены не торгуемые акции, вызов исключения.
        if non_tradable:
            raise ValueError(f'Акции, не доступные для торговли: {non_tradable}.')
    
    
    def execute_order(client, lot_count, direction, account_id, figi, ticker):
        """
        Выполнение ордера на покупку или продажу акций через Tinkoff API.
    
        :param client: (Client) Клиент Tinkoff API.
        :param lot_count: (int) Количество лотов для ордера.
        :param direction: (str) Направление ордера ('BUY' или 'SELL').
        :param account_id: (str) Идентификатор аккаунта.
        :param figi: (str) FIGI акции.
        :param ticker: (str) Тикер акции.
        :return: (OrderResponse) Полный ответ ордера.
        """
    
        # Словарь направлений ордера.
        order_directions = {
            'BUY': OrderDirection.ORDER_DIRECTION_BUY,
            'SELL': OrderDirection.ORDER_DIRECTION_SELL,
        }
    
        try:
            # Выполнение ордера.
            order_response = client.orders.post_order(
                quantity=abs(lot_count),
                direction=order_directions[direction],
                account_id=account_id,
                order_type=OrderType.ORDER_TYPE_MARKET,
                instrument_id=figi,
            )
    
            # Получение статуса и дополнительного сообщения ордера.
            report_status, report_message = order_response.execution_report_status, order_response.message
    
            # Если ордер не был исполнен, вызов исключения.
            if report_status != OrderExecutionReportStatus.EXECUTION_REPORT_STATUS_FILL:
                raise RuntimeError(f'Статус: {report_status}, сообщение: {report_message}')
    
        except Exception as e:
            # Обработка исключений с более детальной информацией.
            raise RuntimeError(f'Ошибка при размещении ордера для "{ticker}". {e}.')
    
        # Полный ответ ордера.
        return order_response
    
    
    def log_order_summary(order_data):
        """
        Логирование сводной информации по ордерам.
    
        :param order_data: (dict) Словарь с данными ордера.
        :return: None
        """
    
        # Извлечение общей стоимости и количества ордеров.
        total_cost, order_counts = order_data['total_cost'], order_data['order_counts']
        # Определение знака перед числовым значением стоимости.
        cost_prefix = '-' if total_cost > 0 else '+' if total_cost != 0 else ''
        # Вычисление множества пропущенных акций.
        skipped_stocks = set(order_data['stocks']) - set(order_data['lots'])
    
        # Создание списка сообщений для логирования.
        log_messages = [
            f'Продажи = {order_counts["SELL"]}',
            f'Покупки = {order_counts["BUY"]}',
            f'Всего ордеров = {sum(order_counts.values())}',
            f'Оборот по сделкам = {format_number(round(order_data["total_turnover"]))} RUB',
            f'Комиссия брокера = {format_number(round(order_data["total_commission"]))} RUB',
            f'Общая стоимость = {cost_prefix}{format_number(abs(round(total_cost)))} RUB',
        ]
    
        # Добавление информации о пропущенных акциях, если таковые имеются и общая стоимость не нулевая.
        if skipped_stocks and total_cost != 0:
            log_messages.append(f'Пропущенные позиции: {sorted(skipped_stocks)}.')
    
        # Вывод всех сформированных сообщений в лог.
        for message in log_messages:
            logger.info(message)
    
    
    def process_orders(client, account_id, stocks_info, lots, sandbox, rate, stocks):
        """
        Обработка и выполнение ордеров.
    
        :param client: (Client) Клиент Тинькофф API.
        :param account_id: (str) Идентификатор аккаунта.
        :param stocks_info: (dict) Информация об акциях.
        :param lots: (dict) Количество лотов и стоимость ордера.
        :param sandbox: (bool) Значение, указывающее на работу в песочнице или с реальным портфелем.
        :param rate: (float) Комиссия брокера.
        :param stocks: (dict) Пользовательский словарь с тикерами и процентами их распределения.
        :return: None
        """
    
        # Инициализация переменных для подсчета общих значений по всем ордерам.
        order_counts = defaultdict(int)
        total_turnover = total_commission = total_cost = 0
    
        # Итерация по каждому тикеру и его информации в словаре lots.
        for ticker, lot_info in lots.items():
            # Получение количества лотов и их информации из stocks_info.
            lot_count = lot_info['lots']
            figi, lot_size = stocks_info[ticker]['figi'], stocks_info[ticker]['lot_size']
    
            # Определение направления ордера (покупка или продажа).
            direction = 'BUY' if lot_count > 0 else 'SELL'
            # Выполнение ордера и получение ответа.
            order_response = execute_order(client, lot_count, direction, account_id, figi, ticker)
    
            # Преобразование цены и комиссии ордера из ответа.
            executed_order_price = float(quotation_to_decimal(order_response.executed_order_price))
            initial_commission = float(quotation_to_decimal(order_response.initial_commission))
    
            # Расчет общей стоимости заявки.
            order_value = executed_order_price * order_response.lots_executed * lot_size
            # Расчет комиссии в зависимости от режима работы (песочница или реальный аккаунт).
            commission = (order_value * rate / 100) if sandbox else initial_commission
            # Итоговая сумма ордера с учетом комиссии.
            order_amount = (order_value if lot_count > 0 else -order_value) + commission
    
            # Логирование информации о выполненном ордере.
            logger.info(
                f'{direction} "{ticker}" | '
                f'ЦЕНА = {format_number(executed_order_price)} | '
                f'ЛОТЫ = {order_response.lots_executed} | '
                f'АКЦИИ = {order_response.lots_executed * lot_size} | '
                f'СУММА = {format_number(abs(round(order_value, 2)))} | '
                f'ОРДЕР = {order_response.order_id}'
            )
    
            # Обновление общих значений по всем ордерам.
            order_counts[direction] += 1
            total_commission += commission
            total_turnover += order_value
            total_cost += order_amount
    
            # Пауза в выполнении для избегания чрезмерной нагрузки на API.
            time.sleep(5)
    
        # Логирование сводной информации по всем ордерам.
        log_order_summary({
            'order_counts': order_counts, 'total_commission': total_commission,
            'total_turnover': total_turnover, 'total_cost': total_cost,
            'stocks': stocks, 'lots': lots
        })
    
    
    def main():
        """
        Главная функция программы.
        """
    
        try:
            # Проверка пользовательских данных.
            check_variables()
            # Проверка наличия новых версий библиотек Tinkoff Invest API.
            check_latest_version()
    
            # Выбор URL для работы с песочницей или реальным портфелем.
            endpoint_url = INVEST_GRPC_API_SANDBOX if SANDBOX else INVEST_GRPC_API
    
            # Создание и использование клиента API.
            with Client(TOKEN, target=endpoint_url) as client:
                # Получение идентификатора аккаунта.
                account_id = get_account_id(client)
                # Получение пользовательского портфеля.
                portfolio = get_portfolio(client, account_id)
                # Сбор информации об акциях.
                stocks_info = get_stocks_info(client, STOCKS)
                # Расчет лотов и стоимости ордера.
                lots = lots_and_cost(portfolio, stocks_info, TOTAL, EQUAL, PREF, BALANCE, RATE, STOCKS)
    
                # Проверка баланса.
                check_balance(lots, portfolio, RATE)
                # Проверка ордеров по типу акций.
                check_orders_by_share_type(lots, stocks_info)
                # Проверка ордеров для квалифицированных инвесторов.
                check_for_qualified_investors(lots, stocks_info)
                # Проверка статуса торговли акциями.
                check_trading_status(client, lots, stocks_info)
                # Обработка и выполнение ордеров.
                process_orders(client, account_id, stocks_info, lots, SANDBOX, RATE, STOCKS)
    
        except Exception as e:
            # Логирование ошибок. Для вывода трассировки замените на 'logger.exception(e)'.
            logger.error(e)
            return
    
    
    # Проверка, что скрипт запущен как главный модуль.
    if __name__ == '__main__':
        # Запуск главной функции.
        main()
    

    Сейчас настроен простой вывод лога, но когда вам потребуется трассировка на этапе тестирования, используйте logger.exception(e). Вообще код хорошо прокомментирован и читается как документация, а если комменты не нужны, то их легко можно убрать регуляркой, например в Sublime Text — все строки в едином стиле. А если у вас появятся вопросы, давайте обсудим их в комментариях или на нашем форуме.

    Официальный репозиторий Tinkoff Invest для языка Python