Keep your market catalog in sync without pollingCopy Anchor Link

If you keep your own copy of sufwan-local-test's events and markets, for search, alerts or your own listing pages, you need to know when a market is listed, changes or goes away. Polling List venue markets sorted by updatedAt gets you part of the way, at the cost of a full page read every few seconds. It also misses changes that do not touch the row itself. A market that joins or leaves a matched cluster keeps its updatedAt, so a poller never sees it become comparable across venues.

The catalog feed tells you what changed instead.

How it worksCopy Anchor Link

There are two halves, and you use both:

• The catalog WebSocket channel pushes a catalog_change frame for every event or market that is created, changes or is deleted. It is live only: nothing is replayed after a reconnect.
• GET /catalog/changes returns the same items from any point in the last 24 hours, oldest first, paged by cursor. This is how you fill the gap while you were away.

To sync, subscribe first and hold the frames that arrive. Then page the REST endpoint from your last position minus about a minute of overlap, apply what it returns, and apply the held frames. After that, apply frames as they arrive. Do the same catch-up after every reconnect and every stream_interrupted that lists catalog. The SDK's watchCatalog does all of this for you.

Items are state, not historyCopy Anchor Link

Each item carries an entity's current state, not a single edit. Two edits close together can arrive as one item, and the overlap means you will see some items twice. Both are fine if you apply an item only when it is newer than what you hold for that (entity, id):

import type { CatalogChange } from "@agg-build/sdk";

function isNewer(held: { version: number; changedAt: string } | undefined, item: CatalogChange) {
  if (!held) return true;
  if (item.version !== held.version) return item.version > held.version;
  return Date.parse(item.changedAt) > Date.parse(held.changedAt);
}

A delete carries only the id: no snapshot and venue: null. Deletes are sent even when you filter by venue, because the server may no longer know the venue. Drop the id if you hold it and ignore it otherwise.

Matching fieldsCopy Anchor Link

Every snapshot says where the entity sits in its matched cluster:

Anchors can change. When a cluster gains or loses a member, every affected entity gets a new item, and the anchor may move to another member. Store each entity under its own id and treat clusterAnchorId as a pointer you update, never as a key. These fields follow your app's venue settings: a member whose anchor is on a venue your app hides is reported unmatched, the same way list endpoints show it as its own row. Market snapshots also carry outcomes[].id, which is the id the orderbook socket takes.

Snapshots carry the fields that identify and place an entity: titles, status, dates, categories, outcomes and matching. Prices and volume are not catalog changes. Read those from market data.

Quick start (SDK)Copy Anchor Link

watchCatalog subscribes, catches up, de-duplicates and re-runs the catch-up after reconnects. onChange is called once per newer state of an entity.

import { createAggClient } from "@agg-build/sdk";

const client = createAggClient({
  baseUrl: "https://api.t.sufwankhalid.com",
  wsUrl: "wss://ws.t.sufwankhalid.com/ws",
  appId: process.env.AGG_APP_ID,
  apiKey: process.env.AGG_API_KEY, // server-side only; sent as x-app-api-key
});

// onChange is not awaited. Chain your writes so they land in order.
let writes = Promise.resolve();

const handle = client.watchCatalog({
  since: (await loadCatalogPosition()) ?? new Date(Date.now() - 60 * 60_000),
  onChange: (change) => {
    writes = writes.then(() =>
      change.op === "deleted"
        ? db.remove(change.entity, change.id)
        : db.upsert(change.entity, change.id, change.event ?? change.market),
    );
  },
  onSynced: ({ lastChangedAt }) => console.log("catalog live from", lastChangedAt),
  onError: (err) => console.error("catalog feed", err),
  onResyncRequired: async () => {
    const startedAt = new Date();
    await relistCatalog(); // see "Resync after 24 hours" below
    handle.resync(startedAt);
  },
});

// Save the position only once everything before it is written.
setInterval(() => {
  const position = handle.lastChangedAt;
  if (position) void writes.then(() => saveCatalogPosition(position));
}, 10_000);

await handle.ready; // first catch-up done

