diff --git a/src/cotdata/providers/databento.py b/src/cotdata/providers/databento.py index 87c8aa6..1ba7069 100644 --- a/src/cotdata/providers/databento.py +++ b/src/cotdata/providers/databento.py @@ -665,14 +665,59 @@ 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 _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. + + 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 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, ...]}``. + """ manifest = _load_ingest_manifest() recorded: dict = {} for schema in _SCHEMAS: @@ -703,8 +748,36 @@ 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 (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] + _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..8ac55de 100644 --- a/tests/test_databento_ingest.py +++ b/tests/test_databento_ingest.py @@ -333,3 +333,182 @@ 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()) + + +# ── 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"]