pm4py.stats module#

The pm4py.stats module contains the statistical functionalities offered in pm4py.

pm4py.stats.get_start_activities(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[str, int][source]#

Returns the start activities and their frequencies from a log object.

Parameters:

log – Log object (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping start activity names to their frequencies.

import pm4py

start_activities = pm4py.get_start_activities(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.get_end_activities(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[str, int][source]#

Returns the end activities and their frequencies from a log object.

Parameters:

log – Log object (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping end activity names to their frequencies.

import pm4py

end_activities = pm4py.get_end_activities(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.get_event_attributes(log: EventLog | DataFrame) → List[str][source]#

Returns the list of event-level attributes in the log.

Parameters:: log – Log object (EventLog or pandas DataFrame).
Returns:: A list of event attribute names.

import pm4py

event_attributes = pm4py.get_event_attributes(dataframe)

pm4py.stats.get_trace_attributes(log: EventLog | DataFrame) → List[str][source]#

Returns the list of trace-level attributes in the log.

Parameters:: log – Log object (EventLog or pandas DataFrame).
Returns:: A list of trace attribute names.

import pm4py

trace_attributes = pm4py.get_trace_attributes(dataframe)

pm4py.stats.get_event_attribute_values(log: EventLog | DataFrame, attribute: str, count_once_per_case: bool = False, case_id_key: str = 'case:concept:name') → Dict[str, int][source]#

Returns the values and their frequencies for a specified event attribute.

Parameters:

log – Log object (EventLog or pandas DataFrame).
attribute – The event attribute to analyze.
count_once_per_case – If True, count each attribute value at most once per case.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping attribute values to their frequencies.

import pm4py

activities = pm4py.get_event_attribute_values(
    dataframe,
    'concept:name',
    case_id_key='case:concept:name'
)

pm4py.stats.get_trace_attribute_values(log: EventLog | DataFrame, attribute: str, case_id_key: str = 'case:concept:name') → Dict[str, int][source]#

Returns the values and their frequencies for a specified trace attribute.

Parameters:

log – Log object (EventLog or pandas DataFrame).
attribute – The trace attribute to analyze.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping trace attribute values to their frequencies.

import pm4py

tr_attr_values = pm4py.get_trace_attribute_values(
    dataframe,
    'case:attribute',
    case_id_key='case:concept:name'
)

pm4py.stats.get_variants(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', max_repetitions: int = 9223372036854775807) → Dict[Tuple[str], List[Trace]] | Dict[Tuple[str], int][source]#

Retrieves the variants from the log.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.
max_repetitions – Maximum number of consecutive repetitions for an activity. Reduces variants by limiting consecutive activity repetitions.

Returns:

A dictionary mapping activity tuples to their counts or lists of traces.

import pm4py

variants = pm4py.get_variants(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.get_variants_as_tuples(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', max_repetitions: int = 9223372036854775807) → Dict[Tuple[str], List[Trace]] | Dict[Tuple[str], int][source]#

Retrieves the variants from the log, where the variant keys are tuples.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.
max_repetitions – Maximum number of consecutive repetitions for an activity. Reduces variants by limiting consecutive activity repetitions.

Returns:

A dictionary mapping activity tuples to their counts or lists of traces.

import pm4py

variants = pm4py.get_variants_as_tuples(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.split_by_process_variant(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', variant_column: str = '@@variant_column', index_in_trace_column: str = '@@index_in_trace') → Iterator[Tuple[Collection[str], DataFrame]][source]#

Splits an event log into sub-dataframes for each process variant. The result is an iterator over the variants along with their corresponding sub- dataframes.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.
variant_column – Name of the utility column that stores the variant’s tuple.
index_in_trace_column – Name of the utility column that stores the index of the event in the case.

Returns:

An iterator of tuples, each containing a variant and its corresponding sub-dataframe.

import pandas as pd
import pm4py

dataframe = pd.read_csv('tests/input_data/receipt.csv')
dataframe = pm4py.format_dataframe(dataframe)
for variant, subdf in pm4py.split_by_process_variant(dataframe):
    print(variant)
    print(subdf)

pm4py.stats.get_variants_paths_duration(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', variant_column: str = '@@variant_column', variant_count: str = '@@variant_count', index_in_trace_column: str = '@@index_in_trace', cumulative_occ_path_column: str = '@@cumulative_occ_path_column', times_agg: str = 'mean') → DataFrame[source]#

Associates a pandas DataFrame aggregated by variants and their positions within each variant.

Each row includes:

The variant
The position within the variant
The source activity of the path
The target activity of the path
An aggregation of the times between the two activities (e.g., mean)
The cumulative occurrences of the path within the case

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.
variant_column – Name of the utility column that stores the variant’s tuple.
variant_count – Name of the utility column that stores the variant’s occurrence count.
index_in_trace_column – Name of the utility column that stores the index of the event in the case.
cumulative_occ_path_column – Name of the column that stores the cumulative occurrences of the path within the case.
times_agg – Aggregation function to be used for time differences (e.g., “mean”, “median”).

Returns:

A pandas DataFrame with the aggregated variant paths and durations.

import pandas as pd
import pm4py

dataframe = pd.read_csv('tests/input_data/receipt.csv')
dataframe = pm4py.format_dataframe(dataframe)

var_paths_durs = pm4py.get_variants_paths_duration(dataframe)
print(var_paths_durs)

pm4py.stats.get_stochastic_language(*args, **kwargs) → Dict[List[str], float][source]#

Retrieves the stochastic language from the provided object.

The stochastic language represents the probabilities of different traces or sequences within the process.

Parameters:

args – The input object, which can be a pandas DataFrame, EventLog, accepting Petri net, or ProcessTree.
kwargs – Additional keyword arguments.

Returns:

A dictionary mapping sequences of activities to their probabilities.

import pm4py

# From an event log
log = pm4py.read_xes('tests/input_data/running-example.xes')
language_log = pm4py.get_stochastic_language(log)
print(language_log)

# From a Petri net
net, im, fm = pm4py.read_pnml('tests/input_data/running-example.pnml')
language_model = pm4py.get_stochastic_language(net, im, fm)
print(language_model)

pm4py.stats.get_minimum_self_distances(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[str, int][source]#

Computes the minimum self-distance for each activity observed in an event log.

The self-distance of an activity a in a trace is defined as follows:

In a trace <a>, it’s infinity.
In a trace <a, a>, it’s 0.
In a trace <a, b, a>, it’s 1.
And so on.

The minimum self-distance for an activity is the smallest self-distance observed across all traces.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping each activity to its minimum self-distance.

import pm4py

msd = pm4py.get_minimum_self_distances(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.get_minimum_self_distance_witnesses(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[str, Set[str]][source]#

Derives the minimum self-distance witnesses for each activity.

A ‘witness’ is an activity that occurs between two occurrences of the same activity at the minimum self-distance. For example, if the minimum self-distance of activity a is 2, then in a trace <a, b, c, a>, activities b and c are witnesses of a.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping each activity to a set of its witness activities.

import pm4py

msd_wit = pm4py.get_minimum_self_distance_witnesses(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.get_case_arrival_average(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → float[source]#

Calculates the average time difference between the start times of two consecutive cases.

This metric is based on the definition: Cycle time = Average time between completion of units.

Example: In a manufacturing facility producing 100 units in a 40-hour week, the average throughput rate is 1 unit per 0.4 hours (24 minutes per unit). Therefore, the cycle time is 24 minutes on average.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

The average case arrival time in the same units as the timestamp.

import pm4py

case_arr_avg = pm4py.get_case_arrival_average(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.get_rework_cases_per_activity(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[str, int][source]#

Identifies activities that have rework occurrences, i.e., activities that occur more than once within the same case. The output is a dictionary mapping each such activity to the number of cases in which rework occurred.

Parameters:

log – Log object (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping each activity with rework to the number of cases where rework occurred.

import pm4py

rework = pm4py.get_rework_cases_per_activity(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.get_case_overlap(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → List[int][source]#

Associates each case in the log with the number of cases that are concurrently open.

Parameters:

log – Log object (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A list where each element corresponds to a case and indicates the number of overlapping cases.

import pm4py

overlap = pm4py.get_case_overlap(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

Deprecated since version 2.3.0: This will be removed in 3.0.0. The get_case_overlap function will be removed in a future release.

pm4py.stats.get_cycle_time(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → float[source]#

Calculates the cycle time of the event log.

Cycle time is defined as the average time between the completion of units.

Example: In a manufacturing facility producing 100 units in a 40-hour week, the average throughput rate is 1 unit per 0.4 hours (24 minutes per unit). Therefore, the cycle time is 24 minutes on average.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

The cycle time as a float.

import pm4py

cycle_time = pm4py.get_cycle_time(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.get_service_time(log: EventLog | DataFrame, aggregation_measure: str = 'mean', activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', start_timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[str, float][source]#

Computes the service time for each activity in the event log using the specified aggregation measure.

Service time refers to the duration an activity takes within a case.

Parameters:

log – Event log (EventLog or pandas DataFrame).
aggregation_measure – Aggregation function to apply (e.g., “mean”, “median”, “min”, “max”, “sum”).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
start_timestamp_key – Attribute to be used for the start timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping each activity to its aggregated service time.

import pm4py

log = pm4py.read_xes('tests/input_data/interval_event_log.xes')
mean_serv_time = pm4py.get_service_time(
    log,
    start_timestamp_key='start_timestamp',
    aggregation_measure='mean'
)
print(mean_serv_time)

median_serv_time = pm4py.get_service_time(
    log,
    start_timestamp_key='start_timestamp',
    aggregation_measure='median'
)
print(median_serv_time)

pm4py.stats.get_all_case_durations(log: EventLog | DataFrame, business_hours: bool = False, business_hour_slots=[(25200, 61200), (111600, 147600), (198000, 234000), (284400, 320400), (370800, 406800)], activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → List[float][source]#

Retrieves the durations of all cases in the event log.

Parameters:

log – Event log (EventLog or pandas DataFrame).
business_hours – If True, computes durations based on business hours; otherwise, uses calendar time.
business_hour_slots –
Work schedule of the company as a list of tuples. Each tuple represents a time slot in seconds since the week start. Example:
```
[
    (7 * 60 * 60, 17 * 60 * 60),
    ((24 + 7) * 60 * 60, (24 + 12) * 60 * 60),
    ((24 + 13) * 60 * 60, (24 + 17) * 60 * 60),
]
```
This example means:
- Monday 07:00 - 17:00
- Tuesday 07:00 - 12:00
- Tuesday 13:00 - 17:00
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A sorted list of case durations.

import pm4py

case_durations = pm4py.get_all_case_durations(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.get_case_duration(log: EventLog | DataFrame, case_id: str, business_hours: bool = False, business_hour_slots=[(25200, 61200), (111600, 147600), (198000, 234000), (284400, 320400), (370800, 406800)], activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str | None = None) → float[source]#

Retrieves the duration of a specific case.

Parameters:

log – Event log (EventLog or pandas DataFrame).
case_id – Identifier of the case whose duration is to be retrieved.
business_hours – If True, computes duration based on business hours; otherwise, uses calendar time.
business_hour_slots –
Work schedule of the company as a list of tuples. Each tuple represents a time slot in seconds since the week start. Example:
```
[
    (7 * 60 * 60, 17 * 60 * 60),
    ((24 + 7) * 60 * 60, (24 + 12) * 60 * 60),
    ((24 + 13) * 60 * 60, (24 + 17) * 60 * 60),
]
```
This example means:
- Monday 07:00 - 17:00
- Tuesday 07:00 - 12:00
- Tuesday 13:00 - 17:00
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

The duration of the specified case.

Example:

import pm4py

duration = pm4py.get_case_duration(
    dataframe,
    'case_1',
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.get_frequent_trace_segments(log: EventLog | DataFrame, min_occ: int, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Counter[source]#

Retrieves frequent trace segments (sub-sequences of activities) from an event log. Each trace segment is preceded and followed by “…”, indicating that it can be part of a larger sequence.

Parameters:

log – Event log (EventLog or pandas DataFrame).
min_occ – Minimum number of occurrences for a trace segment to be included.
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A Counter object mapping trace segments to their occurrence counts.

import pm4py

log = pm4py.read_xes("tests/input_data/receipt.xes")
traces = pm4py.get_frequent_trace_segments(log, min_occ=100)
print(traces)

pm4py.stats.get_activity_position_summary(log: EventLog | DataFrame, activity: str, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[int, int][source]#

Summarizes the positions of a specific activity across all cases in the event log.

For each occurrence of the activity, records its position within the trace. For example, if ‘A’ occurs 1000 times in position 1 and 500 times in position 2, the returned dictionary will be {1: 1000, 2: 500}.

Parameters:

log – Event log object (EventLog or pandas DataFrame).
activity – The activity to analyze.
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping positions (0-based index) to the number of times the activity occurs in that position.

import pm4py

act_pos = pm4py.get_activity_position_summary(
    dataframe,
    'Act. A',
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.get_process_cube(feature_table: DataFrame, x_col: str, y_col: str, agg_col: str, parameters: Dict[Any, Any] | None = None) → Tuple[DataFrame, Dict[Any, Any]][source]#

Build a two-dimensional process cube by slicing and aggregating a feature table.

Parameters:

feature_table (pd.DataFrame) – A table containing one row per case and columns for:
- ‘case:concept:name’ (case identifiers)
- Numeric metrics (e.g. ‘@@arrival_rate’, ‘@@finish_rate’)
- Prefix-based indicators (e.g. ‘concept:name_<activity>’, ‘org:resource_<resource>’)
- Aggregation target (e.g. ‘@@sojourn_time’)
Typically constructed via:
```
enriched_df = pm4py.extract_outcome_enriched_dataframe(log_df)
feature_table = pm4py.extract_features_dataframe(enriched_df, include_case_id=True)
```
x_col (str) – Name of the X dimension:
- Numeric mode: If this exact column exists (e.g. ‘@@arrival_rate’), cases will be binned by value into equally sized ranges.
- Prefix mode: If not present, treated as prefix for indicator columns (e.g. ‘concept:name’), grouping each column starting with this prefix. Each bin corresponds to presence/absence of that prefix-based feature.
y_col (str) – Name of the Y dimension (same rules as x_col):
- Numeric if present in columns (e.g. ‘@@finish_rate’).
- Prefix-based otherwise (e.g. ‘case:channel’).
agg_col (str) – Name of the column to aggregate over each cell. Must be present in feature_table. Commonly a duration or count metric, such as ‘@@sojourn_time’.
variant (Variants, optional) – Algorithm variant to use. Defaults to Variants.CLASSIC.
parameters (Dict[Any, Any], optional) – Additional settings for numeric binning and aggregation:
- Parameters.MAX_DIVISIONS_X (int): Number of bins for X in numeric mode.
- Parameters.MAX_DIVISIONS_Y (int): Number of bins for Y in numeric mode.
- Parameters.AGGREGATION_FUNCTION (str): One of ‘mean’, ‘sum’, ‘min’, ‘max’.

Returns:

pivot_df (pd.DataFrame) – Rows are X bins, columns are Y bins; each cell holds the aggregated agg_col value.
cell_case_dict (Dict[Tuple, Set]) – Maps each (x_bin, y_bin) to the set of case IDs included in that cell, allowing drill-down.

Examples

Activity vs. Resource, aggregating sojourn time:

cube_df, cell_cases = pm4py.get_process_cube(
    feature_table,
    x_col="concept:name",             # prefix mode: one bin per activity
    y_col="org:resource",             # prefix mode: one bin per resource
    agg_col="@@sojourn_time",          # total case time per cell
    parameters={
        "aggregation_function": "sum"
    }
)

Binary presence of a specific activity vs. channel:

cube_df2, cell_cases2 = pm4py.get_process_cube(
    feature_table,
    x_col="concept:name_T06Determinenecessityofstopadvice",
    y_col="case:channel",
    agg_col="@@sojourn_time"
)

Numeric arrival vs. finish rate:

cube_df3, cell_cases3 = pm4py.get_process_cube(
    feature_table,
    x_col="@@arrival_rate",
    y_col="@@finish_rate",
    agg_col="@@sojourn_time",
    parameters={
        "max_divisions_x": 5,
        "max_divisions_y": 5,
        "aggregation_function": "mean"
    }
)

pm4py.stats module#

PMTk

PM4Py

Company