Торговый робот на Python для Tinkoff Invest API

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

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

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

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

Этот скрипт на 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 в поисковую строку.

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

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

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

Чтобы правильно работала проверка орфографии в коде, перейдите в настройках в раздел 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
# Импорт модуля для работы со словарями.
from collections import defaultdict
# Импорт функции для получения версии установленного пакета.
from importlib.metadata import version, PackageNotFoundError

# Импорт модуля для запроса данных о пакете через JSON API PyPI.
import requests
# Импорт модуля для сравнения версий пакетов.
from packaging import version as pkg_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,
    BSPB=1,
    CBOM=1,
    CHMF=1,
    ENPG=1,
    FEES=1,
    # FIVE=1,
    FLOT=1,
    GAZP=1,
    # GLTR=1,
    GMKN=1,
    HYDR=1,
    IRAO=1,
    LEAS=1,
    LKOH=1,
    MAGN=1,
    MGNT=1,
    MOEX=1,
    MSNG=1,
    MTLR=1,
    MTLRP=1,
    MTSS=1,
    NLMK=1,
    NVTK=1,
    # OZON=1,
    PHOR=1,
    PIKK=1,
    PLZL=1,
    POSI=1,
    ROSN=1,
    RTKM=1,
    RUAL=1,
    SBER=1,
    SBERP=1,
    SELG=1,
    SGZH=1,
    SMLT=1,
    SNGS=1,
    SNGSP=1,
    TATN=1,
    TATNP=1,
    TCSG=1,
    TRNFP=1,
    UPRO=1,
    VKCO=1,
    VTBR=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']
    # URL PyPI JSON API.
    pypi_url = 'https://pypi.org/pypi/{}/json'

    # Список для хранения информации об обновлениях.
    updates = []

    # Перебор каждого пакета для проверки.
    for pkg in packages:
        try:
            # Получение текущей версии пакета.
            current_version = version(pkg)

            # Получение данных о пакете с PyPI через JSON API.
            response = requests.get(pypi_url.format(pkg))
            if response.status_code == 200:
                data = response.json()
                # Получение последней версии пакета.
                new_version = data['info']['version']

                # Проверка, есть ли новая версия.
                if pkg_version.parse(current_version) < pkg_version.parse(new_version):
                    # Добавление информации о новой версии в список обновлений.
                    updates.append(f'{pkg} ({current_version} -> {new_version})')
            else:
                print(f"Не удалось получить информацию о пакете {pkg} с PyPI.")

        except PackageNotFoundError:
            # Пропуск пакета, если он не установлен.
            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