XYZ-анализ
Привет, Хабр! Сегодня рассмотрим, что такое XYZ‑анализ и как его применять для оптимизации запасов.
Зачем вообще нужен XYZ-анализ?
Иногда бывает такое, что на складе нет нужного товара, и клиент уходит к другому продавцу, либо склад завален вещами, которые никто не купит. XYZ‑анализ помогает распилить ассортимент на три категории по степени предсказуемости спроса:
X‑товары: надёжные «работяги» — стабильный спрос, минимальные колебания. Эти товары нужно всегда держать под рукой.
Y‑товары: росредники, где спрос колеблется, но всё‑таки предсказуем. Здесь важно найти баланс.
Z‑товары: буйные дети — спрос крайне волатилен, и запас держат на минимуме, закупая по предзаказу.
Итак, как определить, к какой категории относится товар? Всё сводится к коэффициенту вариации (CV) — он показывает, насколько сильно варьируется спрос. Формула проста:
где:
— стандартное отклонение спроса,
— среднее значение спроса.
Пример:
Если
— товар идёт в категорию X,
Если
— попадает в Y,
Если
import pandas as pd import numpy as np def classify_item(cv_value, threshold_x=0.5, threshold_y=1.0): """Классификация товара по коэффициенту вариации.""" if cv_value <= threshold_x: return 'X' elif cv_value <= threshold_y: return 'Y' else: return 'Z' def perform_xyz_analysis(dataframe, demand_columns): """ Выполняет XYZ-анализ. :param dataframe: DataFrame с данными по продажам. :param demand_columns: Список колонок с данными спроса. :return: DataFrame с колонками 'mean', 'std', 'cv' и 'category'. """ dataframe['mean'] = dataframe[demand_columns].mean(axis=1) dataframe['std'] = dataframe[demand_columns].std(axis=1) # Защищаемся от деления на ноль: если mean==0, ставим nan dataframe['cv'] = np.where(dataframe['mean'] != 0, dataframe['std'] / dataframe['mean'], np.nan) dataframe['category'] = dataframe['cv'].apply(classify_item) return dataframe # Пример данных – представьте, что это ваши реальные продажи за 4 недели: data = { 'item': ['Товар A', 'Товар B', 'Товар C', 'Товар D', 'Товар E'], 'week_1': [100, 20, 5, 80, 1], 'week_2': [110, 22, 8, 75, 0], 'week_3': [105, 18, 7, 82, 2], 'week_4': [98, 25, 6, 78, 1] } df = pd.DataFrame(data) result_df = perform_xyz_analysis(df, ['week_1', 'week_2', 'week_3', 'week_4']) print(result_df)
Как видите, всё просто: считаете статистику и классифицируете товары. Если товар показывает стабильный спрос — поздравляем, у вас X‑товар, и его запас нужно держать на уровне.
Добавим обработку ошибок и логирование.
import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def safe_mean(series): """Безопасное вычисление среднего.""" try: return series.mean() except Exception as e: logger.error("Ошибка вычисления среднего: %s", e) return np.nan def safe_std(series): """Безопасное вычисление стандартного отклонения.""" try: return series.std() except Exception as e: logger.error("Ошибка вычисления стандартного отклонения: %s", e) return np.nan def perform_xyz_analysis_pro(dataframe, demand_columns): """ Улучшенная версия анализа с логированием и обработкой ошибок. :param dataframe: DataFrame с данными по продажам. :param demand_columns: Список колонок с данными спроса. :return: DataFrame с результатами анализа. """ try: dataframe['mean'] = dataframe[demand_columns].apply(safe_mean, axis=1) dataframe['std'] = dataframe[demand_columns].apply(safe_std, axis=1) dataframe['cv'] = np.where(dataframe['mean'] != 0, dataframe['std'] / dataframe['mean'], np.nan) dataframe['category'] = dataframe['cv'].apply(classify_item) logger.info("XYZ-анализ успешно выполнен для %d товаров", len(dataframe)) except Exception as e: logger.error("Ошибка в perform_xyz_analysis_pro: %s", e) raise e return dataframe result_df_pro = perform_xyz_analysis_pro(df, ['week_1', 'week_2', 'week_3', 'week_4']) print(result_df_pro)
Автоматизация заказов: интеграция с API
Не будем останавливаться на анализе — автоматизируем процесс пополнения запасов для X‑товаров. Ниже пример кода, который отправляет заказ через REST API. При этом есть таймауты, обработка ошибок, логирование и retries:
import requests def place_order(item_id, quantity): """ Размещает заказ через API. :param item_id: Идентификатор товара. :param quantity: Количество для заказа. :return: JSON-ответ API или None. """ api_url = "https://api.example.com/orders" payload = {"item_id": item_id, "quantity": quantity} try: response = requests.post(api_url, json=payload, timeout=5) response.raise_for_status() logger.info("Заказ для товара %s успешно размещён", item_id) return response.json() except requests.RequestException as e: logger.error("Ошибка при размещении заказа для товара %s: %s", item_id, e) return None # Пример вызова для X-товара: order_result = place_order("A123", 500) if order_result: logger.info("Заказ успешно размещён: %s", order_result) else: logger.error("Заказ не был размещён")
Используем стандартную библиотеку requests. Если вдруг API не отвечает, сразу можно увидеть ошибку в логах.
ETL-пайплайн для регулярного анализа запасов
Зачем вручную обновлять данные, когда можно автоматизировать ETL‑процесс? Настроим пайплайн, который каждый день анализирует продажи и обновляет результаты XYZ‑анализа.
import schedule import time def etl_pipeline(): """ETL-процесс для обновления данных анализа запасов.""" # Этап 1: Извлечение данных (Extract) try: df = pd.read_csv('sales_data.csv') logger.info("Данные успешно загружены") except Exception as e: logger.error("Ошибка загрузки данных: %s", e) return # Этап 2: Преобразование данных (Transform) try: df_analyzed = perform_xyz_analysis_pro(df, ['week_1', 'week_2', 'week_3', 'week_4']) logger.info("Данные успешно проанализированы") except Exception as e: logger.error("Ошибка анализа данных: %s", e) return # Этап 3: Загрузка данных (Load) try: df_analyzed.to_csv('analyzed_data.csv', index=False) logger.info("Результаты анализа успешно сохранены") except Exception as e: logger.error("Ошибка сохранения результатов: %s", e) # Планируем выполнение ETL-пайплайна каждый день в 01:00 schedule.every().day.at("01:00").do(etl_pipeline) logger.info("ETL-процесс запущен, ожидаем следующего запуска...") while True: schedule.run_pending() time.sleep(60)
Нюансы
Конечно, XYZ‑анализ — не панацея. Вот что стоит помнить:
Сезонные колебания: если спрос резко меняется, исторические данные могут вводить в заблуждение.
Внешние факторы: маркетинговые акции, экономические кризисы или погодные условия могут нарушить расчёты.
Комплексность модели: иногда бывает проще комбинировать XYZ‑анализ с другими методами, например, с ABC‑анализом или прогнозированием с помощью машинного обучения.
В общем, не стоит слепо доверять любому алгоритму — всегда полезно иметь запасной план.
Совмещенный ABC/XYZ-анализ
Как часто бывает в бизнесе, одной только сегментации по вариативности спроса (XYZ‑анализ) бывает недостаточно для оптимизации запасов. Нужно учитывать ещё и вклад товара в общий оборот или прибыль, здесь хорошо поможет ABC‑анализ.
ABC‑анализ обычно подразумевает, что товары ранжируются по их суммарным продажам (или стоимости), после чего распределяются по группам:
A‑товары — это лидеры по продажам, зачастую составляющие 70–80% общего оборота при сравнительно небольшом количестве позиций.
B‑товары — товары со средней значимостью, их доля может варьироваться, например, 15–20%.
C‑товары — менее значимые позиции, которые вместе составляют оставшиеся 5–10% оборота, но могут быть очень многочисленными.
Объединяя ABC‑анализ с XYZ‑анализом, можно получить матрицу, которая позволяет выстроить индивидуальные стратегии для каждой ячейки, например, для товаров типа «A‑X» (лидеры продаж и стабильный спрос) можно настроить автоматическую перезаказку, а для «C‑Z» — использовать предзаказы и гибкие схемы закупок.
Вычислим показатели XYZ‑анализ (на основе коэффициента вариации), а затем проведём ABC‑анализ, ранжируя товары по суммарным продажам:
import pandas as pd import numpy as np import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Функция для классификации по XYZ-анализу def classify_xyz(cv_value, threshold_x=0.5, threshold_y=1.0): if cv_value <= threshold_x: return 'X' elif cv_value <= threshold_y: return 'Y' else: return 'Z' def perform_xyz_analysis(df, demand_columns): """ Выполняет XYZ-анализ для заданного DataFrame. Добавляет колонки: 'mean', 'std', 'cv' и 'xyz'. """ df['mean'] = df[demand_columns].mean(axis=1) df['std'] = df[demand_columns].std(axis=1) # Предотвращаем деление на 0 df['cv'] = np.where(df['mean'] != 0, df['std'] / df['mean'], np.nan) df['xyz'] = df['cv'].apply(classify_xyz) logger.info("XYZ-анализ выполнен для %d товаров", len(df)) return df # Функция для ABC-анализу: рассчитываем суммарные продажи и ранжируем товары def perform_abc_analysis(df, sales_columns, a_threshold=0.7, b_threshold=0.9): """ Выполняет ABC-анализ для заданного DataFrame. Добавляет колонки: 'total_sales', 'cum_pct' и 'abc'. a_threshold и b_threshold задают пороги для отнесения к группам A и B. """ df['total_sales'] = df[sales_columns].sum(axis=1) df_sorted = df.sort_values(by='total_sales', ascending=False).copy() total_sum = df_sorted['total_sales'].sum() df_sorted['cum_pct'] = df_sorted['total_sales'].cumsum() / total_sum def classify_abc(cum_pct): if cum_pct <= a_threshold: return 'A' elif cum_pct <= b_threshold: return 'B' else: return 'C' df_sorted['abc'] = df_sorted['cum_pct'].apply(classify_abc) logger.info("ABC-анализ выполнен для %d товаров", len(df_sorted)) # Восстанавливаем исходный порядок df_final = df_sorted.sort_index() return df_final # Функция для совмещенного ABC/XYZ-анализа def perform_abc_xyz_analysis(df, demand_columns, sales_columns): """ Выполняет совмещенный ABC/XYZ-анализ. Возвращает DataFrame с колонками: 'abc' и 'xyz' вместе с дополнительными метриками. """ try: df_xyz = perform_xyz_analysis(df.copy(), demand_columns) df_abc = perform_abc_analysis(df.copy(), sales_columns) # Объединяем результаты по индексу (или по уникальному идентификатору, если он есть) df_combined = df_xyz.join(df_abc[['abc']], how='left') logger.info("ABC/XYZ-анализ успешно объединён") return df_combined except Exception as e: logger.error("Ошибка в совмещённом анализе: %s", e) raise e # Пример данных – предположим, что у нас есть продажи за 4 недели data = { 'item': ['Товар A', 'Товар B', 'Товар C', 'Товар D', 'Товар E'], 'week_1': [100, 20, 5, 80, 1], 'week_2': [110, 22, 8, 75, 0], 'week_3': [105, 18, 7, 82, 2], 'week_4': [98, 25, 6, 78, 1] } df = pd.DataFrame(data) # Для XYZ-анализ используем колонки с данными спроса, для ABC – те же продажи result_df = perform_abc_xyz_analysis(df, demand_columns=['week_1', 'week_2', 'week_3', 'week_4'], sales_columns=['week_1', 'week_2', 'week_3', 'week_4']) print(result_df[['item', 'total_sales', 'cum_pct', 'abc', 'mean', 'std', 'cv', 'xyz']])
Мы вычисляем среднее значение и стандартное отклонение для заданного набора данных по спросу. Коэффициент вариации помогает понять, насколько стабилен спрос, а функция classify_xyz распределяет товары по категориям X, Y и Z.
По ABC анализу суммируются продажи за указанный период, товары сортируются по убыванию, затем вычисляется накопленный процент от общего объёма продаж. На основе этих значений происходит классификация: товары, совокупный вклад которых не превышает 70% — получают метку «A», следующие до 90% — «B», а оставшиеся — «C».
Результаты обоих анализов объединяются по индексу DataFrame. Теперь есть два ключевых показателя для каждого товара: его значение по ABC‑анализу (вклад в продажи) и по XYZ‑анализу (стабильность спроса).
Спасибо, что дочитали до конца. Если вы хотите поделиться своим опытом — пишите в комментариях.
Хотите автоматизировать управление запасами? Разберём работу с API в Python на открытом уроке 18 марта «Эффективное использование библиотеки requests». Узнайте, как отправлять запросы, обрабатывать ответы и интегрировать внешние сервисы. Записаться
Полный список открытых уроков по аналитике и не только можно посмотреть в календаре.