Descarga por REST y websockets: paginación, límites y reintentos.

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())

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

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)

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"])

estado, _, cuerpo = pedir("/v1/barras/NOVA", desde="2025-06-02",
                          hasta="2025-07-01", limite=1000)
print(estado, cuerpo)

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)

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

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")

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())

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

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

descargar_reanudable("junio", fallar_en=6, **JUNIO)

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())

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()

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)))

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 <tu clave>"

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

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"))