MoexBuilder: как я создаю библиотеку на Python. Часть 210.11.2024 20:00

98ebaac9d6e6a3ac87ca0e19545f80fe

Вступление

Привет, Хабр! Продолжаю рассказывать о том, как я создаю библиотеку на Python. В этой статья я расскажу о том, как реализовал взаимодействие с ISS MOEX, используя асинхронный подход, а также о том, как был добавлен функционал interval().

Предыдущие статьи на эту тему:

  1. MoexBuilder: как я создаю библиотеку на Python. Часть 1

О проекте

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

Проблема #2. Нужно продумать способ взаимодействия с ISS MOEX

Учитывая, что проект предусматривает большое количество обращений к ISS MOEX, было принято решение реализовывать взаимодействие с помощью библиотек aiohttp и asyncio.

Коротко о асинхронности

Асинхронность подразумевает отсутствие ожидания при выполнение I/O операций (input-output, ввод-вывод). Иными словами, всякий раз, когда в синхронном коде программа обращается к внешним компонентам (например, к БД или к внешнему ресурсу по HTTP, как в нашем случае), то происходит ожидание ответа от внешнего ресурса и только после этого программа продолжит выполнение. В действительности, в момент ожидания ответа практически ничего не препятствует выполнению кода далее (там, где это возможно). Эту проблему и решает асинхронный подход.

P.S. Это лишь короткая справка, для тех, кто не знаком с темой асинхронного программирования.

Вот пример асинхронного взаимодействия из проекта: 

@staticmethod
async def fetch(url: str, session: aiohttp.ClientSession) -> dict:
    """
    Async function which return response to the request in the format JSON.

    Args:
        url: url for send GET-request.
        session: client session from which the request is sent.

    Returns:
        response to the request in the format JSON.
    """
    async with session.get(url) as response:
        return await response.json()

@classmethod
async def generate_requests(cls,
                            urls: dict[str, str],
                            additional_params: dict[str, list[str]]
                            ) -> dict[str, dict]:
    """
    Async function which generates some tasks to create GET-request to ISS MOEX.

    Args:
        urls: Each of element defines the name of the task and the url to use additional parameters to create
            GET-requests.
        additional_params: dictionary that specifies which additional parameters to use when creating GET-request.

    Returns:
        result of the task group execution.
    """
    async with aiohttp.ClientSession() as session:
        async with asyncio.TaskGroup() as tg:
            tasks: list[asyncio.Task] = []
            for task_name, url in urls.items():
                url: str = url.format(*additional_params[task_name])
                tasks.append(tg.create_task(cls.fetch(url, session), name=task_name))
    all_response: dict[str, dict] = {task.get_name(): task.result() for task in tasks}
    return all_response

Кратко опишу, что я тут делаю:

  1. С помощью контекстного менеджера создаю клиентскую сессию (сеанс) для выполнения HTTP-запросов.

  2. С помощью контекстного менеджера создаю группу задач (тасок).

  3. Прохожу циклом по всем шаблонам URL, заполняя каждую переданными значениями.

  4. Создаю саму задачу, даю ей имя и добавляю к общему списку задач.

  5. Полученные данные перебираю в удобном виде «название задачи» — «результат».

На текущий момент вот такие шаблоныURL я использую: 

MOEX_REQUESTS: dict = {
    'MAIN_INFO': 'https://iss.moex.com/iss/securities/{0}.json',
    'COMPOSITION_INFO': 'https://iss.moex.com/iss/statistics/engines/stock/markets/{0}/analytics/{1}/tickers.json',
    'DETAIL_INFO': 'https://iss.moex.com/iss/engines/stock/markets/{0}/securities/{1}/candles.json?from={2}&till={3}'
}

И вот пример вызова из проекта (для индекса IMOEX): 

additional_params: dict[str, list[str]] = {
    'MAIN_INFO': [self.tech_name],
    'COMPOSITION_INFO': [self.tech_type, self.tech_name],
    'DETAIL_INFO': [self.tech_type, self.tech_name, last_trade_day, last_trade_day]
}
self.__tech_full_info: dict[str, dict] = asyncio.run(
    Helper.generate_requests(
        urls=cnst.MOEX_REQUESTS,
        additional_params=additional_params
    )
)

Такой подход позволяет не дожидаться ответа каждого из запросов, а отправить все запросы сразу, возвращаясь к обработке результата по факту получения ответа от ISS MOEX.

Проблема #3. Добавление функционала interval

И вот я добился того, что стало возможно написать так: 

from moex import MOEX


moex = MOEX()
print(moex.is_trading_now)  # Проводятся ли торги в настоящий момент
print(moex.last_trade_day)  # Последний торговый день

imoex = moex.imoex
print(imoex.initialcapitalization)  # Начальная капитализация индекса IMOEX
print(imoex.actual_composition_index_tickers)  # Тикеры акций, которые на данный момент входят в индекс IMOEX

и получить желаемый результат.

Полученный ответ: 

