---
title: "1.4.8 Reproducible market data pipeline"
description: "Daily data pipeline: download, validation against the NYSE calendar, quarantine, idempotent incremental load with a manifest and tests with pytest."
url: "https://pythonparatrading.com/en/python-trading-course/level-1/reproducible-market-data-pipeline/"
lang: "en"
image: "https://pythonparatrading.com/og/en/python-trading-course/level-1/reproducible-market-data-pipeline.png"
---

# 1.4.8 · Reproducible pipeline: download, validation, incremental updates and quality tests

> A system that trades every day needs new data every day, and a bad data point that gets in unchecked ends up in a signal, an order and a loss. This lesson builds a small, complete pipeline on daily NOVA files with deliberate defects: it downloads them, validates them against the real NYSE calendar, sets the bad ones aside, loads the good ones without duplicating anything and checks the result with tests.

- Course: [Level 1: Financial data, markets and analysis](https://pythonparatrading.com/en/python-trading-course/level-1/)
- Module: 1.4 · Financial data engineering
- Duration: 60 min
- Prerequisites: [1.4.7 · Storage: CSV, Parquet, partitioning, DuckDB, TimescaleDB and ArcticDB](https://pythonparatrading.com/en/python-trading-course/level-1/storing-market-data-parquet-duckdb-timeseries/), [0.4.5 · Professional logging and testing with pytest](https://pythonparatrading.com/en/python-trading-course/level-0/python-logging-and-testing-with-pytest/), [1.4.2 · Data cleaning: missing values, bad ticks, gaps and time zones](https://pythonparatrading.com/en/python-trading-course/level-1/cleaning-financial-market-data/)

## By the end of this lesson

- Split a data pipeline into steps (download, validate, load, report) with a state that can be inspected.
- Write explicit validations of schema, order, duplicates, OHLC consistency and complete sessions according to `exchange_calendars`.
- Make the load incremental and idempotent with a manifest of fingerprints, and explain why a watermark alone is not enough.
- Write quality tests on the data already loaded and run them with pytest.
- Place orchestration (cron, Prefect) and what it adds over a script.

## What a data pipeline does

The previous lessons of the module dealt with each piece separately: downloading (1.4.1), cleaning (1.4.2), adjusting
(1.4.3), building bars (1.4.5) and storing (1.4.7). A system that trades every day needs to put them together in a
process that runs on its own, every day, and always does the same thing. That is a **pipeline**, and it has to meet
three conditions:

- **Reproducible.** With the same inputs it gives the same result. Nothing depends on the time it runs, the order in
  which the files arrive or what was left in memory from the previous time.
- **Idempotent.** Running it twice in a row changes nothing the second time. It is what lets you retry without fear
  after a half-finished failure.
- **Watchful.** What comes in is checked against written rules, what does not meet them is set aside and everything
  is recorded.

```mermaid
flowchart TB
  K[NYSE calendar] --> D[Download]
  D --> C[crudo/]
  C --> V{Valid?}
  V -- no --> Q[cuarentena/]
  V -- yes --> L[lago/ Parquet]
  L --> M[Manifest]
  M --> I[Report and tests]
  Q --> I
```

The calendar comes first because it decides which sessions **should** exist. Everything else is compared with it.

## A vendor with defects

To have something to watch, the lesson manufactures a vendor: a folder with one CSV per session, the format in which
many vendors deliver daily data, with NOVA's 5-minute bars for 18 sessions between June 16 and July 11, 2025. The
window has two real calendar traps: on June 19 (Juneteenth) and July 4 the NYSE is closed, and July 3 is a half
session that ends at 13:00 New York time. And four defects planted on purpose, like the ones that really arrive:

```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":            # duplicated row
        dia = pd.concat([dia, dia.iloc[[10]]]).sort_index()
    if s == "2025-06-26":            # one bar missing
        dia = dia.drop(index=40)
    if s == "2025-07-01":            # high below the close
        dia.loc[5, "maximo"] = dia.loc[5, "cierre"] - 0.05
    if s == "2025-07-08":            # not published yet
        continue
    publicar(s, dia)

print(len(ventana), "sessions in the window;",
      len(list(PROVEEDOR.iterdir())), "files published")
print(sorted(p.name for p in PROVEEDOR.iterdir())[:3])
```
```output
18 sessions in the window; 17 files published
['NOVA_2025-06-16.csv', 'NOVA_2025-06-17.csv', 'NOVA_2025-06-18.csv']
```

`sessions_in_range` returns the real NYSE sessions in the interval: 18, without June 19 or July 4. The vendor has
published 17 files; the July 8 one, not yet.

## Download and validate

In a real system, `descargar` would call an API with retries and limits (lesson 1.4.1). Here it copies the vendor's
file to `crudo/` (raw), which stores what was received as is: if something goes wrong later, it can always be
reprocessed without requesting it again. The **fingerprint** (the first characters of the SHA-256 of the content)
identifies the file: if the vendor corrects it, the fingerprint changes.

```python
import shutil

CRUDO = Path("crudo")

def descargar(sesion):
    """Copy the session's file to crudo/. None if it does not exist."""
    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
```

Validation is a list of written rules, each with a message that says what fails. They are the ones from lesson 1.4.2
turned into automatic checks, plus one that can only be done with a calendar: how many bars the session must have and
between which times.

| Rule | What it detects |
|---|---|
| Columns and types | A vendor that changes the format, a time without a zone |
| `ts` unique and increasing | Duplicated rows, badly sorted or mixed files |
| No missing values | Cells lost on export |
| High ≥ open and close; low ≤ open and close | Bad ticks, swapped columns |
| Volume ≥ 0 | Impossible values |
| Bars = those of the calendar | Lost bars, cut sessions, half sessions |
| All inside the session | Times shifted by a wrongly set time zone |

```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):
    """List of problems; empty if the session is correct."""
    if list(df.columns) != COLUMNAS:
        return [f"columns {list(df.columns)}"]
    problemas = []
    if str(df["ts"].dtype) != "datetime64[us, UTC]":
        problemas.append(f"ts of type {df['ts'].dtype}")
    if df["ts"].duplicated().any():
        problemas.append(f"{df['ts'].duplicated().sum()} repeated ts")
    if not df["ts"].is_monotonic_increasing:
        problemas.append("ts not increasing")
    if df[COLUMNAS].isna().any().any():
        problemas.append("missing values")
    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()} inconsistent OHLC bars")
    if (df["volumen"] < 0).any():
        problemas.append("negative volume")
    esperadas = barras_esperadas(sesion)
    unicas = df["ts"].nunique()
    if unicas != esperadas:
        problemas.append(f"{unicas} bars of {esperadas}")
    fuera = ((df["ts"] < nyse.session_open(sesion))
             | (df["ts"] >= nyse.session_close(sesion)))
    if fuera.any():
        problemas.append(f"{fuera.sum()} bars outside the session")
    return problemas

for s in ["2025-06-16", "2025-07-03"]:
    print(s, barras_esperadas(s), "bars expected;",
          validar(leer(descargar(s)), s) or "no problems")
```
```output
2025-06-16 78 bars expected; no problems
2025-07-03 42 bars expected; no problems
```

June 16 expects 78 bars and July 3, 42: the calendar knows that session closes at 13:00. A constant “78 bars per
session” would have rejected the half session and would have said nothing about a holiday with data. Run over every
session in the window, the rules find exactly the planted defects:

```python
informe = {}
for s in ventana.strftime("%Y-%m-%d"):
    ruta = descargar(s)
    informe[s] = ("not published" if ruta is None
                  else "; ".join(validar(leer(ruta), s)) or "ok")
pd.Series(informe)[lambda x: x != "ok"]
```
```output
2025-06-24               1 repeated ts
2025-06-26               77 bars of 78
2025-07-01    1 inconsistent OHLC bars
2025-07-08               not published
dtype: str
```

The duplicated row of June 24 shows up as “1 repeated ts”; the missing bar on the 26th, as “77 bars of 78”; the
impossible high of July 1, as an inconsistent OHLC bar. The other 14 sessions pass every rule.

<aside class="callout" data-kind="nota">

Validating is not correcting. The pipeline does not fill in the missing bar or delete the duplicate: it does not know
whether the error is in the repeated row or in the one that should be in its place. What does not meet the rules is
set aside and reported. Corrections (lesson 1.4.2) are judgment calls, and they are made separately, documented.

</aside>

## Incremental, idempotent load

The pipeline's state lives in a **manifest**: a JSON file with each loaded session, its fingerprint and its number of
rows. On each run, the pipeline walks the sessions the calendar asks for and decides for each one:

```mermaid
stateDiagram-v2
  [*] --> NotPublished
  NotPublished --> Downloaded: file arrives
  Downloaded --> AlreadyThere: same fingerprint
  Downloaded --> Quarantine: fails validation
  Downloaded --> Loaded: passes validation
  Quarantine --> Downloaded: correction
  Loaded --> [*]
  AlreadyThere --> [*]
```

A valid session is written **whole** into its partition (`sesion=2025-06-16/`), deleting whatever was there first.
That way a corrected session replaces the previous one without leaving leftovers and without duplicating bars.

```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("quarantine:", detalle["cuarentena"])
print("watermark:", cargar_manifiesto()["marca_de_agua"])
```
```output
{'cargada': 14, 'ya_estaba': 0, 'cuarentena': 3, 'no_publicada': 1}
quarantine: ['2025-06-24', '2025-06-26', '2025-07-01']
watermark: 2025-07-11
```

Fourteen sessions loaded, three in quarantine and one unpublished. The manifest is written with `sort_keys=True` so
that its content does not depend on the order of operations: two runs with the same inputs leave the same file, byte
for byte.

The **watermark** is the most recent session loaded: July 11. Many pipelines use only that to decide what to request
next time (“everything after the mark”). With this vendor it would be a mistake:

```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("watermark:", man["marca_de_agua"])
print("unloaded sessions below it:", debajo)
```
```output
watermark: 2025-07-11
unloaded sessions below it: ['2025-06-24', '2025-06-26', '2025-07-01', '2025-07-08']
```

Four sessions are below the mark and are not loaded. A pipeline that only asked for what comes after July 11 would
never retry them, and the lake would have four permanent gaps that nobody would see until a backtest gave an odd
result. The watermark tells you how far you have got; completeness is checked session by session against the
calendar.

The run's **quality report** compares what was expected with what was loaded:

```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()} of {len(rep)} sessions complete; "
      f"{rep['cargadas'].sum()} of {rep['esperadas'].sum()} bars")
rep[~rep["completa"] | (rep["esperadas"] != 78)]
```
```output
14 of 18 sessions complete; 1056 of 1368 bars
            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
```

The figure at the top is this report plotted: 1,056 of 1,368 bars loaded, the July 3 half session complete with its
42 bars and four empty sessions, in quarantine or unpublished.

## Quality tests on the lake

Validation looks at each file on the way in. **Quality tests** look at the result: the whole lake, as a backtest will
read it. They check properties of the whole set that no single file can check (that all the calendar's sessions are
there, that there are no duplicates across partitions) and they watch that nothing has damaged the lake after the
load: a process that wrote where it should not, a botched manual correction.

They are written as pytest tests (lesson 0.4.5) in a file and run with `pytest.main`, the same run as
`uv run pytest test_calidad.py` in the terminal. The helper function keeps the lines of the failed tests and the
summary without the time, which changes on every run:

```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"{len(faltan)} sessions missing"

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: 4 sessions missing
1 failed, 4 passed
```

Four tests pass: everything loaded is fine. One fails, `test_calendario_completo`, because four sessions are missing.
It is exactly what you need to know before launching a backtest on this lake, and what an alerting system should send
by email or to a messaging channel. pytest's exit code (1 if something fails) is what an orchestrator uses to mark the
run as failed.

`-p no:cacheprovider` stops pytest from creating its cache folder, and `--tb=no` from printing the details of each
failure; in the terminal you will want to drop both options to see everything.

## Retrying: idempotency and corrections

The test of idempotency is to run again without anything having changed:

```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("lake unchanged:", antes == despues)
```
```output
{'cargada': 0, 'ya_estaba': 14, 'cuarentena': 3, 'no_publicada': 1}
lake unchanged: True
```

The 14 loaded sessions are recognized by their fingerprint and left alone; the three in quarantine are validated
again (and fail again, because the vendor has not corrected them), and the July 8 one is still unpublished. The lake
stays the same. The next day, the vendor publishes the corrections and the missing session:

```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("loaded now:", detalle["cargada"])
```
```output
{'cargada': 4, 'ya_estaba': 14, 'cuarentena': 0, 'no_publicada': 0}
loaded now: ['2025-06-24', '2025-06-26', '2025-07-01', '2025-07-08']
```

*Fig. — What the pipeline does with the window's 18 sessions on each run.* Three stacked horizontal bars of 18 sessions. Run 1: 14 loaded (highlighted), 3 quarantined and 1 not published. Run 2, with no change at the provider: 14 already in, 3 quarantined and 1 not published; nothing is loaded. Run 3, after the provider's correction: 14 already in and 4 loaded.

The four are loaded: the three from quarantine because their new file has a different fingerprint and now passes
validation, and the July 8 one because it now exists. None of the previous 14 is rewritten. The tests confirm it:

```python
print("exit code:", pasar_tests())
rep = informe_calidad("2025-06-16", "2025-07-11")
print(f"{rep['completa'].sum()} of {len(rep)} sessions complete")
```
```output
5 passed
exit code: 0
18 of 18 sessions complete
```

<aside class="callout" data-kind="aviso">

A vendor can also correct a session that was already loaded and valid (a revised volume, a bad tick detected days
later). With this design, the new file has a different fingerprint and replaces the previous partition: the lake
changes backwards. That is right for having the best data, but yesterday's backtest can no longer be reproduced with
today's lake. If that matters, you have to keep the versions (the raw data with its date of receipt, or a versioned
database such as ArcticDB, lesson 1.4.7) and record in each backtest which version it used.

</aside>

## Running it every day: cron and orchestrators

So far the pipeline is a function. For it to run on its own, there are two levels:

**cron**, the task scheduler of Linux and macOS, runs a command at a fixed time. One line in `crontab -e` is enough to
launch the script Monday to Friday at 22:30 server time (after the New York close at any time of year if the server
is on UTC), saving the output to a log file:

```bash
30 22 * * 1-5 cd /srv/datos && uv run python pipeline.py >> pipeline.log 2>&1
```

cron knows nothing about what happens inside: if the script fails, only the log remains. That is why the script must
return an exit code other than 0 when something goes wrong (like `pytest.main`) and send alerts on its own.

**An orchestrator** such as Prefect adds what cron lacks: retries of each step with a wait between them, a history of
runs with their state, alerts when something fails and the ability to relaunch a specific run by hand. The code does
not change much: the lesson's functions are decorated as tasks and flows. It needs Prefect installed and a process to
serve it, so it does not run here:

```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)          # the lesson's function

@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("quality tests failed")

if __name__ == "__main__":
    pipeline_diario.serve(name="nova-diario", cron="30 22 * * 1-5",
                          parameters={"desde": "2025-06-16",
                                      "hasta": "2025-07-11"})
```

`retries` and `retry_delay_seconds` retry the download if the API fails, with increasing waits (the *backoff* idea of
lesson 1.4.1). `serve` registers the flow with a cron schedule and runs it while the process is alive. Raising an
exception if the tests fail marks the run as failed in the history. In real use, the `desde` and `hasta` dates would
be computed from the day of the run (for example, the last 20 sessions), so that every day a window that includes any
possible gaps is checked again.

## Exercises

1. Add to `validar` a rule that rejects sessions whose total volume is less than a tenth of the median of the sessions
   already loaded. Test it with two damaged June 16 files: one missing the second half of its bars and another with
   all its bars but the volume divided by 20 (a vendor that only counts one trading venue).

**Solution.**

```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 = {"half session": dia.iloc[:39],
         "volume / 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} of a median of "
          f"{mediana:,.0f}; suspicious volume: {sospecha}; "
          f"validate: {validar(df, '2025-06-16') or 'ok'}")
```
```output
half session: 3,872,882 of a median of 9,889,275; suspicious volume: False; validate: ['39 bars of 78']
volume / 20: 399,496 of a median of 9,889,275; suspicious volume: True; validate: ok
```

The two rules complement each other. The half session has 39% of the median volume: it does not fall below the 10%
threshold, because NOVA's first half concentrates a lot of volume (the U of lesson 1.4.5), but the calendar rule
already rejects it (“39 bars of 78”). The file with the volume divided by 20 has all its bars and passes every rule in
`validar`; only the volume rule detects it (399,496 shares versus a median of 9,889,275). It is the case of a vendor
that switches without notice from the consolidated tape to that of a single venue (lesson 1.1.1). A threshold like
this is calibrated with the history already loaded, never with the day itself, and it should warn before rejecting: a
day of very low volume can be real.

2. Write one more quality test for `test_calidad.py`: that there is no bar between the close of one session and the
   open of the next (no timestamp outside the calendar's sessions).

**Solution.**

```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("exit code:", pasar_tests())
```
```output
6 passed
exit code: 0
```

All six tests pass: no bar falls outside its session's `[open, close)`, including the July 3 half session, which
closes at 17:00 UTC. This test would catch a typical time zone mistake: bars stored with New York time labeled as
UTC, which would end up four or five hours before the real open. Validation already checks it at load time; the test
watches it afterwards, over the whole lake.

3. A colleague proposes simplifying the pipeline: “every night I delete the lake and reload it in full from the
   vendor; that way I don't need a manifest”. What does the change gain and what does it lose?

**Solution.**

It gains simplicity and a trivial form of idempotency: the lake is always the exact copy of what the vendor has that
night. It loses several things. First, cost and time: with years of data for many assets, downloading everything
every day is slow, uses up API limits and can cost money. Second, control: if one night the vendor publishes a broken
file or is missing a day it used to have, the lake loses good data it already had; with the manifest and quarantine,
the good stays and the bad is set aside. Third, traceability: without a manifest or dated raw data you cannot tell
what changed from one day to the next or which data a backtest was run with. A reasonable middle ground is to reload
a recent window in full (the last few weeks, where corrections arrive) and keep the old data with its manifest.

## Self-assessment

1. The pipeline runs twice in a row without the vendor having published anything new. What should happen on the second run?

   - A. It should load every session again, just in case
   - B. Nothing: it recognizes by the fingerprint that each session is already loaded, writes and duplicates nothing, and the lake stays identical
   - C. It should fail, because the files already exist
   - D. It should delete the lake and rebuild it

   *Answer B: Nothing: it recognizes by the fingerprint that each session is already loaded, writes and duplicates nothing, and the lake stays identical.* That is idempotency: running once or ten times gives the same result. In the lesson, the second run gives 14 sessions “already there”, 0 loaded and the lake unchanged. Without that property, a retry after a failure would duplicate bars.

2. After the first run the watermark is at July 11. If the next run only asked for sessions after that date, what would happen?

   - A. Nothing, because everything before it is loaded
   - B. The July sessions would be loaded twice
   - C. The four sessions that were not loaded (three in quarantine and one unpublished) would never be retried
   - D. The calendar would fail

   *Answer C: The four sessions that were not loaded (three in quarantine and one unpublished) would never be retried.* A watermark says how far you have got, not whether everything below it is complete. That is why the lesson's pipeline stores each loaded session with its fingerprint in the manifest and walks the whole window against the calendar every time.

3. The July 3, 2025 session has 42 five-minute bars. What should validation do?

   - A. Reject it because it is missing 36 bars
   - B. Fill in the missing bars with the last price
   - C. Always flag it as doubtful
   - D. Accept it: it is a half session (close at 13:00 New York time) and the NYSE calendar says that expecting 42 is correct

   *Answer D: Accept it: it is a half session (close at 13:00 New York time) and the NYSE calendar says that expecting 42 is correct.* The expected number of bars comes from the calendar (`session_open` and `session_close` from `exchange_calendars`), not from a constant. A fixed rule of 78 bars would reject half sessions and accept days that should not exist.

4. What do pytest tests on the lake add if each file is already validated before loading it?

   - A. They check properties of the whole set that no single file can check, such as all the calendar's sessions being present, and they watch that nothing has broken the lake after the load
   - B. Nothing: they are redundant
   - C. They make the load faster
   - D. They replace the manifest

   *Answer A: They check properties of the whole set that no single file can check, such as all the calendar's sessions being present, and they watch that nothing has broken the lake after the load.* Validation looks at each session on the way in; tests look at the result. After the first run, every loaded file was fine and even so `test_calendario_completo` fails: 4 sessions are missing. That is the information whoever is about to launch a backtest needs.

## Key points

- A reproducible pipeline separates steps with inputs and outputs on disk: raw, quarantine, clean lake, manifest and report.
- Validating means checking written rules: schema and types, order, duplicates, missing values, OHLC consistency and number of bars according to the exchange calendar, half sessions included.
- What fails validation goes to quarantine, not to the lake; when the vendor fixes it, the fingerprint changes and it is loaded.
- Idempotency: running twice gives the same lake. It is achieved with a manifest of fingerprints per session and by writing each partition whole.
- A watermark alone leaves gaps that are never retried; the calendar says what should be there and the manifest what is there.
- Quality tests on the lake check the whole set; cron or an orchestrator such as Prefect only decide when it runs.

## Resources

### Research

- [Big Data Management 3 - Building a Datalake on the Nasdaq with ArcticDB.](https://pythonparatrading.com/en/big-data-with-arcticdb/) · Research-0021 · 5 min

### Documentation

- [exchange_calendars: market session calendars in Python](https://github.com/gerrymanoim/exchange_calendars) · github.com
- [pytest: how to invoke pytest (also from Python with pytest.main)](https://docs.pytest.org/en/stable/how-to/usage.html) · docs.pytest.org
- [Prefect: how to schedule runs (schedules and serve with cron)](https://docs.prefect.io/v3/how-to-guides/deployments/create-schedules) · docs.prefect.io
- [Prefect: automatic task retries](https://docs.prefect.io/v3/how-to-guides/workflows/retries) · docs.prefect.io

### Data

- [Level 1 data: NOVA 5-minute bars, 2024-2025 (Parquet)](https://pythonparatrading.com/curso/datos/nivel-1/nova_barras_5min.parquet) · Download · nova_barras_5min.parquet

Previous lesson: [1.4.7 · Storage: CSV, Parquet, partitioning, DuckDB, TimescaleDB and ArcticDB](https://pythonparatrading.com/en/python-trading-course/level-1/storing-market-data-parquet-duckdb-timeseries/)

Next lesson: [1.5.1 · Financial statements, ratios, XBRL and SEC EDGAR](https://pythonparatrading.com/en/python-trading-course/level-1/financial-statements-xbrl-and-sec-edgar/)