Modeling trade state transitions with a finite state machine

A settlement analyst re-runs yesterday’s ingestion batch to recover from a crash, and every trade that was already confirmed receives its confirmation event a second time — if the transition logic is a pile of if statements over status columns, half of them silently regress or double-count, and the day’s match run is now wrong. This page builds the finite state machine that makes that replay a no-op: a single declared transition table, guard predicates that hold the business rules, an idempotent apply keyed by event id, and a hash-chained audit trail. It is the concrete implementation behind Trade Lifecycle State Management, which frames why the lifecycle is modeled this way and how it fits the surrounding ingestion pipeline.

The diagram below traces a single event through the machine — from arrival, past the idempotency check and the guard, to either a committed transition with an appended audit record or a rejection.

Prerequisites #

• Python packages: standard library only — enum, dataclasses, hashlib, json, and decimal for the money comparison in the match guard. No third-party dependency is required for the core machine; a real deployment adds a persistence driver (SQLAlchemy, psycopg, or a Parquet writer) behind the audit-append call.
• Data dependencies: each incoming event supplies a trade_id, a deterministic event_id (derived from the source message, not generated fresh on retry), an event name, and a payload carrying whatever the guard needs — for example the counterparty notional for a match, or the open settlement run id for a settle. Trades must already be schema-valid before they reach the machine.
• Permissions / access: append-only write access to the transition-log store and read access to the current trade state. No delete or in-place update permission should be granted to the process — the model depends on history being immutable.

Implementation #

The machine is one small module. The transition table is inert data; guards are pure predicates; apply is the only function that mutates state, and it is idempotent by construction. Money in the match guard uses Decimal so a one-cent notional difference is compared exactly, never through binary float.

from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
from enum import Enum
from typing import Callable


class TradeState(str, Enum):
    CAPTURED = "captured"
    CONFIRMED = "confirmed"
    MATCHED = "matched"
    SETTLED = "settled"
    RESETTLED = "resettled"
    CANCELLED = "cancelled"


class TradeEvent(str, Enum):
    CONFIRM = "confirm"
    MATCH = "match"
    SETTLE = "settle"
    RESETTLE = "resettle"
    CANCEL = "cancel"


# The ONLY legal edges. (state, event) -> next_state; absence means "rejected".
TRANSITIONS: dict[tuple[TradeState, TradeEvent], TradeState] = {
    (TradeState.CAPTURED,  TradeEvent.CONFIRM):  TradeState.CONFIRMED,
    (TradeState.CAPTURED,  TradeEvent.CANCEL):   TradeState.CANCELLED,
    (TradeState.CONFIRMED, TradeEvent.MATCH):    TradeState.MATCHED,
    (TradeState.CONFIRMED, TradeEvent.CANCEL):   TradeState.CANCELLED,
    (TradeState.MATCHED,   TradeEvent.SETTLE):   TradeState.SETTLED,
    (TradeState.SETTLED,   TradeEvent.RESETTLE): TradeState.RESETTLED,
    (TradeState.RESETTLED, TradeEvent.RESETTLE): TradeState.RESETTLED,
}


class TransitionError(Exception):
    """Raised when an event is not a legal edge or its guard fails."""


# --- Guards: pure predicates over (trade, payload). No I/O. ---
def guard_match(trade: "Trade", payload: dict) -> None:
    """A confirmed trade may match only if economic terms agree to the cent."""
    ours = Decimal(str(trade.notional_usd))
    theirs = Decimal(str(payload["counterparty_notional_usd"]))
    if abs(ours - theirs) > Decimal("0.01"):
        raise TransitionError(f"notional mismatch: {ours} vs {theirs}")
    if Decimal(str(trade.volume_mwh)) == Decimal("0"):
        raise TransitionError("zero-volume leg cannot match")


def guard_settle(trade: "Trade", payload: dict) -> None:
    """A matched trade may settle only into an open settlement run."""
    if not payload.get("settlement_run_open", False):
        raise TransitionError("settlement run is not open for this trading day")


