Аналитика процессов (Process Mining) что происходит в реальности

WarningОграничение публичной документации

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

• Полная спецификация метода: Будет доступна только во внутреннем контуре разработки (Confluence / Swagger Enterprise).

1 Введение

Аномально большое повторение цепочки? Баг или фича или инсайт Отличие между удачными и неудачными процессами. Машинное обучение для прогнозирования исхода (Предсказание переходов) Обнаруженные закономерности Реактивный запас мощности

Для аналитики процессов сделаем следующее:

• Спроектируем то как данные движутся от бэкенда на Go в аналитическое хранилище. Наш инструментальный стек: ClickHouse DBMS, Python + PM4Py.
• Определим технические стандарты сбора, хранения и асинхронной обработки событийных логов (Event Logs) для последующего анализа алгоритмами Process Mining.
• Собранные логи от действий пользователей или симулятора от экранов приложения математически сверим со схемами из описания бизнес-процессов.
• Увидим «Process Mining» — то как на самом деле это работает в коде и логах симулятора.
• И на каждый из этапов привелем примеры которые содержат чистый production-ready код FastAPI-микросервиса и DDL-запросы ClickHouse без лишней теории.

2 Architecture_Data_Flow (Сквозной Data Flow: Go бэкенд -> ClickHouse DWH -> FastAPI воркер)

2.1 Схема базы данных (ClickHouse DDL)

Для изоляции транзакционной СУБД от аналитических запросов воркера, все цифровые следы реплицируются в ClickHouse. Данные физически нарезаются на диске по месяцам с помощью PARTITION BY, что исключает сканирование всего исторического архива при выборках.

-- Таблица хранения сырых логов событий процессов
CREATE TABLE business_analytics.simulation_event_logs
(
    -- Ключевой блок Process Mining (Обязательные поля)
    case_id String COMMENT 'Уникальный ID сессии процесса (Case ID)',
    activity_name String COMMENT 'Системное название шага (Activity)',
    event_timestamp DateTime64(3, 'UTC') COMMENT 'Метка времени с точностью до мс (Timestamp)',
    
    -- Продуктовый контекст (Бизнес-метрики)
    user_id String COMMENT 'ID пользователя или бота-симулятора',
    user_role LowCardinality(String) COMMENT 'Роль: User, Bot_Simulator, QA_Automation',
    scenario_id LowCardinality(String) COMMENT 'Идентификатор бизнес-сценария',
    execution_status LowCardinality(String) COMMENT 'Статус выполнения (Success/Failed/Timeout)',
    product_name String COMMENT 'Наименование продукта',
    weight Float32 COMMENT 'Остаточный вес продукта в кг',
    price UInt32 COMMENT 'Стоимость продукта в рублях',
    
    -- Поле синхронизации для CDC-коннектора
    updated_at DateTime64(3, 'UTC') COMMENT 'Время репликации записи из Postgres'
)
ENGINE = ReplacingMergeTree(updated_at)
PARTITION BY toYYYYMM(event_timestamp)
ORDER BY (scenario_id, user_role, case_id, event_timestamp, activity_name)
SETTINGS index_granularity = 8192;

2.2 Спецификация асинхронного API воркера (FastAPI)

Поскольку расчет матриц связей и Conformance Checking на миллионах строк — это блокирующая CPU-bound задача, HTTP-эндпоинты работают в асинхронном режиме Task Queue с последующим опросом статуса (Long Polling).

import os
import uuid
from typing import Dict, Literal, Any
import clickhouse_connect
import pandas as pd
import pm4py
from fastapi import FastAPI, BackgroundTasks, HTTPException, status
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field

app = FastAPI(title="Process Mining Heavy Analytics Engine", version="1.0.0")

# In-memory хранилище задач (в проде заменить на Redis)
tasks_registry: Dict[str, Dict[str, Any]] = {}

class GraphRequest(BaseModel):
    scenario_id: str = Field(..., example="fridge_cooking_flow")
    user_role: Literal["User", "Bot_Simulator", "QA_Automation"] = "User"
    output_format: Literal["mermaid", "bpmn"] = "mermaid"
    noise_threshold: float = Field(default=0.2, ge=0.0, le=1.0)

