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

Вступление

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

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

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

О проекте

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

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

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

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

@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

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

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

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