ReferenceProcess intelligence

Process intelligence

Process mining and intelligence: from event logs to causal questions. Import from ergodic.process.

The descriptive layer recovers the daily workflow over an EventLog: process_map() builds the directly-follows map as a value object (frequencies, gap statistics, and a simplify that records what it dropped), performance() reports cycle times and, when a lifecycle column exists, the waiting and service split, variant_table() ranks the trace variants, and the filter vocabulary (filter_time(), filter_variants(), filter_endpoints(), filter_activities(), filter_follows(), filter_performance()) composes, with each filter saying whether it drops events or cases.

The causal layer is the point. case_table() encodes the log into one row per case with timestamp-gated features, the named process confounders (workload, calendar), and an auto-built DomainKnowledge whose tiers forbid post-treatment adjustment mechanically; its output feeds discover(), identify_effect(), and estimate_effect() unchanged. kpi_panel() aggregates the log into a PanelData (or a single series), so a process change becomes a natural experiment for did(), event_study(), synthetic_control(), and interrupted_time_series().

The model layer goes past the map. discover_model() is the inductive miner: it cuts the log into a block-structured ProcessTree (build one by hand with seq(), xor(), par(), loop(), and TAU) whose to_net() gives a WorkflowNet that is sound by construction, with is_sound() checking any net for real. conformance() replays the log against a model, token replay for the fast fitness and optimal alignments for the per-variant deviations. And decision_points() with decision_table() make the model causal: an exclusive split is a decision, the branch taken is a treatment, the features known at the decision moment are the honest adjusters, and positivity ships as a first-class report instead of a footnote.

Also exported from ergodic: case_table, conformance, decision_points, decision_table, discover_model, kpi_panel, process_map.

END#

END: str = '__end__'

Target of the synthetic arc out of each case's last activity.

START#

START: str = '__start__'

Source of the synthetic arc into each case's first activity.

TAU#

TAU: ProcessTree = TAU

The silent step: a leaf that takes no time and leaves no event.

ActivityStats#

class ActivityStats(
    activity: str,
    count: int,
    mean_wait: float | None,
    mean_service: float | None,
) -> None

Per-activity timing: event count, mean wait before it, mean service inside it.

count is how many events carry the activity. mean_wait is None when no event of the activity has anything earlier in its case to wait on; mean_service is None on complete-only logs, where one timestamp per activity cannot split waiting from working, and for activities with no matched start/complete pair.

ArcStats#

class ArcStats(count: int, mean_gap: float, median_gap: float) -> None

Traversal count and gap timing for one arc of the map.

mean_gap and median_gap are seconds between the two events. A START or END arc has no second event, so both are NaN there (the one shared math.nan object, which keeps identically built maps equal).

CaseTable#

class CaseTable(
    data: TabularData,
    knowledge: DomainKnowledge,
    treatment: str,
    outcome: str,
    covariates: tuple[str, ...],
    when: str,
    n_cases: int,
    n_dropped_short: int = 0,
    n_dropped_treated: int = 0,
) -> None

An event log flattened to one row per case, ready for the causal stack.

data holds the covariates (sorted), the treatment, and the outcome, one row per kept case in case order. knowledge carries the tier order covariates, then treatment, then outcome. The drop counters are zero outside the landmark modes.

ConformanceResult#

class ConformanceResult(
    fitness: float,
    alignment_fitness: float,
    n_cases: int,
    n_variants: int,
    n_fitting: int,
    share_fitting: float,
    variants: tuple[VariantConformance, ...],
) -> None

How well a log fits a model, by token replay and by alignments.

fitness is the token-replay log fitness in Rozinat's form: the missing/consumed and remaining/produced sums aggregate over all cases before the two halves average, so frequent variants weigh more. alignment_fitness is the case-weighted mean of the per-variant alignment fitness. A case is fitting when its variant aligns at cost zero; n_fitting and share_fitting count those. variants holds one VariantConformance per distinct trace, most frequent first, ties by sequence.

DecisionPoint#

class DecisionPoint(
    path: tuple[int, ...],
    branches: tuple[str, ...],
    alphabets: tuple[tuple[str, ...], ...],
    counts: tuple[int, ...],
    shares: tuple[float, ...],
    n_cases: int,
    n_skipped: int,
    n_ambiguous: int,
) -> None