class TaskResponse(BaseModel):
    task_id: str
    status: Literal["PENDING", "PROCESSING", "SUCCESS", "FAILED"]
    message: str

def execute_process_mining_pipeline(task_id: str, req: GraphRequest):
    """Фоновая CPU-bound задача выгрузки, фильтрации и майнинга логов"""
    try:
        tasks_registry[task_id]["status"] = "PROCESSING"
        
        # 1. Подключение к кликхаусу и выгрузка плоского DataFrame
        client = clickhouse_connect.get_client(host='localhost', port=8123, username='default', password='')
        query = """
            SELECT case_id, activity_name, event_timestamp 
            FROM business_analytics.simulation_event_logs 
            WHERE scenario_id = {scen} AND user_role = {role}
            ORDER BY case_id ASC, event_timestamp ASC, activity_name ASC
        """
        df = client.query_df(query, parameters={'scen': req.scenario_id, 'role': req.user_role})
        
        if df.empty:
            raise ValueError(f"Event Log пуст для сценария {req.scenario_id} и роли {req.user_role}")

        # 2. Фикс параллелизма горутин и маппинг под требования PM4Py
        df = df.sort_values(by=['case_id', 'event_timestamp', 'activity_name'])
        rank = df.groupby('case_id')['event_timestamp'].rank(method='first') - 1
        df['event_timestamp'] = df['event_timestamp'] + pd.to_timedelta(rank, unit='us')
        
        df_log = pm4py.format_dataframe(df, case_id='case_id', activity_key='activity_name', timestamp_key='event_timestamp')

        # 3. Генерация результата в зависимости от запрошенного формата
        if req.output_format == "mermaid":
            # Используем Heuristic Miner для генерации кода графа DFG
            from pm4py.algo.discovery.heuristic import algorithm as heuristics_miner
            parameters = {heuristics_miner.Variants.CLASSIC.value.Parameters.DEPENDENCY_THRESH: 0.5}
            heu_net = heuristics_miner.apply_net(df_log, parameters=parameters)
            
            # Конвертируем эвристическую сеть в разметку Mermaid
            # (Здесь вызывается внутренний парсер формирования структуры 'graph TD')
            mermaid_code = "graph TD\n" + "\n".join([f"  {edge.start_node} --> {edge.end_node}" for edge in heu_net.edges])
            
            tasks_registry[task_id].update({
                "status": "SUCCESS",
                "result_data": {"mermaid_code": mermaid_code}
            })
        
        elif req.output_format == "bpmn":
            # Используем Inductive Miner для построения дерева и экспорта в .bpmn
            from pm4py.algo.discovery.inductive import algorithm as inductive_miner
            from pm4py.objects.bpmn.exporter import algorithm as bpmn_exporter
            
            parameters = {inductive_miner.Variants.IMf.value.Parameters.NOISE_THRESHOLD: req.noise_threshold}
            process_tree = inductive_miner.apply_tree(df_log, parameters=parameters)
            bpmn_model = pm4py.convert_to_bpmn(process_tree)
            
            file_path = f"/tmp/process_model_{task_id}.bpmn"
            bpmn_exporter.apply(bpmn_model, file_path)
            
            tasks_registry[task_id].update({
                "status": "SUCCESS",
                "file_path": file_path
            })
            
    except Exception as e:
        tasks_registry[task_id].update({
            "status": "FAILED",
            "message": str(e)
        })

@app.post("/api/v1/analytics/build-graph", status_code=status.HTTP_202_ACCEPTED, response_model=TaskResponse)
async def start_mining_job(request: GraphRequest, background_tasks: BackgroundTasks):
    """Точка входа для бэкенда Go Core. Регистрирует задачу и переводит поток в фон."""
    task_id = f"job_{uuid.uuid4().hex[:8]}"
    tasks_registry[task_id] = {
        "status": "PENDING", "format": request.output_format,
        "result_data": None, "file_path": None, "message": ""
    }
    
    background_tasks.add_task(execute_process_mining_pipeline, task_id=task_id, req=request)
    
    return {
        "task_id": task_id, "status": "PENDING",
        "message": "Задача добавлена в очередь вычислений воркера."
    }

