--- title: "1.4.1 REST and websocket downloads with retries" description: "Cursor pagination, rate limits, retries with backoff and jitter, resumable downloads and streams with heartbeats and gaps, on a local server." url: "https://pythonparatrading.com/en/python-trading-course/level-1/rest-websockets-pagination-and-retries/" lang: "en" image: "https://pythonparatrading.com/og/en/python-trading-course/level-1/rest-websockets-pagination-and-retries.png" --- # 1.4.1 · Downloading over REST and websockets: pagination, rate limits and retries > Pulling data from an API looks like one line of code until the API cuts you off on page twelve. This lesson builds a local server that behaves like a real market API —pagination, rate limits, random errors, a stream that drops messages— and a client that survives all of it. - Course: [Level 1: Financial data, markets and analysis](https://pythonparatrading.com/en/python-trading-course/level-1/) - Module: 1.4 · Financial data engineering - Duration: 60 min - Prerequisites: [0.4.6 · Concurrency: threading, multiprocessing, asyncio and free-threading](https://pythonparatrading.com/en/python-trading-course/level-0/python-concurrency-threading-multiprocessing-asyncio/), [1.3.3 · Retail and prosumer market data APIs](https://pythonparatrading.com/en/python-trading-course/level-1/retail-and-prosumer-market-data-apis/) ## By the end of this lesson - Walk through a cursor-paginated API without losing or repeating rows, and explain why a cursor is safer than an offset. - Tell apart the errors that are retried (429, 5xx, network) from those that are not (400, 401, 403, 404), and honor `Retry-After`. - Schedule retries with exponential backoff, a cap and jitter, and measure how long they wait. - Make a download idempotent and resumable: an interruption halfway leaves neither gaps nor duplicates. - Consume a stream with sequence numbers and heartbeats: detect gaps and dead connections, and reconnect from the last good message. ## Two ways to request data Almost every provider from lesson 1.3 offers its data in two forms. Over **REST**, the client requests a range (“NOVA 5-minute bars for June”) and the server responds with a JSON document; if the range is large, it splits it into pages. Over **streaming** (websockets, in most retail APIs), the client opens a connection that stays open and the server pushes every trade or every quote to it as soon as it happens. ```mermaid flowchart LR C[Client] -- GET page --> A[REST API] A -- JSON + cursor --> C C -- subscription --> W[Stream] W -- message seq n --> C W -- heartbeat --> C ``` | | REST | Streaming (websocket) | |---|---|---| | What for | History, filling gaps, reference data | Real time: trades, quotes, order book | | Who initiates | The client, request by request | The server, message by message | | Unit | Page of N rows with a cursor to the next one | Message with a sequence number | | Typical failure | 429 from too many requests, 5xx, cuts halfway | Lost messages, a dead connection with no warning | | Recovery | Retry the same request after a wait | Reconnect and resume from the last good message | The two are used together. A live system starts with REST (the history it needs to compute its indicators), hooks into the stream and, every time it detects a gap in the stream, goes back to REST to fill it. Everything that follows actually runs. Instead of calling a real provider (that would need a network connection, a key, and would depend on its mood that day), the lesson spins up a **local server** in a thread that imitates a market API: it paginates, rate-limits requests and fails at random. The client uses only the standard library (`urllib`); at the end are the versions with `requests` and `websockets`. ## A market API in a thread The server serves the NOVA 5-minute bars (`nova_barras_5min.parquet`). So that the lesson is reproducible and does not actually wait, client and server share a **simulated clock**: each request “takes” 40 ms and each client wait advances that clock instead of sleeping. It is the same technique used to test code that involves time (injecting the clock); in production, `dormir` would be `time.sleep` and each side would measure its own time. ```python import base64, json, random, threading import pandas as pd barras = pd.read_parquet("nova_barras_5min.parquet") class Reloj: """Simulated clock shared by client and server.""" def __init__(self): self.t = 0.0 self._cerrojo = threading.Lock() def ahora(self): return self.t def dormir(self, segundos): with self._cerrojo: self.t += segundos def cursor_de(ts): # opaque to the client: it must not build it return base64.urlsafe_b64encode(ts.isoformat().encode()).decode() def ts_de(cursor): return pd.Timestamp(base64.urlsafe_b64decode(cursor).decode()) reloj = Reloj() print(len(barras), "bars from", barras.ts.min().date(), "to", barras.ts.max().date()) ``` ```output 38940 bars from 2024-01-02 to 2025-12-31 ``` The API logic goes in a separate class, with no HTTP in it, so it can be read at a glance. It has the three rules that almost every real API has: - **Rate limit.** 10 per 1 s window. The eleventh gets `429 Too Many Requests` with a `Retry-After` header that says how many seconds to wait. Each good response carries `X-RateLimit-Remaining`, the ones left. - **Transient failures.** 8% of requests get `503 Service Unavailable`, from a seeded generator. - **Cursor pagination.** At most `limite` rows per response and a `siguiente` field with the cursor of the next page (`None` on the last one). A `limite` greater than 500 is a client error: `400 Bad Request`. ```python class ApiMercado: POR_VENTANA = 10 # requests per 1 s window LATENCIA = 0.04 # simulated seconds per request def __init__(self, reloj, prob_fallo=0.08, semilla=7): self.reloj, self.prob_fallo = reloj, prob_fallo self.azar = random.Random(semilla) self.ventana, self.usadas = 0.0, 0 self.registro = [] # (instant, status) def responder(self, p): self.reloj.dormir(self.LATENCIA) ahora = self.reloj.ahora() if ahora - self.ventana >= 1.0: # new window self.ventana, self.usadas = ahora, 0 if self.usadas >= self.POR_VENTANA: falta = 1.0 - (ahora - self.ventana) cab = {"Retry-After": str(max(1, round(falta)))} return self._fin(429, {"error": "too many"}, cab) self.usadas += 1 if self.azar.random() < self.prob_fallo: return self._fin(503, {"error": "unavailable"}) limite = int(p.get("limite", 100)) if limite > 500: return self._fin(400, {"error": "limite: maximum 500"}) hasta = pd.Timestamp(p["hasta"], tz="UTC") if "cursor" in p: # after the cursor sel = barras[barras.ts > ts_de(p["cursor"])] else: sel = barras[barras.ts >= pd.Timestamp(p["desde"], tz="UTC")] sel = sel[sel.ts < hasta] pag = sel.head(limite) sig = cursor_de(pag.ts.iloc[-1]) if len(sel) > limite else None datos = [{"t": r.ts.strftime("%Y-%m-%dT%H:%M:%SZ"), "o": r.apertura, "h": r.maximo, "l": r.minimo, "c": r.cierre, "v": r.volumen} for r in pag.itertuples()] cab = {"X-RateLimit-Remaining": str(self.POR_VENTANA - self.usadas)} return self._fin(200, {"datos": datos, "siguiente": sig}, cab) def _fin(self, estado, cuerpo, cab=None): self.registro.append((round(self.reloj.ahora(), 2), estado)) return estado, cab or {}, cuerpo ``` Now the HTTP part: a `ThreadingHTTPServer` from the standard library, on whatever port the system has free and in a background thread (threads and `asyncio`: lesson 0.4.6). The handler translates each `GET` into a call to `responder` and returns the JSON with its status code and headers. The port number changes on every run, so the cell does not print it. ```python from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qsl, urlsplit api = ApiMercado(reloj) class Manejador(BaseHTTPRequestHandler): def do_GET(self): url = urlsplit(self.path) if url.path == "/v1/barras/NOVA": estado, cab, cuerpo = api.responder( dict(parse_qsl(url.query))) else: estado, cab, cuerpo = 404, {}, {"error": "not found"} datos = json.dumps(cuerpo).encode() self.send_response(estado) self.send_header("Content-Type", "application/json") for k, v in cab.items(): self.send_header(k, v) self.send_header("Content-Length", str(len(datos))) self.end_headers() self.wfile.write(datos) def log_message(self, *args): # no line per request pass servidor = ThreadingHTTPServer(("127.0.0.1", 0), Manejador) threading.Thread(target=servidor.serve_forever, daemon=True).start() BASE = f"http://127.0.0.1:{servidor.server_port}" print("Server running:", servidor.server_port > 0) ``` ```output Server running: True ``` With the server running, the client is what is left: a ten-line function with `urllib.request`. One detail of `urllib`: it raises `HTTPError` on any 4xx or 5xx code. To treat all codes alike, the function catches that exception and returns the status code, the headers and the body, whatever the result. Network errors (connection refused, timeout) are let through: they are a different matter and are handled further down. ```python import urllib.error, urllib.request from urllib.parse import urlencode def pedir(ruta, **params): url = f"{BASE}{ruta}?{urlencode(params)}" try: with urllib.request.urlopen(url, timeout=5) as r: return r.status, r.headers, json.load(r) except urllib.error.HTTPError as e: # 4xx and 5xx return e.code, e.headers, json.load(e) JUNIO = dict(desde="2025-06-02", hasta="2025-07-01", limite=100) estado, cab, cuerpo = pedir("/v1/barras/NOVA", **JUNIO) print("Status:", estado) print("Remaining:", cab["X-RateLimit-Remaining"], "requests") print("Rows:", len(cuerpo["datos"])) print("First:", cuerpo["datos"][0]) print("Cursor:", cuerpo["siguiente"]) ``` ```output Status: 200 Remaining: 9 requests Rows: 100 First: {'t': '2025-06-02T13:30:00Z', 'o': 53.35, 'h': 53.48, 'l': 53.28, 'c': 53.46, 'v': 285081} Cursor: MjAyNS0wNi0wM1QxNToxNTowMCswMDowMA== ``` A hundred bars and a cursor. The cursor is an opaque string: on this server it encodes the time of the last bar served, but the client must neither interpret it nor build it, only send it back. Providers change its format without notice. A request with an invalid parameter shows the other kind of error, the one that waiting will not fix: ```python estado, _, cuerpo = pedir("/v1/barras/NOVA", desde="2025-06-02", hasta="2025-07-01", limite=1000) print(estado, cuerpo) ``` ```output 400 {'error': 'limite: maximum 500'} ``` ## Pagination: why a cursor and not an offset There are two common ways to paginate. By **offset** (`offset=200&limit=100`: “give me rows 200 to 299”) or by **cursor** (“give me the 100 after this key”). With market data the offset is fragile: if the provider inserts or corrects bars while you download (it happens often with the current day or after a tape correction), the positions shift and the next page repeats or skips rows without any visible error. The cursor points to a key (here, the time of the last bar) and does not depend on positions. The naive loop requests pages until `siguiente` is `None` and accepts any response as good. Against this API it does not last long: ```python raises def descargar_ingenuo(**params): filas, cursor, n = [], None, 0 while True: n += 1 extra = {"cursor": cursor} if cursor else {} estado, _, cuerpo = pedir("/v1/barras/NOVA", **params, **extra) if estado != 200: raise RuntimeError(f"HTTP {estado} on page {n}") filas += cuerpo["datos"] cursor = cuerpo["siguiente"] if cursor is None: return filas descargar_ingenuo(**JUNIO) ``` ```output RuntimeError: HTTP 503 on page 2 ``` It breaks on the second page with a 503 that, repeated a moment later, would have worked. Worse than breaking would be the opposite: a loop that ignored the status code and carried on without the page, with a hundred-bar gap in the middle of June that nobody would notice until an indicator came out strange. ## What is retried and what is not HTTP status codes say whose problem it is. **4xx** codes are client errors: repeating the same request will give the same error. The exception is **429**: the request is correct, but it arrives too soon. **5xx** codes are the server's and almost always transient. Network errors (timeout, connection reset) have no code, but they are transient too. | Response | Meaning | What the client does | |---|---|---| | 200 | OK | Saves the page and continues with the cursor | | 400, 422 | Invalid parameters | Fails right away: the code has to be fixed | | 401, 403 | No credentials or no permission (plan, license) | Fails right away: retrying does not grant permissions | | 404 | The resource does not exist (misspelled symbol, path) | Fails right away | | 429 | Too many requests | Waits as long as `Retry-After` says and repeats | | 500, 502, 503, 504 | Server failure or overload | Waits with backoff and repeats | | Timeout, connection reset | Network | Waits with backoff and repeats | The diagram shows the complete loop. What matters is in the arrows that go back: the request that is repeated is the **same** one, with the same cursor; it never skips to the next page. ```mermaid flowchart TB P[Request page] --> E{Status?} E -- 200 --> G[Save page] G --> S{Cursor?} S -- yes --> P S -- no --> F[Done] E -- 429 --> R[Wait Retry-After] E -- 5xx or network --> B[Wait backoff] R --> P B --> P E -- 4xx --> X[Permanent failure] ``` ## Exponential backoff, jitter and client pacing Always waiting the same amount between retries is not enough: if the server is overloaded, all the clients that failed come back at once and knock it over again. **Exponential backoff** doubles the wait on each attempt, with a cap so as not to wait for hours. **Jitter** adds randomness. Marc Brooker compared several ways of doing it on the AWS architecture blog; the simplest, *full jitter*, which is among those generating the fewest total requests in his simulation, waits a uniform time between zero and the exponential ceiling: $$ \text{wait}_n \sim \mathcal{U}\big(0,\ \min(\text{cap},\ \text{base}\cdot 2^{n})\big), \qquad n = 0, 1, 2, \dots $$ ```python def espera_backoff(n, base=0.5, tope=30.0, azar=random): return azar.uniform(0, min(tope, base * 2 ** n)) azar = random.Random(42) filas = [] for n in range(9): techo = min(30.0, 0.5 * 2 ** n) muestras = [espera_backoff(n, azar=azar) for _ in range(200)] filas.append({"intento": n + 1, "techo_s": techo, "media_s": sum(muestras) / len(muestras), "max_s": max(muestras)}) tabla_backoff = pd.DataFrame(filas).set_index("intento").round(2) tabla_backoff ``` ```output techo_s media_s max_s intento 1 0.5 0.24 0.50 2 1.0 0.52 1.00 3 2.0 0.99 2.00 4 4.0 2.06 4.00 5 8.0 4.34 7.99 6 16.0 7.46 15.91 7 30.0 14.99 29.84 8 30.0 15.49 29.99 9 30.0 15.40 29.72 ``` The figure at the top draws the table: the ceiling of each attempt, five sampled waits and the mean. The ceiling doubles until it hits the 30 s cap on the seventh attempt; from there on it stays flat. The mean wait comes out very close to half the ceiling (0.24 s with a 0.5 ceiling; 14.99 s with a 30 ceiling), because that is the mean of a uniform between 0 and the ceiling. With six attempts there are five waits: the worst case adds up to 15.5 s (0.5 + 1 + 2 + 4 + 8) and the average case, half that. It is a reasonable order of magnitude for a nightly download script; a live system needs much shorter caps and, above all, to know what to do while it has no data. The function with retries combines all of the above: it retries only what is retryable, honors `Retry-After` when it arrives (if the server says how long to wait, you listen) and gives up after a number of attempts. The clock and the random generator come in as arguments: that way the same code works in production (`time.sleep`, real randomness) and in a reproducible test. ```python REINTENTABLES = {429, 500, 502, 503, 504} def pedir_con_reintentos(ruta, intentos=6, dormir=reloj.dormir, azar=random.Random(1), **params): for n in range(intentos): try: estado, cab, cuerpo = pedir(ruta, **params) except (urllib.error.URLError, TimeoutError, ConnectionError): estado, cab, cuerpo = None, {}, None # network if estado == 200: return cuerpo if estado is not None and estado not in REINTENTABLES: raise RuntimeError(f"HTTP {estado}: {cuerpo['error']}") if n == intentos - 1: break # no wait after the last one espera = espera_backoff(n, azar=azar) if cab.get("Retry-After"): espera = max(espera, float(cab["Retry-After"])) dormir(espera) raise RuntimeError(f"no response after {intentos} attempts") ``` The full download is the naive loop with the new function. Before starting, it notes where the server log stands, in order to count afterwards how many requests of each kind were needed: ```python from collections import Counter def descargar(**params): filas, cursor = [], None while True: extra = {"cursor": cursor} if cursor else {} pag = pedir_con_reintentos("/v1/barras/NOVA", **params, **extra) filas += pag["datos"] cursor = pag["siguiente"] if cursor is None: return filas inicio, t0 = len(api.registro), reloj.ahora() filas = descargar(**JUNIO) estados = Counter(e for _, e in api.registro[inicio:]) print("Rows:", len(filas), "· pages:", -(-len(filas) // 100)) print("Requests by status:", dict(sorted(estados.items()))) print(f"Simulated time: {reloj.ahora() - t0:.1f} s") # what was downloaded must be exactly what is in the source df = pd.DataFrame(filas) df["t"] = pd.to_datetime(df["t"], utc=True) origen = barras[(barras.ts >= "2025-06-02") & (barras.ts < "2025-07-01")] print("Equal to the source:", len(df) == len(origen) and df["t"].is_unique and (df["c"].to_numpy() == origen["cierre"].to_numpy()).all()) ``` ```output Rows: 1560 · pages: 16 Requests by status: {200: 16, 429: 2, 503: 4} Simulated time: 3.9 s Equal to the source: True ``` *Fig. — Requests of the June download on the simulated clock: without and with a client-side limiter.* Two rows of marks, one per request. Without limiter: 22 requests in 3.9 seconds, with two 429 responses (highlighted) each followed by a one-second wait, and four 503s. With limiter: 18 evenly spaced requests in 2.9 seconds, no 429 and two 503s followed by backoff waits. The 16 pages took 22 requests: 2 429 responses and 4 503 responses, each one repeated. In the top row of the figure, each mark is a request on the simulated clock. The 429s arrive when the client uses up the 10 requests of the window, because it asks faster than the API allows, and each one costs a full second of `Retry-After`: of the 3.9 simulated seconds, 2 go on those two waits. The last line is the check that matters: the same rows as in the source, with no repeated times and the same closes. 429s are not free: each one is a wasted request and, in many APIs, insisting on them ends in a temporary block of the key. If the limit is known (it is in the plan's documentation or in headers such as `X-RateLimit-Remaining`), the right thing is not to reach it: a **client-side rate limiter** that spaces out the requests. The simplest one guarantees a minimum interval between two requests; with 10 per second and some margin, 9 per second, one every 111 ms. ```python class Limitador: def __init__(self, por_segundo, reloj): self.intervalo, self.reloj = 1.0 / por_segundo, reloj self.ultima = -1e9 def esperar(self): falta = self.ultima + self.intervalo - self.reloj.ahora() if falta > 0: self.reloj.dormir(falta) self.ultima = self.reloj.ahora() limitador = Limitador(9, reloj) pedir_sin_limite = pedir def pedir(ruta, **params): # same signature, now spaced out limitador.esperar() return pedir_sin_limite(ruta, **params) inicio, t0 = len(api.registro), reloj.ahora() filas_lim = descargar(**JUNIO) estados = Counter(e for _, e in api.registro[inicio:]) print("Rows:", len(filas_lim)) print("Requests by status:", dict(sorted(estados.items()))) print(f"Simulated time: {reloj.ahora() - t0:.1f} s") pedir = pedir_sin_limite # restored as it was ``` ```output Rows: 1560 Requests by status: {200: 16, 503: 2} Simulated time: 2.9 s ``` Not a single 429: 18 requests instead of 22. And, although each request waits its turn, the download finishes sooner, in 2.9 simulated seconds versus 3.9, because it no longer pays the `Retry-After` second of each 429 (the bottom row of the figure). The 503s are still there, two this time, because they do not depend on the client's pace: the limiter avoids the errors you cause yourself and the retries cover the server's. You need both. ## A download that can be interrupted A download of years of one-minute bars means thousands of pages and hours of wall-clock time. At some point it will be cut off: a reboot, the network, an error that runs out after six attempts. Repeating it in full is slow and, if the data is charged per request (lesson 1.3.4), expensive. The solution is to make it **resumable** and **idempotent**: repeating a step that is already done does not change the result. - Each page is saved to its own file with a fixed name (`p0007.json`), **atomically**: it is written to a temporary file and renamed with `os.replace`, which never leaves a half-written file. - After the page, the state is saved: page number and next cursor. If the process dies between the two writes, on resume the same page is requested again and the same file is overwritten with the same content. ```python import os from pathlib import Path def escribir_atomico(ruta, objeto): tmp = ruta.with_suffix(".tmp") tmp.write_text(json.dumps(objeto)) os.replace(tmp, ruta) # atomic on the same disk def descargar_reanudable(carpeta, fallar_en=None, **params): carpeta = Path(carpeta) carpeta.mkdir(exist_ok=True) f_estado = carpeta / "estado.json" estado = (json.loads(f_estado.read_text()) if f_estado.exists() else {"pagina": 0, "cursor": None, "fin": False}) while not estado["fin"]: if estado["pagina"] == fallar_en: raise ConnectionError("simulated network outage") extra = {"cursor": estado["cursor"]} if estado["cursor"] else {} pag = pedir_con_reintentos("/v1/barras/NOVA", **params, **extra) n = estado["pagina"] + 1 escribir_atomico(carpeta / f"p{n:04d}.json", pag["datos"]) estado = {"pagina": n, "cursor": pag["siguiente"], "fin": pag["siguiente"] is None} escribir_atomico(f_estado, estado) return estado ``` The first run is cut off on purpose after the sixth page: ```python raises descargar_reanudable("junio", fallar_en=6, **JUNIO) ``` ```output ConnectionError: simulated network outage ``` ```python print("Pages on disk:", len(list(Path("junio").glob("p*.json")))) antes = len(api.registro) print(descargar_reanudable("junio", **JUNIO)) print("Requests on resume:", len(api.registro) - antes) antes = len(api.registro) descargar_reanudable("junio", **JUNIO) # already finished print("Requests on the third run:", len(api.registro) - antes) paginas = sorted(Path("junio").glob("p*.json")) junio = pd.concat(pd.DataFrame(json.loads(p.read_text())) for p in paginas) print("Rows:", len(junio), "· repeated:", junio["t"].duplicated().sum()) ``` ```output Pages on disk: 6 {'pagina': 16, 'cursor': None, 'fin': True} Requests on resume: 12 Requests on the third run: 0 Rows: 1560 · repeated: 0 ``` The second run starts at page 7, without repeating the six that were already there: 12 requests for the 10 pages that were missing (two 503s along the way). The third does not make a single request. The final set has the 1,560 bars with no duplicates. The same idea, at data-lake scale (one partition per day or per month, and a record of which partitions are complete), is the basis of the incremental update in lesson 1.4.8. ## Streaming: sequence, heartbeats and reconnection A stream is not paginated: it arrives. Its failures are different and more treacherous, because they raise no error: - **Lost messages.** A message that does not arrive (a server reconnection, an overflowing queue) gives no warning. That is why serious feeds number every message: a jump in the sequence gives the loss away. - **Dead connection.** A cable that is cut in the middle does not always close the TCP connection: the client keeps waiting for data that will never come, and a quiet market looks identical to a broken connection. To tell them apart, the server sends **heartbeats** (empty messages every few seconds; the websocket protocol also has its own *ping* and *pong* frames). If nothing arrives, neither data nor heartbeats, within a deadline, the connection is dead. ```mermaid stateDiagram-v2 [*] --> Connecting Connecting --> Receiving: subscribed Receiving --> Receiving: seq n+1 or heartbeat Receiving --> Reconnecting: seq gap Receiving --> Reconnecting: silence Reconnecting --> Connecting: from last seq Receiving --> [*]: end ``` The server in this section uses `asyncio.start_server`, a TCP server from the standard library on the same event loop as the client. The messages are JSON lines (a real websocket adds a framing protocol on top, but the sequence and heartbeat logic is the same). It emits NOVA's first 40 trades of June 2, with a heartbeat every eight, and fails in two ways: on the first connection message 17 is “lost” and on the second it goes silent after 30 without closing the connection. ```python import asyncio ops = pd.read_parquet("nova_operaciones.parquet").head(40) MENSAJES = [{"tipo": "op", "seq": i + 1, "precio": r.precio, "tamano": int(r.tamano)} for i, r in enumerate(ops.itertuples())] conexiones = 0 async def atender(lector, escritor): global conexiones conexiones += 1 desde = json.loads(await lector.readline())["desde"] salida = [] for m in MENSAJES[desde - 1:]: if conexiones == 1 and m["seq"] == 17: continue # lost message if conexiones == 2 and m["seq"] == 31: break # goes silent, no "fin" salida.append(m) if m["seq"] % 8 == 0: salida.append({"tipo": "latido"}) else: salida.append({"tipo": "fin"}) escritor.write("".join(json.dumps(m) + "\n" for m in salida).encode()) await escritor.drain() if conexiones == 2: await mudo.wait() # connection open and silent escritor.close() ``` The client keeps track of the last good sequence number. With each message it checks that it is the next one; if not, it discards it, closes and reconnects asking from the missing one. The read has a deadline (`asyncio.wait_for`, 50 ms here; in a real feed, a few times the heartbeat interval): if it expires, the connection is considered dead. ```python async def consumir(puerto, plazo=0.05): recibidos, eventos, ultima, k = [], [], 0, 0 while True: k += 1 lector, escritor = await asyncio.open_connection( "127.0.0.1", puerto) escritor.write((json.dumps({"desde": ultima + 1}) + "\n").encode()) await escritor.drain() eventos.append(f"connection {k}: from seq {ultima + 1}") try: while True: linea = await asyncio.wait_for(lector.readline(), plazo) m = json.loads(linea) if m["tipo"] == "fin": escritor.close() return recibidos, eventos if m["tipo"] == "latido": continue if m["seq"] != ultima + 1: eventos.append(f" gap: expected {ultima + 1}" f", got {m['seq']}") break recibidos.append(m) ultima = m["seq"] except TimeoutError: eventos.append(f" silence after seq {ultima}") escritor.close() mudo = asyncio.Event() srv = await asyncio.start_server(atender, "127.0.0.1", 0) puerto = srv.sockets[0].getsockname()[1] recibidos, eventos = await consumir(puerto) mudo.set() # releases the silent connection srv.close() await srv.wait_closed() print("\n".join(eventos)) seqs = [m["seq"] for m in recibidos] print("Received:", len(seqs), "· complete:", seqs == list(range(1, 41))) ``` ```output connection 1: from seq 1 gap: expected 17, got 18 connection 2: from seq 17 silence after seq 30 connection 3: from seq 31 Received: 40 · complete: True ``` *Fig. — The 40 stream messages by connection: a sequence gap, a silence and two reconnections.* Three rows, one per connection, with a mark per message at its sequence number. The first connection receives 1 to 16 and then 18 (highlighted), discarded because 17 is missing. The second resumes at 17 and reaches 30; then it goes silent. The third resumes at 31 and ends at 40. Small dots mark the heartbeats, one every eight messages. Three connections for 40 messages. On the first, the client sees 18 arrive when it expected 17, discards it and reconnects from 17. On the second, the server goes silent after 30; the heartbeats do not arrive either, so after 50 ms the client decides the connection is dead and opens another one from 31. The result is the 40 messages, in order and without repeats. Without the sequence check, the client would have carried on one message short (17 missing) and with a 5-minute bar with a false volume; without the deadline, it would have waited forever. Many feeds do not allow resuming from an arbitrary sequence number. Then recovery goes through REST: you request the missing range (the trades between two times) or a full snapshot of the state (the entire order book) and continue with the stream from there. It is the pattern from lesson 1.1.3 for rebuilding an order book: snapshot plus messages, and a new snapshot if the messages have a gap. ## The real libraries In a real project you do not need to write the retry loop by hand. With `requests`, retrying with backoff is done by `urllib3` underneath, configured with a `Retry` object mounted on the session. It honors `Retry-After` and only retries the codes in the list. Its jitter is another variant: it adds to the exponential wait a random amount between 0 and `backoff_jitter`, instead of sampling the entire wait. This block is not run in the lesson (`requests` is not in the course environment and the URL is an example): ```python skip title="descarga_requests.py" import requests from requests.adapters import HTTPAdapter from urllib3.util import Retry reintentos = Retry( total=6, backoff_factor=0.5, # waits of 0, 1, 2, 4… s backoff_max=30, # cap backoff_jitter=0.5, # + U(0, 0.5) s on each wait status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET"], # safe methods only respect_retry_after_header=True, ) sesion = requests.Session() sesion.mount("https://", HTTPAdapter(max_retries=reintentos)) sesion.headers["Authorization"] = "Bearer " url = "https://api.proveedor.example/v1/barras/NOVA" params = {"desde": "2025-06-02", "hasta": "2025-07-01"} while url: r = sesion.get(url, params=params, timeout=10) r.raise_for_status() # 4xx: permanent failure pagina = r.json() ... # save the page url, params = pagina.get("next_url"), None ``` Many providers return, as in the last line, a full URL for the next page (`next_url`) instead of a bare cursor; the logic does not change. For streaming, the `websockets` library handles the *ping* and *pong* frames (`ping_interval`, `ping_timeout`) and, used as an iterator, reconnects on its own with backoff. The sequence check is still your job: ```python skip title="stream_websockets.py" import asyncio, json from websockets.asyncio.client import connect async def stream(url): ultima = 0 async for ws in connect(url, ping_interval=20, ping_timeout=20): await ws.send(json.dumps({"suscribir": "NOVA", "desde": ultima + 1})) async for texto in ws: m = json.loads(texto) if m["seq"] != ultima + 1: break # gap: reconnect or request via REST ultima = m["seq"] ... # process the message asyncio.run(stream("wss://stream.proveedor.example/v1")) ``` ## Exercises 1. Set `prob_fallo` to 0.3 in a new `ApiMercado` (with a new server or by replacing `api`) and repeat the June download. How many requests does it take now, and how much simulated time? Does any retry run out? **Solution.** ```python api = ApiMercado(reloj, prob_fallo=0.3, semilla=3) t0 = reloj.ahora() filas_30 = descargar(**JUNIO) estados = Counter(e for _, e in api.registro) print("Rows:", len(filas_30)) print("Requests by status:", dict(sorted(estados.items()))) print(f"Simulated time: {reloj.ahora() - t0:.1f} s") ``` ```output Rows: 1560 Requests by status: {200: 16, 503: 7} Simulated time: 3.1 s ``` The server's handler looks up `api` in the lesson's namespace every time it serves a request, so replacing it is enough. With 30% failures the download finishes just the same, with the 1,560 rows, and no retry runs out: 23 requests, 7 of them 503s. No 429s, because the backoff waits slow the client down enough not to hit the window limit. The probability of using up six attempts in a row is 0.3⁶ ≈ 0.07% per page: over 16 pages it is unlikely; over 10,000 pages it would happen about seven times. That is why a long download also has to be resumable. 2. Write a version of `espera_backoff` with *equal jitter*: half of the ceiling fixed and the other half random (`ceiling / 2 + U(0, ceiling / 2)`). Compare its mean wait and its minimum wait with those of *full jitter* on attempt 5. **Solution.** ```python def espera_equal(n, base=0.5, tope=30.0, azar=random): techo = min(tope, base * 2 ** n) return techo / 2 + azar.uniform(0, techo / 2) azar = random.Random(3) full = [espera_backoff(4, azar=azar) for _ in range(10_000)] equal = [espera_equal(4, azar=azar) for _ in range(10_000)] for nombre, x in [("full", full), ("equal", equal)]: print(f"{nombre:>5}: mean {sum(x) / len(x):.2f} s, " f"min {min(x):.2f} s, max {max(x):.2f} s") ``` ```output full: mean 4.00 s, min 0.00 s, max 8.00 s equal: mean 6.00 s, min 4.00 s, max 8.00 s ``` On the fifth attempt (`n = 4`) the ceiling is 8 s. *Full jitter* waits 4 s on average and can retry almost immediately; *equal jitter* waits 6 s on average and never less than 4. The first spreads the retries out more and finishes sooner on average; the second guarantees a minimum pause, useful if the server penalizes retries that come too close together. In Brooker's simulation, *equal jitter* did somewhat more work than *full jitter* and took considerably longer. 3. A provider paginates by offset (`offset`) and, while you download the current day, appends new bars at the end. Are rows repeated or lost? And if it corrects (deletes and reinserts) a bar on a page you have already downloaded? Propose how to detect it after the download. **Solution.** If the new bars are appended at the end, the positions of the existing ones do not change: nothing is repeated or lost (only the bars added after the last page are missing, and they will come in with the next update). If a bar on a page already downloaded is deleted and reinserted, and the reinsertion puts it at the end or in another position, everything behind it shifts: the next page skips one row or repeats another. To detect it, after the download you check that the key (the bar's time) is unique and that no times expected according to the session calendar are missing (lesson 1.4.2), and compare the number of rows with the total the API reports, if it gives one. ## Self-assessment 1. An API returns `503 Service Unavailable` on page 7 of 16. What should the client do? - A. Wait for a growing time with some randomness and repeat the same request, with the same cursor - B. Abort the download: the data is corrupt - C. Skip to page 8 so as not to waste time - D. Repeat the request immediately in a loop until it works *Answer A: Wait for a growing time with some randomness and repeat the same request, with the same cursor.* A 5xx is a transient server failure: the same request may work later. It is repeated with the same cursor (nothing is skipped) after a wait with exponential backoff and jitter. Retrying without waiting worsens the overload that probably caused the error. 2. Why is jitter (randomness) added to the exponential backoff wait? - A. To make the code look less deterministic - B. Because many clients that failed at the same time would retry at the same time and overload the server again; randomness spreads the retries out over time - C. Because `time.sleep` does not accept integers - D. To make the average wait longer *Answer B: Because many clients that failed at the same time would retry at the same time and overload the server again; randomness spreads the retries out over time.* Without jitter, all the clients that got the error at the same instant wait exactly the same and arrive together again: a synchronized stampede. With a uniform wait between 0 and the exponential cap, the retries are spread out. 3. An offset-based download (`offset=200&limit=100`) requests page 3 right after the provider inserts 5 corrected bars at the start of the range. What happens? - A. Nothing: the offset is stable - B. The server returns a 400 - C. Page 3 starts 5 rows earlier than it should and repeats rows from page 2 - D. The 5 new bars are lost and nothing else *Answer C: Page 3 starts 5 rows earlier than it should and repeats rows from page 2.* The offset counts positions in a set that has changed: the 5 inserted rows push everything forward, and the next page repeats the last 5 of the previous one (or skips rows, if rows were deleted). A cursor points to the last key seen and does not depend on positions. 4. A streaming client receives the messages with sequence 15, 16 and 18. What is the right thing to do? - A. Process 18 and move on - B. Wait for 17 to arrive on its own - C. Reorder the messages by time - D. Discard 18, flag the gap and recover from 17 (reconnect with resume, or a REST request for that range) *Answer D: Discard 18, flag the gap and recover from 17 (reconnect with resume, or a REST request for that range).* A jump in the sequence means a message was lost. Processing 18 leaves an incorrect state (order book, position, bar) without knowing it. The safe course is to recover the missing range before continuing. 5. The resumable download saves the page first and then the state file with the cursor. If the process dies between the two writes, what happens on resume? - A. The same page is requested again and the same file is overwritten with the same content: no duplicates - B. A page is lost - C. The page is duplicated - D. The download can no longer be resumed *Answer A: The same page is requested again and the same file is overwritten with the same content: no duplicates.* The state still points to the previous cursor, so the same page is requested again. Since each page's file has a fixed name and is written atomically, repeating the step does not change the result: it is idempotent. ## Key points - A data API paginates, rate-limits and fails. The client has to account for all three from the first line. - 429, 5xx and network errors are retried; 400, 401, 403 and 404 are not. If `Retry-After` arrives, it takes precedence over the backoff. - Exponential backoff with a cap and jitter: a uniform wait between 0 and `min(cap, base · 2^n)`. Without randomness, clients retry in sync. - Paginate by cursor, save each page with a fixed name and an atomic write, and the cursor afterwards: that way an interruption halfway resumes without gaps or duplicates. - In a stream, each message carries a sequence number and the server sends heartbeats: a sequence jump is a lost message and a long silence, a dead connection. Both are solved by reconnecting from the last good message. ## Resources ### Documentation - [RFC 9110, HTTP Semantics: status codes and the Retry-After header](https://www.rfc-editor.org/rfc/rfc9110.html) · rfc-editor.org - [RFC 6585: the 429 Too Many Requests status code](https://www.rfc-editor.org/rfc/rfc6585.html) · rfc-editor.org - [RFC 6455, The WebSocket Protocol: ping and pong frames](https://www.rfc-editor.org/rfc/rfc6455.html) · rfc-editor.org - [Marc Brooker (AWS): Exponential Backoff And Jitter](https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/) · aws.amazon.com - [urllib3: Retry (retries and backoff for requests)](https://urllib3.readthedocs.io/en/stable/reference/urllib3.util.html) · urllib3.readthedocs.io - [Python: urllib.request and http.server](https://docs.python.org/3/library/urllib.request.html) · docs.python.org ### Data - [Level 1 data: NOVA 5-minute bars (Parquet)](https://pythonparatrading.com/curso/datos/nivel-1/nova_barras_5min.parquet) · Download · nova_barras_5min.parquet Previous lesson: [1.3.8 · How to choose a market data provider](https://pythonparatrading.com/en/python-trading-course/level-1/how-to-choose-a-data-provider/) Next lesson: [1.4.2 · Data cleaning: missing values, bad ticks, gaps and time zones](https://pythonparatrading.com/en/python-trading-course/level-1/cleaning-financial-market-data/)