Stateful Stream Processing: Edge vs Level Triggers and OHLC Windows

Somewhere between "consume a topic" and "run a real application" sits a question that trips up a lot of first attempts at stream processing: does this consumer need to remember anything? For a price alert, the honest answer is yes, and getting that wrong is the difference between a useful notification and spam.

When I set out to build the same stock ticker twice, once on Kafka and once on Redis Streams, almost everything I learned in the first few weeks was about transport: moving events safely through partitions, consumer groups, delivery guarantees, and replication. I kept putting off the processing side because I had filed it under "the easy part," and it was not.

Transport Versus Processing

A consumer that reads a tick and prints it is doing transport. A consumer that reads a tick, compares it against something it remembers from the last tick, and decides whether to act, is doing stream processing. The defining ingredient is state. Without it, a consumer can only react to the record in front of it. With it, a consumer can react to a transition, which is a far more useful thing to alert on.

The annoying part, if you live in Node like I do: the official Kafka Streams library is JVM-only, and TypeScript has no equivalent worth reaching for. So I hand-rolled the two patterns my ticker needed, a stateful evaluator for alerts and a windowed aggregator for candles. Less convenient than importing a state store, but closer to what you are signing up for when you point a Node consumer at either broker.

The Alert That Fires Once: Edge vs Level

Take a rule like "notify Alice when BBCA rises to or above 9500." You can read that two ways:

• Level: fire on every tick where the price is >= 9500. Simple, stateless, and it will spam Alice with a notification on every single tick for as long as the price stays up there.
• Edge: fire only on the tick where the price crosses from below 9500 to at or above it. One notification per crossing, which is what a user wants from "notify me when."

The level test only needs the current tick:

export function matches(op: Operator, price: number, threshold: number): boolean {
  switch (op) {
    case "<": return price < threshold;
    case "<=": return price <= threshold;
    case "=": return price === threshold;
    case ">=": return price >= threshold;
    case ">": return price > threshold;
  }
}

The edge test needs one more thing: the previous price.

export function crossed(
  op: Operator,
  prevPrice: number,
  price: number,
  threshold: number,
): boolean {
  return !matches(op, prevPrice, threshold) && matches(op, price, threshold);
}

Read that literally: the condition was false a moment ago and is true now. That is a crossing. You cannot compute it from a single tick, because a single tick has no "a moment ago." Something has to remember the last price it saw for that symbol, and in my ticker that something is the AlertEvaluator.

export class AlertEvaluator {
  private readonly lastPrice = new Map<StockSymbol, number>();
  private readonly rulesBySymbol = new Map<StockSymbol, AlertRule[]>();

  process(tick: Tick): Notification[] {
    const prev = this.lastPrice.get(tick.symbol);
    this.lastPrice.set(tick.symbol, tick.price); // update state AFTER reading prev

    const rules = this.rulesBySymbol.get(tick.symbol);
    if (!rules) return [];

    const out: Notification[] = [];
    for (const rule of rules) {
      const fired =
        rule.trigger === "edge"
          ? prev !== undefined && crossed(rule.operator, prev, tick.price, rule.threshold)
          : matches(rule.operator, tick.price, rule.threshold);
      if (fired) out.push(this.toNotification(rule, tick));
    }
    return out;
  }
}

Two details worth pausing on. First, lastPrice is updated right after it is read, before the rules run, so the next tick always compares against what just happened, not against something two ticks old. Second, on the very first tick for a symbol, prev is undefined and an edge rule cannot fire, which is correct: you cannot have crossed a threshold if you have no history to cross from.

To see how much that ordering buys, I replayed the same 200 synthetic ticks twice through the same set of rules, once with every rule set to edge and once with the identical rules set to level. Edge produced 9 notifications. Level produced 194. Same thresholds, same prices, roughly 21 times the number of buzzes on a user's phone. If you have ever muted an app because it would not stop shouting about a stock you were watching, you have met a level trigger that somebody shipped by accident.

sequenceDiagram
    participant T as Tick Stream
    participant E as AlertEvaluator "state: lastPrice"
    T->>E: price=9480 (below 9500)
    Note over E: lastPrice=9480, no fire
    T->>E: price=9505 (crosses 9500)
    Note over E: EDGE fires (was below, now above)
    Note over E: LEVEL fires too
    T->>E: price=9510 (still above 9500)
    Note over E: EDGE stays silent (no new crossing)
    Note over E: LEVEL fires again

Why State Drags Co-Partitioning Back In

The lastPrice map lives inside one AlertEvaluator instance, in memory, per process. In a partitioned deployment you do not run one giant evaluator for all ten symbols; you run one per pod, and each pod owns a subset of partitions. The map only ever needs to hold state for the symbols whose partition that pod consumes.

That works cleanly only because of co-partitioning: market-updates and notifications are both keyed by symbol and provisioned with the same six partitions, so the pod that owns partition 2 for one topic owns partition 2 for the other. The pod holding BBCA's tick stream is the pod holding BBCA's alert rules. State stays local, correct, and un-shared. Without that alignment, the lastPrice map would either have to live in a shared store like Redis, or every rebalance would risk resetting it and losing the previous price right when a crossing was about to happen.

I had co-partitioning filed under "makes joins cheap." Stateful processing turned out to be the stronger argument for it: state that lives on the wrong pod is state you have to fetch, replicate, or lose.

