Real-Time Token OHLCV Candle Streaming

event: token_ohlcv
id: 1757930400
data: {"chain":"ethereum","token_id":"0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48","interval":"60s","timestamp":"2026-09-15T08:40:00Z","open":0.9998,"high":1.0003,"low":0.9996,"close":1.0001,"avg":1.0000,"volume_usd":15230.50,"txns":12}

candles per hour  <=  min( 3600 / interval_seconds , swaps per hour )

// candle-tail.mjs: Node 18+, no dependencies. Needs DEXPAPRIKA_API_KEY.
const KEY = process.env.DEXPAPRIKA_API_KEY;
const CHAIN = "ethereum";
const TOKEN = "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48";
const INTERVAL = "60s"; // 1s costs 60x this for the same trades

// Candles are keyed by timestamp and OVERWRITTEN, never appended: a late swap
// republishes a sealed candle under the same timestamp with new values, and
// resuming redelivers the candle at the resume point.
const candles = new Map();

let lastEventId = null;
let lastBeat = Date.now();

async function tail() {
  const url = new URL("https://streaming-pro.dexpaprika.com/sse/ohlcv");
  url.searchParams.set("method", "token_ohlcv");
  url.searchParams.set("chain", CHAIN);
  url.searchParams.set("address", TOKEN);
  url.searchParams.set("interval", INTERVAL);

  const headers = {
    // The key is the whole Authorization value, with no scheme word in front of it.
    authorization: KEY,
    accept: "text/event-stream",
  };
  // Older than 15 minutes is clamped rather than refused, so this is always safe
  // to send. The same value as `since` would be a 400.
  if (lastEventId) headers["last-event-id"] = lastEventId;

  const res = await fetch(url, { headers });
  if (!res.ok) {
    // 403 = plan or missing header, 401 = key rejected, 400 = bad parameters.
    // None of them get better by retrying unchanged.
    throw new Error(`${res.status} ${await res.text()}`);
  }

  const decoder = new TextDecoder();
  let buffer = "";

  for await (const chunk of res.body) {
    buffer += decoder.decode(chunk, { stream: true });
    const frames = buffer.split("\n\n");
    buffer = frames.pop() ?? "";

    for (const frame of frames) {
      const name = frame.match(/^event: (.+)$/m)?.[1];
      const data = frame.match(/^data: (.+)$/m)?.[1];
      const id = frame.match(/^id: (.+)$/m)?.[1];
      if (!name || !data) continue;

      if (name === "ping") { lastBeat = Date.now(); continue; }
      if (name === "warning") { console.warn("[warning]", data); continue; }
      if (name === "error") { throw new Error(`stream error: ${data}`); }
      if (name !== "token_ohlcv") continue;

      const c = JSON.parse(data);
      const known = candles.has(c.timestamp);
      candles.set(c.timestamp, c); // overwrite, do not append
      if (id) lastEventId = id;

      console.log(
        `${known ? "revised" : "sealed "} ${c.timestamp}  ` +
          `o ${c.open}  h ${c.high}  l ${c.low}  c ${c.close}  ` +
          `$${c.volume_usd.toFixed(2)} in ${c.txns} swaps`,
      );
    }
  }
}

// Pings every 15s, so 45s of silence means the socket is gone. A token with no
// trades is silent too, which is exactly why the heartbeat is the liveness test.
setInterval(() => {
  if (Date.now() - lastBeat > 45_000) {
    console.error("no heartbeat, exiting for the supervisor to restart");
    process.exit(1);
  }
}, 5_000);

let delay = 1_000;
for (;;) {
  try {
    await tail();
    delay = 1_000;
  } catch (err) {
    console.error(String(err));
    if (/^4(0[0134])/.test(String(err))) process.exit(1); // fix the request, do not retry
    await new Promise((r) => setTimeout(r, delay + Math.random() * 500));
    delay = Math.min(delay * 2, 60_000);
  }
}