Energy Settlement Reconciliation Automation Hub

Event sourcing for ETRM trade capture

An analyst asks why a trade booked at 40 MWh is now settling at 35, and the mutable trade table has no answer — the UPDATE that changed it overwrote the old value and kept no record of who amended it, when, or why. That erased history is the failure mode event sourcing removes. Instead of storing the current state of a trade and mutating it in place, you store the ordered sequence of events that produced that state — captured, amended, confirmed, cancelled — and derive the current view by replaying them. This guide sits under ETRM System Architecture and covers the capture side specifically: how to model trade events, how to fold them back into a current position, and how snapshots keep replay cheap. The lifecycle rules that decide which transitions are even legal are the subject of Trade Lifecycle State Management; here we focus on persisting the transitions as an immutable log.

The diagram shows the shape of the pattern: commands append events to a per-trade stream, and a reducer folds that stream — optionally starting from a snapshot — into the current trade state.

Event-sourced trade capture and state reconstruction A command flows into an append step that writes an immutable event to a per-trade append-only event log. To read current state, a reducer folds the ordered events into a materialized trade state. A snapshot store sits between the log and the reducer: the reducer loads the latest snapshot and folds only the events after it, keeping replay fast. Commandbook / amend Append eventvalidate first Event logappend-only stream Reducerfold events Trade statecurrent view Snapshot storestate at seq N seed fold from snapshot

The core idea is a left fold. Current state is a pure function of an initial value and the ordered events: \(S_n = \text{fold}(S_0, [e_1, e_2, \dots, e_n])\), where each event applies a small deterministic change and \(S_0\) is the empty trade. Because the log is the source of truth and the state is derived, the state can always be thrown away and rebuilt.

Prerequisites

  • Python packages: the standard library carries the essentials — dataclasses, enum, decimal, hashlib, and json. sqlalchemy (2.x) or plain psycopg persists the event and snapshot tables; any store that supports an append-only insert works.
  • An append-only event store: a table with a per-trade stream_id, a monotonic sequence per stream, an immutable payload, and a UNIQUE (stream_id, sequence) constraint that enforces ordering and blocks a second writer from claiming the same slot.
  • Data dependencies: each command must carry the trade’s stream_id (usually trade_id), the actor and reason for audit, and the economic fields an event changes (node_id, settlement_interval, mwh, price).
  • Permissions: append scope on the event store and read scope on snapshots. No component ever needs UPDATE or DELETE on the event table — revoking those rights is what makes the log tamper-evident at the database level.

Implementation

Model each event as an immutable record. The event type names what happened in the past tense, and the payload carries only the fields that event changed. Money and volume are Decimal from the moment they enter the log so no float ever touches a persisted amount.

from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
from enum import Enum

class TradeEvent(str, Enum):
    CAPTURED  = "TradeCaptured"
    AMENDED   = "TradeAmended"
    CONFIRMED = "TradeConfirmed"
    CANCELLED = "TradeCancelled"

@dataclass(frozen=True)                # frozen: an appended event is immutable
class Event:
    stream_id: str                     # trade_id this event belongs to
    sequence: int                      # 1-based position within the stream
    event_type: TradeEvent
    payload: dict                      # only the fields this event sets/changes
    actor: str                         # who caused it (audit)
    reason: str                        # why (audit)
    occurred_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

Appending is a validated insert. The reducer is where state lives, so appending only needs to assign the next sequence and reject anything that would corrupt ordering. The UNIQUE (stream_id, sequence) constraint is the real guard: if two workers race to append sequence 5, the database rejects the loser.

def append_event(conn, stream_id: str, event_type: TradeEvent,
                 payload: dict, actor: str, reason: str) -> Event:
    """Append one immutable event to a trade's stream. Never updates in place."""
    # Next sequence = current max for this stream + 1.
    row = conn.execute(
        "SELECT COALESCE(MAX(sequence), 0) FROM trade_events WHERE stream_id = %s",
        (stream_id,),
    ).fetchone()
    next_seq = int(row[0]) + 1
    evt = Event(stream_id, next_seq, event_type, payload, actor, reason)
    conn.execute(
        "INSERT INTO trade_events "
        "(stream_id, sequence, event_type, payload, actor, reason, occurred_at) "
        "VALUES (%s, %s, %s, %s, %s, %s, %s)",       # UNIQUE(stream_id, sequence)
        (evt.stream_id, evt.sequence, evt.event_type.value,
         _json(evt.payload), evt.actor, evt.reason, evt.occurred_at),
    )
    return evt

Rebuilding current state is a fold over the ordered events. The reducer is a pure function: given a state and one event, it returns the next state and never performs I/O. Purity is what makes replay reproducible — the same events always fold to the same state.

@dataclass(frozen=True)
class TradeState:
    trade_id: str
    node_id: str | None = None
    settlement_interval: str | None = None
    mwh: Decimal = Decimal("0")
    price: Decimal = Decimal("0")        # $/MWh
    status: str = "NEW"
    version: int = 0                     # advances with every applied event

def apply_event(state: TradeState, evt: Event) -> TradeState:
    """Pure reducer: fold one event into the running state."""
    p = evt.payload
    base = {**state.__dict__, "version": evt.sequence}
    if evt.event_type is TradeEvent.CAPTURED:
        return TradeState(trade_id=evt.stream_id,
                          node_id=p["node_id"],
                          settlement_interval=p["settlement_interval"],
                          mwh=Decimal(str(p["mwh"])),
                          price=Decimal(str(p["price"])),
                          status="CAPTURED", version=evt.sequence)
    if evt.event_type is TradeEvent.AMENDED:
        # Only the fields present in the amendment payload change.
        if "mwh" in p:   base["mwh"] = Decimal(str(p["mwh"]))
        if "price" in p: base["price"] = Decimal(str(p["price"]))
        base["status"] = "AMENDED"
        return TradeState(**base)
    if evt.event_type is TradeEvent.CONFIRMED:
        base["status"] = "CONFIRMED"
        return TradeState(**base)
    if evt.event_type is TradeEvent.CANCELLED:
        base["status"] = "CANCELLED"
        return TradeState(**base)
    raise ValueError(f"unknown event type: {evt.event_type}")

