From a8a2104ea2dc22adfa21a2438586484cd7d41225 Mon Sep 17 00:00:00 2001 From: Matt Spinola Date: Sun, 26 Jul 2026 18:16:13 -0400 Subject: [PATCH 1/2] feat: reconcile the ingest manifest in BOTH directions reconcile_manifest() backfilled the manifest from the raw parquets on disk, which fixes a manifest that has fallen BEHIND the disk: an interrupted run leaves fetched tables unrecorded and a restart re-downloads them. The mirror image was unhandled and is the more dangerous one. ingest() derives each table's start date from the manifest's last_date and skips the table when that is already current: rec = manifest.get(key, {}) start = rec["last_date"] + 1 day if rec else cold_start if start > end: continue # "already current" It never checks the parquet exists. So a manifest entry whose file is gone (a partial copy, a store move, a deleted file) is treated as current and NEVER fetched again. No error, no rows, a permanent hole in a paid dataset. The first failure costs money re-downloading data you already have. This one leaves you believing you have data you do not. reconcile_manifest now prunes those entries so the next run re-fetches them, and returns {"recorded": {...}, "pruned": [...]}. --reconcile-databento reports the prune separately and says plainly what would otherwise have happened. prune=False keeps the old behaviour. Note the manifest legitimately runs ahead of the disk WHILE a batch ingest is in flight, since the batch API completes and then lands its files in bulk. Observed on a live run: 11 files at one moment, 145 ten minutes later. So this is a between-runs repair, not something to run against a live ingest. Suite 121 passed (3 new tests), ruff clean. Co-Authored-By: Claude Opus 5 --- src/cotdata/providers/databento.py | 46 ++++++++++++++++---- src/cotdata/update.py | 25 ++++++++--- tests/test_databento_ingest.py | 70 ++++++++++++++++++++++++++++++ 3 files changed, 125 insertions(+), 16 deletions(-) diff --git a/src/cotdata/providers/databento.py b/src/cotdata/providers/databento.py index 87c8aa6..c0330db 100644 --- a/src/cotdata/providers/databento.py +++ b/src/cotdata/providers/databento.py @@ -665,14 +665,28 @@ def _ingest_n1_stats_windowed(client, dataset, s, manifest, end, window) -> int: return added -def reconcile_manifest() -> dict: - """Rebuild the ingest manifest from the raw parquet files actually on disk. - - ingest() writes the raw parquets incrementally, but a run interrupted before it - persisted the manifest can leave the manifest behind what is on disk — a restart - would then re-download already-fetched tables. This backfills the manifest from the - raw store so a restart resumes correctly. Reads only local files, never the API. - Returns ``{manifest_key: last_date}`` for every table it recorded.""" +def reconcile_manifest(*, prune: bool = True) -> dict: + """Make the ingest manifest match the raw parquet files actually on disk. + + The manifest is the resume ledger: ingest() computes each table's start date from + ``last_date`` and skips the table entirely when that is already current. It never + checks that the file exists. So the manifest drifting out of step with the disk is + silently destructive in BOTH directions, and this fixes both. + + * **Manifest behind disk.** ingest() writes raw parquets incrementally, so a run + interrupted before it persisted the manifest leaves fetched tables unrecorded and + a restart re-downloads them. Backfilled from the files. + * **Manifest ahead of disk.** An entry whose parquet is missing (a partial copy, a + store move, a deleted file) still carries a current ``last_date``, so a restart + marks it "already current" and NEVER fetches it. No error, no row, a permanent + hole in a paid dataset. Pruned, so the next run re-fetches it. + + The second case is the dangerous one: the first costs money re-downloading data you + already have, the second leaves you believing you have data you do not. + + Reads only local files, never the API. Returns + ``{"recorded": {key: last_date}, "pruned": [key, ...]}``. + """ manifest = _load_ingest_manifest() recorded: dict = {} for schema in _SCHEMAS: @@ -703,8 +717,22 @@ def reconcile_manifest() -> dict: "reconciled": True, } recorded[key] = newest + + pruned: list = [] + if prune: + for key in sorted(manifest): + try: + sym_feed, schema = key.rsplit(":", 1) + except ValueError: + continue + sub = "ohlcv" if schema == "ohlcv-1d" else "statistics" + if not (raw_root() / sub / f"{sym_feed}.parquet").exists(): + pruned.append(key) + for key in pruned: + del manifest[key] + _write_ingest_manifest(manifest) - return recorded + return {"recorded": recorded, "pruned": pruned} # ── Batch ingest: robust large pulls via the databento batch API ───────────── diff --git a/src/cotdata/update.py b/src/cotdata/update.py index d77daec..270d6cc 100644 --- a/src/cotdata/update.py +++ b/src/cotdata/update.py @@ -87,9 +87,12 @@ def main(argv=None, half=None) -> None: help="Prune manifest entries whose parquet file is missing (ghosts " "from old naming), refresh status.json, and exit. Never touches data.") p.add_argument("--reconcile-databento", action="store_true", - help="Rebuild the databento raw-store ingest manifest from the parquet files " - "on disk, so an interrupted --ingest-databento resumes instead of " - "re-downloading. Reads only local files, no API. Exit.") + help="Make the databento ingest manifest match the parquet files on disk, " + "in BOTH directions: record tables an interrupted run left " + "unrecorded (so a restart does not re-download them), and prune " + "entries whose file is missing (so a restart does not skip them as " + "'already current' and leave a silent hole). Local files only, no " + "API. Exit.") args = p.parse_args(argv) if half: _reject_other_half(p, args, half) @@ -134,14 +137,22 @@ def main(argv=None, half=None) -> None: if args.reconcile_databento: from .providers import databento - recorded = databento.reconcile_manifest() - if not recorded: - print("databento reconcile: no raw parquet files found — nothing to record.") - else: + res = databento.reconcile_manifest() + recorded, pruned = res["recorded"], res["pruned"] + if not recorded and not pruned: + print("databento reconcile: manifest already matches the raw store.") + if recorded: syms = sorted({k.split(".n.")[0] for k in recorded}) print(f"databento reconcile: recorded {len(recorded)} raw table(s) across " f"{len(syms)} symbol(s) into the ingest manifest.") print(" symbols: " + ", ".join(syms)) + if pruned: + syms = sorted({k.split(".n.")[0] for k in pruned}) + print(f"databento reconcile: PRUNED {len(pruned)} manifest entr" + f"{'y' if len(pruned) == 1 else 'ies'} with no parquet on disk, across " + f"{len(syms)} symbol(s). The next --ingest-databento will re-fetch these; " + f"without the prune it would have skipped them as 'already current'.") + print(" symbols: " + ", ".join(syms)) return if not (args.prices or args.metadata or args.prices_yahoo or args.ingest_databento diff --git a/tests/test_databento_ingest.py b/tests/test_databento_ingest.py index 636a4c3..8054f95 100644 --- a/tests/test_databento_ingest.py +++ b/tests/test_databento_ingest.py @@ -333,3 +333,73 @@ def test_reconcile_manifest_rebuilds_from_raw_store(tmp_path, monkeypatch): client2 = _FakeClient(_frames(dates)) ingest(symbols=["ES"], client=client2, end="2020-01-05", cold_start="2020-01-01") assert client2.calls == [] + + +# ── reconcile: the manifest must match the disk in BOTH directions ──────── +def _raw_layout(tmp_path, monkeypatch): + monkeypatch.setenv("COTDATA_STORE", str(tmp_path)) + root = tmp_path / "_raw" / "databento" + (root / "ohlcv").mkdir(parents=True) + (root / "statistics").mkdir(parents=True) + return root + + +def _write_raw(root, sub, name, dates): + """ts_event is the INDEX for both schemas, matching what _append_raw writes.""" + import pandas as pd + idx = pd.DatetimeIndex(pd.to_datetime(dates), name="ts_event") + df = pd.DataFrame({"open": range(len(dates)), "high": range(len(dates)), + "low": range(len(dates)), "close": range(len(dates)), + "volume": range(len(dates))}, index=idx) + df.to_parquet(root / sub / f"{name}.parquet") + + +def test_reconcile_prunes_entries_whose_parquet_is_missing(tmp_path, monkeypatch): + """The dangerous direction. ingest() derives each table's start from last_date and + skips it when already current, without ever checking the file exists. A manifest + entry with no file therefore means that table is NEVER fetched again: no error, no + rows, a permanent hole in a paid dataset.""" + import json + + from cotdata.providers import databento + root = _raw_layout(tmp_path, monkeypatch) + _write_raw(root, "ohlcv", "ES.n.0", ["2026-07-01", "2026-07-02"]) + (root / "ingest_manifest.json").write_text(json.dumps({ + "ES.n.0:ohlcv-1d": {"last_date": "2026-07-02", "n_rows": 2}, + "GC.n.0:ohlcv-1d": {"last_date": "2026-07-02", "n_rows": 99}, # no file + "GC.n.0:statistics": {"last_date": "2026-07-02", "n_rows": 99}, # no file + })) + + res = databento.reconcile_manifest() + assert set(res["pruned"]) == {"GC.n.0:ohlcv-1d", "GC.n.0:statistics"} + + m = json.loads((root / "ingest_manifest.json").read_text()) + assert "ES.n.0:ohlcv-1d" in m # has a file, kept + assert "GC.n.0:ohlcv-1d" not in m # pruned, so the next run re-fetches + + +def test_reconcile_still_records_files_missing_from_the_manifest(tmp_path, monkeypatch): + """The original direction: a run interrupted before persisting the manifest.""" + import json + + from cotdata.providers import databento + root = _raw_layout(tmp_path, monkeypatch) + _write_raw(root, "ohlcv", "ES.n.0", ["2026-07-01", "2026-07-02"]) + (root / "ingest_manifest.json").write_text(json.dumps({})) + + res = databento.reconcile_manifest() + assert "ES.n.0:ohlcv-1d" in res["recorded"] + assert res["pruned"] == [] + + +def test_reconcile_prune_can_be_turned_off(tmp_path, monkeypatch): + import json + + from cotdata.providers import databento + root = _raw_layout(tmp_path, monkeypatch) + (root / "ingest_manifest.json").write_text(json.dumps( + {"GC.n.0:ohlcv-1d": {"last_date": "2026-07-02", "n_rows": 99}})) + + res = databento.reconcile_manifest(prune=False) + assert res["pruned"] == [] + assert "GC.n.0:ohlcv-1d" in json.loads((root / "ingest_manifest.json").read_text()) From ee2c67cca2bee578679ac4641763c2e2e2a9a04d Mon Sep 17 00:00:00 2001 From: Matt Spinola Date: Mon, 27 Jul 2026 19:24:01 -0400 Subject: [PATCH 2/2] fix: an empty-window advance marker is not a ghost The prune direction deleted a manifest entry whenever its parquet was missing. That is wrong for one entry shape, and it is the shape this very feature makes dangerous. A paged pull that meets an empty window writes `{last_date: ...}` with NO parquet, deliberately (the `elif chunk_days` branch of _ingest_range): "Advance past it so a re-run doesn't refetch the empty span forever." For a symbol whose data begins after the requested floor, EVERY window is empty, so that marker is the only thing the manifest ever gets and no parquet directory is created at all. Reproduced against the real writer: MANIFEST: {"ES.n.0:statistics": {"last_date": "2020-01-05"}, "ES.n.1:statistics": {"last_date": "2020-01-05"}} ohlcv/: (absent) statistics/: (absent) Pruning on "no file" alone deletes exactly those markers and restores the refetch loop they exist to prevent, on every run, against a paid API. This function exists to stop a silent hole in paid data; as written it would have traded that for a silent spend. The test is now "claims rows AND has no file". `_claims_rows` reads n_rows, so a record of fetched data is a candidate and a bare advance marker is not. It errs toward KEEPING on any shape it does not recognise, including the newer `windowed` / `batch` / `reconciled` fields: a wrongly-kept entry costs one skipped table that --ingest-databento reports, a wrongly-pruned marker costs a paid refetch every run thereafter. Five tests, and all five fail against the previous predicate (verified by reverting it). The end-to-end one drives the real ingest rather than a hand-built manifest; its first version used data starting INSIDE the range, which produced a marker the next window immediately overwrote, so it skipped and proved nothing. Found via a cross-session note from the databento work. That note predicted a merge conflict, which did not exist (GitHub reports MERGEABLE/CLEAN), but was right that the two reconcile directions had to be checked against each other's manifest fields. Co-Authored-By: Claude Opus 5 --- src/cotdata/providers/databento.py | 57 +++++++++++++-- tests/test_databento_ingest.py | 109 +++++++++++++++++++++++++++++ 2 files changed, 160 insertions(+), 6 deletions(-) diff --git a/src/cotdata/providers/databento.py b/src/cotdata/providers/databento.py index c0330db..1ba7069 100644 --- a/src/cotdata/providers/databento.py +++ b/src/cotdata/providers/databento.py @@ -665,6 +665,29 @@ def _ingest_n1_stats_windowed(client, dataset, s, manifest, end, window) -> int: return added +def _claims_rows(entry) -> bool: + """Does this manifest entry assert that data was actually fetched? + + The ingest manifest holds two shapes, and only one of them implies a file: + + * a **record** of a fetched table, carrying `first_date` and `n_rows` + * an **advance marker**, `{last_date: ...}` and nothing else, written when a + paged pull meets an empty window so a re-run skips that span instead of + refetching it forever + + Only a record can be a ghost. Treated as a record when it claims rows, so an + entry that is malformed or from a future writer errs toward being KEPT: a + wrongly-kept entry costs one skipped table that `--ingest-databento` reports, + while a wrongly-pruned marker costs a paid refetch on every run thereafter. + """ + if not isinstance(entry, dict): + return False + try: + return int(entry.get("n_rows") or 0) > 0 + except (TypeError, ValueError): + return False + + def reconcile_manifest(*, prune: bool = True) -> dict: """Make the ingest manifest match the raw parquet files actually on disk. @@ -676,14 +699,22 @@ def reconcile_manifest(*, prune: bool = True) -> dict: * **Manifest behind disk.** ingest() writes raw parquets incrementally, so a run interrupted before it persisted the manifest leaves fetched tables unrecorded and a restart re-downloads them. Backfilled from the files. - * **Manifest ahead of disk.** An entry whose parquet is missing (a partial copy, a - store move, a deleted file) still carries a current ``last_date``, so a restart - marks it "already current" and NEVER fetches it. No error, no row, a permanent - hole in a paid dataset. Pruned, so the next run re-fetches it. + * **Manifest ahead of disk.** An entry that claims rows but whose parquet is missing + (a partial copy, a store move, a deleted file) still carries a current + ``last_date``, so a restart marks it "already current" and NEVER fetches it. No + error, no row, a permanent hole in a paid dataset. Pruned, so the next run + re-fetches it. The second case is the dangerous one: the first costs money re-downloading data you already have, the second leaves you believing you have data you do not. + **Empty-window advance markers are not ghosts and are never pruned.** A paged pull + that meets an empty window writes ``{last_date: ...}`` with no parquet on purpose, + so a re-run does not refetch that span forever; for a symbol whose data starts after + the GLBX floor, the marker is all there is. Pruning on "no file" alone would delete + them and restore the loop, so the test is "claims rows AND has no file". See + ``_claims_rows``. + Reads only local files, never the API. Returns ``{"recorded": {key: last_date}, "pruned": [key, ...]}``. """ @@ -726,8 +757,22 @@ def reconcile_manifest(*, prune: bool = True) -> dict: except ValueError: continue sub = "ohlcv" if schema == "ohlcv-1d" else "statistics" - if not (raw_root() / sub / f"{sym_feed}.parquet").exists(): - pruned.append(key) + if (raw_root() / sub / f"{sym_feed}.parquet").exists(): + continue + if not _claims_rows(manifest[key]): + # NOT a ghost. A paged pull that meets an empty window writes a + # `last_date` advance marker with no parquet, deliberately, so a + # re-run does not refetch that empty span forever (see the + # `elif chunk_days` branch in _ingest_range). For a symbol whose + # data starts after the GLBX floor, every early window is empty + # and the marker is ALL there is. + # + # Pruning on "no file" alone would delete exactly those markers and + # restore the refetch loop they exist to prevent, on every run, + # against a paid API. This function exists to stop a silent hole in + # paid data; deleting a marker would trade it for a silent spend. + continue + pruned.append(key) for key in pruned: del manifest[key] diff --git a/tests/test_databento_ingest.py b/tests/test_databento_ingest.py index 8054f95..8ac55de 100644 --- a/tests/test_databento_ingest.py +++ b/tests/test_databento_ingest.py @@ -403,3 +403,112 @@ def test_reconcile_prune_can_be_turned_off(tmp_path, monkeypatch): res = databento.reconcile_manifest(prune=False) assert res["pruned"] == [] assert "GC.n.0:ohlcv-1d" in json.loads((root / "ingest_manifest.json").read_text()) + + +# ── advance markers are not ghosts ─────────────────────────────────────────── +# A paged pull that meets an empty window writes {last_date: ...} with NO parquet, on +# purpose, so a re-run does not refetch that empty span forever. For a symbol whose +# data starts after the GLBX floor, that marker is all there is. Pruning on "no file" +# alone deletes exactly those and restores the loop they exist to prevent, every run, +# against a paid API. + +def test_an_empty_window_marker_survives_the_prune(tmp_path, monkeypatch): + import json + + from cotdata.providers import databento + root = _raw_layout(tmp_path, monkeypatch) + _write_raw(root, "ohlcv", "ES.n.0", ["2026-07-01"]) + (root / "ingest_manifest.json").write_text(json.dumps({ + "ES.n.0:ohlcv-1d": {"last_date": "2026-07-01", "n_rows": 1}, + # The marker: last_date only, no first_date, no n_rows, no file. + "BTC.n.0:statistics": {"last_date": "2020-12-31"}, + })) + + res = databento.reconcile_manifest() + + assert res["pruned"] == [], "an advance marker is not a ghost" + m = json.loads((root / "ingest_manifest.json").read_text()) + assert m["BTC.n.0:statistics"]["last_date"] == "2020-12-31", "marker was destroyed" + + +def test_a_real_ghost_beside_a_marker_is_still_pruned(tmp_path, monkeypatch): + """The narrowing must not cost the prune its purpose.""" + import json + + from cotdata.providers import databento + root = _raw_layout(tmp_path, monkeypatch) + (root / "ingest_manifest.json").write_text(json.dumps({ + "BTC.n.0:statistics": {"last_date": "2020-12-31"}, # marker + "GC.n.0:ohlcv-1d": {"last_date": "2026-07-02", "n_rows": 99, + "first_date": "2020-01-01"}, # ghost + })) + + res = databento.reconcile_manifest() + + assert res["pruned"] == ["GC.n.0:ohlcv-1d"] + m = json.loads((root / "ingest_manifest.json").read_text()) + assert "BTC.n.0:statistics" in m and "GC.n.0:ohlcv-1d" not in m + + +def test_a_zero_row_entry_is_treated_as_a_marker(tmp_path, monkeypatch): + """n_rows == 0 asserts no data, so there is nothing for a missing file to contradict.""" + import json + + from cotdata.providers import databento + root = _raw_layout(tmp_path, monkeypatch) + (root / "ingest_manifest.json").write_text(json.dumps({ + "BTC.n.0:statistics": {"last_date": "2020-12-31", "n_rows": 0}, + })) + + assert databento.reconcile_manifest()["pruned"] == [] + + +def test_an_unrecognised_entry_shape_errs_toward_keeping(tmp_path, monkeypatch): + """A wrongly-KEPT entry costs one skipped table that --ingest-databento reports. A + wrongly-PRUNED marker costs a paid refetch on every run after. Fail the cheap way.""" + import json + + from cotdata.providers import databento + root = _raw_layout(tmp_path, monkeypatch) + (root / "ingest_manifest.json").write_text(json.dumps({ + "BTC.n.0:statistics": {"last_date": "2020-12-31", "n_rows": "lots"}, + "ETH.n.0:statistics": {"windowed": True, "batch": True, "reconciled": True}, + })) + + assert databento.reconcile_manifest()["pruned"] == [] + + +def test_the_marker_the_real_ingest_writes_survives_reconcile(tmp_path, monkeypatch): + """End to end against the actual writer, not a hand-built manifest. + + The GLBX-floor case: a symbol whose data begins after the whole requested range, so + EVERY window is empty and the `elif chunk_days` branch is the only thing that ever + writes. The result is a manifest of pure advance markers and no parquet directory + at all. Under a "no file means ghost" rule reconcile deletes both entries and the + next ingest re-scans the empty span, on every run, against a paid API. + + A first version of this test used data that started INSIDE the range. That produced + a marker on the first window which the next window immediately overwrote with real + rows, so the assertion had nothing to assert and the test skipped. A skipped test is + not evidence. + """ + import cotdata.providers.databento as dbmod + monkeypatch.setenv("COTDATA_DATABENTO_RAW", str(tmp_path)) + monkeypatch.setattr(dbmod, "_STATS_CHUNK_DAYS", 2) + client = _FakeClient(_frames(pd.date_range("2021-06-01", periods=3))) + + dbmod.ingest(symbols=["ES"], client=client, end="2020-01-06", + cold_start="2020-01-01") + + man_path = tmp_path / "ingest_manifest.json" + markers = {k: v for k, v in json.loads(man_path.read_text()).items() + if not v.get("n_rows")} + assert markers, "fixture no longer produces an advance marker; the test is vacuous" + assert not (tmp_path / "ohlcv").exists(), "expected no parquet for an all-empty pull" + + dbmod.reconcile_manifest() + + after = json.loads(man_path.read_text()) + for key, val in markers.items(): + assert key in after, f"reconcile destroyed the advance marker {key}" + assert after[key]["last_date"] == val["last_date"]