@app.get("/api/v1/analytics/status/{task_id}")
async def check_task_status(task_id: str):
    """Эндпоинт для Long Polling со стороны Go Core."""
    if task_id not in tasks_registry:
        raise HTTPException(status_code=404, detail="Аналитическая задача не найдена")
        
    task = tasks_registry[task_id]
    
    if task["status"] in ["PENDING", "PROCESSING"]:
        return {"task_id": task_id, "status": task["status"]}
    if task["status"] == "FAILED":
        return {"task_id": task_id, "status": "FAILED", "error": task["message"]}
        
    if task["status"] == "SUCCESS":
        if task["format"] == "mermaid":
            return {"task_id": task_id, "status": "SUCCESS", "type": "mermaid", "data": task["result_data"]}
        if task["format"] == "bpmn":
            return {"task_id": task_id, "status": "SUCCESS", "type": "bpmn", "download_url": f"/api/v1/analytics/download/{task_id}"}

@app.get("/api/v1/analytics/download/{task_id}")
async def download_generated_bpmn(task_id: str):
    """Отдает сгенерированный физический файл модели процесса."""
    if task_id not in tasks_registry or tasks_registry[task_id]["status"] != "SUCCESS":
        raise HTTPException(status_code=400, detail="Файл еще не готов")
        
    file_path = tasks_registry[task_id]["file_path"]
    if not file_path or not os.path.exists(file_path):
        raise HTTPException(status_code=404, detail="Файл стерт или отсутствует на диске воркера")
        
    return FileResponse(path=file_path, media_type="application/xml", filename=f"model_{task_id}.bpmn")

3 Event_Log_Preparation (Подготовка Event Log: Case ID, Activity, Timestamp и борьба с параллелизмом)

В основе любой аналитики методом Process Mining лежит строгое математическое правило: алгоритмы не способны работать с сырыми, неструктурированными текстовыми логами (такими как stdout контейнеров или дампы Nginx).

Чтобы восстановить карту процесса, данные должны быть приведены к формату Event Log (журнал событий). Математическое ядро библиотеки PM4Py требует наличия трех обязательных атрибутов для каждой строки данных. Если отсутствует хотя бы один из них, запуск майнинга технически невозможен.

3.1 Три обязательных атрибута Event Log

3.1.1 Case ID (Идентификатор экземпляра процесса)

Это сквозной ключ, который объединяет разрозненные события в одну логическую цепочку (трассу). Определение границ Case ID — главная задача при проектировании аналитики. В кейсе цифрового холодильника Case ID — это уникальный идентификатор конкретного продукта (например, id_meat_123 для куска говядины или id_milk_789 для пачки молока). Все действия, происходящие с этим продуктом от момента закупки до утилизации, будут иметь одинаковый Case ID.Архитектурный нюанс: При вызове эндпоинта Готовка происходит трансформация сущностей (lineage). Старый продукт (id_meat_123) списывается в ноль, и в этот же момент симулятор генерирует новый Case ID для готового блюда (например, id_soup_999). Для Process Mining это две разные, но последовательные цепочки жизни продуктов.

3.1.2 Activity (Название шага / Активность)

Это атомарное действие, которое зафиксировала стейт-машина системы. Названия активностей должны быть стандартизированы и жестко привязаны к бизнес-логике. В нашей системе это названия вызванных эндпоинтов: Покупка продуктов, Получение списка в наличии, Готовка, Потребление и Выкидывание в мусор.

3.1.3 Timestamp (Точное время выполнения)

Хронологический маркер события. Для Process Mining критически важна высокая точность — формат DateTime64(3) с фиксацией миллисекунд. Без миллисекунд алгоритм не сможет понять, какое действие было совершено раньше, если бот-симулятор или быстрый пользователь выполнили несколько шагов внутри одной секунды.

Проблема параллелизма и логика склейки дубликатов времениПри интеграции бэкенда на Go Core с аналитическим хранилищем ClickHouse возникает классическая инженерная проблема: конкурентность. Когда бэкенд обрабатывает шаги процесса в параллельных горутинах, события могут записаться в базу данных одновременно, с точностью до одной и той же миллисекунды.

3.2 В чем опасность для майнинга?

