Pipeline reproducible: descarga, validación, actualización incremental y tests de calidad.

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

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

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

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

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

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)

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

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

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)

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

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

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