Poll for updates · Signals Infra

Fetch every datapoint and document event since your last run, resuming from a stored cursor.

GET /v1/updates returns the same events the data webhooks deliver, in the same envelope and with the same ids, oldest first. Poll it when you cannot receive webhooks. Store next_cursor after each page and resume from it: every response carries it, even an empty page.

import json
import os
import pathlib
import time

import httpx

BASE_URL = os.environ.get("SIGNALS_BASE_URL", "https://api.signals.ai")
API_KEY = os.environ["SIGNALS_API_KEY"]
STATE = pathlib.Path("signals_cursor.json")


def get(client, url, **kwargs):
    """GET with the rate limit honoured: a 429 waits for its Retry-After."""
    for _attempt in range(6):
        r = client.get(url, **kwargs)
        if r.status_code != 429:
            r.raise_for_status()
            return r
        time.sleep(float(r.headers.get("Retry-After", "1")))
    r.raise_for_status()
    return r


def poll_updates(client, *, since="2026-01-01T00:00:00+00:00", state_path=STATE,
                 tickers=None, topics=None):
    """Every event after the stored cursor (or after `since` on the first run),
    oldest first. Stores the new cursor as it goes."""
    base = {"limit": 500}
    if tickers:
        base["tickers"] = ",".join(tickers)
    if topics:
        base["topics"] = ",".join(topics)
    state_path = pathlib.Path(state_path)
    if state_path.exists():
        params = {**base, "cursor": json.loads(state_path.read_text())["cursor"]}
    else:
        params = {**base, "since": since}
    events = []
    while True:
        r = get(client, "/v1/updates", params=params)
        page = r.json()
        events.extend(page["data"])
        cursor = page["pagination"]["next_cursor"]
        if cursor:
            state_path.write_text(json.dumps({"cursor": cursor}))
        if not page["pagination"]["has_more"]:
            return events
        params = {**base, "cursor": cursor}


if __name__ == "__main__":
    headers = {"Authorization": f"Bearer {API_KEY}"}
    with httpx.Client(base_url=BASE_URL, headers=headers, timeout=30) as client:
        for event in poll_updates(client):
            data = event["data"]
            print(event["id"], event["type"], data.get("ticker"),
                  data.get("series_key"), data.get("fiscal_period"), data.get("value"))

Run it on a schedule (every few minutes is plenty). A run that finds nothing prints nothing and keeps its cursor.