Привет, Хабр! Сегодня расскажу, как оперативно обновлять цены в интернет-магазине, чтобы не оказаться в убытке. Для этого будем использовать Битрикс24 в связке с сервисом рассылки SMS-уведомлений МТС Exolve. Владелец бизнеса или менеджер будет постоянно получать данные об изменениях на валютном рынке и принимать решения на их основе.

Битрикс24

Это популярная система, и мы рассмотрим интеграцию с ней в качестве примера. У системы много инструментов для создания и управления веб-сайтами, интернет-магазинами и корпоративными порталами.

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

Крупные компании выбирают Битрикс24 за хорошую SEO-оптимизацию, аналитику и автоматизацию бизнес-процессов.

Обрабатываемые события

Целесообразно отслеживать только значительные изменения на валютном рынке: скачок курсов на заданную относительную величину (например, 5%); выход за пределы коридора значений — например, превышение психологически важной отметки. Также отслеживаем разницу среднего курса в заданном промежутке по сравнению с аналогичным предыдущим.

Получение данных

Текущий курс на любую дату или его значения в заданный промежуток можно получить при помощи API Центробанка России. Для этого в запросе указываем начальную и конечную дату, код валюты. Пример:

http://www.cbr.ru/scripts/XML_dynamic.asp?date_req1=02/03/2001&date_req2=14/03/2001&VAL_NM_RQ=R01235

Запрос строим при помощи функции, которая формирует строку аргументов из заданных дат и кода валюты. Коды валют заданы самим ЦБ. Начальная и конечная дата по умолчанию устанавливаются сегодняшними: если они одинаковы, возвращается одно значение.

CB_URL = "http://www.cbr.ru/scripts/XML_dynamic.asp"
DOLLAR_TAG = 'R01235'




def get_rates(date_req1: datetime.date | None = None, date_req2: datetime.date | None = None, currency: str = DOLLAR_TAG):
   """
   Запрашивает курс валют с сайта ЦБ РФ в формате .xml;
   Парсит полученный .xml, добавляя нужную информацию ее в список;
  
   :return: text of response
   :rtype: str;
   """
   if date_req1 is None:
       date_req1 = date.today()
   if date_req2 is None:
       date_req2 = date_req1
  
   params = {
       'date_req1': date_req1.strftime("%d/%m/%Y"),
       'date_req2': date_req2.strftime("%d/%m/%Y"),
       'VAL_NM_RQ': currency
   }


   # Запрашиваем и парсим данные
   response = requests.get(CB_URL, params)
   encoded_text = response.text
   return encoded_text

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

Пример ответа:

<?xml version="1.0" encoding="windows-1251"?>
<ValCurs ID="R01235" DateRange1="02.03.2001" DateRange2="14.03.2001" name="Foreign Currency Market Dynamic">
<Record Date="02.03.2001" Id="R01235"><Nominal>1</Nominal><Value>28,6200</Value><VunitRate>28,62</VunitRate></Record>
<Record Date="03.03.2001" Id="R01235"><Nominal>1</Nominal><Value>28,6500</Value><VunitRate>28,65</VunitRate></Record>
<Record Date="06.03.2001" Id="R01235"><Nominal>1</Nominal><Value>28,6600</Value><VunitRate>28,66</VunitRate></Record>
<Record Date="07.03.2001" Id="R01235"><Nominal>1</Nominal><Value>28,6300</Value><VunitRate>28,63</VunitRate></Record>
<Record Date="08.03.2001" Id="R01235"><Nominal>1</Nominal><Value>28,6200</Value><VunitRate>28,62</VunitRate></Record>
<Record Date="12.03.2001" Id="R01235"><Nominal>1</Nominal><Value>28,6200</Value><VunitRate>28,62</VunitRate></Record>
<Record Date="13.03.2001" Id="R01235"><Nominal>1</Nominal><Value>28,6700</Value><VunitRate>28,67</VunitRate></Record>
<Record Date="14.03.2001" Id="R01235"><Nominal>1</Nominal><Value>28,6500</Value><VunitRate>28,65</VunitRate></Record>
</ValCurs>

Подробнее про запросы и ответы можно прочитать в документации. Мы же преобразуем ответ в список значений при помощи функции:

def parse_to_dict(parsed):
   # Достаем нужную информацию о валютах и добавляем ее в список
   values = []
   for rec in parsed.iter("Record"):
       nominal = rec.find("Nominal").text
       value = rec.find("Value").text.replace(",", ".")
       values.append(float(value)/float(nominal))
   return values

