--- title: "1.4.8 Pipeline de datos reproducible" description: "Pipeline de datos diario: descarga, validación con el calendario de la NYSE, cuarentena, carga incremental idempotente con manifiesto y tests con pytest." url: "https://pythonparatrading.com/curso-python-trading/nivel-1/pipeline-de-datos-reproducible/" lang: "es" image: "https://pythonparatrading.com/og/curso-python-trading/nivel-1/pipeline-de-datos-reproducible.png" --- # 1.4.8 · Pipeline reproducible: descarga, validación, actualización incremental y tests de calidad > Un sistema que opera cada día necesita datos nuevos cada día, y un dato malo que entra sin control acaba en una señal, en una orden y en una pérdida. Esta lección construye un pipeline pequeño y completo sobre ficheros diarios de NOVA con defectos a propósito: los descarga, los valida contra el calendario real de la NYSE, aparta los malos, carga los buenos sin duplicar nada y comprueba el resultado con tests. - 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: [1.4.7 · Almacenamiento: CSV, Parquet, particionado, DuckDB, TimescaleDB y ArcticDB](https://pythonparatrading.com/curso-python-trading/nivel-1/almacenamiento-parquet-duckdb-y-bases-temporales/), [0.4.5 · Logging profesional y testing con pytest](https://pythonparatrading.com/curso-python-trading/nivel-0/logging-y-testing-con-pytest/), [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/) ## Al terminar esta lección - Separar un pipeline de datos en pasos (descargar, validar, cargar, informar) con un estado que se pueda inspeccionar. - Escribir validaciones explícitas de esquema, orden, duplicados, coherencia OHLC y sesiones completas según `exchange_calendars`. - Hacer la carga incremental e idempotente con un manifiesto de huellas, y explicar por qué una marca de agua sola no basta. - Escribir tests de calidad sobre los datos ya cargados y ejecutarlos con pytest. - Situar la orquestación (cron, Prefect) y lo que añade frente a un script. ## Qué hace un pipeline de datos Las lecciones anteriores del módulo trataron cada pieza por separado: descargar (1.4.1), limpiar (1.4.2), ajustar (1.4.3), construir barras (1.4.5) y guardar (1.4.7). Un sistema que opera cada día necesita juntarlas en un proceso que se ejecute solo, todos los días, y que haga siempre lo mismo. Eso es un **pipeline**, y tiene que cumplir tres condiciones: - **Reproducible.** Con las mismas entradas da el mismo resultado. Nada depende de la hora a la que se ejecuta, del orden en que llegan los ficheros o de lo que quedó en memoria de la vez anterior. - **Idempotente.** Ejecutarlo dos veces seguidas no cambia nada la segunda vez. Es lo que permite reintentar sin miedo después de un fallo a medias. - **Vigilante.** Lo que entra se comprueba contra reglas escritas, lo que no las cumple se aparta y queda constancia de todo. ```mermaid flowchart TB K[Calendario NYSE] --> D[Descargar] D --> C[crudo/] C --> V{¿Válida?} V -- no --> Q[cuarentena/] V -- sí --> L[lago/ Parquet] L --> M[Manifiesto] M --> I[Informe y tests] Q --> I ``` El calendario va primero porque decide qué sesiones **debería** haber. Todo lo demás se compara con él. ## Un proveedor con defectos Para tener algo que vigilar, la lección fabrica un proveedor: una carpeta con un CSV por sesión, el formato en que muchos proveedores entregan datos diarios, con las barras de 5 minutos de NOVA de 18 sesiones entre el 16 de junio y el 11 de julio de 2025. La ventana tiene dos trampas reales del calendario: el 19 de junio (Juneteenth) y el 4 de julio la NYSE cierra, y el 3 de julio es una media sesión que termina a las 13:00 de Nueva York. Y cuatro defectos puestos a propósito, como los que llegan de verdad: ```python import hashlib import json from pathlib import Path import exchange_calendars as xcals import pandas as pd nyse = xcals.get_calendar("XNYS") barras = pd.read_parquet("nova_barras_5min.parquet") ny = barras["ts"].dt.tz_convert("America/New_York") barras["sesion"] = ny.dt.strftime("%Y-%m-%d") PROVEEDOR = Path("proveedor") PROVEEDOR.mkdir() def publicar(sesion, df): df.drop(columns="sesion").to_csv( PROVEEDOR / f"NOVA_{sesion}.csv", index=False, date_format="%Y-%m-%dT%H:%M:%SZ") ventana = nyse.sessions_in_range("2025-06-16", "2025-07-11") for s in ventana.strftime("%Y-%m-%d"): dia = barras[barras["sesion"] == s].reset_index(drop=True) if s == "2025-06-24": # fila duplicada dia = pd.concat([dia, dia.iloc[[10]]]).sort_index() if s == "2025-06-26": # falta una barra dia = dia.drop(index=40) if s == "2025-07-01": # máximo por debajo del cierre dia.loc[5, "maximo"] = dia.loc[5, "cierre"] - 0.05 if s == "2025-07-08": # todavía no publicada continue publicar(s, dia) print(len(ventana), "sesiones en la ventana;", len(list(PROVEEDOR.iterdir())), "ficheros publicados") print(sorted(p.name for p in PROVEEDOR.iterdir())[:3]) ``` ```output 18 sesiones en la ventana; 17 ficheros publicados ['NOVA_2025-06-16.csv', 'NOVA_2025-06-17.csv', 'NOVA_2025-06-18.csv'] ``` `sessions_in_range` devuelve las sesiones reales de la NYSE en el intervalo: 18, sin el 19 de junio ni el 4 de julio. El proveedor ha publicado 17 ficheros; el del 8 de julio todavía no. ## Descargar y validar En un sistema real, `descargar` llamaría a una API con reintentos y límites (lección 1.4.1). Aquí copia el fichero del proveedor a `crudo/`, que guarda lo recibido tal cual: si algo sale mal más adelante, siempre se puede volver a procesar sin pedirlo otra vez. La **huella** (los primeros caracteres del SHA-256 del contenido) identifica el fichero: si el proveedor lo corrige, la huella cambia. ```python import shutil CRUDO = Path("crudo") def descargar(sesion): """Copia el fichero de la sesión a crudo/. None si no existe.""" origen = PROVEEDOR / f"NOVA_{sesion}.csv" if not origen.exists(): return None CRUDO.mkdir(exist_ok=True) return Path(shutil.copy(origen, CRUDO / origen.name)) def huella(ruta): return hashlib.sha256(ruta.read_bytes()).hexdigest()[:12] ruta = descargar("2025-06-16") print(ruta, huella(ruta)) ``` ```output crudo/NOVA_2025-06-16.csv b754df50e97a ``` La validación es una lista de reglas escritas, cada una con un mensaje que diga qué falla. Son las de la lección 1.4.2 convertidas en comprobaciones automáticas, más una que solo puede hacerse con un calendario: cuántas barras debe tener la sesión y entre qué horas. | Regla | Qué detecta | |---|---| | Columnas y tipos | Un proveedor que cambia el formato, una hora sin zona | | `ts` sin repetir y creciente | Filas duplicadas, ficheros mal ordenados o mezclados | | Sin vacíos | Celdas perdidas al exportar | | Máximo ≥ apertura y cierre; mínimo ≤ apertura y cierre | Ticks erróneos, columnas cambiadas | | Volumen ≥ 0 | Valores imposibles | | Barras = las del calendario | Barras perdidas, sesiones cortadas, medias sesiones | | Todas dentro de la sesión | Horas desplazadas por una zona horaria mal puesta | ```python COLUMNAS = ["ts", "apertura", "maximo", "minimo", "cierre", "volumen"] def leer(ruta): return pd.read_csv(ruta, parse_dates=["ts"]) def barras_esperadas(sesion): apertura = nyse.session_open(sesion) cierre = nyse.session_close(sesion) return int((cierre - apertura) / pd.Timedelta("5min")) def validar(df, sesion): """Lista de problemas; vacía si la sesión es correcta.""" if list(df.columns) != COLUMNAS: return [f"columnas {list(df.columns)}"] problemas = [] if str(df["ts"].dtype) != "datetime64[us, UTC]": problemas.append(f"ts de tipo {df['ts'].dtype}") if df["ts"].duplicated().any(): problemas.append(f"{df['ts'].duplicated().sum()} ts repetidas") if not df["ts"].is_monotonic_increasing: problemas.append("ts no creciente") if df[COLUMNAS].isna().any().any(): problemas.append("valores vacíos") techo = df[["apertura", "cierre"]].max(axis=1) suelo = df[["apertura", "cierre"]].min(axis=1) malas = (df["maximo"] < techo) | (df["minimo"] > suelo) if malas.any(): problemas.append(f"{malas.sum()} barras OHLC incoherentes") if (df["volumen"] < 0).any(): problemas.append("volumen negativo") esperadas = barras_esperadas(sesion) unicas = df["ts"].nunique() if unicas != esperadas: problemas.append(f"{unicas} barras de {esperadas}") fuera = ((df["ts"] < nyse.session_open(sesion)) | (df["ts"] >= nyse.session_close(sesion))) if fuera.any(): problemas.append(f"{fuera.sum()} barras fuera de la sesión") return problemas for s in ["2025-06-16", "2025-07-03"]: print(s, barras_esperadas(s), "barras esperadas;", validar(leer(descargar(s)), s) or "sin problemas") ``` ```output 2025-06-16 78 barras esperadas; sin problemas 2025-07-03 42 barras esperadas; sin problemas ``` El 16 de junio espera 78 barras y el 3 de julio, 42: el calendario sabe que esa sesión cierra a las 13:00. Una constante «78 barras por sesión» habría rechazado la media sesión y no habría dicho nada de un día festivo con datos. Pasadas por todas las sesiones de la ventana, las reglas encuentran exactamente los defectos puestos: ```python informe = {} for s in ventana.strftime("%Y-%m-%d"): ruta = descargar(s) informe[s] = ("no publicada" if ruta is None else "; ".join(validar(leer(ruta), s)) or "ok") pd.Series(informe)[lambda x: x != "ok"] ``` ```output 2025-06-24 1 ts repetidas 2025-06-26 77 barras de 78 2025-07-01 1 barras OHLC incoherentes 2025-07-08 no publicada dtype: str ``` La fila duplicada del 24 de junio aparece como «1 ts repetidas»; la barra que falta el 26, como «77 barras de 78»; el máximo imposible del 1 de julio, como una barra OHLC incoherente. Las otras 14 sesiones pasan todas las reglas. ## Carga incremental e idempotente El estado del pipeline vive en un **manifiesto**: un fichero JSON con cada sesión cargada, su huella y su número de filas. En cada ejecución, el pipeline recorre las sesiones que pide el calendario y para cada una decide: ```mermaid stateDiagram-v2 [*] --> NoPublicada NoPublicada --> Descargada: llega el fichero Descargada --> YaEstaba: misma huella Descargada --> Cuarentena: no valida Descargada --> Cargada: valida Cuarentena --> Descargada: corrección Cargada --> [*] YaEstaba --> [*] ``` Una sesión válida se escribe **entera** en su partición (`sesion=2025-06-16/`), borrando antes lo que hubiera. Así una sesión corregida reemplaza a la anterior sin dejar restos y sin duplicar barras. ```python LAGO, CUARENTENA = Path("lago/barras_5min"), Path("cuarentena") MANIFIESTO = Path("manifiesto.json") def cargar_manifiesto(): if MANIFIESTO.exists(): return json.loads(MANIFIESTO.read_text()) return {"cargadas": {}} def ejecutar(desde, hasta): man = cargar_manifiesto() res = {"cargada": [], "ya_estaba": [], "cuarentena": [], "no_publicada": []} for s in nyse.sessions_in_range(desde, hasta).strftime("%Y-%m-%d"): ruta = descargar(s) if ruta is None: res["no_publicada"].append(s) continue h = huella(ruta) if man["cargadas"].get(s, {}).get("huella") == h: res["ya_estaba"].append(s) continue df = leer(ruta) problemas = validar(df, s) if problemas: CUARENTENA.mkdir(exist_ok=True) shutil.copy(ruta, CUARENTENA / ruta.name) res["cuarentena"].append(s) continue destino = LAGO / f"sesion={s}" shutil.rmtree(destino, ignore_errors=True) destino.mkdir(parents=True) df.to_parquet(destino / "parte-0.parquet", compression="zstd") man["cargadas"][s] = {"huella": h, "filas": len(df)} res["cargada"].append(s) man["marca_de_agua"] = max(man["cargadas"], default=None) MANIFIESTO.write_text(json.dumps(man, indent=1, sort_keys=True)) return {k: len(v) for k, v in res.items()}, res resumen, detalle = ejecutar("2025-06-16", "2025-07-11") print(resumen) print("cuarentena:", detalle["cuarentena"]) print("marca de agua:", cargar_manifiesto()["marca_de_agua"]) ``` ```output {'cargada': 14, 'ya_estaba': 0, 'cuarentena': 3, 'no_publicada': 1} cuarentena: ['2025-06-24', '2025-06-26', '2025-07-01'] marca de agua: 2025-07-11 ``` Catorce sesiones cargadas, tres en cuarentena y una sin publicar. El manifiesto se escribe con `sort_keys=True` para que su contenido no dependa del orden de las operaciones: dos ejecuciones con las mismas entradas dejan el mismo fichero, byte a byte. La **marca de agua** (*watermark*) es la sesión más reciente cargada: el 11 de julio. Muchos pipelines usan solo eso para decidir qué pedir la próxima vez («todo lo posterior a la marca»). Con este proveedor sería un error: ```python man = cargar_manifiesto() sesiones = nyse.sessions_in_range("2025-06-16", "2025-07-11") debajo = [s for s in sesiones.strftime("%Y-%m-%d") if s <= man["marca_de_agua"] and s not in man["cargadas"]] print("marca de agua:", man["marca_de_agua"]) print("sesiones por debajo sin cargar:", debajo) ``` ```output marca de agua: 2025-07-11 sesiones por debajo sin cargar: ['2025-06-24', '2025-06-26', '2025-07-01', '2025-07-08'] ``` Cuatro sesiones están por debajo de la marca y no están cargadas. Un pipeline que solo pidiera lo posterior al 11 de julio no las volvería a intentar nunca, y el lago tendría cuatro huecos permanentes que nadie vería hasta que un backtest diera un resultado raro. La marca de agua sirve para saber hasta dónde se ha llegado; la completitud se comprueba sesión a sesión contra el calendario. El **informe de calidad** de la ejecución compara lo esperado con lo cargado: ```python def informe_calidad(desde, hasta): man = cargar_manifiesto()["cargadas"] filas = [] for s in nyse.sessions_in_range(desde, hasta).strftime("%Y-%m-%d"): cargadas = man.get(s, {}).get("filas", 0) filas.append({"sesion": s, "esperadas": barras_esperadas(s), "cargadas": cargadas}) rep = pd.DataFrame(filas).set_index("sesion") rep["completa"] = rep["cargadas"] == rep["esperadas"] return rep rep = informe_calidad("2025-06-16", "2025-07-11") print(f"{rep['completa'].sum()} de {len(rep)} sesiones completas; " f"{rep['cargadas'].sum()} de {rep['esperadas'].sum()} barras") rep[~rep["completa"] | (rep["esperadas"] != 78)] ``` ```output 14 de 18 sesiones completas; 1056 de 1368 barras esperadas cargadas completa sesion 2025-06-24 78 0 False 2025-06-26 78 0 False 2025-07-01 78 0 False 2025-07-03 42 42 True 2025-07-08 78 0 False ``` La figura de la cabecera es este informe dibujado: 1.056 de 1.368 barras cargadas, la media sesión del 3 de julio completa con sus 42 barras y cuatro sesiones vacías, en cuarentena o sin publicar. ## Tests de calidad sobre el lago La validación mira cada fichero al entrar. Los **tests de calidad** miran el resultado: el lago entero, tal como lo leerá un backtest. Comprueban propiedades del conjunto que ningún fichero suelto puede comprobar (que estén todas las sesiones del calendario, que no haya duplicados entre particiones) y vigilan que nada haya estropeado el lago después de la carga: un proceso que escribió donde no debía, una corrección manual mal hecha. Se escriben como tests de pytest (lección 0.4.5) en un fichero y se ejecutan con `pytest.main`, la misma ejecución que `uv run pytest test_calidad.py` en la terminal. La función auxiliar se queda con las líneas de los tests fallidos y con el resumen sin el tiempo, que cambia en cada ejecución: ```python import contextlib import io import re import sys import pytest sys.dont_write_bytecode = True Path("test_calidad.py").write_text(""" import exchange_calendars as xcals import pandas as pd import pytest NYSE = xcals.get_calendar("XNYS") SESIONES = NYSE.sessions_in_range("2025-06-16", "2025-07-11") @pytest.fixture(scope="module") def lago(): df = pd.read_parquet("lago/barras_5min") df["sesion"] = df["sesion"].astype(str) return df def test_calendario_completo(lago): faltan = sorted(set(SESIONES.strftime("%Y-%m-%d")) - set(lago["sesion"])) assert not faltan, f"faltan {len(faltan)} sesiones" def test_barras_por_sesion(lago): n = lago.groupby("sesion").size() for s, filas in n.items(): abre, cierra = NYSE.session_open(s), NYSE.session_close(s) assert filas == (cierra - abre) / pd.Timedelta("5min"), s def test_sin_duplicados(lago): assert not lago["ts"].duplicated().any() def test_ohlc_coherente(lago): assert (lago["maximo"] >= lago[["apertura", "cierre"]] .max(axis=1)).all() assert (lago["minimo"] <= lago[["apertura", "cierre"]] .min(axis=1)).all() def test_ts_en_utc(lago): assert str(lago["ts"].dt.tz) == "UTC" """) def pasar_tests(): sys.modules.pop("test_calidad", None) salida = io.StringIO() with contextlib.redirect_stdout(salida): codigo = pytest.main(["-q", "-p", "no:cacheprovider", "-rf", "--tb=no", "test_calidad.py"]) lineas = salida.getvalue().strip().splitlines() for linea in lineas: if linea.startswith("FAILED"): print(linea) print(re.sub(r" in [\d.]+s$", "", lineas[-1])) return int(codigo) pasar_tests(); ``` ```output FAILED test_calidad.py::test_calendario_completo - AssertionError: faltan 4 sesiones 1 failed, 4 passed ``` Cuatro tests pasan: todo lo cargado está bien. Uno falla, `test_calendario_completo`, porque faltan cuatro sesiones. Es exactamente lo que hay que saber antes de lanzar un backtest sobre este lago, y lo que un sistema de alertas debería enviar por correo o por un canal de mensajes. El código de salida de pytest (1 si algo falla) es lo que un orquestador usa para marcar la ejecución como fallida. `-p no:cacheprovider` evita que pytest cree su carpeta de caché, y `--tb=no` que imprima los detalles de cada fallo; en la terminal conviene quitar las dos opciones para ver todo. ## Reintentar: idempotencia y correcciones La prueba de la idempotencia es ejecutar otra vez sin que haya cambiado nada: ```python antes = sorted(p.as_posix() for p in LAGO.rglob("*.parquet")) resumen, _ = ejecutar("2025-06-16", "2025-07-11") despues = sorted(p.as_posix() for p in LAGO.rglob("*.parquet")) print(resumen) print("lago sin cambios:", antes == despues) ``` ```output {'cargada': 0, 'ya_estaba': 14, 'cuarentena': 3, 'no_publicada': 1} lago sin cambios: True ``` Las 14 sesiones cargadas se reconocen por su huella y no se tocan; las tres de cuarentena se vuelven a validar (y vuelven a fallar, porque el proveedor no las ha corregido), y la del 8 de julio sigue sin publicar. El lago queda igual. Al día siguiente, el proveedor publica las correcciones y la sesión que faltaba: ```python for s in ["2025-06-24", "2025-06-26", "2025-07-01", "2025-07-08"]: publicar(s, barras[barras["sesion"] == s]) resumen, detalle = ejecutar("2025-06-16", "2025-07-11") print(resumen) print("cargadas ahora:", detalle["cargada"]) ``` ```output {'cargada': 4, 'ya_estaba': 14, 'cuarentena': 0, 'no_publicada': 0} cargadas ahora: ['2025-06-24', '2025-06-26', '2025-07-01', '2025-07-08'] ``` *Fig. — Qué hace el pipeline con las 18 sesiones de la ventana en cada ejecución.* Tres barras horizontales apiladas de 18 sesiones. Ejecución 1: 14 cargadas (destacadas), 3 en cuarentena y 1 no publicada. Ejecución 2, sin cambios en el proveedor: 14 que ya estaban, 3 en cuarentena y 1 no publicada; no carga nada. Ejecución 3, tras la corrección del proveedor: 14 que ya estaban y 4 cargadas. Las cuatro se cargan: las tres de cuarentena porque su fichero nuevo tiene otra huella y ahora valida, y la del 8 de julio porque ya existe. Ninguna de las 14 anteriores se reescribe. Los tests lo confirman: ```python print("código de salida:", pasar_tests()) rep = informe_calidad("2025-06-16", "2025-07-11") print(f"{rep['completa'].sum()} de {len(rep)} sesiones completas") ``` ```output 5 passed código de salida: 0 18 de 18 sesiones completas ``` ## Ejecutarlo cada día: cron y orquestadores Hasta aquí el pipeline es una función. Para que corra solo, hay dos niveles: **cron**, el programador de tareas de Linux y macOS, ejecuta un comando a una hora fija. Una línea en `crontab -e` basta para lanzar el script de lunes a viernes a las 22:30 de la hora del servidor (después del cierre de Nueva York en cualquier época del año si el servidor está en UTC), guardando la salida en un fichero de registro: ```bash 30 22 * * 1-5 cd /srv/datos && uv run python pipeline.py >> pipeline.log 2>&1 ``` cron no sabe nada de lo que pasa dentro: si el script falla, solo queda el registro. Por eso el script debe devolver un código de salida distinto de 0 cuando algo va mal (como `pytest.main`) y avisar por su cuenta. **Un orquestador** como Prefect añade lo que cron no tiene: reintentos de cada paso con espera entre ellos, un historial de ejecuciones con su estado, avisos cuando algo falla y la posibilidad de relanzar a mano una ejecución concreta. El código no cambia mucho: las funciones de la lección se decoran como tareas y flujos. Necesita Prefect instalado y un proceso que lo sirva, así que no se ejecuta aquí: ```python skip title="flujo_prefect.py" from prefect import flow, task @task(retries=4, retry_delay_seconds=[60, 120, 300, 600]) def descargar_sesion(sesion: str): return descargar(sesion) # la función de la lección @flow(log_prints=True) def pipeline_diario(desde: str, hasta: str): for s in nyse.sessions_in_range(desde, hasta).strftime("%Y-%m-%d"): descargar_sesion(s) resumen, _ = ejecutar(desde, hasta) print(resumen) if pasar_tests() != 0: raise RuntimeError("tests de calidad fallidos") if __name__ == "__main__": pipeline_diario.serve(name="nova-diario", cron="30 22 * * 1-5", parameters={"desde": "2025-06-16", "hasta": "2025-07-11"}) ``` `retries` y `retry_delay_seconds` reintentan la descarga si la API falla, con esperas crecientes (la idea del *backoff* de la lección 1.4.1). `serve` registra el flujo con una programación cron y lo ejecuta mientras el proceso esté vivo. Lanzar una excepción si los tests fallan hace que la ejecución quede marcada como fallida en el historial. En un uso real, las fechas `desde` y `hasta` se calcularían a partir del día de la ejecución (por ejemplo, las últimas 20 sesiones), para que cada día se vuelva a comprobar una ventana que incluya los posibles huecos. ## Ejercicios 1. Añade a `validar` una regla que rechace sesiones cuyo volumen total sea menor que la décima parte de la mediana de las sesiones ya cargadas. Pruébala con dos ficheros del 16 de junio estropeados: uno al que le falta la segunda mitad de las barras y otro con todas las barras pero con el volumen dividido entre 20 (un proveedor que solo cuenta un centro de negociación). **Solución.** ```python def volumen_sospechoso(df, lago=LAGO, fraccion=0.1): historia = pd.read_parquet(lago) mediana = historia.groupby("sesion", observed=True)[ "volumen"].sum().median() return df["volumen"].sum() < fraccion * mediana, mediana dia = leer(CRUDO / "NOVA_2025-06-16.csv") casos = {"media sesión": dia.iloc[:39], "volumen / 20": dia.assign(volumen=dia["volumen"] // 20)} for nombre, df in casos.items(): sospecha, mediana = volumen_sospechoso(df) print(f"{nombre}: {df['volumen'].sum():,.0f} de una mediana de " f"{mediana:,.0f}; volumen sospechoso: {sospecha}; " f"validar: {validar(df, '2025-06-16') or 'ok'}") ``` ```output media sesión: 3,872,882 de una mediana de 9,889,275; volumen sospechoso: False; validar: ['39 barras de 78'] volumen / 20: 399,496 de una mediana de 9,889,275; volumen sospechoso: True; validar: ok ``` Las dos reglas se complementan. La media sesión tiene un 39 % del volumen mediano: no baja del umbral del 10 %, porque la primera mitad de NOVA concentra mucho volumen (la U de la lección 1.4.5), pero la regla del calendario ya la rechaza («39 barras de 78»). El fichero con el volumen dividido entre 20 tiene todas sus barras y pasa todas las reglas de `validar`; solo la regla de volumen lo detecta (399.496 acciones frente a una mediana de 9.889.275). Es el caso de un proveedor que cambia sin avisar de cinta consolidada a la de un solo centro (lección 1.1.1). Un umbral así se calibra con la historia ya cargada, nunca con el propio día, y conviene que avise antes de rechazar: un día de volumen muy bajo puede ser real. 2. Escribe un test de calidad más para `test_calidad.py`: que entre el cierre de una sesión y la apertura de la siguiente no haya ninguna barra (ninguna marca de tiempo fuera de las sesiones del calendario). **Solución.** ```python with open("test_calidad.py", "a") as f: f.write(''' def test_todo_dentro_de_sesion(lago): for s, grupo in lago.groupby("sesion"): abre, cierra = NYSE.session_open(s), NYSE.session_close(s) assert grupo["ts"].between(abre, cierra, inclusive="left").all(), s ''') print("código de salida:", pasar_tests()) ``` ```output 6 passed código de salida: 0 ``` Los seis tests pasan: ninguna barra cae fuera de `[apertura, cierre)` de su sesión, también en la media sesión del 3 de julio, que cierra a las 17:00 UTC. Este test detectaría un error típico de zona horaria: barras guardadas con la hora de Nueva York etiquetada como UTC, que quedarían cuatro o cinco horas antes de la apertura real. La validación ya lo comprueba al cargar; el test lo vigila después, sobre todo el lago. 3. Un compañero propone simplificar el pipeline: «cada noche borro el lago y lo vuelvo a cargar entero desde el proveedor; así no necesito manifiesto». ¿Qué gana y qué pierde? **Solución.** Gana sencillez y una forma trivial de idempotencia: el lago siempre es la copia exacta de lo que el proveedor tiene esa noche. Pierde varias cosas. Primero, coste y tiempo: con años de datos de muchos activos, descargar todo cada día es lento, gasta límites de la API y puede costar dinero. Segundo, control: si una noche el proveedor publica un fichero roto o le falta un día que antes tenía, el lago pierde datos buenos que ya estaban; con el manifiesto y la cuarentena, lo bueno se queda y lo malo se aparta. Tercero, trazabilidad: sin manifiesto ni crudo con fecha no se sabe qué cambió de un día a otro ni con qué datos se hizo un backtest. Un término medio razonable es recargar entera una ventana reciente (las últimas semanas, donde llegan las correcciones) y mantener lo antiguo con su manifiesto. ## Autoevaluación 1. El pipeline se ejecuta dos veces seguidas sin que el proveedor haya publicado nada nuevo. ¿Qué debe ocurrir en la segunda ejecución? - A. Que vuelva a cargar todas las sesiones, por si acaso - B. Nada: reconoce por la huella que cada sesión ya está cargada, no escribe ni duplica, y el lago queda idéntico - C. Que falle, porque los ficheros ya existen - D. Que borre el lago y lo reconstruya *Respuesta B: Nada: reconoce por la huella que cada sesión ya está cargada, no escribe ni duplica, y el lago queda idéntico.* Es la idempotencia: ejecutar una vez o diez da el mismo resultado. En la lección, la segunda ejecución da 14 sesiones «ya estaban», 0 cargadas y el lago sin cambios. Sin esa propiedad, un reintento tras un fallo duplicaría barras. 2. Tras la primera ejecución la marca de agua está en el 11 de julio. Si la siguiente ejecución solo pidiera sesiones posteriores a esa fecha, ¿qué pasaría? - A. Nada, porque todo lo anterior está cargado - B. Las cuatro sesiones que no se cargaron (tres en cuarentena y una sin publicar) no se volverían a intentar nunca - C. Se cargarían dos veces las sesiones de julio - D. Fallaría el calendario *Respuesta B: Las cuatro sesiones que no se cargaron (tres en cuarentena y una sin publicar) no se volverían a intentar nunca.* Una marca de agua dice hasta dónde se ha llegado, no si todo lo de debajo está completo. Por eso el pipeline de la lección guarda en el manifiesto cada sesión cargada con su huella y recorre cada vez toda la ventana contra el calendario. 3. La sesión del 3 de julio de 2025 tiene 42 barras de 5 minutos. ¿Qué debe hacer la validación? - A. Rechazarla porque le faltan 36 barras - B. Aceptarla: es una media sesión (cierre a las 13:00 de Nueva York) y el calendario de la NYSE dice que esperar 42 es lo correcto - C. Rellenar las barras que faltan con el último precio - D. Marcarla como dudosa siempre *Respuesta B: Aceptarla: es una media sesión (cierre a las 13:00 de Nueva York) y el calendario de la NYSE dice que esperar 42 es lo correcto.* El número de barras esperado sale del calendario (`session_open` y `session_close` de `exchange_calendars`), no de una constante. Una regla fija de 78 barras rechazaría las medias sesiones y aceptaría días que no deberían existir. 4. ¿Qué aportan los tests de pytest sobre el lago si cada fichero ya se valida antes de cargarlo? - A. Nada: son redundantes - B. Comprueban propiedades del conjunto que ningún fichero suelto puede comprobar, como que estén todas las sesiones del calendario, y vigilan que nada haya roto el lago después de la carga - C. Hacen que la carga sea más rápida - D. Sustituyen al manifiesto *Respuesta B: Comprueban propiedades del conjunto que ningún fichero suelto puede comprobar, como que estén todas las sesiones del calendario, y vigilan que nada haya roto el lago después de la carga.* La validación mira cada sesión al entrar; los tests miran el resultado. Tras la primera ejecución, todos los ficheros cargados estaban bien y aun así `test_calendario_completo` falla: faltan 4 sesiones. Es la información que necesita quien va a lanzar un backtest. ## Ideas clave - Un pipeline reproducible separa pasos con entradas y salidas en disco: crudo, cuarentena, lago limpio, manifiesto e informe. - Validar es comprobar reglas escritas: esquema y tipos, orden, duplicados, vacíos, coherencia OHLC y número de barras según el calendario de la bolsa, medias sesiones incluidas. - Lo que no pasa la validación va a cuarentena, no al lago; cuando el proveedor lo corrige, la huella cambia y se carga. - Idempotencia: ejecutar dos veces da el mismo lago. Se consigue con un manifiesto de huellas por sesión y escribiendo cada partición entera. - Una marca de agua sola deja huecos sin volver a intentar; el calendario dice qué debería haber y el manifiesto qué hay. - Los tests de calidad sobre el lago comprueban el conjunto; cron o un orquestador como Prefect solo deciden cuándo se ejecuta. ## Recursos ### Research - [Big Data Management 3 - Creando un Datalake sobre el Nasdaq con ArcticDB.](https://pythonparatrading.com/big-data-con-arcticdb/) · Research-0021 · 5 min ### Documentación - [exchange_calendars: calendarios de sesiones de mercado en Python](https://github.com/gerrymanoim/exchange_calendars) · github.com - [pytest: cómo invocar pytest (también desde Python con pytest.main)](https://docs.pytest.org/en/stable/how-to/usage.html) · docs.pytest.org - [Prefect: cómo programar ejecuciones (schedules y serve con cron)](https://docs.prefect.io/v3/how-to-guides/deployments/create-schedules) · docs.prefect.io - [Prefect: reintentos automáticos de tareas](https://docs.prefect.io/v3/how-to-guides/workflows/retries) · docs.prefect.io ### Datos - [Datos del nivel 1: barras de 5 minutos de NOVA, 2024-2025 (Parquet)](https://pythonparatrading.com/curso/datos/nivel-1/nova_barras_5min.parquet) · Descarga · nova_barras_5min.parquet Lección anterior: [1.4.7 · Almacenamiento: CSV, Parquet, particionado, DuckDB, TimescaleDB y ArcticDB](https://pythonparatrading.com/curso-python-trading/nivel-1/almacenamiento-parquet-duckdb-y-bases-temporales/) Lección siguiente: [1.5.1 · Estados financieros, ratios, XBRL y SEC EDGAR](https://pythonparatrading.com/curso-python-trading/nivel-1/estados-financieros-xbrl-y-edgar/)