If in Python-worker come two records with equal time (for example, 10:00:00.125 — Cooking and 10:00:00.125 — Getting list in stock), library PM4Py under each new launch script will sort them random. On graph process will arise false anomaly: connection will start chaotically jump in both sides, creating non-existent in reality loops and breaking Conformance Checking.

Решение: Сортировка на этапе SQL-выборки и микро-сдвиг в Pandas

Чтобы гарантировать стабильность графа, мы применяем двухэтапную склейку данных.

3.2.0.1 Этап 1: Строгая сортировка в ClickHouse.

При выгрузке логов мы принудительно заставляем СУБД выстраивать стабильный порядок строк по трем полям. Если время совпадает, записи выстроятся по алфавитному порядку названий активностей:

SELECT 
    case_id, 
    activity_name, 
    event_timestamp, 
    product_name, 
    weight, 
    price
FROM business_analytics.simulation_event_logs
WHERE scenario_id = 'fridge_cooking_flow'
-- Гарантируем детерминированный порядок выдачи строк для Python
ORDER BY case_id ASC, event_timestamp ASC, activity_name ASC;

3.2.0.2 Этап 2: Искусственный сдвиг дубликатов в Pandas (Python).

Перед тем как передать DataFrame в алгоритмы PM4Py, воркер проверяет хронологию внутри каждого case_id. Если обнаруживаются одинаковые таймстампы, код искусственно добавляет к каждому последующему шагу по 1 микросекунде, выстраивая их в идеальную, читаемую для майнера очередь.

import pandas as pd
import pm4py

def prepare_event_log(df: pd.DataFrame) -> pd.DataFrame:
    # 1. Сортируем данные для гарантии стабильности
    df = df.sort_values(by=['case_id', 'event_timestamp', 'activity_name'])
    
    # 2. Находим дубликаты времени внутри каждой группы case_id и генерируем порядковый шаг (1, 2, 3...)
    rank = df.groupby('case_id')['event_timestamp'].rank(method='first') - 1
    
    # 3. Добавляем к таймстампу микросекунды на основе ранга, убирая коллизии параллелизма
    df['event_timestamp'] = df['event_timestamp'] + pd.to_timedelta(rank, unit='us')
    
    # 4. Приводим DataFrame к внутреннему стандарту библиотеки PM4Py
    df_formatted = pm4py.format_dataframe(
        df, 
        case_id='case_id', 
        activity_key='activity_name', 
        timestamp_key='event_timestamp'
    )
    return df_formatted

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

4 Noise_Filtering_Algorithms (Алгоритмы фильтрации шума: Heuristic Miner и Inductive Miner)

Если импортировать подготовленный на первом шаге Event Log в алгоритмы генерации графов без предварительной обработки, аналитик столкнется со «спагетти-синдромом». Реальные логи стохастического симулятора или живых пользователей всегда содержат шум: разовые сетевые таймауты бэкенда, случайные двойные клики по кнопкам интерфейса во Flutter или единичные аномальные сессии.

Математические алгоритмы Process Mining честно отобразят каждую аномальную связь, превратив итоговую карту в хаотичную, нечитаемую «паутину». Чтобы граф оставался инструментом бизнес-аналитики, на этапе вычислений в Python-воркере применяется калибровка двух базовых алгоритмов фильтрации: Heuristic Miner и Inductive Miner.

4.1 1. Тюнинг Heuristic Miner через dependency_threshold

Алгоритм Heuristic Miner ориентирован на построение графов зависимостей (DFG). Его ключевая особенность — он учитывает не просто абсолютную частоту повторения шагов, а силу причинно-следственной связи (causality) между активностями.

Сила связи между шагом A и шагом B рассчитывается на основе их взаимного расположения во всех трассах лога. Если шаг B почти всегда идет сразу за A, но A никогда не идет за B, сила связи стремится к 1.0. Если они хаотично меняются местами, сила связи падает до нуля.

Главным рычагом управления шумом здесь выступает параметр dependency_threshold (порог зависимости), принимающий значения от 0.0 до 1.0:

Порог 0.9–1.0 (Режим Happy Path): Алгоритм оставляет на графе только «железобетонные» переходы, которые повторяются постоянно. Это позволяет мгновенно увидеть идеальный процесс, каким его задумал бизнес. Минус — полностью срезаются легитимные альтернативные сценарии.

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