Функция на вход получает объект корневого узла документа. Этот объект получаем из строки, содержащей xml-документ, с помощью функции:

import xml.etree.ElementTree as ET




def parser_from_string(text):
   xml_parser = ET.XMLParser(encoding="utf-8")
   parsed = ET.fromstring(text, parser=xml_parser)
   return parsed

Здесь мы специально не используем лишние сторонние библиотеки, а стараемся ограничиться встроенными структурами и функциями, чтобы примером могли воспользоваться все. Ценители алгоритмов и машинного обучения могут преобразовать список в массив NumPy или Pandas DataFrame и реализовать более сложную аналитику.

Обнаружение и учёт событий

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

Удобнее всего хранить параметры системы в Data-классе. Этот мощный инструмент нужен для контроля типов данных, задания значений по умолчанию, преобразований в и из словаря и JSON. Последняя опция удобна для обновлений настроек извне и хранения их в файле. Наша реализация Data-класса:

@dataclass
class Settings:
   currency: str = DOLLAR_TAG
   frame_length: int = 10  # Дни, длина окна анализа
   average_difference: float = 100  # проценты
   upper_threshold: float | None = None # Верхняя граница коридора
   lower_threshold: float | None = None # Нижняя граница коридора
   jump_threshold: float = 5  # проценты
   frame_validity_threshold = 0.6  # Минимальная длина окна в долях от заданной длины frame_length

Курс ЦБ обновляется раз в сутки, поэтому система не может присылать сообщения чаще. Но нет смысла получать одни и те же оповещения: если курс валют превысил важную отметку и значение не изменилось на следующий день, то руководитель об этом уже в курсе. 

Для действующего магазина можно продумать и более сложную систему анализа, но здесь мы ограничимся записью зафиксированных событий за заданный промежуток времени. Известные события игнорируются. Более ранние автоматически удаляются. Новые — фиксируются, при этом отправляется SMS заданному абоненту через сервис МТС Exolve. Для реализации учёта событий используем Data-класс:

@dataclass
class EventsList:
   jump: datetime.date | None = None
   lower_exceeding: datetime.date | None = None
   upper_exceeding: datetime.date | None = None
   average_diff: datetime.date | None = None
   weighted_diff: datetime.date | None = None


   def push_event(self, name: str, date: datetime.date, settings: Settings):
       assert name in [f.name for f in dataclasses.fields(self)]
       previous_date = self.__getattribute__(name)
       cutoff_date = date-timedelta(settings.frame_length)
       if previous_date is not None and previous_date < cutoff_date:
           self.__setattr__(name, None)
       previous_date = self.__getattribute__(name)
       if previous_date is None:
           self.__setattr__(name, date)
           result = send_SMS(NOTIFICATION_DICTIONARY[name])
           print(result)

Далее рассмотрим процедуру отправки сообщений, которая вызывается здесь.

Отправка сообщений по новому событию

Для отправки SMS используется функция send_SMS. Взаимодействие с системой происходит через API МТС Exolve. При формировании запроса необходимо указать обязательные параметры: номер отправителя и получателя и текст самого сообщения. Эти данные должны передаваться в формате JSON. Кроме того, заголовок запроса обязан включать поле 'Authorization' с ключевым словом 'Bearer' для определения типа авторизации, за которым следует сам ключ, отделённый пробелом. 

Конфиденциальные данные, такие как номера телефонов и ключ API, лучше хранить в переменных окружения. Их легко будет получить в коде:

exolve_account_phone = os.environ['EXOLVE_PHONE']
manager_phone = os.environ['MANAGER_PHONE']
sms_api_key = os.environ['MTS_API_KEY']
bitrix_code = os.environ['BITRIX_CODE']

Функция:

import requests
SMS_ENDPOINT = r'https://api.exolve.ru/messaging/v1/SendSMS'




def send_SMS(send_str: str):
   payload = {'number': exolve_account_phone, 'destination': manager_phone, 'text': send_str}
   r = requests.post(SMS_ENDPOINT, headers={'Authorization': 'Bearer '+sms_api_key}, data=json.dumps(payload))
   print(r.text)
   return r.text, r.status_code

Чтобы удобнее принимать решение, график курсов за текущий и предыдущий период выводится на экран:

