Kamal: Created page with "Once a state machine specification has been designed, its Python implementation is a mostly mechanical transcription. Every state, event, and condition in the code points back..."

2026-07-13T18:34:35Z

Created page with "Once a state machine specification has been designed, its Python implementation is a mostly mechanical transcription. Every state, event, and condition in the code points back..."

New page

Once a state machine specification has been designed, its Python implementation
is a mostly mechanical transcription. Every state, event, and condition in the
code points back to a counterpart in the specification, and nothing appears in
the code that the specification doesn't call for.

This article defines the transcription. It assumes familiarity with
[[State Machine Specifications]] and continues its running example,
<code>PeggedOrder</code>.

== The execution model ==

The entire machine runs on a single <code>beam.RoutineTaskQueue</code>.
Every event such as market data, execution reports, timer expiries, the order's
own cancel request is delivered onto that queue and processed one task at a
time, in order of arrival.

== The class ==

<pre>
import beam
import nexus

class PeggedOrder:
"""Trades an order passively at a price pegged to the touch, honoring the
order's limit price. The order's type is superseded: every submission is a
limit order.
"""

def __init__(self, clients, order_fields, peg_difference):
"""Constructs a PeggedOrder.
@param clients The clients used to access Nexus services.
@param order_fields The specification of the order to execute.
@param peg_difference The difference applied to the pegged price.
"""
self._clients = clients
self._order_fields = nexus.OrderFields(order_fields)
self._peg_difference = peg_difference
self._state = None
self._failure = None
self._orders = beam.QueueWriterPublisher()
self._tasks = beam.RoutineTaskQueue()
self._tasks.push(self._s0)
</pre>

Conventions, in order of appearance:

* Import only what the file uses.
* The class docstring states the goal, never the mechanics or implementation details; the constructor docstring documents each parameter with <code>@param</code> lines copied verbatim from the specification's Parameters section. Methods common to all orders (<code>cancel</code>, <code>wait</code>) are not documented.
* The constructor is responsible for starting the state machine in state S0 by pushing <code>self._s0</code> onto the task queue.
* Every field and internal method takes a <code>_</code> prefix. Mutable parameters are stored as copies (<code>nexus.OrderFields(order_fields)</code>) so the caller can't mutate a running machine.
* State machines for order types should have an <code>orders</code> publisher created in the constructor that is used to publish any submitted orders so they can be traced back to the state machine.

Constructor parameters are exposed as read-only properties, returning copies
of mutable types to avoid inadvertant mutation:

<pre>
@property
def order_fields(self):
"""Returns a copy of the specification of the order being executed."""
return nexus.OrderFields(self._order_fields)

@property
def peg_difference(self):
"""Returns the difference applied to the pegged price."""
return self._peg_difference
</pre>

<code>cancel</code> and <code>wait</code> complete the remainder of the public API:

<pre>
def cancel(self):
try:
self._tasks.push(self._on_cancel)
except beam.PipeBrokenException:
pass

def wait(self):
self._tasks.wait()
</pre>

<code>cancel</code> delivers the global cancel event through the same serialized
task queue as everything else, so it can never race with another event.

== Transcribing states ==

State <code>Sn</code> becomes method <code>_sn</code>. Its first line
records the state and its remaining lines are the specification's action,
followed by the state's conditions in label order. Each condition is an
<code>if</code>/<code>elif</code> returning the call to its target state,
marked with a comment naming the condition — the only comments in the file
are these specification markers.

<pre>
def _s1(self):
self._state = 1
if self._filled_quantity == self._order_fields.quantity:

# C0
return self._s6()

# epsilon
return self._s2()
</pre>

The bare <code>return self._s2()</code> is S1's ε transition: after
the conditions fail, the transition fires unconditionally. A state's action
transcribes nearly verbatim because the specification is already written in
real API syntax:

<pre>
def _s2(self):
self._state = 2
self._submission_price = self._peg_price
fields = nexus.OrderFields(self._order_fields)
fields.type = nexus.OrderType.LIMIT
fields.price = self._submission_price
fields.quantity = self._order_fields.quantity - self._filled_quantity
self._order = self._clients.order_execution_client.submit(fields)
self._order.publisher.monitor(
self._tasks.get_slot(self._on_execution_report))
self._orders.push(self._order)
</pre>

Two lines here are implicit from the state machine specification, the
<code>monitor</code> call is the method that publishes the order's events
onto the task queue, and <code>self._orders.push</code> is how the PeggedOrder
publishes orders to external observers.

Terminal states close everything the machine owns as follows:

<pre>
def _s3(self):
self._state = 3
self._orders.close(self._failure)
self._tasks.close()

def _s6(self):
self._state = 6
self._orders.close()
self._tasks.close()
</pre>

The red terminal state closes the orders publisher '''with the failure''', so
observers receive the original exception and the blue terminal closes it
normally. Closing the task queue is what unblocks <code>wait</code>.

== Transcribing events ==

Each event source gets a handler wired through
<code>self._tasks.get_slot(...)</code>. The handler's top transcribes the
event bodies that always execute — the base event's body first — and the
rest dispatches on <code>self._state</code>, checking each refinement's
guard in label order, with a comment marking each transition's event:

<pre>
def _on_execution_report(self, report):

# E1
self._filled_quantity += report.last_quantity
if self._state == 2:
if report.status == nexus.OrderStatus.REJECTED:

# E2
self._failure = RuntimeError(report.text)
return self._s3()
elif report.status == nexus.OrderStatus.NEW:

# E3
return self._s4()
elif self._state == 4:
if nexus.is_terminal(report.status):

# E4
return self._s1()
</pre>

The structure mirrors the specification exactly: E1's accounting runs on every
report because E1 happened, regardless of state; the state dispatch then
performs at most one transition, and checking REJECTED before
<code>is_terminal</code> preserves the label order that resolves their
overlap. States with no edge for an event simply have no branch — the body
still ran, the machine stays put.

A derived variable computed in an event body is recomputed in the handler,
before the dispatch:

<pre>
def _on_bbo_quote(self, quote):

# E0
self._bbo = quote
self._peg_price = nexus.pick(self._order_fields.side,
self._bbo.ask.price, self._bbo.bid.price) - \
nexus.direction(self._order_fields.side) * self._peg_difference
if self._order_fields.price != nexus.Money.ZERO:
self._peg_price = nexus.pick(self._order_fields.side,
max(self._peg_price, self._order_fields.price),
min(self._peg_price, self._order_fields.price))
if self._state == 0:

# E0
return self._s1()
elif self._state == 4:

# E0
return self._s4()
</pre>

The global cancel event's handler is implemented similarly:

<pre>
def _on_cancel(self):
if self._state == 0:

# E5
return self._s6()
elif self._state == 2:

# E5
return self._s7()
</pre>

State Machine Implementation - Revision history

Kamal: Created page with "Once a state machine specification has been designed, its Python implementation is a mostly mechanical transcription. Every state, event, and condition in the code points back..."