The handle has stop(), lastChangedAt, resync(since) and ready. Persist lastChangedAt somewhere durable once the changes before it are stored, and pass it back as since when your process restarts. An older position is always safe: it only re-reads more. Without socket, watchCatalog opens its own connection and stop() closes it. To share a connection you already have, pass it as socket. On a backend, give the client your apiKey and the connection authenticates with it. See Server connections.

First syncCopy Anchor Link

Before your first watch, you hold nothing. Note the time, list everything once with List venue events and List venue markets, then start watchCatalog with since set to the time you noted. Changes made while you were listing arrive through the catch-up.

ReactCopy Anchor Link

useCatalogFeed from @agg-build/hooks runs watchCatalog for the life of the component, on its own connection (the client needs wsUrl). It takes the same options plus enabled, and returns { status, lastChangedAt, error, resync }. status is idle, syncing, live, error or resync_required.

import { useCatalogFeed } from "@agg-build/hooks";

function CatalogSync({ since }: { since: string }) {
  const { status, lastChangedAt, error } = useCatalogFeed({
    since,
    venues: ["polymarket", "kalshi"],
    matchedOnly: true,
    onChange: (change) => catalogStore.apply(change),
  });

  if (error) return <p>Catalog sync error: {String(error)}</p>;
  return (
    <p>
      Catalog {status}, up to {lastChangedAt ?? "start"}
    </p>
  );
}

Pass a fixed since, such as the position you loaded at startup. Changing since, venues, entities or matchedOnly restarts the feed; changing the callbacks does not. After resync_required, re-list and call the returned resync(startedAt).

Most catalogs belong on a server. Use the hook for live UI, such as a "new markets" list, not as your system of record.

Raw protocolCopy Anchor Link

Any language works. Connect, subscribe, catch up, then apply frames. The same isNewer check makes the overlap harmless.

1. Subscribe. Open wss://ws.t.sufwankhalid.com/ws?appId=YOUR_APP_ID. From a backend, send your key in the x-app-api-key handshake header and no Origin. After connected, send:

{ "action": "subscribe", "channel": "catalog", "venues": ["polymarket", "kalshi"], "id": "cat-1" }

Wait for { "type": "subscribed", "channel": "catalog", "id": "cat-1" }. Hold any catalog_change frames that arrive until the catch-up finishes. Frame shapes are in WebSocket Protocol.

2. Catch up. Page from your saved position minus 60 seconds until hasMore is false. Repeat multi-value parameters (venues=a&venues=b); a comma list is rejected.

let cursor: string | undefined;
const since = new Date(Date.parse(savedChangedAt) - 60_000);
do {
  const page = await client.getCatalogChanges(
    cursor ? { cursor } : { since, venues: ["polymarket", "kalshi"] },
  );
  for (const change of page.data) applyIfNewer(change);
  cursor = page.hasMore ? page.nextCursor : undefined;
} while (cursor);

{
  "data": [
    {
      "entity": "market",
      "id": "vm_8f2c",
      "op": "created",
      "version": 1,
      "changedAt": "2026-10-21T12:00:41.512Z",
      "venue": "kalshi",
      "market": { "id": "vm_8f2c", "venueEventId": "ve_51aa", "question": "…", "matched": false, "clusterAnchorId": null, "clusterVenues": ["kalshi"], "…": "…" }
    }
  ],
  "nextCursor": "MTc5MjU5MzI0MTUxMjAwMHxtYXJrZXQ6dm1fOGYyYw",
  "hasMore": false
}

nextCursor is opaque and always set. A page can hold fewer than limit items, or none, while hasMore is still true, so stop on hasMore, not on an empty page. limit is 1 to 500 and defaults to 200. Full parameters are in List catalog changes.

3. Go live. Apply the held frames with the same check, then apply each new frame as it arrives. Save the newest changedAt you applied.

4. Repeat the catch-up from your saved position minus 60 seconds after every reconnect and every stream_interrupted that lists catalog. Keep holding frames while it runs.

A Python backend, for example (httpx and websockets 14+). load_position() returns your saved changedAt as epoch seconds, and apply_if_newer() is the check above plus your write:

import asyncio, json, os
from datetime import datetime, timezone
import httpx, websockets