Для этого можно использовать мощную библиотеку matplotlib. Она входит в число обязательных для изучения для тех, кто занимается обработкой данных. Лучше использовать её в паре с NumPy или Pandas. Там же, в окне с графиком, можно увидеть сообщения обо всех событиях и кнопку, которая позволит руководителю автоматически пересчитать цены. Пример реализации:

import numpy as np
from matplotlib.widgets import Button
from matplotlib import gridspec




def plot_results(dates_earl, data_earl, dates_curr, data_curr, settings: Settings, events: EventsList):
   spec = gridspec.GridSpec(ncols=1, nrows=2, height_ratios=[2, 1])
   figure = plt.figure()
   ax = figure.add_subplot(spec[0])
   ax.plot(dates_earl, data_earl, '-o', label='Курс')
   ax.plot(dates_curr, data_curr, '-o')
   ax.plot(dates_earl, np.ones_like(data_earl)*np.mean(data_earl), '--', label='Среднее значение')
   ax.plot(dates_curr, np.ones_like(data_curr)*np.mean(data_curr), '--')
   label = 'Границы коридора'
   if settings.lower_threshold:
       ax.plot(dates_curr, np.ones_like(data_curr)*settings.lower_threshold, '--', label=label)
       label = None
   if settings.upper_threshold:
    ax.plot(dates_curr, np.ones_like(data_curr)*settings.upper_threshold, '--', label=label)
   if events.jump is not None:
       ax.plot([events.jump, events.jump], ax.ylims())
   ax.legend()
   ax.tick_params("x", rotation=90)


   control_ax = figure.add_subplot(spec[1])
   text = ''
   for field in dataclasses.fields(events):
       if events.__getattribute__(field.name) is not None:
           text += MESSAGE_DICTIONARY[field.name]+f'\n'
   if len(text) == 0:
       text = 'Никаких событий не зафиксировано.'
   control_ax.text(0, 0.7, text)
   apply_button.on_clicked(lambda x: send_hook_to_bitrix(x, data_curr))
   plt.tight_layout(pad=1)
   plt.show()

Интеграция с Битрикс24

Для разработки и тестирования достаточно бесплатной облачной версии Битрикс24. Вам выделят домен, дадут возможность создавать простейшие сервисы из шаблонов. Для этого выберем на левой панели вкладку «Маркет», пролистаем вниз до каталога приложений, где будет несколько бесплатных шаблонов.

Для примера возьмём интернет-магазин мебели. Переходим на панели слева во вкладку «Магазин», сверху находим «Товары и Склады». Пока здесь пусто, но внесём товар в каталог.

Получение и обновление данных

Чтобы информация в системе обновлялась, нужно настроить доступ к нашему сайту. Для этого можно использовать полный протокол OAuth 2.0 или воспользоваться созданием локальных вебхуков. Выберем второе. При этом будет создан эндпоинт, содержащий в своём адресе ключ доступа: https://<your_site>/rest/<user_id>/<key>.

Пример из документации: https://doc-test-b24.bitrix24.ru/rest/1/173glortu42lvpju.

Для создания хука переходим в раздел «Разработчикам», выбираем «Входящий вебхук», подключаем тариф.

Важный этап — настройка прав. Нужно выбрать подсистемы, к ним будет доступ по вебхуку. Иначе сервер вернёт ошибку 400. В этом окне указан и адрес, по которому можно направлять хуки. Его нужно скопировать, затем сохранить изменения. Ключ (последняя часть адреса) лучше хранить не в репозитории, а в переменной окружения той системы, где будет работать приложение слежения за курсами.

Обращаться к Битрикс24 можно через REST API напрямую или через библиотеку fast_bitrix. Она позволяет написать более читаемый код.

Рассмотрим простую операцию — получение данных из системы. Для начала создаём объект класса Bitrix, который требует единственный аргумент — URL эндпоинта. Последний состоит из частей, описанных выше, в разделе о создании вебхуков. При получении информации по REST API нужно запрашивать данные по каждому элементу по его идентификатору. Оболочка fast_bitrix предлагает метод get_all, которым мы и воспользуемся. Он вернёт список всех объектов, принимая на вход только название требуемого ресурса.

Для обновления данных отправляется запрос, содержащий структуру из двух полей: идентификатор объекта и поле Fields с именем обновляемого поля и его новым значением. В общем виде приведём эту структуру, хотя в каждом случае она будет отличаться, даже именем поля с идентификатором:

