petri_net_nn.anomalies

anomalies

Synthetic anomaly generators for XES traces.

§7.2 of the architecture spec promises anomaly detection grounded in the Petri-net substrate — a process instance that produces an unusual activation pattern, relative to the trained distribution, is anomalous. To quantify that promise we need controlled anomalies: traces that deliberately violate the structure in known ways. These generators produce them.

Each function takes a normal XESTrace and returns a new XESTrace with the corruption applied; the input trace is not mutated. The corruption types correspond to the kinds of process deviations that §10 Step 4 calls out as evaluation targets:

• drop_event — a step is skipped entirely;
• insert_event — a spurious step is added;
• swap_event_labels — branch flipping (a step from the wrong BPMN branch is recorded);
• shuffle_events — the right steps fire but out of order.

A frequency-baseline detector lives here too, so anomaly evaluations can compare the structured Petri-net detection against a model that sees only marginal event frequencies — isolating the contribution of the structural prior plus attribute conditioning.

FrequencyBaseline

FrequencyBaseline(*, smoothing=0.5)

Marginal-frequency anomaly detector.

Fits a unigram distribution over event labels in a training set of traces; the anomaly score of a new trace is the negative log probability of its events under that distribution, summed across events. Unseen labels get a small smoothing mass.

This baseline deliberately ignores trace-level attributes and the process structure. Its purpose in Phase 7 is to provide a contrast: branch-flip anomalies that the structured Petri-net detector can catch — because it conditions on the input marking derived from trace attributes — are invisible to the frequency baseline, which sees only that the events are familiar event labels.

Source code in petri_net_nn/anomalies.py

def __init__(self, *, smoothing: float = 0.5) -> None:
    if smoothing <= 0:
        raise ValueError("smoothing must be positive")
    self.smoothing = smoothing
    self._log_probs: dict[str, float] = {}
    self._log_unseen: float = 0.0

drop_event

drop_event(trace, index=-1)

Return a copy of trace with one event removed (last by default). Empty input traces are returned unchanged — there is nothing to drop.

Source code in petri_net_nn/anomalies.py

def drop_event(trace: XESTrace, index: int = -1) -> XESTrace:
    """Return a copy of ``trace`` with one event removed (last by
    default). Empty input traces are returned unchanged — there is
    nothing to drop."""
    if not trace.events:
        return _copy_trace(trace)
    new = _copy_trace(trace)
    if index < 0:
        index += len(new.events)
    del new.events[index]
    return new

insert_event

insert_event(trace, label, *, index=None)

Return a copy of trace with a new event named label inserted at index (end of the trace by default).

Source code in petri_net_nn/anomalies.py

def insert_event(
    trace: XESTrace, label: str, *, index: int | None = None
) -> XESTrace:
    """Return a copy of ``trace`` with a new event named ``label``
    inserted at ``index`` (end of the trace by default)."""
    new = _copy_trace(trace)
    pos = len(new.events) if index is None else index
    new.events.insert(pos, XESEvent(name=label))
    return new

swap_event_labels

swap_event_labels(trace, label_a, label_b)

Return a copy of trace with every occurrence of label_a replaced by label_b and vice versa. Models BPMN branch flipping: the wrong branch's task name appears in the trace.

Source code in petri_net_nn/anomalies.py

def swap_event_labels(
    trace: XESTrace, label_a: str, label_b: str
) -> XESTrace:
    """Return a copy of ``trace`` with every occurrence of ``label_a``
    replaced by ``label_b`` and vice versa. Models BPMN branch
    flipping: the *wrong* branch's task name appears in the trace."""
    swap = {label_a: label_b, label_b: label_a}
    new = _copy_trace(trace)
    for e in new.events:
        e.name = swap.get(e.name, e.name)
    return new

shuffle_events

shuffle_events(trace, *, seed=None)

Return a copy of trace with its events reordered randomly. Useful for processes where the order matters (sequential / AND-join patterns); for single-event traces it is a no-op.

Source code in petri_net_nn/anomalies.py

def shuffle_events(trace: XESTrace, *, seed: int | None = None) -> XESTrace:
    """Return a copy of ``trace`` with its events reordered randomly.
    Useful for processes where the order matters (sequential / AND-join
    patterns); for single-event traces it is a no-op."""
    new = _copy_trace(trace)
    if len(new.events) <= 1:
        return new
    rng = random.Random(seed)
    rng.shuffle(new.events)
    return new