feat: reconcile the ingest manifest in BOTH directions

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

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

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