One xor split of a process tree, with the branch each case took.

path locates the xor node in the nodes() convention. The branch tuples line up positionally: branches[j] names the j-th child (a single visible leaf by its label, any other child by its sorted alphabet joined with +, a silent child as TAU), alphabets[j] is its activity alphabet, and counts[j] and shares[j] say how many assigned cases took it. A case is assigned the branch owning its first event inside the split's combined alphabet; events in two or more branches make it ambiguous (n_ambiguous, excluded from the counts). A case touching no branch counts as the silent branch when the split has one and is always reached; otherwise it lands in n_skipped, because the log cannot tell a silent branch from a split never reached.

DecisionTable#

class DecisionTable(
    data: TabularData,
    knowledge: DomainKnowledge,
    outcome: str,
    covariates: tuple[str, ...],
    point: DecisionPoint,
    positivity: PositivityReport,
    n_cases: int,
    n_dropped_ambiguous: int,
    treatment: str = 'branch',
) -> None

A decision point encoded for the causal stack, one row per assigned case.

data holds the sorted covariates, the categorical branch treatment, and the outcome, one row per unambiguously assigned case in case order. knowledge carries the tier order covariates, then branch, then outcome. point is the underlying DecisionPoint, positivity the coverage diagnostic, and n_dropped_ambiguous counts the cases excluded for touching two or more branches.

PositivityReport#

class PositivityReport(
    branches: tuple[str, ...],
    counts: tuple[int, ...],
    shares: tuple[float, ...],
    min_share: float,
    overlap_share: float | None,
    flagged: bool,
) -> None

Per-branch coverage of a decision table: the positivity diagnostic.

counts and shares are per branch among the included cases, and min_share is the smallest share. For a two-branch split, overlap_share is the share of cases whose in-sample propensity (a logistic regression of the branch on the covariate matrix) lies inside [0.05, 0.95]; it is None for wider splits, and 1.0 when there are no covariates, since the propensity is then the base rate for every case and the covariates cannot starve either branch.

flagged is True when min_share falls below 0.05 (one branch holds under five percent of the cases), any branch holds fewer than 10 cases (too few for a stable branch mean), or overlap_share exists and falls below 0.8 (for over a fifth of the cases the covariates nearly decide the branch, so the comparison there rests on extrapolation). A flag is a warning that the branches barely overlap, not a verdict that an estimate is wrong.

ProcessMap#

class ProcessMap(
    activities: tuple[str, ...],
    arcs: Mapping[tuple[str, str], ArcStats],
    n_cases: int,
    n_events: int,
    dropped_arcs: int = 0,
    dropped_share: float = 0.0,
) -> None

A directly-follows map: activities, timed arcs, and filter provenance.

activities lists the real activities, sorted. arcs maps each (source, target) pair to its ArcStats and iterates in sorted key order; it holds one (START, a) arc per first activity and one (b, END) arc per last, with per-case counts aggregated. A freshly built map has dropped_arcs and dropped_share at zero; they record what simplify() has removed and accumulate over repeated calls.

ProcessMap.frequency#

def ProcessMap.frequency(a: str, b: str) -> int

How many times the arc a -> b was traversed, 0 when it is absent.

ProcessMap.simplify#

def ProcessMap.simplify(keep: float = 0.8) -> ProcessMap

A thinner map keeping the busiest arcs that cover keep of the traversals.

keep must lie in (0, 1]. START and END arcs always stay, so every case still enters and leaves the map. Real arcs are ranked by count, heaviest first (ties broken by the sorted pair), and the smallest prefix whose counts reach keep times the total real traversals survives; the rest are dropped. Every activity stays, isolated ones included.

Filtering is visible, never silent: dropped_arcs and dropped_share on the returned map say how many arcs are gone and what share of the original log's real traversals they carried, and both accumulate across repeated calls.

ProcessMap.to_dict#

def ProcessMap.to_dict() -> dict[str, Any]

The map as plain data: the activities and arcs as lists, plus the counts.

ProcessModel#

class ProcessModel(
    tree: ProcessTree,
    net: WorkflowNet,
    fallthroughs: tuple[tuple[int, ...], ...],
    n_filtered_arcs: int,
    n_dropped_events: int,
    dropped_share: float,
    noise: float,
) -> None