Порог 0.0–0.2 (Режим хаоса): На граф выводятся абсолютно все переходы. Схема мгновенно превращается в «спагетти».

Реализация калибровки эвристического майнера в PM4Py:

from pm4py.algo.discovery.heuristic import algorithm as heuristics_miner
import pandas as pd

def apply_heuristic_filtering(df_formatted: pd.DataFrame):
    # Настраиваем конфигурацию фильтрации под "грязный" лог холодильника
    parameters = {
        # 1. Отсекаем все переходы, где сила причинно-следственной связи ниже 65%
        heuristics_miner.Variants.CLASSIC.value.Parameters.DEPENDENCY_THRESH: 0.65,
        
        # 2. Полностью игнорируем редкие шаги, которые встретились менее 15 раз во всем логе
        heuristics_miner.Variants.CLASSIC.value.Parameters.MIN_ACT_COUNT: 15,
        
        # 3. Фильтруем "петли в себе" (двойные клики в приложении из-за сетевых задержек)
        heuristics_miner.Variants.CLASSIC.value.Parameters.LOOP_LENGTH_TWO_THRESH: 0.3
    }
    
    # Извлекаем очищенную эвристическую сеть
    heu_net = heuristics_miner.apply_net(df_formatted, parameters=parameters)
    return heu_net

4.2 2. Тюнинг Inductive Miner через noise_threshold

В отличие от эвристического подхода, алгоритм Inductive Miner рекурсивно разделяет Event Log на подпроцессы, выстраивая строгое дерево процессов (Process Tree). Это математическая гарантия того, что итоговая модель будет sound (без логических тупиков, бесконечных циклов и зависших шлюзов), что критически важно для последующей конвертации в нотацию BPMN.

Основным инструментом фильтрации здесь является параметр noise_threshold (порог шума):

Порог 0.0: Алгоритм обязан учесть абсолютно каждую аномальную трассу из симулятора. Если хотя бы один бот из миллиона из-за сетевого сбоя перескочил с Покупки на Выкидывание, алгоритм усложнит дерево процессов десятком компенсационных шлюзов, делая модель нечитаемой.

Порог 0.2 (Оптимальный баланс): Алгоритм берет 80% наиболее частотных и стабильных цепочек выполнения, а 20% редких отклонений отбрасывает. На выходе получается чистый, понятный бизнесу каркас реального процесса.

Порог 0.8: Отрезает вообще всё, кроме одной самой популярной цепочки, чрезмерно примитивизируя логику.Реализация индуктивного майнинга с фильтрацией шума:

from pm4py.algo.discovery.inductive import algorithm as inductive_miner
import pandas as pd

def discover_clean_process_tree(df_formatted: pd.DataFrame, noise_level: float = 0.2):
    # Применяем индуктивный майнер (вариант IMf — с поддержкой фильтрации частот)
    parameters = {
        inductive_miner.Variants.IMf.value.Parameters.NOISE_THRESHOLD: noise_level
    }
    
    # Генерируем строгое дерево процессов, очищенное от заданного процента аномалий
    process_tree = inductive_miner.apply_tree(df_formatted, parameters=parameters)
    
    # Конвертируем очищенный каркас в стандартную BPMN модель
    bpmn_model = pm4py.convert_to_bpmn(process_tree)
    return bpmn_model

5 Process_Discovery_As_Is (Построение карт реальных процессов: частотные и временные графы DFG)

5.1 (Проектирование “As-Is”)

Этап Discovery (выявление) — это отправная точка визуализации в Process Mining. На этом шаге алгоритмы считывают массив подготовленных и отфильтрованных данных из ClickHouse, чтобы автоматически построить карту процесса «As-Is» (Как есть на самом деле). Аналитику не нужно строить гипотезы — структура процесса генерируется на основе фактов.

Основным инструментом визуализации на этом этапе является направленный граф зависимостей (DFG — Directly-Follows Graph). Узлами графа выступают эндпоинты бизнес-логики (Покупка, Готовка, Потребление), а стрелками — переходы между ними. Граф DFG анализируется в двух независимых проекциях: частотной и временной.

5.1.1 1. Частотный анализ (Frequency)

