Skip to main content

Sustained-condition alarms (sliding window)

A limit on a single reading fires on every transient. A pump's discharge pressure spikes on every start-up and every slug of gas, the alarm fires four hundred times a month, and somebody switches it off; on the one occasion it mattered, nobody was watching.

The question worth acting on is not is it above 4.5 but has it stayed above 4.5, and that is a question about a stretch of time. This page builds that rule end to end: seed a signal, replay it to prove the rule behaves, then run the same code against a live subscription.

Seed the data first

Run generator L on Generate sample data. It writes pump_p101_discharge_bar: six hours at one reading a minute, four brief spikes well above the limit, and one eighty-minute excursion just above it. Seeding is Python, as it is for every scenario here; the rule itself is below in all three languages.

1. The rule, in four numbers​

SettingValueWhy
LIMIT4.5 barThe existing alarm limit. Unchanged, deliberately
WINDOW20 minutesLonger than any transient worth ignoring, shorter than the time you need to act
CLEAR_BELOW4.2 barHysteresis. Raise and clear thresholds must differ or the rule flaps at the boundary
MIN_READINGS15A window with holes proves nothing. At one reading a minute, twenty minutes should bring about twenty

2. The detector​

One object, holding the last WINDOW of readings and whether a finding is currently open. Nothing in it knows where the readings came from, which is what lets the same code run over history and over a live stream.

import java.time.*;
import java.util.*;

record Reading(Instant at, double value) {}

class SustainedHigh {
    static final double LIMIT = 4.5, CLEAR_BELOW = 4.2;
    static final int MIN_READINGS = 15;
    static final Duration WINDOW = Duration.ofMinutes(20);

    private final Deque<Reading> recent = new ArrayDeque<>();
    private final String seriesId;
    private boolean open = false;

    SustainedHigh(String seriesId) { this.seriesId = seriesId; }

    EventModel push(Instant at, double value) {
        if (recent.stream().anyMatch(r -> r.at().equals(at))) {
            return null;                 // a redelivery, not a new reading
        }
        recent.addLast(new Reading(at, value));
        Instant newest = recent.peekLast().at();
        while (recent.peekFirst().at().isBefore(newest.minus(WINDOW))) {
            recent.removeFirst();        // evict on carried time, not the clock
        }
        return evaluate();
    }

    private EventModel evaluate() {
        if (recent.size() < MIN_READINGS) return null;
        Duration span = Duration.between(
                recent.peekFirst().at(), recent.peekLast().at());
        if (span.toMillis() < WINDOW.toMillis() * 0.9) return null;   // not full
        if (!open && recent.stream().allMatch(r -> r.value() > LIMIT)) {
            open = true;
            return raiseEvent();
        }
        if (open && recent.stream().allMatch(r -> r.value() < CLEAR_BELOW)) {
            open = false;                // clear only once it is properly back
        }
        return null;
    }

    private EventModel raiseEvent() {
        Instant at = recent.peekLast().at();
        double peak = recent.stream()
                .mapToDouble(Reading::value).max().orElseThrow();

        EventModel event = new EventModel();
        event.setExternalId(
                "sustained_high_" + seriesId + "_" + at.getEpochSecond());
        event.setType("sustained_high");
        event.setStatus("open");
        event.setEventTime(ZonedDateTime.ofInstant(at, ZoneOffset.UTC));
        event.setMetadata(Map.of(
                "series", seriesId, "limit", String.valueOf(LIMIT),
                "window_minutes", "20", "peak", String.format("%.2f", peak)));
        client.events().create(List.of(event));
        return event;
    }
}

The "window not filled" check is worth a second look. The buffer only ever holds readings within WINDOW of the newest one, so its span can never exceed the window; what it can be is short, either because the process has only just started or because readings stopped arriving. Comparing against most of a window rather than exactly one keeps the rule working at any sampling rate.

3. Replay the history, and see what fires​

The seeded six hours are the test. Pull them back and push each reading through the detector in order. A datapoint carries its timestamp and value as strings, so both are parsed on the way in.

final String SERIES = "pump_p101_discharge_bar";