A mined process model and the provenance of everything removed.

tree is the block-structured model and net is its workflow net, tree.to_net(), sound by construction. The rest records what the miner filtered or dropped on the way (filtering is visible, never silent):

• fallthroughs: paths, in nodes() convention (child indices from the root), of the subtrees where no cut was valid and the miner fell through to the flower model, which accepts its activities in any order. Empty when every subtree came from a real cut.
• n_filtered_arcs: directly-follows edges removed by the infrequent filter, summed over all recursions. Zero when noise is 0.
• n_dropped_events: events discarded while projecting traces through an exclusive-choice cut, where each trace goes whole to one branch and its events outside that branch are dropped.
• dropped_share: n_dropped_events over the log's total events.
• noise: the filter level the model was mined with.

ProcessPerformance#

class ProcessPerformance(
    n_cases: int,
    n_events: int,
    mean_duration: float,
    median_duration: float,
    p90_duration: float,
    activities: tuple[ActivityStats, ...],
    lifecycle: bool,
    n_unmatched: int = 0,
) -> None

Case cycle times plus the per-activity timing table.

Durations are each case's last event minus its first, in seconds. activities is sorted by activity name. lifecycle says whether a waiting/service split was possible, and n_unmatched counts lifecycle rows that found no partner.

ProcessTree#

class ProcessTree(
    op: str | None = None,
    label: str | None = None,
    children: tuple[ProcessTree, ...] = (),
) -> None

One node of a process tree.

A leaf has no children and carries an activity label (or None, which is the silent step TAU). An operator node has op set, no label, and at least two children; loop takes exactly two, the do body and the redo body. A seq child of a seq is spliced into its parent (sequence is associative), and the children of xor and par are stored sorted by their activity alphabets (a silent child first), so two trees that differ only in that order or in how sequences were nested are equal objects. Nested xor and par nodes are kept as built, so xor(a, xor(b, c)) and xor(a, b, c) describe the same behaviour but are not equal.

ProcessTree.is_leaf#

property ProcessTree.is_leaf: bool

Whether this node is a leaf (an activity or the silent step).

ProcessTree.is_tau#

property ProcessTree.is_tau: bool

Whether this node is the silent step.

ProcessTree.activities#

def ProcessTree.activities() -> tuple[str, ...]

The activity labels in this subtree, sorted and unique.

ProcessTree.nodes#

def ProcessTree.nodes() -> tuple[tuple[tuple[int, ...], ProcessTree], ...]

Every node with its path from the root, depth first.

A path is the tuple of child indices leading to the node; the root's path is ().

ProcessTree.subtree#

def ProcessTree.subtree(path: tuple[int, ...]) -> ProcessTree

The node at path, where each entry picks a child by index.

ProcessTree.to_net#

def ProcessTree.to_net() -> WorkflowNet

The workflow net of this tree, sound by construction.

Variant#

class Variant(
    sequence: tuple[str, ...],
    n_cases: int,
    share: float,
    mean_duration: float,
    median_duration: float,
) -> None

One distinct activity sequence and its case statistics.

VariantConformance#

class VariantConformance(
    sequence: tuple[str, ...],
    n_cases: int,
    replay_fitness: float,
    missing: int,
    remaining: int,
    consumed: int,
    produced: int,
    cost: int,
    worst_cost: int,
    alignment_fitness: float,
    alignment: tuple[tuple[str, str], ...],
) -> None

Replay and alignment results for one distinct trace.

sequence is the trace and n_cases how many cases follow it. The token-replay counters come next: missing tokens had to be added, remaining were left behind, and consumed and produced count every token the replay moved, the initial token and the final take from the sink included. replay_fitness is Rozinat's half-and-half of the two ratios. cost is the optimal alignment cost, worst_cost the reference cost of deleting the whole trace and running the model's cheapest visible path instead, and alignment_fitness is 1 - cost / worst_cost. alignment lists the moves: ("sync", a) where trace and model agree, ("log", a) for an event the model cannot mirror, ("model", a) for a required visible model step the trace skipped. Silent model steps cost nothing and are omitted from the list.

WorkflowNet#