Turning a Firehose Into a Chart: Tumbling Windows

Alerts consume ticks one at a time. Charts need something coarser: candles. Ask anyone who has looked at a stock chart and they know the shape, an open, a high, a low, and a close for some fixed slice of time. Building that from a raw tick stream is a windowing problem.

A tumbling window is a fixed-size, non-overlapping bucket of time, say every 1000 milliseconds. Every tick that arrives gets folded into whichever bucket its timestamp belongs to. When a tick's timestamp belongs to a bucket later than the one currently open, the open bucket is done. It gets closed, emitted, and a new bucket starts.

OhlcAggregator implements exactly that, per symbol:

export class OhlcAggregator {
  private readonly current = new Map<StockSymbol, Candle>();

  constructor(private readonly windowMs: number) {}

  add(tick: Tick): Candle | null {
    const windowStart = Math.floor(tick.ts / this.windowMs) * this.windowMs;
    const cur = this.current.get(tick.symbol);

    if (!cur || cur.windowStart !== windowStart) {
      const closed = cur && cur.windowStart !== windowStart ? cur : null;
      this.current.set(tick.symbol, {
        symbol: tick.symbol,
        windowStart,
        open: tick.price,
        high: tick.price,
        low: tick.price,
        close: tick.price,
        volume: tick.volume,
        count: 1,
      });
      return closed;
    }

    cur.high = Math.max(cur.high, tick.price);
    cur.low = Math.min(cur.low, tick.price);
    cur.close = tick.price;
    cur.volume += tick.volume;
    cur.count++;
    return null;
  }

  flush(): Candle[] {
    const out = [...this.current.values()];
    this.current.clear();
    return out;
  }
}

windowStart is computed by flooring the tick's timestamp down to the nearest window boundary, Math.floor(tick.ts / windowMs) * windowMs. That is the whole trick for tumbling windows: you never have to explicitly schedule a "close the window" timer, the arrival of a tick belonging to the next bucket is what closes the previous one. The tradeoff is that a window only closes when something pushes it closed, which is why flush() exists, to force-close whatever is still open when the stream ends (or the process shuts down).

graph LR
    subgraph "window 09:00:00"
        A["tick O=9500"] --> B["tick H=9520"] --> C["tick L=9490"]
    end
    subgraph "window 09:00:01"
        D["tick opens new candle"]
    end
    C -->|"tick.ts crosses boundary"| D
    C -.->|closed candle emitted| E["Candle: O/H/L/C, volume, count"]

What 200 Ticks Look Like as Candles

I ran both patterns back to back over the same 200-tick, two-symbol stream, using synthetic timestamps spaced 10ms apart so the windows would land deterministically. With a 50ms window, that firehose collapsed into 80 closed candles, each one an open, high, low, close, and volume for a slice of time instead of a wall of individual prints. That reduction, from raw prints to a manageable series, is the reason charting libraries never plot raw ticks directly.

Tumbling was the only shape I built, though the others are worth knowing about: hopping windows overlap by moving forward in smaller steps than their size, so a tick can belong to more than one window; sliding windows recompute continuously as new events arrive; session windows close after a gap of inactivity rather than a fixed clock boundary. Tumbling is the simplest and the right default for OHLC, because candles are supposed to be disjoint, non-overlapping slices of time by definition.

State Is Yours to Manage

Kafka Streams on the JVM backs its state stores with changelog topics, so if a pod dies, another pod can rebuild the exact same lastPrice map or candle-in-progress by replaying the changelog. Rolling my own, I get none of that. If a pod crashes mid-window and its in-memory state is gone, two options survive scrutiny: make the updates idempotent by recomputing state from the source ticks instead of trusting an in-memory accumulator, or use transactions to commit processing output and consumer offsets together, so a replay after a crash reprocesses exactly the ticks it needs to and no more. Either way, state you keep in memory for speed is state you need a story for losing.

Redis Streams gives you the primitives to build the same two patterns: a Lua script can do an atomic per-key read-modify-write for the edge check, and a sorted set can hold a windowed aggregate. You are still assembling it from smaller pieces. Neither broker hands you a windowed state store in Node, and both hand you enough to build one yourself, which has been the running theme of this whole exercise.

The Short Version

• An edge trigger needs the previous value to detect a crossing; a level trigger only needs the current value and is stateless but noisy. Same rules, 9 notifications versus 194 over the same 200 ticks.
• AlertEvaluator keeps a per-symbol lastPrice map and reads it before updating it, and that ordering is what makes crossing detection correct.
• A tumbling window is a fixed, non-overlapping bucket; a tick belonging to a later bucket is what closes the previous one and emits it as a candle.
• OhlcAggregator reduced a 200-tick firehose to 80 chartable candles at a 50ms window, by folding open, high, low, close, and volume per bucket.
• Co-partitioning does more than enable joins. It is what lets a stateful processor's in-memory state stay correct without a shared store, because the pod holding the ticks is the pod holding the state.
• Rolling your own stateful processing means owning the failure story: idempotent recomputation or transactional offset commits, since no changelog-backed state store does it for you.

Everything above I verified by squinting at log output, which is a miserable way to get a feel for a running system. So the next thing I built was a live dashboard over Server-Sent Events: ticks, alerts, and partition ownership all moving in a browser tab while the brokers work underneath.