RetrieveFilter filter = new RetrieveFilter();
filter.setExternalId(SERIES);
filter.setStart(ZonedDateTime.now().minusHours(7));
filter.setEnd(ZonedDateTime.now());
filter.setLimit(10000);

DataRetriever<RetrieveFilter> request = new DataRetriever<>();
request.setItems(List.of(filter));
var points = client.timeseries().retrieve(request)
        .getItems().get(0).getDatapoints();

SustainedHigh detector = new SustainedHigh(SERIES);
final double LIMIT = SustainedHigh.LIMIT;
int raised = 0, crossings = 0;
double previous = 0.0;
for (var p : points) {
    Instant at = Instant.ofEpochMilli(Long.parseLong(p.getTimestamp()));
    double value = Double.parseDouble(p.getValue());
    if (previous <= LIMIT && value > LIMIT) crossings++;
    previous = value;
    if (detector.push(at, value) != null) raised++;
}
System.out.printf("%d readings, %d crossings, %d event(s) raised%n",
        points.size(), crossings, raised);
// 360 readings, 5 crossings, 1 event(s) raised
What "working" looks like

Five crossings, one event. The four transients each cross the limit and each would have fired the old alarm; none of them survives twenty minutes, so none of them raises anything here. The excursion does, twenty minutes after it began, which is the price the window charges and the reason it is worth paying.

4. Verify it landed​

The detector writes real events, so the check is a query rather than a print.

EventRetreiver retriever = new EventRetreiver();
retriever.setLimit(100);
retriever.getFilter().setType(List.of("sustained_high"));

DataWrapper<EventModel> found = client.events().filter(retriever);
found.getItems().forEach(e ->
        System.out.println(e.getExternalId() + " " + e.getMetadata()));

if (found.getItems().size() != 1) {
    throw new AssertionError(
            "expected one finding, found " + found.getItems().size());
}

Re-running the replay is safe to try, and instructive: the detector raises the same finding again with the same external id, and the platform collapses it rather than storing a duplicate. Events are keyed by id, datapoints by series and timestamp.

5. Run the same detector live​

Swap the source. Nothing in the detector changes: push is fed by a subscription instead of by a list, and each message is acknowledged after it has been handled, so a crash replays it rather than losing it.

datapoints(...) below is your own unpacking of the delivered payload, which carries the action and the datapoints it affected; the SDK hands it to you rather than deciding what your rule wants from it.

Subscription sub = new Subscription();
sub.setExternalId("pump_p101");
sub.setName("Pump P-101");
sub.setTimeseries(List.of(IdCollection.createFromExternalId(SERIES)));
client.subscriptions().create(List.of(sub));

SustainedHigh detector = new SustainedHigh(SERIES);

try (SubscriptionListener listener =
             client.subscriptions().listen(List.of("pump_p101"))) {
    while (running) {
        SubscriptionMessage msg = listener.poll(Duration.ofSeconds(5));
        if (msg == null) continue;
        for (Reading r : datapoints(msg.payload())) {
            EventModel event = detector.push(r.at(), r.value());
            if (event != null) {
                System.out.println("raised " + event.getExternalId());
            }
        }
        listener.ack(msg.messageId());          // only once it is durably handled
    }
}

Two habits make the difference between this working on a desk and working on a plant:

  • Fill the window before trusting it. On start-up, replay the last twenty minutes through push exactly as section 3 does. Without it the rule is blind for its first twenty minutes, which is precisely when somebody has restarted it during an incident.
  • Treat a thin window as a finding of its own. MIN_READINGS makes the detector stay quiet when readings stop arriving, which is right, but silence is not the same as health. Raise a separate, quieter event about the instrument or the link.

6. The same shape, other rules​

Only the evaluate step changes:

QuestionThe change
Sustained below a limitInvert the comparisons
Rate of change over the windowCompare first and last, divide by the span
A ratio between two series driftingOne detector fed by two series, keyed on the nearest timestamps
Above the limit for most of the window, not all of itCount instead of "all", and raise on a fraction

When no threshold over any window can express what you are looking for, because the readings never leave their limits and it is the shape that changed, that is where LSTM anomaly detection starts earning its keep.

Further reading​

See also​