class WorkflowNet(
    places: tuple[str, ...],
    transitions: tuple[tuple[str, str | None], ...],
    arcs: tuple[tuple[str, str], ...],
    source: str,
    sink: str,
) -> None

A workflow net: places, labeled or silent transitions, and arcs.

transitions maps ids to labels as (id, label) pairs, with None for a silent transition; ids and place names live in one namespace and must not collide. Arcs connect a place to a transition or a transition to a place, once each (these are weight-one nets). The constructor sorts the fields into canonical order and validates the workflow-net shape: a single source with no inputs, a single sink with no outputs, and every node on some directed path from source to sink.

WorkflowNet.initial_marking#

property WorkflowNet.initial_marking: Marking

One token on the source place.

WorkflowNet.final_marking#

property WorkflowNet.final_marking: Marking

One token on the sink place.

WorkflowNet.index#

def WorkflowNet.index() -> NetIndex

The pre/post maps for the firing rule; build once per algorithm run.

case_table#

def case_table(
    log: EventLog,
    treatment: str,
    outcome: str = 'cycle_time',
    *,
    covariates: str | Iterable[str] = 'auto',
    when: str | tuple[str, int | float] = 'anytime',
) -> CaseTable

Encode an event log as one row per case: covariates, treatment, outcome.

Returns: A CaseTable: the TabularData (columns in the order sorted covariates, treatment, outcome; one row per kept case, in case order), the tiered DomainKnowledge, and the drop counters of the landmark modes.

conformance#

def conformance(
    log: EventLog,
    model: ProcessModel | ProcessTree | WorkflowNet,
    *,
    max_states: int = 100000,
) -> ConformanceResult

Check log against model by token replay and by optimal alignments.

A mined ProcessModel contributes its net, and a process tree converts to its workflow net up front. Each distinct variant replays and aligns once, then weights by its case count: fitness aggregates the replay token sums over all cases (Rozinat's form), alignment_fitness is the case-weighted mean of the per-variant alignment fitness, and a case is fitting when its variant aligns at cost zero. max_states caps every state search (the silent closures inside replay, the alignment search, and the model's cheapest-run search); past the cap a ValueError raises rather than returning a partial answer. A model that cannot reach its final marking at all raises too, since no alignment exists for it.

decision_points#

def decision_points(
    log: EventLog,
    model: ProcessTree | ProcessModel,
) -> tuple[DecisionPoint, ...]

Locate every xor split of model and the branch each case of log took.

model is a process tree or a mined ProcessModel (its tree is read). Returns one DecisionPoint per xor node, in depth-first order. The tree must carry each activity label in exactly one leaf (a mined tree does, by construction); a duplicated label raises, since branch membership would then be ambiguous by design rather than by data.

decision_table#

def decision_table(
    log: EventLog,
    model: ProcessTree | ProcessModel,
    *,
    point: tuple[int, ...] | DecisionPoint | None = None,
    outcome: str = 'remaining_time',
    covariates: str | Sequence[str] = 'auto',
) -> DecisionTable

Encode one xor split of model as a causal table: covariates, branch, outcome.

The landmark is per case the time of its first event inside the split's combined alphabet, the moment the case reveals its branch. Only cases assigned exactly one branch enter the table; cases touching two or more branches are dropped and counted in n_dropped_ambiguous. A split with a silent (tau or empty-alphabet) branch raises: taking it leaves no event, so the decision moment is unobservable without alignment-based replay.

Returns: A DecisionTable: the TabularData (columns in the order sorted covariates, branch, outcome; one row per kept case, in case order), the tiered DomainKnowledge, the DecisionPoint, and the PositivityReport.

discover_model#

def discover_model(log: EventLog, *, noise: float = 0.0) -> ProcessModel

Mine a block-structured process model from log.

The miner works on the log's variants, weighting each distinct trace by how many cases follow it, so a large log mines as fast as its variant table. noise must lie in [0, 1): at 0 the log is taken literally, and a larger value drops, at every recursion and before cut detection, each directly-follows edge carrying less than noise times the heaviest edge leaving the same activity. The returned ProcessModel carries the tree, its workflow net, and the provenance counters.

filter_activities#

def filter_activities(
    log: EventLog,
    *,
    keep: Iterable[str] | None = None,
    mandatory: Iterable[str] | None = None,
    forbidden: Iterable[str] | None = None,
) -> EventLog