[
       {
           "ID": "ID1",
           "fields":
           {
               "field_name1": data1,
               "field_name2": data2
           }
       },
       {
           "ID": "ID2",
           "fields":
           {
               "field_name1": data1,
               "field_name2": data2
           }
       },
   ]

Строить запросы начнём с более простого примера — обновления курсов валют в CRM. В этом случае будем хранить в CRM те курсы, при которых рассчитывались цены: в момент закупки товаров, после пересчёта цен. Обновим в таблице курс доллара последним его значением:

from fast_bitrix24 import Bitrix
BITRIX_URL = r'https://b24-d40p9s.bitrix24.ru/rest/1/'+BITRIX_CODE




def do_update(data):
   """
       Выполняет обновление данных курсов валют на портале Битрикс24;
       Возвращает значение предыдущего курса;


       :return: Previous dollar course
       :rtype: float
   """
  
   # Запрашиваем список всех валют с портала
   endpoint = Bitrix(BITRIX_URL)
   currency_get = endpoint.get_all("crm.currency.list")


   # Отправляем запрос на обновление курса валют
   update_data = [
       {
           "ID": 'USD',
           "fields":
           {
               "AMOUNT": data[-1]
           }
       }
   ]
   endpoint.call("crm.currency.update", items=update_data)


   for cur in currency_get:
       if cur['CURRENCY'] == 'USD':
           return float(cur['AMOUNT']), endpoint
   raise

Функция возвращает предыдущий курс доллара, полученный до обновления таблицы в системе.

Аналогично обновим сведения о товарах. Для этого получим курсы, при которых устанавливались цены в прошлый раз. Затем получим таблицу цен товаров, обратившись к каталогу. Пересчёт цен осуществляется по пропорции:

price_new = price_old*course_old/course_new.

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

def send_hook_to_bitrix(event, data):
   past_course, endpoint = do_update(data)
   ratio = data[-1]/past_course
   prices = endpoint.get_all("catalog.price.list")


   # Обновляем цены
   for p in prices:
       p['price'] *= ratio


   # Формируем данные об обновляемых полях
   update_data = [
       {
           "id": item["id"],
           "fields":
               {
                   "price": item["price"],
               }
       }
       for item in prices
   ]
   endpoint.call("catalog.price.update", items=update_data)
   print('Finished updating')

Обращаем ещё раз внимание на отличия в запросах обновления данных. При обновлении курса валют запрос содержал поля ID, Fields. Во втором же случае — поля id, Fields. Те имена полей, что отличаются в запросах и содержат идентификатор объекта, по которому обновляются сведения, совпадают с именами таких же полей в ответе метода get_all. Имена чувствительны к регистрам, в случае несовпадения сервер вернёт ошибку:

{
"order0000000000": 
{"error": 100,
"error_description": "Could not find value for parameter {id}"
}
}

В случае успеха возвращается JSON с обновлёнными объектами:

[{
"catalogGroupId": 2,
"currency": "RUB",
"extraId": null,
"id": 2,
"price": 2700,
"priceScale": 25000,
"productId": 4,
"quantityFrom": null,
"quantityTo": null,
"timestampX": "2025-01-23T21:26:51+03:00"
},
{
"catalogGroupId": 2,
"currency": "RUB",
"extraId": null,
"id": 4,
"price": 2700,
"priceScale": 25000,
"productId": 2,
"quantityFrom": null,
"quantityTo": null,
"timestampX": "2025-01-23T21:26:51+03:00"
}]

На сайте таблица с каталогом обновилась:

Пайплайн целиком выглядит так:

def main():
   current_settings = Settings()
   events = EventsList()
   # Retrieve frame for unary operations
   dates = data_frame(days=current_settings.frame_length)
   data, dates_curr = fetch_frame_values(*dates)
   unary_checks(data, current_settings, events)
   # Retrieve frame for binary operations
   dates = data_frame(days=current_settings.frame_length, base_date=dates[0])
   data_prev, dates_earl = fetch_frame_values(*dates)
   binary_checks(data, data_prev, current_settings, events)
   plot_results(dates_earl, data_prev, dates_curr, data, current_settings, events)

Выводы

Мы показали пример системы оповещения по SMS через МТС Exolve, которая автоматически отслеживает изменения курсов валют. Инструмент помогает оперативно принимать решения, а интеграция с сервисом Битрикс24 позволяет быстро обновлять информацию в магазине.