Are you an LLM? Read llms.txt for a summary of the docs, or llms-full.txt for the full context.
Skip to content

Price alert stream

Template: price-alert-stream · category: monitoring · risk: monitor_only

Subscribe to a public price WebSocket feed, filter by symbol and a price threshold, and notify a channel — as an ordinary, durable rflow workflow. No networks, no signer: neither embedded engine boots.

This is the reference use of the trigger.stream WebSocket trigger.

Honest scope

This is durable automation, not HFT. A matched message flows through rflow's normal path — it claims a run, respects the workflow's throttle and rate limit, and is journaled with full history and recovery. It is not a low-latency trading engine, and a third-party stream that does not carry a stable message id is not exactly-once (a reconnect can then re-deliver).

When to use it

  • watch a market and page an operator when a symbol crosses a level
  • prove a streaming intake path before wiring reads/sends behind it
  • turn an exchange/oracle/vendor feed into normal, replayable rflow runs

How it works

# recipe: partial
trigger:
  stream:
    websocket:
      url: ${STREAM_WS_URL}          # e.g. wss://stream.binance.com:9443/ws
      subscribe:
        method: SUBSCRIBE
        params: ["ethusdt@ticker"]
        id: 1
      where: "${{ message.s == 'ETHUSDT' and wei(message.c, 8) > wei('3500', 8) }}"
      idempotency_key: "${{ message.E }}"
      idempotency_ttl: 1h
      heartbeat: 30s
      reconnect:
        min_backoff: 1s
        max_backoff: 60s
      throttle:
        threshold:
          count: 1
          window: 10m
        cooldown: 10m
  • where is evaluated over the message root (the JSON payload after message_json_path, default $). Only matching messages fire; false fires nothing. wei(message.c, 8) parses the last price (a decimal string) to an exact integer so the comparison is precise.
  • idempotency_key makes the feed's event time the claim identity: a reconnect that re-delivers the same event within idempotency_ttl creates no new run. Drop it (or point it at a feed without stable ids) and every matched message fires a fresh run — documented at-least-once.
  • throttle caps a chatty feed to one alert per 10 minutes while the price stays above the level.

Reliability + redaction

The client connects with a timeout, reconnects with bounded exponential backoff on any drop, pings on the heartbeat interval, and drops any frame over max_message_bytes (default 1 MiB) with a warning. The run payload is { args: <message>, stream: { url_host, received_at } } — a token in the URL is redacted to the host only and never lands in a payload or a log line.

Generate it

rflow new --template price-alert-stream
cd price-alert-stream
cp .env.example .env    # STREAM_WS_URL, TG_BOT_TOKEN, TG_CHAT_ID
docker compose up -d
rflow validate && rflow start

Rehearse offline

rflow test price-alert --fixture fixtures/ticker-message.json

The fixture is the extracted message object, so the where filter and the notification run without touching the internet.