Drops cases: by mandatory and forbidden activities; keep then drops events.

A case is dropped when it misses any mandatory activity or contains any forbidden one, judged on its full trace. When keep is given, the surviving cases then lose every event whose activity is not in it, and a case left with no events vanishes. At least one argument is required, and an activity name the log has never seen raises.

filter_endpoints#

def filter_endpoints(
    log: EventLog,
    *,
    start_in: Iterable[str] | None = None,
    end_in: Iterable[str] | None = None,
) -> EventLog

Drops cases: keep cases that start and end in the given activity sets.

start_in constrains a trace's first activity and end_in its last; a None set leaves that end unconstrained. The standard way to drop incomplete cases.

filter_follows#

def filter_follows(
    log: EventLog,
    a: str,
    b: str,
    *,
    directly: bool = True,
    within: float | None = None,
) -> EventLog

Drops cases: keep cases where activity b follows activity a.

With directly (the default) b must be the very next event after an a; otherwise any later event counts. within bounds the gap in seconds from the a event to the b event, so a pair further apart does not count. Both activities must be known to the log.

filter_performance#

def filter_performance(
    log: EventLog,
    *,
    min_duration: float | None = None,
    max_duration: float | None = None,
    min_events: int | None = None,
    max_events: int | None = None,
) -> EventLog

Drops cases: by case duration in seconds (first to last event) and event count.

Bounds are inclusive and a None bound is open; give at least one.

filter_time#

def filter_time(
    log: EventLog,
    *,
    start: Any | None = None,
    end: Any | None = None,
    mode: str = 'intersecting',
) -> EventLog

Drops cases: keep cases meeting the window on event times; mode "trim" drops events.

start and end accept whatever compares to the log's stamps, a datetime-like or a number, and a None bound is open. Three modes: "contained" keeps cases whose every event falls inside [start, end], "intersecting" (the default) keeps cases with at least one event inside, and "trim" keeps only the events inside, so a case losing all its events vanishes.

filter_variants#

def filter_variants(
    log: EventLog,
    *,
    top: int | None = None,
    min_cases: int | None = None,
) -> EventLog

Drops cases: keep cases whose full activity sequence is a frequent variant.

Give exactly one selector. top keeps the top most frequent variants (count descending, ties by sequence, the variant_table() ranking); min_cases keeps every variant with at least that many cases.

is_sound#

def is_sound(net: WorkflowNet, *, max_states: int = 100000) -> bool

Whether net is sound in van der Aalst's classical sense.

Three conditions, all decided on the real state space. Option to complete: from every marking reachable from the initial marking, the final marking is still reachable. Proper completion: every reachable marking that puts a token on the sink is exactly the final marking, so no run finishes with tokens left behind. No dead transitions: every transition fires on some reachable path.

The check explores markings breadth-first from initial_marking, recording which marking fires which transition into which marking. Proper completion is then read off the explored markings, dead transitions off the union of fired transition ids, and option to complete by walking the recorded state graph backward from the final marking: every explored marking must reach it, which also settles whether the final marking is reachable at all.

A sound net is bounded, so the exploration terminates. When the number of distinct markings passes max_states, the net is either unbounded (and so unsound) or genuinely this big, and the function raises a ValueError rather than guess; raise the cap if the net really is this big.

kpi_panel#

def kpi_panel(
    log: EventLog,
    period: str | float,
    *,
    unit: str | None = None,
    kpis: Iterable[str] = ('cycle_time',),
) -> PanelData | TimeSeriesData

Aggregate an event log into per-period KPI rows for the panel estimators.

Each case belongs to the bucket of its first event and, when unit is given, to its unit, read from a case-constant column of the log (a column that varies within a case raises). Every aggregate in a row is over the cases that start in that unit-period cell. A cell where no case starts is simply absent; the panel designs handle the unbalanced grid.

period picks the bucketing. With datetime stamps pass "day", "week", or "month"; the bucket labels are np.datetime64 values: the stamp's calendar day (datetime64[D]), the Monday of the stamp's week (datetime64[D]), or the stamp's calendar month (datetime64[M]). With numeric stamps pass a positive width w; a case starting at t lands in the bucket labeled floor(t / w) * w, in the stamps' own units. A calendar name on numeric stamps raises TypeError, as does a numeric width on datetime stamps, and a width at or below zero raises ValueError.

