--- title: "1.4.1 Descarga por REST y websockets, con reintentos" description: "Paginación por cursor, límites de peticiones, reintentos con backoff y jitter, descarga reanudable y streams con latidos y huecos, en un servidor local." url: "https://pythonparatrading.com/curso-python-trading/nivel-1/descarga-rest-websockets-y-reintentos/" lang: "es" image: "https://pythonparatrading.com/og/curso-python-trading/nivel-1/descarga-rest-websockets-y-reintentos.png" --- # 1.4.1 · Descarga por REST y websockets: paginación, límites y reintentos > Bajar datos de una API parece una línea de código hasta que la API corta a la página doce. Esta lección construye un servidor local que se comporta como una API de mercado real —paginación, límites, errores aleatorios, un stream que pierde mensajes— y un cliente que sobrevive a todo ello. - Curso: [Nivel 1: Datos financieros, mercados y análisis](https://pythonparatrading.com/curso-python-trading/nivel-1/) - Módulo: 1.4 · Ingeniería de datos financieros - Duración: 60 min - Requisitos: [0.4.6 · Concurrencia: threading, multiprocessing, asyncio y free-threading](https://pythonparatrading.com/curso-python-trading/nivel-0/concurrencia-threading-multiprocessing-asyncio/), [1.3.3 · APIs de datos retail y prosumer](https://pythonparatrading.com/curso-python-trading/nivel-1/apis-de-datos-retail-y-prosumer/) ## Al terminar esta lección - Recorrer una API paginada por cursor sin perder ni repetir filas, y explicar por qué el cursor es más seguro que el desplazamiento. - Distinguir los errores que se reintentan (429, 5xx, red) de los que no (400, 401, 403, 404) y respetar `Retry-After`. - Programar reintentos con backoff exponencial, tope y jitter, y medir cuánto esperan. - Hacer una descarga idempotente y reanudable: un corte a mitad no deja huecos ni duplicados. - Consumir un stream con números de secuencia y latidos: detectar huecos y conexiones muertas, y reconectar desde el último mensaje bueno. ## Dos maneras de pedir datos Casi todos los proveedores de la lección 1.3 ofrecen sus datos de dos formas. Por **REST**, el cliente pide un tramo («barras de 5 minutos de NOVA de junio») y el servidor responde con un documento JSON; si el tramo es grande, lo reparte en páginas. Por **streaming** (websockets, en la mayoría de las APIs retail), el cliente abre una conexión que se queda abierta y el servidor le empuja cada operación o cada cotización en cuanto ocurre. ```mermaid flowchart LR C[Cliente] -- GET página --> A[API REST] A -- JSON + cursor --> C C -- suscripción --> W[Stream] W -- mensaje seq n --> C W -- latido --> C ``` | | REST | Streaming (websocket) | |---|---|---| | Para qué | Histórico, rellenar huecos, datos de referencia | Tiempo real: operaciones, cotizaciones, libro | | Quién inicia | El cliente, petición a petición | El servidor, mensaje a mensaje | | Unidad | Página de N filas con un cursor a la siguiente | Mensaje con número de secuencia | | Fallo típico | 429 por exceso de peticiones, 5xx, cortes a mitad | Mensajes perdidos, conexión muerta sin aviso | | Recuperación | Reintentar la misma petición con espera | Reconectar y reanudar desde el último mensaje bueno | Las dos se usan juntas. Un sistema en vivo arranca con REST (el histórico necesario para calcular sus indicadores), se engancha al stream y, cada vez que detecta un hueco en el stream, vuelve a REST para rellenarlo. Todo lo que sigue se ejecuta de verdad. En lugar de llamar a un proveedor real (necesitaría red, una clave y dependería de su humor ese día), la lección levanta un **servidor local** en un hilo que imita a una API de mercado: pagina, limita las peticiones y falla al azar. El cliente usa solo la biblioteca estándar (`urllib`); al final están las versiones con `requests` y `websockets`. ## Una API de mercado en un hilo El servidor sirve las barras de 5 minutos de NOVA (`nova_barras_5min.parquet`). Para que la lección sea reproducible y no espere de verdad, cliente y servidor comparten un **reloj simulado**: cada petición «tarda» 40 ms y cada espera del cliente adelanta ese reloj en lugar de dormir. Es la misma técnica que se usa para probar código con tiempos (inyectar el reloj); en producción, `dormir` sería `time.sleep` y cada lado mediría su propio tiempo. ```python import base64, json, random, threading import pandas as pd barras = pd.read_parquet("nova_barras_5min.parquet") class Reloj: """Reloj simulado que comparten cliente y servidor.""" 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): # opaco para el cliente: no debe construirlo 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), "barras de", barras.ts.min().date(), "a", barras.ts.max().date()) ``` ```output 38940 barras de 2024-01-02 a 2025-12-31 ``` La lógica de la API va en una clase aparte, sin nada de HTTP, para poder leerla de un vistazo. Tiene las tres reglas que tienen casi todas las APIs reales: - **Límite de peticiones.** 10 por ventana de 1 s. La undécima recibe `429 Too Many Requests` con una cabecera `Retry-After` que dice cuántos segundos esperar. Cada respuesta buena trae `X-RateLimit-Remaining`, las que quedan. - **Fallos transitorios.** Un 8 % de las peticiones recibe `503 Service Unavailable`, con un generador con semilla. - **Paginación por cursor.** Como mucho `limite` filas por respuesta y un campo `siguiente` con el cursor de la página siguiente (`None` en la última). Un `limite` mayor de 500 es un error del cliente: `400 Bad Request`. ```python class ApiMercado: POR_VENTANA = 10 # peticiones por ventana de 1 s LATENCIA = 0.04 # segundos simulados por petición 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 = [] # (instante, estado) def responder(self, p): self.reloj.dormir(self.LATENCIA) ahora = self.reloj.ahora() if ahora - self.ventana >= 1.0: # ventana nueva 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": "demasiadas"}, cab) self.usadas += 1 if self.azar.random() < self.prob_fallo: return self._fin(503, {"error": "no disponible"}) limite = int(p.get("limite", 100)) if limite > 500: return self._fin(400, {"error": "limite máximo 500"}) hasta = pd.Timestamp(p["hasta"], tz="UTC") if "cursor" in p: # después del 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 ``` Ahora la parte HTTP: un `ThreadingHTTPServer` de la biblioteca estándar, en el puerto que el sistema tenga libre y en un hilo de fondo (hilos y `asyncio`: lección 0.4.6). El manejador traduce cada `GET` a una llamada a `responder` y devuelve el JSON con su código y sus cabeceras. El número de puerto cambia en cada ejecución, así que la celda no lo imprime. ```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": "no existe"} 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): # sin una línea por petición 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("Servidor en marcha:", servidor.server_port > 0) ``` ```output Servidor en marcha: True ``` Con el servidor en marcha falta el cliente: una función de diez líneas con `urllib.request`. Un detalle de `urllib`: lanza `HTTPError` con cualquier código 4xx o 5xx. Para tratar todos los códigos igual, la función captura esa excepción y devuelve el código, las cabeceras y el cuerpo, sea cual sea el resultado. Los errores de red (conexión rechazada, tiempo agotado) sí se dejan pasar: son otra cosa y se tratan más abajo. ```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 y 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("Estado:", estado) print("Quedan:", cab["X-RateLimit-Remaining"], "peticiones") print("Filas:", len(cuerpo["datos"])) print("Primera:", cuerpo["datos"][0]) print("Cursor:", cuerpo["siguiente"]) ``` ```output Estado: 200 Quedan: 9 peticiones Filas: 100 Primera: {'t': '2025-06-02T13:30:00Z', 'o': 53.35, 'h': 53.48, 'l': 53.28, 'c': 53.46, 'v': 285081} Cursor: MjAyNS0wNi0wM1QxNToxNTowMCswMDowMA== ``` Cien barras y un cursor. El cursor es una cadena opaca: en este servidor codifica la hora de la última barra servida, pero el cliente no debe interpretarlo ni fabricarlo, solo devolverlo. Los proveedores cambian su formato sin avisar. Una petición con un parámetro inválido muestra el otro tipo de error, el que no tiene arreglo esperando: ```python estado, _, cuerpo = pedir("/v1/barras/NOVA", desde="2025-06-02", hasta="2025-07-01", limite=1000) print(estado, cuerpo) ``` ```output 400 {'error': 'limite máximo 500'} ``` ## Paginación: por qué un cursor y no un desplazamiento Hay dos maneras habituales de paginar. Por **desplazamiento** (`offset=200&limit=100`: «dame las filas 200 a 299») o por **cursor** («dame las 100 siguientes a esta clave»). Con datos de mercado el desplazamiento es frágil: si el proveedor inserta o corrige barras mientras se descarga (pasa a menudo con el día en curso o tras una corrección de la cinta), las posiciones se mueven y la página siguiente repite o se salta filas sin ningún error visible. El cursor apunta a una clave (aquí, la hora de la última barra) y no depende de las posiciones. El bucle ingenuo pide páginas hasta que `siguiente` sea `None` y da por buena cualquier respuesta. Contra esta API dura poco: ```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} en la página {n}") filas += cuerpo["datos"] cursor = cuerpo["siguiente"] if cursor is None: return filas descargar_ingenuo(**JUNIO) ``` ```output RuntimeError: HTTP 503 en la página 2 ``` Se rompe en la segunda página con un 503 que, repetido un instante después, habría funcionado. Peor que romperse sería lo contrario: un bucle que ignorase el código de estado y siguiera sin la página, con un hueco de cien barras en mitad de junio que nadie notaría hasta que un indicador saliese raro. ## Qué se reintenta y qué no Los códigos de estado HTTP dicen de quién es el problema. Los **4xx** son errores del cliente: repetir la misma petición dará el mismo error. La excepción es el **429**: la petición es correcta, pero llega demasiado pronto. Los **5xx** son del servidor y casi siempre transitorios. Los errores de red (tiempo agotado, conexión cortada) no tienen código, pero también son transitorios. | Respuesta | Significado | Qué hace el cliente | |---|---|---| | 200 | Correcto | Guarda la página y sigue con el cursor | | 400, 422 | Parámetros inválidos | Falla en el acto: hay que corregir el código | | 401, 403 | Sin credenciales o sin permiso (plan, licencia) | Falla en el acto: reintentar no da permisos | | 404 | El recurso no existe (símbolo mal escrito, ruta) | Falla en el acto | | 429 | Demasiadas peticiones | Espera lo que diga `Retry-After` y repite | | 500, 502, 503, 504 | Fallo o sobrecarga del servidor | Espera con backoff y repite | | Tiempo agotado, conexión cortada | Red | Espera con backoff y repite | El diagrama recoge el bucle completo. Lo importante está en las flechas que vuelven: la petición que se repite es la **misma**, con el mismo cursor; nunca se salta a la página siguiente. ```mermaid flowchart TB P[Pedir página] --> E{¿Estado?} E -- 200 --> G[Guardar página] G --> S{¿Hay cursor?} S -- sí --> P S -- no --> F[Fin] E -- 429 --> R[Espera Retry-After] E -- 5xx o red --> B[Esperar backoff] R --> P B --> P E -- 4xx --> X[Fallo definitivo] ``` ## Backoff exponencial, jitter y ritmo del cliente Esperar siempre lo mismo entre reintentos no basta: si el servidor está sobrecargado, todos los clientes que fallaron vuelven a la vez y lo tumban otra vez. El **backoff exponencial** dobla la espera en cada intento, con un tope para no esperar horas. El **jitter** añade azar. Marc Brooker comparó varias formas de hacerlo en el blog de arquitectura de AWS; la más sencilla, la *full jitter*, que está entre las que menos peticiones totales generan en su simulación, espera un tiempo uniforme entre cero y el techo exponencial: $$ \text{espera}_n \sim \mathcal{U}\big(0,\ \min(\text{tope},\ \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 ``` La figura de la cabecera dibuja la tabla: el techo de cada intento, cinco esperas sorteadas y la media. El techo se dobla hasta chocar con el tope de 30 s en el séptimo intento; a partir de ahí se queda plano. La espera media sale muy cerca de la mitad del techo (0,24 s con techo 0,5; 14,99 s con techo 30), porque esa es la media de una uniforme entre 0 y el techo. Con seis intentos hay cinco esperas: el peor caso suma 15,5 s (0,5 + 1 + 2 + 4 + 8) y el caso medio, la mitad. Es el orden de magnitud razonable para un script de descarga nocturno; un sistema en vivo necesita topes mucho más cortos y, sobre todo, saber qué hacer mientras no tiene datos. La función con reintentos combina todo lo anterior: reintenta solo lo reintentable, respeta `Retry-After` cuando llega (si el servidor dice cuánto esperar, se le hace caso) y se rinde tras un número de intentos. El reloj y el generador de azar entran como argumentos: así el mismo código sirve en producción (`time.sleep`, azar real) y en una prueba reproducible. ```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 # red 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 # sin espera tras el último espera = espera_backoff(n, azar=azar) if cab.get("Retry-After"): espera = max(espera, float(cab["Retry-After"])) dormir(espera) raise RuntimeError(f"sin respuesta tras {intentos} intentos") ``` La descarga completa es el bucle ingenuo con la función nueva. Antes de empezar se anota dónde está el registro del servidor, para contar después cuántas peticiones de cada tipo hicieron falta: ```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("Filas:", len(filas), "· páginas:", -(-len(filas) // 100)) print("Peticiones por estado:", dict(sorted(estados.items()))) print(f"Tiempo simulado: {reloj.ahora() - t0:.1f} s") # lo descargado tiene que ser exactamente lo que hay en el origen 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("Iguales al origen:", len(df) == len(origen) and df["t"].is_unique and (df["c"].to_numpy() == origen["cierre"].to_numpy()).all()) ``` ```output Filas: 1560 · páginas: 16 Peticiones por estado: {200: 16, 429: 2, 503: 4} Tiempo simulado: 3.9 s Iguales al origen: True ``` *Fig. — Peticiones de la descarga de junio en el reloj simulado: sin limitador del cliente y con él.* Dos filas de marcas, una por petición. Sin limitador: 22 peticiones en 3,9 segundos, con dos respuestas 429 (destacadas) seguidas de una espera de un segundo y cuatro 503. Con limitador: 18 peticiones regularmente espaciadas en 2,9 segundos, sin ningún 429 y con dos 503 seguidos de esperas de backoff. Las 16 páginas han necesitado 22 peticiones: 2 respuestas 429 y 4 respuestas 503, cada una repetida. En la fila de arriba de la figura, cada marca es una petición en el reloj simulado. Los 429 llegan cuando el cliente agota las 10 peticiones de la ventana, porque pide más deprisa de lo que la API permite, y cada uno cuesta un segundo entero de `Retry-After`: de los 3,9 s simulados, 2 se van en esas dos esperas. La última línea es la comprobación que importa: mismas filas que en el origen, sin horas repetidas y con los mismos cierres. Los 429 no son gratis: cada uno es una petición perdida y, en muchas APIs, insistir con ellas acaba en un bloqueo temporal de la clave. Si se conoce el límite (viene en la documentación del plan o en cabeceras como `X-RateLimit-Remaining`), lo correcto es no llegar a él: un **limitador del lado del cliente** que espacie las peticiones. El más simple garantiza un intervalo mínimo entre dos peticiones; con 10 por segundo y algo de margen, 9 por segundo, una cada 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): # misma firma, ahora espaciada 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("Filas:", len(filas_lim)) print("Peticiones por estado:", dict(sorted(estados.items()))) print(f"Tiempo simulado: {reloj.ahora() - t0:.1f} s") pedir = pedir_sin_limite # se deja como estaba ``` ```output Filas: 1560 Peticiones por estado: {200: 16, 503: 2} Tiempo simulado: 2.9 s ``` Sin un solo 429: 18 peticiones en lugar de 22. Y, aunque cada petición espera su turno, la descarga termina antes, en 2,9 s simulados frente a 3,9, porque ya no paga el segundo de `Retry-After` de cada 429 (la fila de abajo de la figura). Los 503 siguen ahí, dos esta vez, porque no dependen del ritmo del cliente: el limitador evita los errores que provoca uno mismo y los reintentos cubren los del servidor. Hacen falta las dos cosas. ## Una descarga que se puede cortar Una descarga de años de barras de un minuto son miles de páginas y horas de reloj. En algún momento se cortará: un reinicio, la red, un error que se agota tras seis intentos. Repetirla entera es lento y, si los datos cuestan por petición (lección 1.3.4), caro. La solución es hacerla **reanudable** e **idempotente**: repetir un paso ya hecho no cambia el resultado. - Cada página se guarda en su propio archivo con un nombre fijo (`p0007.json`), de forma **atómica**: se escribe en un temporal y se renombra con `os.replace`, que no deja nunca un archivo a medias. - Después de la página se guarda el estado: número de página y cursor siguiente. Si el proceso muere entre las dos escrituras, al reanudar se vuelve a pedir la misma página y se sobrescribe el mismo archivo con el mismo contenido. ```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) # atómico en el mismo disco 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("corte simulado de la red") 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 ``` La primera ejecución se corta a propósito después de la sexta página: ```python raises descargar_reanudable("junio", fallar_en=6, **JUNIO) ``` ```output ConnectionError: corte simulado de la red ``` ```python print("Páginas en disco:", len(list(Path("junio").glob("p*.json")))) antes = len(api.registro) print(descargar_reanudable("junio", **JUNIO)) print("Peticiones al reanudar:", len(api.registro) - antes) antes = len(api.registro) descargar_reanudable("junio", **JUNIO) # ya estaba terminada print("Peticiones en la tercera ejecución:", 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("Filas:", len(junio), "· repetidas:", junio["t"].duplicated().sum()) ``` ```output Páginas en disco: 6 {'pagina': 16, 'cursor': None, 'fin': True} Peticiones al reanudar: 12 Peticiones en la tercera ejecución: 0 Filas: 1560 · repetidas: 0 ``` La segunda ejecución empieza en la página 7, sin repetir las seis que ya estaban: 12 peticiones para las 10 páginas que faltaban (dos 503 por el camino). La tercera no hace ni una petición. El conjunto final tiene las 1.560 barras sin duplicados. La misma idea, a escala de un data lake (una partición por día o por mes, y un registro de qué particiones están completas), es la base de la actualización incremental de la lección 1.4.8. ## Streaming: secuencia, latidos y reconexión Un stream no se pagina: llega. Sus fallos son distintos y más traicioneros, porque no dan error: - **Mensajes perdidos.** Un mensaje que no llega (reconexión del servidor, cola desbordada) no avisa. Por eso los feeds serios numeran cada mensaje: un salto en la secuencia delata la pérdida. - **Conexión muerta.** Un cable que se corta en medio no siempre cierra la conexión TCP: el cliente se queda esperando datos que no vendrán, y un mercado tranquilo parece idéntico a una conexión rota. Para distinguirlos, el servidor manda **latidos** (mensajes vacíos cada pocos segundos; el protocolo websocket tiene además tramas *ping* y *pong* propias). Si no llega nada, ni datos ni latidos, durante un plazo, la conexión está muerta. ```mermaid stateDiagram-v2 [*] --> Conectando Conectando --> Recibiendo: suscrito Recibiendo --> Recibiendo: seq n+1 o latido Recibiendo --> Reconectando: hueco en seq Recibiendo --> Reconectando: silencio Reconectando --> Conectando: desde última seq Recibiendo --> [*]: fin ``` El servidor de esta sección usa `asyncio.start_server`, un servidor TCP de la biblioteca estándar en el mismo bucle de eventos que el cliente. Los mensajes son líneas JSON (un websocket real añade un protocolo de tramas encima, pero la lógica de secuencia y latidos es la misma). Emite las 40 primeras operaciones de NOVA del 2 de junio, con un latido cada ocho, y falla de dos maneras: en la primera conexión se «pierde» el mensaje 17 y en la segunda se queda mudo después del 30 sin cerrar la conexión. ```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 # mensaje perdido if conexiones == 2 and m["seq"] == 31: break # se calla, sin "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() # conexión abierta y muda escritor.close() ``` El cliente lleva la cuenta de la última secuencia buena. Con cada mensaje comprueba que sea la siguiente; si no, lo descarta, cierra y reconecta pidiendo desde la que falta. La lectura lleva un plazo (`asyncio.wait_for`, 50 ms aquí; en un feed real, unas veces el intervalo de latido): si vence, la conexión se da por muerta. ```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"conexión {k}: desde 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" hueco: esperaba {ultima + 1}" f", llegó {m['seq']}") break recibidos.append(m) ultima = m["seq"] except TimeoutError: eventos.append(f" silencio tras 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() # libera la conexión muda srv.close() await srv.wait_closed() print("\n".join(eventos)) seqs = [m["seq"] for m in recibidos] print("Recibidos:", len(seqs), "· completos:", seqs == list(range(1, 41))) ``` ```output conexión 1: desde seq 1 hueco: esperaba 17, llegó 18 conexión 2: desde seq 17 silencio tras seq 30 conexión 3: desde seq 31 Recibidos: 40 · completos: True ``` *Fig. — Los 40 mensajes del stream por conexión: un hueco en la secuencia, un silencio y dos reconexiones.* Tres filas, una por conexión, con una marca por mensaje según su número de secuencia. La primera conexión recibe del 1 al 16 y después el 18 (destacado), que se descarta porque falta el 17. La segunda reanuda en el 17 y llega al 30; después se calla. La tercera reanuda en el 31 y termina en el 40. Puntos pequeños marcan los latidos, uno cada ocho mensajes. Tres conexiones para 40 mensajes. En la primera, el cliente ve llegar el 18 cuando esperaba el 17, lo descarta y reconecta desde el 17. En la segunda, el servidor se calla después del 30; los latidos tampoco llegan, así que a los 50 ms el cliente decide que la conexión está muerta y abre otra desde el 31. El resultado son los 40 mensajes, en orden y sin repetir. Sin la comprobación de secuencia, el cliente habría seguido con el 17 de menos y una barra de 5 minutos con un volumen falso; sin el plazo, se habría quedado esperando para siempre. Muchos feeds no permiten reanudar desde una secuencia arbitraria. Entonces la recuperación es por REST: se pide el tramo que falta (las operaciones entre dos horas) o una foto completa del estado (el libro de órdenes entero) y se sigue con el stream a partir de ella. Es el patrón de la lección 1.1.3 para reconstruir un libro: foto más mensajes, y foto nueva si los mensajes tienen un hueco. ## Las bibliotecas de verdad En un proyecto real no hace falta escribir el bucle de reintentos a mano. Con `requests`, el reintento con backoff lo hace `urllib3` por debajo, configurado con un objeto `Retry` montado en la sesión. Respeta `Retry-After` y solo reintenta los códigos de la lista. Su jitter es otra variante: suma a la espera exponencial un azar entre 0 y `backoff_jitter`, en lugar de sortear la espera entera. Este bloque no se ejecuta en la lección (`requests` no está en el entorno del curso y la URL es de ejemplo): ```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, # esperas de 0, 1, 2, 4… s backoff_max=30, # tope backoff_jitter=0.5, # + U(0; 0,5) s en cada espera status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET"], # solo métodos seguros 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: fallo definitivo pagina = r.json() ... # guardar la página url, params = pagina.get("next_url"), None ``` Muchos proveedores devuelven, como en la última línea, una URL completa de la página siguiente (`next_url`) en lugar de un cursor suelto; la lógica no cambia. Para el streaming, la biblioteca `websockets` gestiona las tramas *ping* y *pong* (`ping_interval`, `ping_timeout`) y, usada como iterador, reconecta sola con backoff. La comprobación de secuencia sigue siendo cosa tuya: ```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 # hueco: reconectar o pedir por REST ultima = m["seq"] ... # procesar el mensaje asyncio.run(stream("wss://stream.proveedor.example/v1")) ``` ## Ejercicios 1. Cambia `prob_fallo` a 0,3 en una `ApiMercado` nueva (con un servidor nuevo o sustituyendo `api`) y repite la descarga de junio. ¿Cuántas peticiones hacen falta ahora y cuánto tiempo simulado? ¿Llega a agotarse algún reintento? **Solución.** ```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("Filas:", len(filas_30)) print("Peticiones por estado:", dict(sorted(estados.items()))) print(f"Tiempo simulado: {reloj.ahora() - t0:.1f} s") ``` ```output Filas: 1560 Peticiones por estado: {200: 16, 503: 7} Tiempo simulado: 3.1 s ``` El manejador del servidor busca `api` en el espacio de nombres de la lección cada vez que atiende una petición, así que basta con sustituirla. Con un 30 % de fallos la descarga termina igual, con las 1.560 filas, y ningún reintento se agota: 23 peticiones, 7 de ellas 503. Ningún 429, porque las esperas del backoff frenan al cliente lo bastante para no llegar al límite de la ventana. La probabilidad de agotar seis intentos seguidos es 0,3⁶ ≈ 0,07 % por página: en 16 páginas es improbable; en 10.000 páginas pasaría unas siete veces. Por eso una descarga larga tiene que ser además reanudable. 2. Escribe una versión de `espera_backoff` con *equal jitter*: la mitad del techo fija y la otra mitad al azar (`techo / 2 + U(0, techo / 2)`). Compara su espera media y su espera mínima con las de la *full jitter* en el intento 5. **Solución.** ```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}: media {sum(x) / len(x):.2f} s, " f"mínima {min(x):.2f} s, máxima {max(x):.2f} s") ``` ```output full: media 4.00 s, mínima 0.00 s, máxima 8.00 s equal: media 6.00 s, mínima 4.00 s, máxima 8.00 s ``` En el quinto intento (`n = 4`) el techo es 8 s. La *full jitter* espera de media 4 s y puede reintentar casi en el acto; la *equal jitter* espera de media 6 s y nunca menos de 4. La primera dispersa más los reintentos y termina antes de media; la segunda garantiza una pausa mínima, útil si el servidor castiga los reintentos muy seguidos. En la simulación de Brooker, la *equal jitter* hacía algo más de trabajo que la *full jitter* y tardaba bastante más. 3. Un proveedor pagina por desplazamiento (`offset`) y, mientras descargas el día en curso, añade barras nuevas al final. ¿Se repiten o se pierden filas? ¿Y si corrige (borra y vuelve a insertar) una barra de una página que ya has bajado? Propón cómo detectarlo después de la descarga. **Solución.** Si las barras nuevas se añaden al final, las posiciones de las ya existentes no cambian: no se repite ni se pierde nada (solo faltan las añadidas después de la última página, que entrarán en la siguiente actualización). Si se borra y se reinserta una barra de una página ya bajada, y la reinserción la deja al final o en otra posición, todo lo que había detrás se desplaza: la página siguiente se salta una fila o repite otra. Para detectarlo, después de la descarga se comprueba que la clave (la hora de la barra) sea única y que no falten horas esperadas según el calendario de la sesión (lección 1.4.2), y se compara el número de filas con el total que declare la API, si lo da. ## Autoevaluación 1. Una API devuelve `503 Service Unavailable` en la página 7 de 16. ¿Qué debe hacer el cliente? - A. Abortar la descarga: los datos están corruptos - B. Esperar un tiempo creciente con algo de azar y repetir la misma petición, con el mismo cursor - C. Saltar a la página 8 para no perder tiempo - D. Repetir la petición inmediatamente en bucle hasta que funcione *Respuesta B: Esperar un tiempo creciente con algo de azar y repetir la misma petición, con el mismo cursor.* Un 5xx es un fallo transitorio del servidor: la misma petición puede funcionar después. Se repite con el mismo cursor (no se salta nada) tras una espera con backoff exponencial y jitter. Reintentar sin esperar empeora la sobrecarga que probablemente causó el error. 2. ¿Por qué se añade jitter (azar) a la espera del backoff exponencial? - A. Para que el código parezca menos determinista - B. Porque muchos clientes que fallaron a la vez reintentarían a la vez y volverían a saturar el servidor; el azar reparte los reintentos en el tiempo - C. Porque `time.sleep` no admite números enteros - D. Para que la espera media sea más larga *Respuesta B: Porque muchos clientes que fallaron a la vez reintentarían a la vez y volverían a saturar el servidor; el azar reparte los reintentos en el tiempo.* Sin jitter, todos los clientes que recibieron el error en el mismo instante esperan exactamente lo mismo y llegan juntos otra vez: una avalancha sincronizada. Con espera uniforme entre 0 y el tope exponencial, los reintentos se dispersan. 3. Una descarga por desplazamiento (`offset=200&limit=100`) pide la página 3 justo después de que el proveedor inserte 5 barras corregidas al principio del rango. ¿Qué pasa? - A. Nada: el desplazamiento es estable - B. La página 3 empieza 5 filas antes de lo que debería y repite filas de la página 2 - C. El servidor devuelve un 400 - D. Se pierden las 5 barras nuevas y nada más *Respuesta B: La página 3 empieza 5 filas antes de lo que debería y repite filas de la página 2.* El desplazamiento cuenta posiciones en un conjunto que ha cambiado: las 5 filas insertadas empujan todo hacia delante y la página siguiente repite las 5 últimas de la anterior (o se salta filas, si se borraron). El cursor apunta a la última clave vista y no depende de las posiciones. 4. Un cliente de streaming recibe los mensajes con secuencia 15, 16 y 18. ¿Qué es lo correcto? - A. Procesar el 18 y seguir - B. Descartar el 18, marcar el hueco y recuperar desde el 17 (reconexión con reanudación o petición REST del tramo) - C. Esperar a que el 17 llegue solo - D. Reordenar los mensajes por hora *Respuesta B: Descartar el 18, marcar el hueco y recuperar desde el 17 (reconexión con reanudación o petición REST del tramo).* Un salto en la secuencia significa que se perdió un mensaje. Procesar el 18 deja un estado (libro, posición, barra) incorrecto sin saberlo. Lo seguro es recuperar el tramo que falta antes de seguir. 5. La descarga reanudable guarda primero la página y después el archivo de estado con el cursor. Si el proceso muere entre las dos escrituras, ¿qué ocurre al reanudar? - A. Se pierde una página - B. Se vuelve a pedir la misma página y se sobrescribe el mismo archivo con el mismo contenido: no hay duplicados - C. Se duplica la página - D. La descarga ya no se puede reanudar *Respuesta B: Se vuelve a pedir la misma página y se sobrescribe el mismo archivo con el mismo contenido: no hay duplicados.* El estado todavía apunta al cursor anterior, así que se pide otra vez la misma página. Como el archivo de cada página tiene un nombre fijo y se escribe de forma atómica, repetir el paso no cambia el resultado: es idempotente. ## Ideas clave - Una API de datos pagina, limita y falla. El cliente tiene que contar con las tres cosas desde la primera línea. - Se reintentan 429, 5xx y errores de red; no se reintentan 400, 401, 403 ni 404. Si llega `Retry-After`, manda sobre el backoff. - Backoff exponencial con tope y jitter: espera uniforme entre 0 y `min(tope, base · 2^n)`. Sin azar, los clientes reintentan sincronizados. - Paginar por cursor, guardar cada página con nombre fijo y escritura atómica, y el cursor después: así un corte a mitad se reanuda sin huecos ni duplicados. - En un stream, cada mensaje lleva secuencia y el servidor manda latidos: un salto de secuencia es un mensaje perdido y un silencio largo, una conexión muerta. Las dos cosas se resuelven reconectando desde el último mensaje bueno. ## Recursos ### Documentación - [RFC 9110, HTTP Semantics: códigos de estado y cabecera Retry-After](https://www.rfc-editor.org/rfc/rfc9110.html) · rfc-editor.org - [RFC 6585: el código 429 Too Many Requests](https://www.rfc-editor.org/rfc/rfc6585.html) · rfc-editor.org - [RFC 6455, The WebSocket Protocol: tramas ping y pong](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 (reintentos y backoff para requests)](https://urllib3.readthedocs.io/en/stable/reference/urllib3.util.html) · urllib3.readthedocs.io - [Python: urllib.request y http.server](https://docs.python.org/3/library/urllib.request.html) · docs.python.org ### Datos - [Datos del nivel 1: barras de 5 minutos de NOVA (Parquet)](https://pythonparatrading.com/curso/datos/nivel-1/nova_barras_5min.parquet) · Descarga · nova_barras_5min.parquet Lección anterior: [1.3.8 · Cómo elegir proveedor de datos](https://pythonparatrading.com/curso-python-trading/nivel-1/como-elegir-proveedor-de-datos/) Lección siguiente: [1.4.2 · Limpieza de datos: faltantes, ticks erróneos, huecos y zonas horarias](https://pythonparatrading.com/curso-python-trading/nivel-1/limpieza-de-datos-financieros/)