Частотная проекция отвечает на вопрос: «Какими путями чаще всего идут пользователи и боты симулятора?». Алгоритм подсчитывает абсолютное количество сессий (case_id), прошедших через каждый узел и каждую стрелку перехода.

Это позволяет мгновенно локализовать скрытые поведенческие паттерны:

Толщина стрелок: Чем чаще повторяется переход, тем жирнее стрелка на графе. Популярные маршруты формируют магистраль процесса (Happy Path).

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

5.1.2 2. Временной анализ (Performance) и поиск узких мест

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

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

Человеческий фактор: Продукты слишком долго лежат без действия (например, между Покупкой и Готовкой проходит в среднем 5 дней), что увеличивает риск их порчи.

Технический фактор: Бэкенд на Go Core слишком долго обрабатывает тяжелую операцию трансформации ингредиентов в новое блюдо (задержка на шаге Готовка).

5.1.3 3. Реализация генерации графа DFG в Python

Для расчета графа воркер FastAPI использует функцию discover_dfg. Чтобы не перегружать память графическими рендерами, код рассчитывает сухую математическую матрицу связей, которую можно сразу отдать на фронтенд в формате JSON или вставить в документацию как код Mermaid.

import pm4py
import pandas as pd
from typing import Dict, Any

def generate_directly_follows_graph(df_formatted: pd.DataFrame) -> Dict[str, Any]:
    """
    Генерирует частотную и временную матрицы графа процесса для визуализации
    """
    # 1. Извлекаем базовый граф DFG (частотная проекция)
    dfg_frequency, start_activities, end_activities = pm4py.discover_dfg(
        df_formatted,
        case_id_key='case_id',
        activity_key='activity_name',
        timestamp_key='event_timestamp'
    )
    
    # 2. Извлекаем временной граф DFG (проекция Performance — медианное время в секундах)
    dfg_performance = pm4py.discover_dfg_performance(
        df_formatted,
        case_id_key='case_id',
        activity_key='activity_name',
        timestamp_key='event_timestamp'
    )
    
    # 3. Формируем структуру данных для передачи по API бэкенду Go
    graph_metadata = {
        "start_nodes": list(start_activities.keys()),
        "end_nodes": list(end_activities.keys()),
        "edges_frequency": [
            {"from": edge[0], "to": edge[1], "count": count} 
            for edge, count in dfg_frequency.items()
        ],
        "edges_performance": [
            {"from": edge[0], "to": edge[1], "median_duration_sec": duration} 
            for edge, duration in dfg_performance.items()
        ]
    }
    
    return graph_metadata

5.2 Анализ поведенческих паттернов (На примере цифрового холодильника)

В отличие от стандартных BI-инструментов, которые группируют данные в плоские таблицы расходов, Process Mining извлекает из ClickHouse нелинейные поведенческие паттерны. Используя дополнительные атрибуты Event Log — «Продукт», «Вес», «Количество» и «Цена» — библиотека PM4Py позволяет разложить хаотичные действия пользователей и стохастического симулятора на понятные бизнесу сценарии.

5.2.1 1. Паттерн «Пассивного просмотра» (Залипание в интерфейсе)

На частотных графах DFG этот паттерн отображается как выраженное узкое место — жесткая циклическая петля на эндпоинте Получение списка в наличии.

Что происходит в системе: Бот или реальный пользователь вызывает метод чтения остатков холодильника по 5–12 раз подряд (параметр Количество вызовов растет), однако параметры Вес и Цена продуктов внутри сессии остаются абсолютно неизменными.

Бизнес-инсайт: Пользователь открывает приложение («залипает» в интерфейсе), проверяя наличие еды, но не переходит к активным действиям. В нотации BPMN этот сценарий невозможно адекватно спрогнозировать, так как он описывает пассивное состояние ожидания.

Продуктовое решение: Выявление этой петли алгоритмом PM4Py служит автоматическим триггером для маркетинга. При фиксации 4-го пустого просмотра остатков за сутки система генерирует событие push-уведомления: «У вас в наличии есть Куриное филе и Томаты. Предлагаем приготовить Салат “Цезарь”. Показать рецепт?». Бизнес конвертирует пассивное внимание в целевое действие.