API = "https://api.t.sufwankhalid.com"
app_id, app_key = os.environ["AGG_APP_ID"], os.environ["AGG_API_KEY"]
HEADERS = {"x-app-id": app_id, "x-app-api-key": app_key}
VENUES = ["polymarket", "kalshi"]

class ResyncRequired(Exception):
    pass

async def catch_up(http):
    since = datetime.fromtimestamp(load_position() - 60, timezone.utc).isoformat()
    params = {"since": since, "venues": VENUES}
    while True:
        r = await http.get(f"{API}/catalog/changes", params=params, headers=HEADERS)
        if r.status_code == 410:
            raise ResyncRequired()
        r.raise_for_status()
        page = r.json()
        for change in page["data"]:
            apply_if_newer(change)
        if not page["hasMore"]:
            return
        params = {"cursor": page["nextCursor"]}

async def run():
    url = f"wss://ws.t.sufwankhalid.com/ws?appId={app_id}"
    async with httpx.AsyncClient() as http, websockets.connect(
        url, additional_headers={"x-app-api-key": app_key}
    ) as ws:
        held, task, again = [], None, False
        async for raw in ws:
            msg = json.loads(raw)
            kind = msg.get("type")
            if kind == "connected":
                await ws.send(json.dumps({"action": "subscribe", "channel": "catalog", "venues": VENUES}))
            elif kind == "catalog_change":
                held.append(msg)
            elif (kind == "subscribed" and msg.get("channel") == "catalog") or (
                kind == "stream_interrupted" and "catalog" in msg.get("channels", [])
            ):
                again = True  # catch up after every ack and every interruption
            if task is not None and task.done():
                task.result()  # raises ResyncRequired or the HTTP error
                task = None
            if again and task is None:
                again = False
                task = asyncio.create_task(catch_up(http))
            if task is None:  # live: apply held and new frames directly
                for change in held:
                    apply_if_newer(change)
                held.clear()

The held frames are applied on the first message after the catch-up finishes. The gateway sends a heartbeat at least every 30 seconds, so they never wait longer than that.

Reconnect with backoff when the socket closes, as in Reconnection. The new connection runs the catch-up again.

Resync after 24 hoursCopy Anchor Link

GET /catalog/changes keeps 24 hours. A since, or a cursor, older than that returns 410:

{
  "message": "Position is older than the catalog feed retention window; re-list with GET /venue-events and GET /venue-markets, then resume from when that re-list started",
  "code": "resync_required"
}

Note the time, re-list everything with List venue events and List venue markets, and resume the feed from the time you noted. In the SDK, that is onResyncRequired followed by handle.resync(startedAt); isCatalogResyncRequired(err) recognizes the error from getCatalogChanges. Keep the frames that arrive while you re-list; the catch-up after resync covers them anyway.

Limits and timingCopy Anchor Link

• 24-hour window. Resume positions older than that need a resync.
• No replay on the socket. Frames sent while you were disconnected are not resent. The REST catch-up is the only way to fill a gap.
• Latency. A frame follows a recorded change within a few seconds. How soon a new listing is recorded depends on how often sufwan-local-test discovers that venue's catalog: typically 30 to 60 seconds after the venue lists a market.
• Your app's settings apply. Venues and categories disabled for your app are left out of both halves, except deletes, which only carry an id.
• Settings changes are not replayed. Disabling a venue or category for your app hides future items; it does not send deletes for what you already hold. Re-list after changing them.
• Market categories. A market's categories are its event's. When only the event's categories change, the event gets a new item and its markets do not; read categories from the event when they matter.
• Filters. Up to 64 venues per subscribe. Subscribing again on the same connection replaces your filter. There is no server-side matched-only filter; the SDK's matchedOnly filters on the client.
• Rate limits. GET /catalog/changes has its own Catalog feed group in Rate limits, separate from Discovery. A catch-up runs once per reconnect, not on a timer: at current volume, expect roughly 40 pages of 500 items for every hour you were disconnected. Once you are on the feed, you should not need to poll GET /venue-events or GET /venue-markets for new listings.
• Connections. The catalog channel works on browser and API-key connections. Backends connect with x-app-api-key and no Origin; the connection limits are in WebSocket limits.

RelatedCopy Anchor Link