False  # is_trading_now
'2024-11-08'  # last_trade_day
240287712872.71  # initialcapitalization
['AFKS', 'AFLT', 'AGRO', 'ALRS', 'ASTR', 'BSPB', 'CBOM', 'CHMF', 'ENPG', 'FEES', 'FIVE', 'FLOT', 'GAZP', 'GMKN', 'HYDR', 'IRAO', 'LEAS', 'LKOH', 'MAGN', 'MGNT', 'MOEX', 'MSNG', 'MTLR', 'MTLRP', 'MTSS', 'NLMK', 'NVTK', 'OZON', 'PHOR', 'PIKK', 'PLZL', 'POSI', 'ROSN', 'RTKM', 'RUAL', 'SBER', 'SBERP', 'SELG', 'SMLT', 'SNGS', 'SNGSP', 'TATN', 'TATNP', 'TCSG', 'TRNFP', 'UPRO', 'VKCO', 'VTBR', 'YDEX']  # actual_composition_index_tickers

Это уже хорошо, но все еще весьма скудный функционал.

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

Легко сказать, но не так легко сделать.

Первое, на что стоило обратить внимание, что для поиска, например, max, min, avg в указанном интервале потребуются все значения из интервала. Казалось бы, что можно отправить запрос вида: 

'DETAIL_INFO': 'https://iss.moex.com/iss/engines/stock/markets/index/securities/IMOEX/candles.json?from=2024-03-08&till=2024-11-10'

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

  1. Объем возвращаемой информации ограниченный. Что-то около 10 дней, но при этом ограничение накладывается не ровно «по дням», из-за чего можно получить «кусок» дня с правой границы диапазона.

  2. В случае, если границы интервала выпадают на не торговый день, то ответ приходит на ближайший «вперед» торговый день.

То есть в нашем примере данные возвращаются с 2024–03–11 по 2024–03–22 (частично).

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

Вторая проблема менее прозрачна. Хотелось бы управлять этим делом, а именно, определять, в какую сторону смещать границу интервала, если указан не торговый день. И смещать ли вообще. Поэтому я решил так:

Функция interval() возвращает экземпляр классаInterval, для которого реализованы свойства: max_value, min_value, avg_value. При этом функция interval() принимает следующие параметры:

  • period_from — дата начала интервала, за который требуется получить данные;

  • period_to — дата окончания интервала, за который требуется получить данные (по умолчания равен last_trade_day);

  • return_datetime_str — флаг, определяющий возвращаемый тип данных для даты (по умолчанию True— даты возвращаются в виде строк).

  • soft_search — режим поиска (по умолчанию равен None — будет возбуждено кастомное исключение SpecifiedDayIsNotTradingDay, указывающее, что указанная правая и/или левая граница (-ы) интервала не является торговым днем. Можно установить в качестве значения forward и, в таком случае указание не торговых дней в качестве границ интервала не будет возбуждать исключение, а будет искать ближайший «вперед» торговый день. Соответственно, значение back будет искать ближайший «назад» торговый день.

Таким образом, стало возможно это: 

from moex import MOEX


moex = MOEX()
print(moex.is_trading_now)  # Проводятся ли торги в настоящий момент
print(moex.last_trade_day)  # Последний торговый день

imoex = moex.imoex
print(imoex.initialcapitalization)  # Начальная капитализация индекса IMOEX
print(imoex.actual_composition_index_tickers)  # Тикеры акций, которые на данный момент входят в индекс IMOEX

interval_imoex = imoex.interval('2024-03-08', soft_search='back')  # Создать объект Interval для индекса IMOEX. Если указанные границы интервала являются не торговыми днями, будет произведено смещение назад до ближайшего торгового дня
print(interval_imoex.max_value)  # Словарь с данными о максимальном значении индекса IMOEX в указанный период
print(interval_imoex.min_value)  # Словарь с данными о минимальном значении индекса IMOEX в указанный период
print(interval_imoex.avg_value)  # Словарь с данными о среднем значении индекса IMOEX в указанный период

Полученный ответ: 

False  # is_trading_now
'2024-11-08'  # last_trade_day
240287712872.71  # initialcapitalization
['AFKS', 'AFLT', 'AGRO', 'ALRS', 'ASTR', 'BSPB', 'CBOM', 'CHMF', 'ENPG', 'FEES', 'FIVE', 'FLOT', 'GAZP', 'GMKN', 'HYDR', 'IRAO', 'LEAS', 'LKOH', 'MAGN', 'MGNT', 'MOEX', 'MSNG', 'MTLR', 'MTLRP', 'MTSS', 'NLMK', 'NVTK', 'OZON', 'PHOR', 'PIKK', 'PLZL', 'POSI', 'ROSN', 'RTKM', 'RUAL', 'SBER', 'SBERP', 'SELG', 'SMLT', 'SNGS', 'SNGSP', 'TATN', 'TATNP', 'TCSG', 'TRNFP', 'UPRO', 'VKCO', 'VTBR', 'YDEX']  # actual_composition_index_tickers
{'from': '2024-05-20 10:00:00', 'to': '2024-05-20 10:09:59', 'value': 3515.11}  # max_value
{'from': '2024-09-03 18:10:00', 'to': '2024-09-03 18:19:59', 'value': 2516.17}  # min_value
{'from': '2024-03-07', 'to': '2024-11-08', 'value': 3054.71}  # avg_value

Этим уже вполне можно пользоваться в собственных проектах не думая о логике «под капотом».

Спасибо за внимание!

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

© Habrahabr.ru