state_machine

sqllocks_spindle.simulation.state_machine

Business workflow events — state machines with dwell times.

Generate realistic event streams for business processes where entities transition through defined states (e.g. order: created->confirmed->shipped-> delivered->returned) with configurable dwell times between transitions.

Usage::

from sqllocks_spindle.simulation.state_machine import (
    WorkflowSimulator, WorkflowConfig, StateDefinition, TransitionRule,
)

states = [
    StateDefinition("created", is_initial=True),
    StateDefinition("confirmed"),
    StateDefinition("shipped"),
    StateDefinition("delivered", is_terminal=True),
    StateDefinition("cancelled", is_terminal=True),
    StateDefinition("returned", is_terminal=True),
]
transitions = [
    TransitionRule("created", "confirmed", probability=0.90, dwell_hours_mean=2.0, dwell_hours_std=0.5),
    TransitionRule("created", "cancelled", probability=0.10, dwell_hours_mean=1.0, dwell_hours_std=0.3),
    TransitionRule("confirmed", "shipped", probability=0.95, dwell_hours_mean=24.0, dwell_hours_std=6.0),
    TransitionRule("confirmed", "cancelled", probability=0.05, dwell_hours_mean=4.0, dwell_hours_std=1.0),
    TransitionRule("shipped", "delivered", probability=0.92, dwell_hours_mean=72.0, dwell_hours_std=24.0),
    TransitionRule("shipped", "returned", probability=0.08, dwell_hours_mean=120.0, dwell_hours_std=36.0),
]
cfg = WorkflowConfig(states=states, transitions=transitions, entity_count=1000)
result = WorkflowSimulator(config=cfg).run()

Classes

StateDefinition dataclass

A single state in a business workflow.

Parameters:

Name	Type	Description	Default
name	str	Unique name for this state.	required
is_initial	bool	Whether entities can start in this state.	False
is_terminal	bool	Whether this state ends the workflow (no outgoing transitions).	False
metadata	dict[str, Any]	Arbitrary key-value pairs attached to the state.	dict()

TransitionRule dataclass

A directed edge in the workflow state machine.

Parameters:

Name	Type	Description	Default
from_state	str	Source state name.	required
to_state	str	Destination state name.	required
probability	float	Relative weight for this transition (normalised per source state).	1.0
dwell_hours_mean	float	Mean hours spent in from_state before this transition fires.	1.0
dwell_hours_std	float	Standard deviation for the dwell-time normal distribution.	0.5
min_dwell_hours	float	Hard floor on sampled dwell time.	0.1

WorkflowConfig dataclass

Configuration for :class:WorkflowSimulator.

Parameters:

Name	Type	Description	Default
states	list[StateDefinition]	List of state definitions for the workflow.	list()
transitions	list[TransitionRule]	List of transition rules connecting states.	list()
entity_count	int	Number of entities to simulate.	100
entity_prefix	str	Prefix for generated entity IDs.	'entity'
start_time	str	ISO-format start timestamp for the simulation.	'2024-01-01T00:00:00'
max_transitions_per_entity	int	Safety limit to prevent infinite loops.	20
anomaly_enabled	bool	Whether to inject anomalous transitions.	True
anomaly_skip_probability	float	Chance of skipping a state in the workflow.	0.02
anomaly_stuck_probability	float	Chance of an entity getting stuck (no further transitions).	0.01
anomaly_backward_probability	float	Chance of transitioning backward to a previous state.	0.01
seed	int	Random seed for reproducibility.	42

WorkflowResult dataclass

Result of :meth:WorkflowSimulator.run.

Attributes:

Name	Type	Description
events	DataFrame	All transition events sorted by (entity_id, transitioned_at).
entity_summary	DataFrame	Per-entity summary with final state and timing.
state_distribution	dict[str, int]	Count of entities in each final state.
stats	dict[str, Any]	Aggregate statistics dictionary.

WorkflowSimulator

Simulate business-process event streams using a state machine.

Entities are created in an initial state and transition through the workflow according to probabilistic :class:TransitionRule definitions until they reach a terminal state, get stuck (anomaly), or hit the per-entity transition safety limit.

Parameters:

Name	Type	Description	Default
config	WorkflowConfig	:class:WorkflowConfig describing the workflow graph and simulation parameters.	required

Example::

cfg = WorkflowConfig(states=states, transitions=transitions, entity_count=500)
sim = WorkflowSimulator(config=cfg)
result = sim.run()
result.events.head()

Methods:

run()

Execute the full simulation and return a :class:WorkflowResult.

Functions:

get_preset_workflow(name)

Return (states, transitions) for a named preset workflow.

Available presets: order_fulfillment, support_ticket, employee_onboarding.

Raises:

Type	Description
KeyError	If name is not a recognised preset.