def rebuild(events: list[Event], seed: TradeState | None = None) -> TradeState:
    """Left fold: current state = fold(seed, ordered events)."""
    state = seed or TradeState(trade_id=events[0].stream_id)
    for evt in sorted(events, key=lambda e: e.sequence):
        state = apply_event(state, evt)
    return state

Folding from sequence 1 every time is fine for a trade with a handful of amendments, but a stream that accumulates thousands of events — a heavily re-priced position over a long delivery term — makes reconstruction slow. Snapshots solve this: periodically persist the folded state at a known sequence, then on the next read load the snapshot and fold only the events after it. The snapshot is a cache derived from the log, never a substitute for it, so it can always be discarded and recomputed.

def load_current(conn, stream_id: str, snapshot_every: int = 100) -> TradeState:
    """Load the newest snapshot, then fold only the tail of newer events."""
    snap = conn.execute(
        "SELECT sequence, state FROM trade_snapshots "
        "WHERE stream_id = %s ORDER BY sequence DESC LIMIT 1",
        (stream_id,),
    ).fetchone()

    seed, from_seq = (None, 0)
    if snap:
        seed = _state_from_json(snap[1])          # materialized TradeState
        from_seq = int(snap[0])

    tail = _load_events(conn, stream_id, after_sequence=from_seq)
    state = rebuild(tail, seed=seed) if tail else seed

    # Write a fresh snapshot once enough new events have accrued.
    if state and (state.version - from_seq) >= snapshot_every:
        conn.execute(
            "INSERT INTO trade_snapshots (stream_id, sequence, state) "
            "VALUES (%s, %s, %s) ON CONFLICT (stream_id, sequence) DO NOTHING",
            (stream_id, state.version, _state_to_json(state)),
        )
    return state

A subtlety worth stating: correcting a trade never means editing or deleting a past event. To reverse a bad amendment you append a compensating event — another TradeAmended that restores the prior value, or a TradeCancelled — so the log grows forward-only and the mistake plus its correction both remain visible to an auditor.

Verification

Confirm the log and the derived state agree before relying on them:

  1. Replay determinism. Fold a stream’s events twice; both rebuild calls must yield an identical TradeState. Any difference means a reducer branch performed hidden I/O or depended on wall-clock time.
  2. Snapshot equivalence. Assert that rebuild(all_events) equals load_current(...) for the same stream — the snapshot-accelerated path must never diverge from a full replay.
  3. Sequence integrity. For each stream_id, the sequences must form a gap-free 1..N. A missing number signals a lost append; a duplicate signals the UNIQUE constraint was not enforced.
  4. Audit reconstruction. Given any past sequence, folding events 1..k must reproduce the exact state as of that point, so a query like “what were the economics before the amendment on sequence 7” is answerable.
def test_replay_is_deterministic():
    events = _sample_stream(trade_id="T1")     # captured, amended, confirmed
    assert rebuild(events) == rebuild(events)

def test_snapshot_matches_full_replay(conn):
    stream = "T1"
    full = rebuild(_load_events(conn, stream, after_sequence=0))
    fast = load_current(conn, stream)
    assert full == fast

Compliance Note

An event-sourced trade log directly serves the audit-trail obligations of energy-market settlement. FERC’s Open Access framework and the RTO settlement manuals (PJM Manual 28, ERCOT Nodal Protocols Section 9, the CAISO Business Practice Manual for Settlements) require that every settled quantity be traceable to its origin and to each amendment along the way — precisely what an immutable, sequenced log preserves and a mutable table destroys. Storing the actor and reason on each event, revoking UPDATE/DELETE on the event table, and correcting only through compensating events aligns with the tamper-evidence expectations of the NERC Critical Infrastructure Protection standards for logging. Validate retention against your market’s dispute and resettlement windows so the full event history for a delivery month survives at least through its final true-up, and treat snapshots as disposable derived data that need no independent retention guarantee.

Frequently Asked Questions

How is event sourcing different from just keeping an audit log alongside a trades table?

An audit log is a side record of changes; the mutable trades table is still the authority, and the two can drift. In event sourcing the log is the authority and the current-state view is derived from it, so they cannot disagree by construction. You gain the ability to reconstruct the trade as of any past point, to correct by appending rather than overwriting, and to rebuild the entire read model from scratch if its schema changes — none of which a bolt-on audit table gives you.

When do I need snapshots, and can they ever be wrong?

You need snapshots once folding a stream from event one becomes too slow for your read latency — typically a position that has accumulated hundreds or thousands of events over a long delivery term. A snapshot is only a cache of the fold at a given sequence, so if the reducer logic changes, the snapshot can encode stale logic; guard against that by versioning the snapshot with the reducer version and discarding snapshots whose version no longer matches, forcing a clean replay from the log.

How do I correct a bad trade event without breaking the immutable log?

Never edit or delete the offending event. Append a compensating event that expresses the correction — an amendment that sets the field back to its correct value, or a cancellation if the trade should not exist. The reducer folds the original and the correction in order, so the current state is right while the full history, including the mistake, stays visible for audit. This is why the event table carries no UPDATE or DELETE grant in production.