5.2.2 2. Паттерн «Трансформации сущностей» (Продуктовая родословная / Lineage)

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

Process Mining решает эту задачу через механизм сквозного отслеживания родословной данных (Data Lineage) с контролем метрики Вес:

graph TD
    classDef ingredient fill:#EDF2F7,stroke:#4A5568,color:#2D3748;
    classDef dish fill:#E6FFFA,stroke:#319795,stroke-width:2px,color:#234E52;
    classDef logic fill:#FEFCBF,stroke:#B7791F,stroke-width:1px;

    Sub1["<b>Case_ID: meat_123</b><br/>Продукт: Куриное филе<br/>Вес: 0.8 кг"]:::ingredient
    Sub2["<b>Case_ID: veg_789</b><br/>Продукт: Томаты<br/>Вес: 0.5 кг"]:::ingredient
    
    Op["Эндпоинт: <b>Готовка</b><br/>(Списание сырья в ноль)"]:::logic
    
    NewSub["<b>Case_ID: dish_abc55</b><br/>Продукт: Домашнее рагу<br/>Вес: 1.1 кг"]:::dish

    Sub1 --> Op
    Sub2 --> Op
    Op --> NewSub

5.2.2.1 Логика работы алгоритма:

При вызове эндпоинта Готовка симулятор одновременно списывает вес исходных ингредиентов (meat_123 и veg_789) до 0.0. В эту же миллисекунду бэкенд на Go генерирует новое событие поступления с новым уникальным идентификатором case_id: dish_abc55 и именем product_name: Домашнее рагу.

5.2.2.2 Как это анализирует PM4Py:

Чтобы не разорвать единый процесс жизнедеятельности кухни, Python-воркер использует технику маппинга связанных идентификаторов (Object-Centric Process Mining). Алгоритм связывает трассы ингредиентов и готового блюда через общий контекст сессии готовки.

6 Conformance_Checking (Анализ отклонений от BPMN: верификация багов бэкенда и скрытых фич API)

6.1 Conformance Checking: Сравнение реальности с моделью BPMN

Если этап Discovery позволяет увидеть процесс в его фактическом состоянии («As-Is»), то задача Conformance Checking (проверка соответствия) — сопоставить эту реальность с эталонной моделью «To-Be» (Как должно быть), спроектированной аналитиками в нотации BPMN.

Этот этап является ключевым для бизнеса, так как он автоматически выявляет любые расхождения между заложенными алгоритмами и логами стохастического симулятора, не требуя ручного аудита кода бэкенда.

6.1.1 1. Математика Token Replay и метрика Fitness Score

Основным математическим методом проверки соответствия является Token Replay (Воспроизведение токенов). Алгоритм работает следующим образом:

Эталонная BPMN-модель преобразуется в математический граф — Сеть Петри (Petri Net), где шаги процесса становятся переходами, а условия — позициями.

В начальную позицию сети Ппетри помещается виртуальный маркер (токен).

Алгоритм последовательно берет реальную цепочку событий (case_id) из ClickHouse и пытается «протащить» токен по переходам сети Петри.

В процессе воспроизведения токенов фиксируются четыре метрики:

c (consumed): Сколько токенов было корректно потреблено позициями.

p (produced): Сколько токенов было успешно сгенерировано.

m (missing): Сколько раз токена не оказалось в нужной позиции для выполнения реального шага (пропуск шага или нарушение порядка).

r (remaining): Сколько лишних токенов осталось в сети после завершения процесса (незавершенные сессии, тупики).

На основе этих параметров вычисляется интегральный показатель соответствия — Fitness Score (индекс прилегания), который варьируется от 0.0 до 1.0:[=(1-)+(1-)]

6.1.2 2. Реализация проверки соответствия в PM4Py

Фактический код воркера на Python принимает на вход отформатированный DataFrame из ClickHouse и XML-файл идеального BPMN-процесса, рассчитывая метрику Fitness для каждой сессии.

import pm4py
import pandas as pd
from typing import Dict, Any