def guard_resettle(trade: "Trade", payload: dict) -> None:
    """A true-up is legal only inside the tariff dispute window."""
    if not payload.get("within_dispute_window", False):
        raise TransitionError("resettlement outside tariff dispute window")


GUARDS: dict[TradeEvent, Callable[["Trade", dict], None]] = {
    TradeEvent.MATCH: guard_match,
    TradeEvent.SETTLE: guard_settle,
    TradeEvent.RESETTLE: guard_resettle,
    # CONFIRM and CANCEL carry no economic precondition here.
}


@dataclass
class TransitionRecord:
    trade_id: str
    event_id: str
    event: str
    from_state: str
    to_state: str
    occurred_at: str
    prev_hash: str
    row_hash: str = ""

    def compute_hash(self) -> str:
        """SHA-256 over the record's fields plus the previous row's hash."""
        payload = {
            "trade_id": self.trade_id,
            "event_id": self.event_id,
            "event": self.event,
            "from_state": self.from_state,
            "to_state": self.to_state,
            "occurred_at": self.occurred_at,
            "prev_hash": self.prev_hash,
        }
        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(canonical.encode()).hexdigest()


@dataclass
class Trade:
    trade_id: str
    notional_usd: str          # kept as string; cast to Decimal in guards
    volume_mwh: str
    state: TradeState = TradeState.CAPTURED
    version: int = 0           # optimistic-concurrency counter
    seen_event_ids: set[str] = field(default_factory=set)
    audit_log: list[TransitionRecord] = field(default_factory=list)

    def apply(self, event: TradeEvent, event_id: str, payload: dict) -> TradeState:
        """Idempotently apply one event. Duplicate event_id -> no-op."""
        # 1. Idempotency: a redelivered event changes nothing.
        if event_id in self.seen_event_ids:
            return self.state

        # 2. Legality: the edge must exist in the declared table.
        key = (self.state, event)
        if key not in TRANSITIONS:
            raise TransitionError(
                f"illegal transition: {self.state.value} -x-> {event.value}"
            )
        next_state = TRANSITIONS[key]

        # 3. Guard: the business precondition must hold.
        guard = GUARDS.get(event)
        if guard is not None:
            guard(self, payload)   # raises TransitionError on failure

        # 4. Commit + append a hash-linked audit record.
        prev_hash = self.audit_log[-1].row_hash if self.audit_log else "genesis"
        record = TransitionRecord(
            trade_id=self.trade_id,
            event_id=event_id,
            event=event.value,
            from_state=self.state.value,
            to_state=next_state.value,
            occurred_at=datetime.now(timezone.utc).isoformat(),
            prev_hash=prev_hash,
        )
        record.row_hash = record.compute_hash()

        self.state = next_state
        self.version += 1
        self.seen_event_ids.add(event_id)
        self.audit_log.append(record)
        return self.state

The four numbered steps inside apply are the whole contract: idempotency first so a replay short-circuits before any side effect, then legality against the table, then the guard, and only then a commit that advances state and extends the hash chain. Because the guard runs after the legality check, a guard never has to defend against an impossible starting state — it only encodes the economic rule for an edge that already exists.

A short driver shows the happy path plus a duplicate and an illegal event:

trade = Trade(trade_id="PWR-2026-03-08-0042", notional_usd="18450.00", volume_mwh="25")

trade.apply(TradeEvent.CONFIRM, event_id="evt-conf-1", payload={})
trade.apply(TradeEvent.MATCH, event_id="evt-match-1",
            payload={"counterparty_notional_usd": "18450.00"})
trade.apply(TradeEvent.SETTLE, event_id="evt-settle-1",
            payload={"settlement_run_open": True})

