Real-Time Pool Reserves Streaming

{
  "chain": "ethereum",
  "pool_id": "0xe0554a476a092703abdb3ef35c80e0d76d32939f",
  "block": 25098911,
  "tokens": [
    {
      "token_id": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
      "reserve": 1897184450989,
      "delta": -4915987385,
      "price_usd": 1.0000070290769338,
      "reserve_usd": 1897197.78,
      "delta_usd": -4916.02
    },
    {
      "token_id": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
      "reserve": 1575557282142409643389,
      "delta": 2180365462384895407,
      "price_usd": 2255.21,
      "reserve_usd": 3553216.99,
      "delta_usd": 4917.19
    }
  ],
  "total_reserve_usd": 5450414.77,
  "total_delta_usd": 1.16
}
event: reserve_update

curl -N "https://streaming.dexpaprika.com/sse/reserves?method=pool_reserves&chain=ethereum&address=0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640"

data: {"chain":"ethereum","pool_id":"0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640","block":25098912,"tokens":[...],"total_reserve_usd":5042113.41,"total_delta_usd":-128.44}
event: reserve_update

import json
import requests

url = "https://streaming.dexpaprika.com/sse/reserves"
params = {
    "method": "token_reserves",
    "chain": "ethereum",
    "address": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",  # USDC
}

with requests.get(url, params=params, stream=True) as response:
    response.raise_for_status()

    # Buffer one SSE message at a time, then dispatch. The wire emits
    # `data:` before `event:`, so grouping is mandatory: a line-by-line
    # parser would never see the event type.
    msg_lines = []
    for raw in response.iter_lines(decode_unicode=True):
        if raw:
            msg_lines.append(raw)
            continue

        event_type = "message"
        data_str = None
        for ml in msg_lines:
            if ml.startswith("event:"):
                event_type = ml.split(":", 1)[1].strip()
            elif ml.startswith("data:"):
                data_str = ml[5:].lstrip()
        msg_lines = []

        if event_type != "reserve_update" or data_str is None:
            continue

        event = json.loads(data_str)
        pool = event["pool_id"]
        block = event["block"]
        net = event["total_delta_usd"]
        print(f"block {block}  pool {pool[:10]}...  net {net:+,.2f} USD")

block 25098905  pool 0x168e9b3c...  net +2.24 USD
block 25098905  pool 0x0f91b8bb...  net +0.14 USD
block 25098905  pool 0xff4ce5aa...  net +0.04 USD
block 25098905  pool 0xe0554a47...  net -0.03 USD
...

async function subscribe() {
  const response = await fetch("https://streaming.dexpaprika.com/sse/reserves", {
    method: "POST",
    headers: {
      "Accept": "text/event-stream",
      "Content-Type": "application/json",
    },
    body: JSON.stringify([
      {
        chain: "ethereum",
        address: "0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640",
        method: "pool_reserves",
      },
      {
        chain: "ethereum",
        address: "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
        method: "token_reserves",
      },
    ]),
  });

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

  for await (const chunk of response.body) {
    buffer += decoder.decode(chunk, { stream: true });

    // SSE messages are separated by a blank line.
    const messages = buffer.split("\n\n");
    buffer = messages.pop() ?? "";

    for (const message of messages) {
      const lines = message.split("\n");
      const eventLine = lines.find((l) => l.startsWith("event:"));
      const dataLine = lines.find((l) => l.startsWith("data:"));
      if (!dataLine) continue;

      // Four event types may arrive: `reserve_update`, `ping`, `warning`, `error`.
      // Filter for the one we care about; unrecognised events should be no-ops.
      const eventType = eventLine ? eventLine.slice(6).trim() : "message";
      if (eventType !== "reserve_update") continue;

      const payload = JSON.parse(dataLine.slice(5).trim());
      handleReserveUpdate(payload);
    }
  }
}

function handleReserveUpdate(event) {
  console.log(`pool ${event.pool_id}  block ${event.block}  Δ ${event.total_delta_usd.toFixed(2)} USD`);
}

subscribe();