Live trades over WebSocket
The trades channel: every swap of a token, a pool or a wallet, pushed as blocks are indexed, in GET /trades's row shape.
Subscribe to trades with exactly one of token, pool or wallet. Each trades message carries the scope's trades of one indexed batch of blocks, newest first, as the rows of GET /trades for the same scope: from the token's side with token, else from each swap's base; with the trader's labels.
Subscribe:
{"op":"subscribe","id":"sol","channel":"trades","chain":"solana","token":"So11111111111111111111111111111111111111112"}Captured on a test instance: a token's feed, trimmed to 1 trade:
{"type":"subscribed","id":"t1","channel":"trades","chain":"evm:56","token":"0xa9fcda9bd321adeb7cbbbd30ef2dd0ed1d657777"}
{
"id": "t1",
"type": "trades",
"data": [
{
"time": 1790618628209,
"block": 1000751,
"txIndex": 0,
"eventIndex": 4,
"tx": "0xa8c3338c4c9514edbed29ea1fce567486d9ae1bf1062b6996f15d1149cbd035f",
"pool": "0x702e6533e0182381606adbe3ccbb77f682ccc58b",
"protocol": "pancakeswap-v2",
"side": "buy",
"token": {
"address": "0xa9fcda9bd321adeb7cbbbd30ef2dd0ed1d657777",
"symbol": "TAX",
"name": "Tax Token",
"decimals": 18
},
"amount": "50000",
"counter": {
"address": "0xbb4cdb9cbd36b01bd1cbaebf2de08d9173bc095c",
"symbol": "WBNB",
"name": "Wrapped BNB",
"decimals": 18
},
"counterAmount": "0.5",
"priceUsd": 0.006,
"amountUsd": 300,
"priced": true,
"trader": "0x3304e22ddaa22bcdc5fca2269b418046ae7b566a",
"labels": [
"sniper",
"fresh"
]
}
]
}Ordering and gaps
- Within a message trades are newest first, and messages arrive in the order their blocks were indexed.
- Nothing is replayed. Swaps indexed before
subscribed, or during an outage (resubscribed), were not sent: on both, fetch the recent trades withGET /tradesand merge them, keyed bytx,poolandeventIndex. - A trade is sent as soon as it is known, and once. On Solana that is when it executes (
processed), on Base when its flashblock is built, on other chains when its block arrives. Until its block is sealed it carriespreconfirmed: true, and REST returns it about half a second later on Solana. - A
retractmessage names trades already sent that did not happen:{"type":"retract","data":[{"block","tx","pool","eventIndex"}]}. The chain abandoned the slot or reorganised, and REST will not return them. Drop those rows, matchingblockas well: the same transaction may land again in another block, and arrives then as a new trade.
Clients
All three take a REST snapshot on every (re)subscription, merge without duplicates, and reconnect with backoff.
Live clients, Browser:
// Live trades of one token in a web page: one socket, a trades subscription, a REST snapshot
// on every (re)subscription, and reconnection. Same origin as the Aurelia console, whose /api
// forwards to the API: your signed-in session is the credential.
const API = "/api";
const CHAIN = "solana";
const TOKEN = "So11111111111111111111111111111111111111112"; // WSOL
const list = document.body.appendChild(document.createElement("ol"));
let trades = []; // newest first
const key = (t) => `${t.tx}:${t.pool}:${t.eventIndex}`;
const newest = (a, b) =>
b.time - a.time || b.block - a.block || b.txIndex - a.txIndex || b.eventIndex - a.eventIndex;
function merge(batch) {
const byKey = new Map(trades.map((t) => [key(t), t]));
for (const t of batch) byKey.set(key(t), t);
trades = [...byKey.values()].sort(newest).slice(0, 200);
list.replaceChildren(...trades.slice(0, 50).map((t) => {
const li = document.createElement("li");
const usd = t.priceUsd === null ? "unpriced" : `$${t.priceUsd.toPrecision(4)}`;
li.textContent = `${new Date(t.time).toLocaleTimeString()} ${t.side} ${t.amount} at ${usd} ${t.labels.join(" ")}`;
return li;
}));
}
// Swaps indexed while this subscription was not live were not sent: fetch them.
async function snapshot() {
const res = await fetch(`${API}/trades?chain=${CHAIN}&token=${TOKEN}&limit=50`);
if (res.ok) merge((await res.json()).trades);
}
let delay = 1000;
function connect() {
const url = new URL(`${API}/live`, location.href);
url.protocol = url.protocol === "https:" ? "wss:" : "ws:";
const ws = new WebSocket(url);
ws.onopen = () => ws.send(JSON.stringify({ op: "subscribe", id: "sol", channel: "trades", chain: CHAIN, token: TOKEN }));
ws.onmessage = (e) => {
const msg = JSON.parse(e.data);
if (msg.type === "subscribed") delay = 1000;
if (msg.type === "subscribed" || msg.type === "resubscribed") snapshot();
else if (msg.type === "trades") merge(msg.data);
else if (msg.type === "error") console.error(msg.status, msg.error); // that subscription ended
};
ws.onclose = (e) => {
if (e.code === 4401 || e.code === 4402) return console.error("refused:", e.reason); // fix the key or add credit: do not retry
let wait = delay;
if (e.code === 4429) {
try { wait = Math.max(wait, JSON.parse(e.reason).retryAfter * 1000 || 60_000); } catch {}
}
setTimeout(connect, wait);
delay = Math.min(delay * 2, 30_000);
};
}
connect();Live clients, Bun / Node:
// live.mjs: `bun live.mjs` or `node live.mjs` (Node 22+ has fetch and WebSocket built in).
// Prints a pool's swaps as they are indexed, and reconnects when the connection drops.
const API = process.env.AURELIA_API ?? "https://app.aurelia.build/api";
const KEY = process.env.AURELIA_KEY; // your API key, from your account page
const CHAIN = "solana";
const POOL = "Czfq3xZZDmsdGdUyrNLtRhGc47cXcZtLG4crryfu44zE"; // SOL / USDC, Orca
const seen = new Set();
function print(batch) {
for (const t of [...batch].reverse()) { // batches are newest first; print oldest first
const key = `${t.tx}:${t.pool}:${t.eventIndex}`;
if (seen.has(key)) continue;
if (seen.size > 10_000) seen.clear();
seen.add(key);
const usd = t.priced ? `$${t.priceUsd.toFixed(4)}` : "unpriced";
console.log(new Date(t.time).toISOString(), t.side, t.amount, t.token.symbol, usd, t.labels.join(" "));
}
}
// Swaps indexed while this subscription was not live were not sent: fetch them.
async function snapshot() {
const res = await fetch(`${API}/trades?chain=${CHAIN}&pool=${POOL}&limit=20`, {
headers: { authorization: `Bearer ${KEY}` },
});
if (res.ok) print((await res.json()).trades);
else console.error("snapshot failed:", res.status, await res.text());
}
let delay = 1000;
function connect() {
const ws = new WebSocket(`${API.replace(/^http/, "ws")}/live?apiKey=${KEY}`);
ws.onopen = () => ws.send(JSON.stringify({ op: "subscribe", id: "pool", channel: "trades", chain: CHAIN, pool: POOL }));
ws.onmessage = (e) => {
const msg = JSON.parse(e.data);
if (msg.type === "subscribed") delay = 1000;
if (msg.type === "subscribed" || msg.type === "resubscribed") snapshot();
else if (msg.type === "trades") print(msg.data);
else if (msg.type === "error") console.error("subscription ended:", msg.status, msg.error);
};
ws.onclose = (e) => {
if (e.code === 4401 || e.code === 4402) {
console.error("refused:", e.reason); // bad credentials or no credit: do not retry
process.exit(1);
}
let wait = delay;
if (e.code === 4429) {
try { wait = Math.max(wait, JSON.parse(e.reason).retryAfter * 1000 || 60_000); } catch {}
}
console.error(`closed (${e.code}), reconnecting in ${wait} ms`);
setTimeout(connect, wait);
delay = Math.min(delay * 2, 30_000);
};
}
connect();Live clients, Python:
# live.py: pip install requests websockets, then python live.py
# Prints a pool's swaps as they are indexed, and reconnects when the connection drops.
import asyncio, json, os, sys
from datetime import datetime, timezone
import requests
import websockets
API = os.environ.get("AURELIA_API", "https://app.aurelia.build/api")
KEY = os.environ["AURELIA_KEY"] # your API key, from your account page
CHAIN = "solana"
POOL = "Czfq3xZZDmsdGdUyrNLtRhGc47cXcZtLG4crryfu44zE" # SOL / USDC, Orca
seen = set()
def show(batch):
for t in reversed(batch): # batches are newest first; print oldest first
key = (t["tx"], t["pool"], t["eventIndex"])
if key in seen:
continue
seen.add(key)
when = datetime.fromtimestamp(t["time"] / 1000, timezone.utc).isoformat()
usd = f"${t['priceUsd']:.4f}" if t["priced"] else "unpriced"
print(when, t["side"], t["amount"], t["token"]["symbol"], usd, " ".join(t["labels"]), flush=True)
def snapshot():
# Swaps indexed while this subscription was not live were not sent: fetch them.
headers = {"Authorization": f"Bearer {KEY}"}
r = requests.get(f"{API}/trades", params={"chain": CHAIN, "pool": POOL, "limit": 20}, headers=headers, timeout=10)
r.raise_for_status()
show(r.json()["trades"])
async def main():
url = API.replace("http", "ws", 1) + f"/live?apiKey={KEY}"
subscribe = {"op": "subscribe", "id": "pool", "channel": "trades", "chain": CHAIN, "pool": POOL}
delay = 1
while True:
wait = delay
try:
async with websockets.connect(url) as ws:
await ws.send(json.dumps(subscribe))
async for raw in ws:
msg = json.loads(raw)
if msg["type"] == "subscribed":
delay = 1
if msg["type"] in ("subscribed", "resubscribed"):
await asyncio.to_thread(snapshot)
elif msg["type"] == "trades":
show(msg["data"])
elif msg["type"] == "error":
print("subscription ended:", msg["status"], msg["error"], file=sys.stderr)
except (OSError, websockets.WebSocketException) as e:
close = getattr(e, "rcvd", None)
if close and close.code in (4401, 4402):
sys.exit(f"refused: {close.reason}") # bad credentials or no credit: do not retry
if close and close.code == 4429:
try:
wait = max(wait, json.loads(close.reason).get("retryAfter") or 60)
except ValueError:
wait = max(wait, 60)
print(f"disconnected ({e}), reconnecting in {wait} s", file=sys.stderr)
await asyncio.sleep(wait)
delay = min(delay * 2, 30)
asyncio.run(main())