# Redelivered confirmation — a no-op, state and log unchanged.
before = (trade.state, len(trade.audit_log))
trade.apply(TradeEvent.CONFIRM, event_id="evt-conf-1", payload={})
assert (trade.state, len(trade.audit_log)) == before

# Illegal: a settled trade cannot be cancelled.
try:
    trade.apply(TradeEvent.CANCEL, event_id="evt-cancel-1", payload={})
except TransitionError as exc:
    print("rejected as expected:", exc)

Verification #

Confirm the machine behaves before wiring it to a real event stream:

1. Terminal state. After confirm → match → settle, trade.state is TradeState.SETTLED and len(trade.audit_log) == 3. The driver above asserts this directly.
2. Idempotency. Re-applying an event with a seen event_id leaves both trade.state and len(trade.audit_log) unchanged. This is the proof that batch replay is safe.
3. Illegal edges rejected. Any (state, event) pair absent from TRANSITIONS raises TransitionError; a settled trade rejecting cancel is the canonical case.
4. Hash-chain integrity. Recompute the chain and assert each row’s prev_hash equals its predecessor’s row_hash, and that recomputing compute_hash() reproduces the stored digest — proof that no record was altered, inserted, or dropped.

def verify_audit_chain(audit_log: list[TransitionRecord]) -> None:
    """Assert the transition log is an unbroken, untampered hash chain."""
    expected_prev = "genesis"
    for record in audit_log:
        assert record.prev_hash == expected_prev, (
            f"chain break at {record.event_id}: "
            f"{record.prev_hash} != {expected_prev}"
        )
        assert record.compute_hash() == record.row_hash, (
            f"tampered record: {record.event_id}"
        )
        expected_prev = record.row_hash

Running verify_audit_chain(trade.audit_log) on the driver’s trade returns without error; mutating any to_state after the fact and re-running raises immediately, because every downstream digest depends on the altered field.

Compliance Note #

This state model is the recordkeeping substrate FERC-jurisdictional sellers rely on to reconstruct a transaction as it stood on any date. FERC’s recordkeeping rules and the Electric Quarterly Report expect a complete, defensible history of each deal, not merely its final disposition — the append-only transition log with a SHA-256 hash chain satisfies that by making every state change retained and tamper-evident. The captured → confirmed edge is the system’s representation of the confirmation handshake defined by NAESB Wholesale Electric Quadrant business practices, and the guard on that path should be validated against the confirmation-timing windows in the applicable business practice manual. Whether a settle or a resettle edge is legal for a given trading day is governed by the market operator’s settlement calendar, so the settlement_run_open and within_dispute_window flags in the guard payloads must be sourced from the authoritative calendar rather than inferred, and each transition’s occurred_at timestamp must be recorded in UTC to survive audit across daylight-saving boundaries.

Frequently Asked Questions #

Why derive the event_id from the source message instead of generating it? #

Idempotency only works if a retried delivery of the same logical event carries the same id. If the id were generated fresh each time the event is produced, a redelivered confirmation would look like a brand-new event and be applied twice. Deriving event_id deterministically from the source message — for example a hash of the counterparty message id and event type — guarantees the machine recognizes the duplicate and short-circuits to a no-op.

How do I add a new state or transition without breaking existing trades? #

Add the state to the TradeState enum and the new edge to the TRANSITIONS table; if the edge has a business precondition, add a guard and register it in GUARDS. Because the table is the single source of truth and existing edges are untouched, trades already in flight keep their legal paths. Run the exhaustive edge test afterward so any accidental extra edge is caught before deployment.

Should the audit log live in memory like this example? #

No — the in-memory list and set here keep the example self-contained. In production, seen_event_ids and audit_log are backed by a durable, append-only store, and the version counter drives optimistic-concurrency control so two workers cannot clobber the same trade. The apply logic is unchanged; only the read of current state and the append of the transition record cross the persistence boundary.

Related #

• Trade Lifecycle State Management
• Implementing Pydantic for energy trade validation
• Automating ETRM sync with Python requests