Process mining, Python, Анализ процессов, Графы

Вводим правила для Графа или алгоритм rules checking для process mining на Python

Время прочтения: 2 мин.

Давайте поподробнее. На вход алгоритму подается лог файл, содержащий 3 столбца (id_column, dt_column, activity_column), который мы преобразовываем в pandas Data Frame. Далее, для более удобного восприятия информации, нам необходимо максимально подробно и правильно поставить задачу: «В результате алгоритма мы хотим видеть массив данных, содержащий только те цепочки событий графа процесса, в которых обязательно есть либо интересующий нас переход между шагами процесса, либо определенный шаг, либо и то и другое условие вместе.» Для этого напишем функцию с несколькими входными параметрами: исходный датафрейм, список ребер графа, список вершин графа.

def rules_cheking(frame, edges=None, activity=None):

Для начала создадим новый датафрейм из исходного, в котором будет новый столбец ‘next_activity’, который мы заполним таким образом, чтобы там был каждое следующее за исходным действие в рамках одного идентификатора. Образовавшиеся в этом столбце значения ‘nan’ заполним строкой ‘end’, это значит, что шагов после этого нет.

    rules_check = pd.DataFrame(columns=['id_column', 'activity_column', 'dt_column', 'next_activity'])
    rules_check[['id_column', 'activity_column', 'dt_column']] = frame[['id_column', 'activity_column', 'dt_column']]
    rules_check['next_activity'] = frame.groupby('id_column')['activity_column'].shift(-1)
    rules_check['next_activity'].fillna('end', inplace=True)

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

        if (activity != None) & (edges == None) :
        id_rules = rules_check.loc[rules_check['activity_column'].isin(activity)]['id_column']
        rules_check = frame[frame['id_column'].isin(id_rules)]

Второе условие будет выполняться, если вас интересуют только некоторые переходы — edges.

       elif (activity == None) & (edges != None) :
        id_rules = rules_check.loc[rules_check[['activity_column', 'next_activity']].apply(tuple, 1).isin(
        pd.DataFrame(edges, columns=['activity_column', 'next_activity']).apply(tuple, 1))]['id_column']
        rules_check = frame[frame['id_column'].isin(id_rules)]

 Иначе поиск будет выполняться по обоим заданным условиям одновременно.   

        else:
        id_rules_edges = rules_check.loc[rules_check[['activity_column', 'next_activity']].apply(tuple, 1).isin(
        pd.DataFrame(edges, columns=['activity_column', 'next_activity']).apply(tuple, 1))]['id_column']
        rules_check_edges = frame[frame['id_column'].isin(id_rules_edges)]
        id_rules_all = rules_check.loc[rules_check['activity_column'].isin(activity)]['id_column']
        rules_check = frame[frame['id_column'].isin(id_rules_all)]

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

    return rules_check

Полученный набор данных вы можете дальше анализировать другими алгоритмами Process Mining, например, найти временные аномалии или построить граф, с учетом ваших правил.

Советуем почитать