Alignments and fitness
An alignment matches a trace against the best path through the model. Where both agree, it makes a synchronous move. A log move is an event the model cannot explain, a model move is an activity the model required but the trace lacks. Each deviation costs 1 by default, so the cost of an alignment counts the deviations.
Rust4PM aligns each trace variant once and weights the result by how often the variant occurs, which keeps this fast on logs with many traces.
import r4pm
import r4pm.petri_net
log = r4pm.import_item("EventLog", "data/road-traffic-fines.xes.gz")
projection = r4pm.bindings.log_to_activity_projection(log)
net = r4pm.bindings.discover_alphaplusplusplus(
projection,
{
"balance_thresh": 0.1,
"fitness_thresh": 0.8,
"replay_thresh": 0.0,
"log_repair_skip_df_thresh_rel": 4.0,
"log_repair_loop_df_thresh_rel": 4.0,
"absolute_df_clean_thresh": 1,
"relative_df_clean_thresh": 0.01,
},
)
# Any Petri net works here, e.g. one from r4pm.petri_net.import_pnml("model.pnml").
alignments = r4pm.bindings.align_variants(net, projection)
fitness = r4pm.bindings.compute_fitness(alignments, net)
print(f"log fitness {fitness['log_fitness']:.3f}")
print(f"{fitness['perfectly_fitting_frac']:.1%} of traces fit perfectly")
# Each variant gets one alignment. Its cost counts the deviations.
deviating = [a for a in alignments if a["result"]["Ok"]["cost"] > 0]
deviating.sort(key=lambda a: -a["frequency"])
for a in deviating[:3]:
print(f"{a['frequency']} traces, cost {a['result']['Ok']['cost']}: {' > '.join(a['activities'])}")log fitness 0.886
51.7% of traces fit perfectly
46371 traces, cost 1: Create Fine > Payment
9520 traces, cost 1: Create Fine > Send Fine > Insert Fine Notification > Add penalty > Payment
3736 traces, cost 2: Create Fine > Send Fine > Insert Fine Notification > Add penalty > Payment > PaymentThe second most frequent variant, a fine paid right away, costs 1, so the discovered net does not allow this exact sequence. Finding where a model and the recorded behavior disagree is what alignments are for.
Results
align_variants returns one entry per variant with activities, frequency and result. A successful result holds the moves (synchronous, log or model moves with the transitions involved), the total cost and how many search states were visited.
compute_fitness aggregates them:
| Field | Meaning |
|---|---|
log_fitness | 1 minus the total cost, divided by the worst case: every event as a log move plus the shortest model run for every trace |
average_fitness | The same ratio per trace, averaged over all traces |
perfectly_fitting_frac | Fraction of traces with cost 0 |
total_costs | Sum of all alignment costs, weighted by variant frequency |
Options
Both functions take AlignmentOptions, which are optional in Python:
cost_fnwithmodel_move_cost,log_move_cost(both 1 by default),sync_move_costandsilent_move_cost(both 0). Raisemodel_move_costabovelog_move_costwhen skipping a required activity is worse than doing an extra one.max_states, the search limit per alignment (100,000 by default). If an alignment needs more states, its variant gets an error result andcompute_fitnessfails withSearchError(LimitReached). Raise the limit, at the cost of time and memory, or simplify the model. With Alpha+++'s default thresholds on this log, the 100,000 limit is too low and 1,000,000 works.
In Python, pass the options as a dict:
options = {
"cost_fn": {"model_move_cost": 1, "log_move_cost": 1, "sync_move_cost": 0, "silent_move_cost": 0},
"max_states": 1_000_000,
}
alignments = r4pm.bindings.align_variants(net, projection, options)
fitness = r4pm.bindings.compute_fitness(alignments, net, options)Your own model
The snippet discovers its model to stay self-contained. Any Petri net works:
- Python:
net = r4pm.petri_net.import_pnml("model.pnml"), orr4pm.petri_net.from_pm4py(...)for a PM4Py net. - Rust:
PetriNet::import_from_path("model.pnml")?.
The net needs an initial and a final marking.
References: align_variants, compute_fitness.