XYZ-анализ

Привет, Хабр! Сегодня рассмотрим, что такое XYZ‑анализ и как его применять для оптимизации запасов.

Зачем вообще нужен XYZ-анализ?

Иногда бывает такое, что на складе нет нужного товара, и клиент уходит к другому продавцу,  либо склад завален вещами, которые никто не купит. XYZ‑анализ помогает распилить ассортимент на три категории по степени предсказуемости спроса:

  • X‑товары: надёжные «работяги» — стабильный спрос, минимальные колебания. Эти товары нужно всегда держать под рукой.

  • Y‑товары: росредники, где спрос колеблется, но всё‑таки предсказуем. Здесь важно найти баланс.

  • Z‑товары: буйные дети — спрос крайне волатилен, и запас держат на минимуме, закупая по предзаказу.

Итак, как определить, к какой категории относится товар? Всё сводится к коэффициенту вариации (CV) — он показывает, насколько сильно варьируется спрос. Формула проста:

CV = \frac{\sigma}{\mu}

где:

  • σ— стандартное отклонение спроса,

  • μ— среднее значение спроса.

Пример:

  • Если CV≤0.5CV — товар идёт в категорию X,

  • Если 0.5<CV≤1.0 — попадает в Y,

  • Если CV>1.0» src=«https://habrastorage.org/getpro/habr/upload_files/e5c/b60/e41/e5cb60e41d364200ac06777151020a49.svg» /> — это <strong>Z</strong>.</p></li></ul><h3>XYZ-анализ на практике</h3><p>Начнём с простого скрипта, который: </p><ol><li><p>Загрузит данные о продажах, </p></li><li><p>Рассчитает среднее значение, стандартное отклонение и коэффициент вариации, </p></li><li><p>Распределит товары по категориям.</p></li></ol><pre><code class=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». Узнайте, как отправлять запросы, обрабатывать ответы и интегрировать внешние сервисы. Записаться

    Полный список открытых уроков по аналитике и не только можно посмотреть в календаре.

© Habrahabr.ru