Each name in kpis becomes one float column, named after itself, in the order given. Three names are derived from the log: "cycle_time" is the mean case duration among the cell's cases, in seconds for datetime stamps and in the stamps' own units otherwise; "n_cases" counts the cases that start in the cell; "n_events" is the mean number of events per case. Any other name reads a case-constant numeric or boolean column of the log and aggregates it by its mean in the cell, so a 0/1 case flag that switches on a period boundary comes out as an exact 0/1 column, ready to be a did treatment; a non-numeric column raises ValueError. The three derived names are reserved: when the log also carries a column by a requested derived name, the request is ambiguous and raises ValueError.

The output columns are the unit column (named after the source column), a "period" column holding the bucket labels, then the kpi columns. A kpi named like the unit column or "period" collides with those index columns and raises, as does a unit column itself named "period". With unit given the result is a PanelData keyed by (unit, "period"), the shape did, event_study, and synthetic_control consume. With unit=None the whole log is one series and the result is a TimeSeriesData with time="period", so interrupted_time_series consumes it directly.

Two caveats. A case is attributed to the bucket where it starts, so a case that runs long spills its effect into earlier buckets: its whole duration counts where it began, even when it finished much later. And the aggregation assumes the log is complete, with no still-open cases; a censored log drops exactly the long cases near its end, which biases late-bucket cycle times downward.

loop#

def loop(do: str | ProcessTree, redo: str | ProcessTree) -> ProcessTree

A loop node: do runs, then any number of redo then do rounds.

net_of_tree#

def net_of_tree(tree: ProcessTree) -> WorkflowNet

The workflow net of a process tree, by structural recursion.

Each subtree is built between an entry and an exit place: a leaf becomes one transition (silent for TAU), seq chains its children through fresh places, xor lets every child share the entry and exit, par brackets per-child places with silent split and join transitions, and loop enters through a silent step into a do body whose exit either leaves through a silent step or runs the redo body back to the do entry. The construction is sound for every tree (and is_sound() will verify it for real).

par#

def par(*children: str | ProcessTree) -> ProcessTree

A parallel node: the children interleave. Strings become leaves.

performance#

def performance(
    log: EventLog,
    *,
    lifecycle: str | None = None,
) -> ProcessPerformance

Summarize case cycle times and per-activity waiting and service.

Case durations run first event to last, in seconds, summarized as the mean, median, and 90th percentile. Per activity, count is how many events carry it.

With lifecycle=None the log has one timestamp per activity, which cannot split waiting from working: mean_wait is the mean gap from the previous event in the same case (a case's first event contributes no wait), mean_service is None, and the result's lifecycle is False.

With lifecycle="column" the named column must exist (KeyError otherwise). Rows valued "start" and "complete" are paired per case and activity in time order, each start with the next complete of the same case and activity: service is complete minus start, wait is start minus the previous complete event in the case (a start with no earlier complete contributes no wait), and rows that never find a partner are counted in n_unmatched. Other lifecycle values are ignored. Durations still run first event to last over all events.

process_map#

def process_map(log: EventLog) -> ProcessMap

Build the directly-follows map of log with counts and gap timing.

Every consecutive event pair within a case traverses the arc between its two activities; each real arc reports its count and the mean and median gap in seconds (datetime and numeric timestamps alike). Each case also traverses (START, first activity) and (last activity, END), whose gaps are NaN. An activity named like a boundary sentinel raises a ValueError, since the boundary arcs would then be ambiguous.

seq#

def seq(*children: str | ProcessTree) -> ProcessTree

A sequence node: the children run left to right. Strings become leaves.

variant_table#

def variant_table(
    log: EventLog,
    *,
    top: int | None = None,
) -> tuple[Variant, ...]

The distinct variants, most frequent first.

Ranked by case count descending, ties by sequence. share is the fraction of all cases on the variant, and the durations are first-to-last event times in seconds over the variant's cases. top truncates the table; None keeps every variant.

xor#

def xor(*children: str | ProcessTree) -> ProcessTree

An exclusive choice node: exactly one child runs. Strings become leaves.