Lesson 1.4.8 · 60 min

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.

Fig. — First pipeline run: bars loaded into the lake against those expected from the NYSE calendar, sessions from 16 June to 11 July 2025

08/08

Module: 1.4 · Financial data engineering
Duration: 60 min
Prerequisites: 1.4.7 · Storage: CSV, Parquet, partitioning, DuckDB, TimescaleDB and ArcticDB, 0.4.5 · Professional logging and testing with pytest, 1.4.2 · Data cleaning: missing values, bad ticks, gaps and time zones
Author: Jesús Cuesta
Updated: 04 Nov 2026

Level 1 lessons

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 source

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:

IN [01]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])

OUT [01]A simulated vendor: one CSV per session

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.

IN [02]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))

OUT [02]Download: from the vendor to crudo/

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

IN [03]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")

OUT [03]Validate: explicit checks

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:

IN [04]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"]

OUT [04]All published sessions, validated

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.

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 source

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.

IN [05]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"])

OUT [05]The incremental pipeline, with a manifest

{'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:

IN [06]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)

OUT [06]What a watermark alone does not see

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:

IN [07]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)]

OUT [07]Quality report of the run

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:

IN [08]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();

OUT [08]Quality tests on the lake, with pytest

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:

IN [09]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)

OUT [09]Second run: idempotent

{'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:

IN [10]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"])

OUT [10]The vendor corrects and publishes what was missing

{'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

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:

IN [11]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")

OUT [11]The tests, after the correction

5 passed
exit code: 0
18 of 18 sessions complete

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:

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:

IN [12]flujo_prefect.pypython

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

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

IN [13]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'}")

OUT [13]An anomalous volume rule

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.

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

IN [14]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())

OUT [14]No bar outside the sessions

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.

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.

Each answer is folded under its question

Self-assessment

The pipeline runs twice in a row without the vendor having published anything new. What should happen on the second run?
- AIt should load every session again, just in case
- BNothing: it recognizes by the fingerprint that each session is already loaded, writes and duplicates nothing, and the lake stays identical
- CIt should fail, because the files already exist
- DIt should delete the lake and rebuild it
Show answer
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.
After the first run the watermark is at July 11. If the next run only asked for sessions after that date, what would happen?
- ANothing, because everything before it is loaded
- BThe July sessions would be loaded twice
- CThe four sessions that were not loaded (three in quarantine and one unpublished) would never be retried
- DThe calendar would fail
Show answer
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.
The July 3, 2025 session has 42 five-minute bars. What should validation do?
- AReject it because it is missing 36 bars
- BFill in the missing bars with the last price
- CAlways flag it as doubtful
- DAccept it: it is a half session (close at 13:00 New York time) and the NYSE calendar says that expecting 42 is correct
Show answer
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.
What do pytest tests on the lake add if each file is already validated before loading it?
- AThey 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
- BNothing: they are redundant
- CThey make the load faster
- DThey replace the manifest
Show answer
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

What a data pipeline does

A vendor with defects

Download and validate

Incremental, idempotent load

Quality tests on the lake

Retrying: idempotency and corrections

Running it every day: cron and orchestrators

Exercises

Self-assessment.

Key points.

Resources.

Research

Documentation

Data

Self-assessment

Key points

Resources