def evaluate_process_conformance(df_formatted: pd.DataFrame, bpmn_file_path: str) -> Dict[str, Any]:
    """
    Сверяет лог из ClickHouse с идеальной BPMN схемой методом Token Replay
    """
    # 1. Импортируем эталонную модель BPMN
    bpmn_graph = pm4py.read_bpmn(bpmn_file_path)
    
    # 2. Конвертируем BPMN в Сеть Петри для математического расчета
    net, initial_marking, final_marking = pm4py.convert_to_petri_net(bpmn_graph)
    
    # 3. Запускаем Token Replay диагностику
    replayed_traces = pm4py.conformance_diagnostics_token_replay(
        df_formatted, net, initial_marking, final_marking
    )
    
    # 4. Рассчитываем общий Fitness Score по всему логу
    log_fitness = pm4py.evaluate_fitness_token_replay(
        df_formatted, net, initial_marking, final_marking
    )
    
    return {
        "global_fitness_score": log_fitness['log_fitness'],
        "percentage_of_perfect_traces": log_fitness['percentage_of_fitting_traces'],
        "raw_traces_diagnostics": replayed_traces
    }

6.1.3 3. Отклонение от BPMN: Баг или Фича?

Главный аналитический инсайт Conformance Checking заключается в том, что обнаруженное алгоритмом падение Fitness Score — это не всегда ошибка. Процесс-майнинг делит все отклонения от BPMN на два критических класса:

6.1.3.1 Когда отклонение — это баг (Логическая ошибка):

Лог симулятора показывает цепочку [Покупка] ──► [Готовка] ──► [Мусор], при этом вес продукта обнулился на шаге «Готовка», но эндпоинт создания нового готового блюда (Супа) не зафиксировал транзакцию. Token Replay подсветит это как MISSING_TOKEN в позиции поступления готовой еды. Это явный баг бэкенда на Go Core — стейт-машина оборвала связь и потеряла продукт.

6.1.3.2 Когда отклонение — это фича (Недокументированная логика):

Симулятор выдает цепочку [Покупка] ──► [Потребление], полностью минуя промежуточный шаг [Получение списка в наличии]. Интерфейс мобильного приложения Flutter физически не имеет кнопки «Съесть прямо в магазине» — предполагалось, что пользователь обязан сначала открыть холодильник (проверить список). Однако разработчики бэкенда добавили в API метод «Быстрого списания» для внешних интеграций, но забыли внести эту ветку в исходную BPMN-диаграмму.

6.2 Ценность стохастической симуляции для майнинга

Запуск стохастического симулятора на этапе проектирования ИТ-системы часто воспринимается разработчиками как избыточное усложнение. Однако в контексте внедрения Process Mining генерация синтетических логов — это не просто способ наполнить базу данных, а критически важный инструмент верификации всей аналитической архитектуры до её вывода в реальный продакшен.

Главная ценность стохастического симулятора для майнинга заключается в возможности управляемого моделирования хаоса. Если бы симулятор генерировал логи исключительно по идеальному «солнечному» сценарию (Happy Path), заложенному в ТЗ, алгоритмы PM4Py строили бы тривиальные линейные графы. В таких условиях бизнес не смог бы оценить практическую пользу технологии.

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

Стресс-тестирование конвейера данных (Data Pipeline): Разработчики получают возможность проверить, как колоночная архитектура ClickHouse (ReplacingMergeTree, логика секционирования по месяцам) и алгоритмы сглаживания параллелизма в Python-воркере справляются с миллионами конкурентных транзакций в секунду.

Калибровка математических фильтров: На синтетическом «грязном» логе аналитики могут заранее подобрать идеальные пороги для параметров dependency_threshold и noise_threshold. Это гарантирует, что к моменту запуска системы на реальных пользователях алгоритмы будут выдавать читаемые, очищенные от спагетти-синдрома графы.

Наглядная демонстрация ROI для бизнеса: Руководство компании видит ценность Process Mining не на абстрактных примерах, а на живом кейсе своего продукта. Демонстрация того, как алгоритм Token Replay автоматически находит 14% скрытых потерь из-за багов шлюза или предлагает изменить UX интерфейса Flutter для повышения конверсии, является главным аргументом в пользу внедрения технологии.

Таким образом, симулятор превращает заготовку технической статьи в полноценный полигон для испытаний, доказывая, что Process Mining — это не просто инструмент пост-анализа, а фундамент для создания адаптивных, ориентированных на реальное поведение систем.