State Machine Implementation

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, PeggedOrder.

The execution model

The entire machine runs on a single beam.RoutineTaskQueue. 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

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)

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 @param lines copied verbatim from the specification's Parameters section. Methods common to all orders (cancel, wait) are not documented.
• The constructor is responsible for starting the state machine in state S0 by pushing self._s0 onto the task queue.
• Every field and internal method takes a _ prefix. Mutable parameters are stored as copies (nexus.OrderFields(order_fields)) so the caller can't mutate a running machine.
• State machines for order types should have an orders 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:

  @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

cancel and wait complete the remainder of the public API:

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

  def wait(self):
    self._tasks.wait()

cancel 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 Sn becomes method _sn. 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 if/elif returning the call to its target state, marked with a comment naming the condition — the only comments in the file are these specification markers.

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

      # C0
      return self._s6()

    # epsilon
    return self._s2()

The bare return self._s2() 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:

  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)

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

Terminal states close everything the machine owns as follows:

  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()

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 wait.

Transcribing events

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

  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()

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 is_terminal 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:

  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()

The global cancel event's handler is implemented similarly:

  def _on_cancel(self):
    if self._state == 0:

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

      # E5
      return self._s7()