diff --git a/src/cotdata/providers/databento.py b/src/cotdata/providers/databento.py
index 8be6060..2070fe7 100644
--- a/src/cotdata/providers/databento.py
+++ b/src/cotdata/providers/databento.py
@@ -418,26 +418,53 @@ def _client_from_env():
     return db.Historical(key=key)
 
 
-def _fetch(client, dataset: str, dbsym: str, schema: str, start: str, end: str) -> pd.DataFrame:
-    with warnings.catch_warnings():
-        warnings.filterwarnings("ignore", message=".*No data found.*")
-        warnings.filterwarnings("ignore", message=".*did not resolve.*")
-        # Databento flags a handful of historically "degraded" sessions (e.g. 2014-06-11)
-        # on every request. It's a known data-condition note, not an error, and it floods
-        # the ingest/cron logs — quiet it.
-        warnings.filterwarnings("ignore", message=".*reduced quality.*")
-        data = client.timeseries.get_range(
-            dataset=dataset, symbols=[dbsym], stype_in="continuous",
-            schema=schema, start=start, end=end)
-    return data.to_df()
+def _fetch(client, dataset: str, dbsym: str, schema: str, start: str, end: str,
+           *, retries: int = 3, backoff: float = 5.0) -> pd.DataFrame:
+    """One databento get_range, with retry + linear backoff on transient network
+    failures (read timeouts, dropped/aborted streams) — the historical API throws these
+    often on large pulls. Raises the last error only after ``retries`` attempts."""
+    last = None
+    for attempt in range(1, retries + 1):
+        try:
+            with warnings.catch_warnings():
+                warnings.filterwarnings("ignore", message=".*No data found.*")
+                warnings.filterwarnings("ignore", message=".*did not resolve.*")
+                # Databento flags a handful of historically "degraded" sessions (e.g.
+                # 2014-06-11) on every request. A known data-condition note, not an error,
+                # and it floods the logs — quiet it.
+                warnings.filterwarnings("ignore", message=".*reduced quality.*")
+                data = client.timeseries.get_range(
+                    dataset=dataset, symbols=[dbsym], stype_in="continuous",
+                    schema=schema, start=start, end=end)
+            return data.to_df()
+        except Exception as e:  # noqa: BLE001 — databento/network is flaky; retry transient
+            last = e
+            if attempt < retries:
+                time.sleep(backoff * attempt)
+    raise last
+
+
+def _fmt_hms(seconds: float) -> str:
+    """Compact H:MM:SS for progress logging."""
+    s = int(seconds)
+    return f"{s // 3600}:{(s % 3600) // 60:02d}:{s % 60:02d}"
 
 
 def ingest(symbols=None, *, client=None, dataset="GLBX.MDP3", end=None,
-           cold_start=GLBX_HISTORY_FLOOR) -> dict:
+           cold_start=GLBX_HISTORY_FLOOR, n1_stats_window=None) -> dict:
     """Fetch raw databento daily bars (.n.0 + .n.1) and statistics into the raw
     store, append-only. Resumes each (symbol, feed, schema) from its manifest
-    last_date+1. Scoped to registry symbols that databento can serve (a non-null
-    ``databento`` mapping); pass ``symbols`` to narrow further.
+    last_date+1 — the manifest is persisted after EVERY fetch, so an interrupted run
+    (a from-inception ``statistics`` pull can take minutes) resumes where it left off
+    instead of re-downloading. Logs per-asset progress ``[i/N]`` and timing. Scoped to
+    registry symbols that databento can serve (a non-null ``databento`` mapping); pass
+    ``symbols`` to narrow further.
+
+    ``n1_stats_window`` (days): when set, the n.1 ``statistics`` schema is fetched only in
+    ±window-day windows around each roll date (found from the on-disk n.0 ohlcv) instead
+    of its full history. n.1 settlement is only used at rolls, so this drops the biggest
+    avoidable download with NO back-adjustment accuracy loss. The full n.0 statistics
+    (daily settlement + OI) and n.1 ohlcv are unaffected.
 
     `client` is injectable (databento.Historical-shaped) for tests; the default
     builds one from DATABENTO_API_KEY. Returns {kind, ok, symbols, rows}."""
@@ -455,11 +482,23 @@ def ingest(symbols=None, *, client=None, dataset="GLBX.MDP3", end=None,
     end = end or (pd.Timestamp.now().normalize() - pd.Timedelta(days=1)).strftime("%Y-%m-%d")
     manifest = _load_ingest_manifest()
     total_rows, failed = 0, 0
+    n = len(targets)
+    run_start = time.monotonic()
 
-    for s in targets:
+    for i, s in enumerate(targets, 1):
+        sym_start = time.monotonic()
+        sym_rows = 0
+        print(f"[{i}/{n}] {s.internal}: ingesting  (run elapsed {_fmt_hms(time.monotonic() - run_start)})")
         for feed in _FEEDS:
             dbsym = f"{s.databento}{feed}"
             for schema in _SCHEMAS:
+                if (n1_stats_window is not None and feed == ".n.1"
+                        and schema == "statistics"):
+                    added = _ingest_n1_stats_windowed(
+                        client, dataset, s, manifest, end, n1_stats_window)
+                    sym_rows += added
+                    total_rows += added
+                    continue
                 key = f"{s.internal}{feed}:{schema}"
                 rec = manifest.get(key, {})
                 last = rec.get("last_date")
@@ -467,12 +506,16 @@ def ingest(symbols=None, *, client=None, dataset="GLBX.MDP3", end=None,
                          if last else cold_start)
                 if pd.Timestamp(start) < pd.Timestamp(GLBX_HISTORY_FLOOR):
                     start = GLBX_HISTORY_FLOOR
-                if pd.Timestamp(start) > pd.Timestamp(end):
-                    continue  # already current
+                # Already current. databento requires start < end; a start == end query
+                # 422s (data_time_range_start_on_or_after_end), so skip on >=.
+                if pd.Timestamp(start) >= pd.Timestamp(end):
+                    continue
+                t0 = time.monotonic()
                 try:
                     raw = _fetch(client, dataset, dbsym, schema, start, end)
                 except Exception as e:  # noqa: BLE001 — databento/network is flaky
-                    print(f"{s.internal}{feed} {schema}: databento fetch failed ({start}..{end}) — {e}")
+                    print(f"    {s.internal}{feed} {schema}: fetch failed ({start}..{end}) "
+                          f"after {time.monotonic() - t0:.1f}s — {e}")
                     failed += 1
                     continue
                 if raw is None or raw.empty:
@@ -488,10 +531,281 @@ def ingest(symbols=None, *, client=None, dataset="GLBX.MDP3", end=None,
                     "n_rows": int(rec.get("n_rows", 0)) + len(raw),
                     "fetched_at": dt.datetime.utcnow().isoformat(timespec="seconds") + "Z",
                 }
+                # Persist after EACH fetch so an interrupted run resumes here, not from
+                # scratch. The write is a small atomic JSON, negligible next to the fetch.
+                _write_ingest_manifest(manifest)
+                sym_rows += len(raw)
                 total_rows += len(raw)
-                print(f"{s.internal}{feed} {schema}: +{len(raw):5d} rows ({start}..{newest}) -> raw store")
+                print(f"    {s.internal}{feed} {schema}: +{len(raw):>8,} rows "
+                      f"({start}..{newest}) in {time.monotonic() - t0:.1f}s")
+        print(f"[{i}/{n}] {s.internal}: done in {_fmt_hms(time.monotonic() - sym_start)} "
+              f"({sym_rows:,} rows)")
 
     _write_ingest_manifest(manifest)
+    print(f"databento ingest: {n} symbols, {total_rows:,} rows in "
+          f"{_fmt_hms(time.monotonic() - run_start)}"
+          + (f"  ({failed} fetch failure(s))" if failed else ""))
+    return {"kind": "ingest_databento", "ok": failed == 0, "symbols": n, "rows": total_rows}
+
+
+def _roll_dates_on_disk(symbol: str) -> pd.DatetimeIndex:
+    """Roll dates for a symbol, read from the raw n.0 ohlcv already on disk (the last
+    session before the front contract's ``instrument_id`` changes). Empty if n.0 ohlcv is
+    absent — the windowed n.1 stats fetch needs n.0 ohlcv ingested first (it always is,
+    since .n.0/ohlcv-1d is fetched before .n.1/statistics in the loop)."""
+    n0 = _read_ohlcv(symbol, ".n.0")
+    if n0.empty:
+        return pd.DatetimeIndex([])
+    key = _roll_key(n0)
+    if key is None:
+        return pd.DatetimeIndex([])
+    ids = n0[key]
+    is_roll = ids.ne(ids.shift(-1)) & ids.shift(-1).notna()
+    return n0.index[is_roll]
+
+
+def _ingest_n1_stats_windowed(client, dataset, s, manifest, end, window) -> int:
+    """Fetch n.1 ``statistics`` only in ±``window``-day windows around each roll date,
+    instead of the full history. Settlement is only read at rolls, so this is accuracy-
+    neutral. Resumes from the newest roll already covered. Returns rows added."""
+    key = f"{s.internal}.n.1:statistics"
+    rolls = _roll_dates_on_disk(s.internal)
+    if len(rolls) == 0:
+        print(f"    {s.internal}.n.1 statistics: no roll dates yet (n.0 ohlcv missing) — skipped")
+        return 0
+    rec = manifest.get(key, {})
+    covered = pd.Timestamp(rec["last_date"]) if rec.get("last_date") else None
+    todo = [d for d in rolls if covered is None or d > covered]
+    if not todo:
+        return 0
+    dbsym = f"{s.databento}.n.1"
+    end_ts = pd.Timestamp(end)
+    added, t0 = 0, time.monotonic()
+    # Advance the resume watermark only through the last CONTIGUOUS success, so a transient
+    # mid-history window failure is retried on the next run rather than silently leaving that
+    # roll's gap unmeasured.
+    watermark, ok = covered, True
+    for d in todo:
+        wstart = d - pd.Timedelta(days=window)
+        wend = min(d + pd.Timedelta(days=window + 1), end_ts)   # exclusive end
+        if wstart >= wend:
+            continue
+        try:
+            raw = _fetch(client, dataset, dbsym, "statistics",
+                         wstart.strftime("%Y-%m-%d"), wend.strftime("%Y-%m-%d"))
+        except Exception as e:  # noqa: BLE001 — databento/network is flaky
+            print(f"    {s.internal}.n.1 statistics [{d.date()}]: fetch failed — {e}")
+            ok = False
+            continue
+        if raw is not None and not raw.empty:
+            raw = _normalize(raw, "statistics")
+            if not raw.empty:
+                _append_raw(s.internal, ".n.1", "statistics", raw)
+                added += len(raw)
+        if ok:
+            watermark = d
+    if watermark is not None:
+        manifest[key] = {
+            "first_date": rec.get("first_date") or str(todo[0].date()),
+            "last_date": str(pd.Timestamp(watermark).date()),
+            "n_rows": int(rec.get("n_rows", 0)) + added,
+            "fetched_at": dt.datetime.utcnow().isoformat(timespec="seconds") + "Z",
+            "windowed": True,
+        }
+        _write_ingest_manifest(manifest)
+    print(f"    {s.internal}.n.1 statistics: +{added:>8,} rows across {len(todo)} roll "
+          f"window(s) (±{window}d) in {time.monotonic() - t0:.1f}s"
+          + ("" if ok else "  (some windows failed — retry on re-run)"))
+    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."""
+    manifest = _load_ingest_manifest()
+    recorded: dict = {}
+    for schema in _SCHEMAS:
+        sub = "ohlcv" if schema == "ohlcv-1d" else "statistics"
+        d = raw_root() / sub
+        if not d.exists():
+            continue
+        for p in sorted(d.glob("*.parquet")):
+            name = p.stem  # "<symbol><feed>", e.g. "6E.n.0"
+            feed = next((f for f in _FEEDS if name.endswith(f)), None)
+            if feed is None:
+                continue
+            symbol = name[: -len(feed)]
+            try:
+                df = pd.read_parquet(p)
+            except Exception as e:  # noqa: BLE001
+                print(f"reconcile: could not read {p.name} — {e}")
+                continue
+            if df.empty:
+                continue
+            first, newest = _date_bounds(df, schema)
+            key = f"{symbol}{feed}:{schema}"
+            manifest[key] = {
+                "first_date": first,
+                "last_date": newest,
+                "n_rows": int(len(df)),
+                "fetched_at": dt.datetime.utcnow().isoformat(timespec="seconds") + "Z",
+                "reconciled": True,
+            }
+            recorded[key] = newest
+    _write_ingest_manifest(manifest)
+    return recorded
+
+
+# ── Batch ingest: robust large pulls via the databento batch API ─────────────
+# The streaming timeseries.get_range chokes on from-inception statistics (read timeouts,
+# dropped streams). The batch API prepares result files server-side and delivers them as
+# a download (robust, resumable), which is what it is designed for. Same raw store +
+# manifest as the streaming ingest, so build()/reconcile() are unchanged. Statistics are
+# fetched FULL here (no windowing — batch handles the volume).
+
+def _batch_fetch(client, dataset, dbsyms, schema, start, end, *, poll=30, timeout=14400) -> dict:
+    """Submit one databento BATCH job for many continuous symbols over [start, end), wait
+    for it, download the CSV(s), and return ``{dbsym: raw_df}`` grouped by the ``symbol``
+    column. Raises on job failure/timeout."""
+    import tempfile
+
+    def _jid(j):
+        return j.get("id") if isinstance(j, dict) else getattr(j, "id", None)
+
+    def _state(j):
+        return (j.get("state") if isinstance(j, dict) else getattr(j, "state", None)) or "unknown"
+
+    job = client.batch.submit_job(
+        dataset=dataset, symbols=list(dbsyms), stype_in="continuous", schema=schema,
+        start=start, end=end, encoding="csv", split_symbols=False, delivery="download")
+    job_id = _jid(job)
+    if not job_id:
+        raise RuntimeError("batch submit_job returned no job id")
+
+    waited = 0
+    while True:
+        me = next((j for j in client.batch.list_jobs() if _jid(j) == job_id), None)
+        st = _state(me)
+        if st == "done":
+            break
+        if st == "expired" or "fail" in str(st).lower():
+            raise RuntimeError(f"batch job {job_id} state={st}")
+        if waited >= timeout:
+            raise TimeoutError(f"batch job {job_id} not done after {timeout}s (state={st})")
+        time.sleep(poll)
+        waited += poll
+
+    frames: dict = {}
+    with tempfile.TemporaryDirectory() as tmp:
+        for path in client.batch.download(job_id=job_id, output_dir=tmp):
+            if ".csv" not in str(path):
+                continue
+            try:
+                df = pd.read_csv(path)                       # pandas infers .zst/.gz compression
+            except Exception as e:  # noqa: BLE001
+                print(f"batch: could not read {path} — {e}")
+                continue
+            if df.empty or "symbol" not in df.columns:
+                continue
+            for dbsym, g in df.groupby("symbol"):
+                frames.setdefault(dbsym, []).append(g)
+    return {k: pd.concat(v, ignore_index=True) for k, v in frames.items()}
+
+
+def _normalize_batch_csv(df: pd.DataFrame, schema: str) -> pd.DataFrame:
+    """Bring a batch CSV frame to the same shape the streaming ``_normalize`` produces
+    (ohlcv: tz-naive daily Date index; statistics: tz-naive columns)."""
+    df = df.copy()
+    for c in ("ts_event", "ts_ref"):
+        if c in df.columns:
+            df[c] = _to_naive(df[c])
+    if schema == "ohlcv-1d":
+        df = df.set_index("ts_event")
+        df.index = df.index.normalize()
+        df.index.name = "Date"
+        return df[~df.index.duplicated(keep="last")].sort_index()
+    return df.drop_duplicates().reset_index(drop=True)
+
+
+def ingest_batch(symbols=None, *, client=None, dataset="GLBX.MDP3", end=None,
+                 cold_start=GLBX_HISTORY_FLOOR, poll=30) -> dict:
+    """Batch-API variant of ingest(): fetch each (feed, schema) as a databento batch job
+    (download-to-file) instead of streaming — robust for large from-inception pulls. One
+    job per (feed, schema) covers every symbol still needing it, from the earliest resume
+    point; per-symbol rows are appended and the manifest advanced individually, so resume
+    and build are unchanged. Returns {kind, ok, symbols, rows}."""
+    targets = [s for s in all_symbols()
+               if s.databento and s.price_source in (None, "databento")
+               and (symbols is None or s.internal in symbols)]
+    if not targets:
+        print("databento batch ingest: no databento-capable symbols"
+              + (f" among {symbols}" if symbols else ""))
+        return {"kind": "ingest_databento", "ok": True, "symbols": 0, "rows": 0}
+    if client is None:
+        client = _client_from_env()
+
+    end = end or (pd.Timestamp.now().normalize() - pd.Timedelta(days=1)).strftime("%Y-%m-%d")
+    end_ts = pd.Timestamp(end)
+    manifest = _load_ingest_manifest()
+    total_rows, failed = 0, 0
+    run_start = time.monotonic()
+
+    for feed in _FEEDS:
+        for schema in _SCHEMAS:
+            needed, starts = {}, []
+            for s in targets:
+                last = manifest.get(f"{s.internal}{feed}:{schema}", {}).get("last_date")
+                st = (pd.Timestamp(last) + pd.Timedelta(days=1)) if last else pd.Timestamp(cold_start)
+                if st < pd.Timestamp(GLBX_HISTORY_FLOOR):
+                    st = pd.Timestamp(GLBX_HISTORY_FLOOR)
+                if st >= end_ts:
+                    continue                                # already current
+                needed[f"{s.databento}{feed}"] = s.internal
+                starts.append(st)
+            if not needed:
+                continue
+            job_start = min(starts).strftime("%Y-%m-%d")
+            t0 = time.monotonic()
+            print(f"batch {feed} {schema}: {len(needed)} symbols {job_start}..{end} — submitting job...")
+            try:
+                frames = _batch_fetch(client, dataset, list(needed), schema, job_start, end, poll=poll)
+            except Exception as e:  # noqa: BLE001 — batch/network is flaky
+                print(f"batch {feed} {schema}: FAILED — {e}")
+                failed += 1
+                continue
+            wrote = 0
+            for dbsym, raw in frames.items():
+                internal = needed.get(dbsym)
+                if internal is None:
+                    continue
+                raw = _normalize_batch_csv(raw, schema)
+                if raw.empty:
+                    continue
+                _append_raw(internal, feed, schema, raw)
+                first, newest = _date_bounds(raw, schema)
+                key = f"{internal}{feed}:{schema}"
+                rec = manifest.get(key, {})
+                manifest[key] = {
+                    "first_date": rec.get("first_date") or first,
+                    "last_date": newest,
+                    "n_rows": int(rec.get("n_rows", 0)) + len(raw),
+                    "fetched_at": dt.datetime.utcnow().isoformat(timespec="seconds") + "Z",
+                    "batch": True,
+                }
+                wrote += len(raw)
+            _write_ingest_manifest(manifest)
+            total_rows += wrote
+            print(f"batch {feed} {schema}: +{wrote:,} rows for {len(frames)} symbols in "
+                  f"{_fmt_hms(time.monotonic() - t0)}")
+
+    print(f"databento batch ingest: {len(targets)} symbols, {total_rows:,} rows in "
+          f"{_fmt_hms(time.monotonic() - run_start)}"
+          + (f"  ({failed} job failure(s))" if failed else ""))
     return {"kind": "ingest_databento", "ok": failed == 0, "symbols": len(targets), "rows": total_rows}
 
 
diff --git a/src/cotdata/update.py b/src/cotdata/update.py
index 58ab577..e29c8e7 100644
--- a/src/cotdata/update.py
+++ b/src/cotdata/update.py
@@ -24,6 +24,16 @@ def main() -> None:
     p.add_argument("--build-databento", action="store_true",
                    help="Databento Stage 2 (free, no API): build back-adjusted prices from the "
                         "raw store into the cotdata store. Run after --ingest-databento.")
+    p.add_argument("--windowed-n1-stats", nargs="?", type=int, const=3, default=None, metavar="DAYS",
+                   help="With --ingest-databento: fetch the n.1 statistics schema only in "
+                        "±DAYS windows around roll dates (default 3) instead of full history. "
+                        "n.1 settlement is only needed at rolls, so this cuts the biggest "
+                        "avoidable download with no accuracy loss. Recommended for the backfill.")
+    p.add_argument("--batch", action="store_true",
+                   help="With --ingest-databento: use the databento BATCH API (prepare + "
+                        "download files) instead of streaming. Far more robust for large "
+                        "from-inception pulls (avoids the streaming read-timeouts); statistics "
+                        "are fetched full. Slower to start (async job) but reliable.")
     p.add_argument("--cot-legacy", action="store_true", help="Update CFTC COT Legacy (cross-platform).")
     p.add_argument("--cot-disagg", action="store_true", help="Update CFTC COT Disaggregated Futures-Only (cross-platform).")
     p.add_argument("--cot-tff", action="store_true", help="Update Traders in Financial Futures (TFF) COT (cross-platform).")
@@ -46,6 +56,10 @@ def main() -> None:
     p.add_argument("--reconcile", action="store_true",
                    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.")
     args = p.parse_args()
 
     config.store_root()  # fail fast if COTDATA_STORE unset
@@ -71,6 +85,18 @@ def main() -> None:
                                                "at": _dt.datetime.utcnow().isoformat(timespec="seconds") + "Z"})
         return
 
+    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:
+            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))
+        return
+
     if not (args.prices or args.metadata or args.prices_yahoo or args.ingest_databento
             or args.build_databento or args.cot_legacy or args.cot_disagg or args.cot_tff or args.cot_all):
         p.error("nothing to do — pass --check, --prices, --prices-yahoo, --ingest-databento, "
@@ -110,7 +136,10 @@ def main() -> None:
 
     if args.ingest_databento:
         from .providers import databento
-        r = databento.ingest(symbols=args.symbols)
+        if args.batch:
+            r = databento.ingest_batch(symbols=args.symbols)
+        else:
+            r = databento.ingest(symbols=args.symbols, n1_stats_window=args.windowed_n1_stats)
         kinds.append("ingest_databento")
         if not (r or {}).get("ok", True):
             failed_kinds.append("ingest_databento")
diff --git a/tests/test_databento_build.py b/tests/test_databento_build.py
index c2475b8..a4019be 100644
--- a/tests/test_databento_build.py
+++ b/tests/test_databento_build.py
@@ -134,3 +134,81 @@ def test_build_skips_symbol_missing_from_raw_store(stores):
     # Nothing written for CL → build reports it skipped (and it's not 'ok').
     res = build(["CL"])
     assert res["wrote"] == 0 and res["ok"] is False
+
+
+# ── windowed n.1 statistics: same back-adjustment at a fraction of the download ──
+class _WResp:
+    def __init__(self, df):
+        self._df = df
+
+    def to_df(self):
+        return self._df
+
+
+class _WClient:
+    def __init__(self, frames):
+        self.frames = frames
+        self.calls = []
+
+    @property
+    def timeseries(client_self):
+        class _TS:
+            def get_range(ts_self, *, dataset, symbols, stype_in, schema, start, end):
+                client_self.calls.append((symbols[0], schema, start, end))
+                fr = client_self.frames.get((symbols[0], schema))
+                if fr is None or fr.empty:
+                    return _WResp(pd.DataFrame())
+                naive = fr.index.tz_convert(None)
+                mask = (naive >= pd.Timestamp(start)) & (naive < pd.Timestamp(end))
+                return _WResp(fr[mask])
+        return _TS()
+
+
+def _roll_frames(dates):
+    idx = pd.to_datetime(dates).tz_localize("UTC")
+    idx.name = "ts_event"
+
+    def ohlcv(close, iid):
+        return pd.DataFrame({"open": close, "high": close, "low": close, "close": close,
+                             "volume": [1000] * len(close), "instrument_id": iid,
+                             "symbol": ["ES.n"] * len(close)}, index=idx)
+
+    def stats(settle):
+        return pd.DataFrame({"ts_ref": idx, "stat_type": 3, "price": settle,
+                             "quantity": float("nan")}, index=idx)
+
+    return {
+        ("ES.n.0", "ohlcv-1d"): ohlcv([100, 101, 102, 110, 111, 112], [10, 10, 10, 20, 20, 20]),
+        ("ES.n.1", "ohlcv-1d"): ohlcv([103, 104, 105, 113, 114, 115], [20, 20, 20, 30, 30, 30]),
+        ("ES.n.0", "statistics"): stats([100.5, 101.5, 102.5, 110.5, 111.5, 112.5]),
+        ("ES.n.1", "statistics"): stats([103.5, 104.5, 105.5, 113.5, 114.5, 115.5]),
+    }
+
+
+def test_windowed_n1_stats_matches_full_backadj(tmp_path, monkeypatch):
+    import cotdata
+    from cotdata.providers.databento import ingest
+    dates = pd.date_range("2020-01-01", periods=6)          # roll at 2020-01-03 (id 10→20)
+
+    def run(tag, window):
+        root = tmp_path / tag
+        monkeypatch.setenv("COTDATA_DATABENTO_RAW", str(root / "raw"))
+        monkeypatch.setenv("COTDATA_STORE", str(root / "store"))
+        client = _WClient(_roll_frames(dates))
+        ingest(symbols=["ES"], client=client, end="2020-01-07", cold_start="2020-01-01",
+               n1_stats_window=window)
+        build(["ES"])
+        return cotdata.get_prices("ES", adjustment="backadj"), client
+
+    full_bad, full_client = run("full", None)
+    win_bad, win_client = run("win", 1)
+
+    # Identical back-adjustment (gap = n1_settle - n0_settle = 105.5 - 102.5 = +3 either way).
+    assert list(win_bad["Close"]) == list(full_bad["Close"]) == [103.5, 104.5, 105.5, 110.5, 111.5, 112.5]
+
+    # ...but the n.1 statistics fetch was a narrow roll window, not the full range.
+    def n1_stat_spans(c):
+        return [(s, e) for (sym, sch, s, e) in c.calls if sym == "ES.n.1" and sch == "statistics"]
+    assert n1_stat_spans(full_client) == [("2020-01-01", "2020-01-07")]   # one full-range pull
+    win_spans = n1_stat_spans(win_client)
+    assert win_spans and all(s >= "2020-01-02" and e <= "2020-01-05" for s, e in win_spans)
diff --git a/tests/test_databento_ingest.py b/tests/test_databento_ingest.py
index 94d55e8..b598e3d 100644
--- a/tests/test_databento_ingest.py
+++ b/tests/test_databento_ingest.py
@@ -7,8 +7,9 @@
 import json
 
 import pandas as pd
+import pytest
 
-from cotdata.providers.databento import ingest
+from cotdata.providers.databento import _ingest_manifest_path, ingest, reconcile_manifest
 
 
 # ── a databento.Historical-shaped fake ───────────────────────────────────────
@@ -136,3 +137,128 @@ def test_ingest_skips_databento_null_symbol(tmp_path, monkeypatch):
     res = ingest(symbols=["CC"], client=client, end="2020-01-05")   # CC is databento: null
     assert res["symbols"] == 0
     assert client.calls == []
+
+
+def test_ingest_skips_when_start_equals_end(tmp_path, monkeypatch):
+    # Regression: an up-to-date symbol computes start == end, which databento rejects
+    # (422 data_time_range_start_on_or_after_end). The guard must skip, not fetch.
+    monkeypatch.setenv("COTDATA_DATABENTO_RAW", str(tmp_path))
+    ingest(symbols=["ES"], client=_FakeClient(_frames(pd.date_range("2020-01-01", periods=5))),
+           end="2020-01-05", cold_start="2020-01-01")               # stores last_date = 2020-01-05
+    # end == stored last_date + 1 → start (last+1) == end → must skip, no API call.
+    client2 = _FakeClient(_frames(pd.date_range("2020-01-01", periods=6)))
+    ingest(symbols=["ES"], client=client2, end="2020-01-06", cold_start="2020-01-01")
+    assert client2.calls == []
+
+
+def test_fetch_retries_transient_failures_then_succeeds():
+    from cotdata.providers.databento import _fetch
+    n = {"calls": 0}
+
+    class _TS:
+        def get_range(self, **k):
+            n["calls"] += 1
+            if n["calls"] < 3:
+                raise ConnectionError("Response ended prematurely")
+            return _FakeResp(pd.DataFrame({"x": [1]}))
+
+    class _C:
+        timeseries = _TS()
+
+    df = _fetch(_C(), "GLBX.MDP3", "ES.n.0", "ohlcv-1d", "2020-01-01", "2020-01-05",
+                retries=3, backoff=0)
+    assert n["calls"] == 3 and not df.empty
+
+
+def test_fetch_raises_after_exhausting_retries():
+    from cotdata.providers.databento import _fetch
+
+    class _TS:
+        def get_range(self, **k):
+            raise TimeoutError("read timed out")
+
+    class _C:
+        timeseries = _TS()
+
+    with pytest.raises(TimeoutError):
+        _fetch(_C(), "GLBX.MDP3", "ES.n.0", "ohlcv-1d", "2020-01-01", "2020-01-05",
+               retries=2, backoff=0)
+
+
+class _FakeBatch:
+    """databento client.batch stand-in: submit → 'done' → download CSV files."""
+    def __init__(self, dates):
+        self.dates = dates
+        self.submitted = []
+
+    def submit_job(self, **k):
+        self.submitted.append(k)
+        return {"id": "j1", "state": "queued"}
+
+    def list_jobs(self):
+        return [{"id": "j1", "state": "done"}]
+
+    def download(self, *, job_id, output_dir):
+        import os
+        k = self.submitted[-1]
+        schema, syms = k["schema"], k["symbols"]
+        rows = []
+        for sym in syms:
+            for i, d in enumerate(self.dates):
+                base = {"ts_event": d.isoformat(), "instrument_id": 10, "symbol": sym}
+                if schema == "ohlcv-1d":
+                    rows.append({**base, "open": 100 + i, "high": 100 + i, "low": 100 + i,
+                                 "close": 100 + i, "volume": 1000})
+                else:
+                    rows.append({**base, "ts_ref": d.isoformat(), "stat_type": 3,
+                                 "price": 100.5 + i, "quantity": 0})
+        p = os.path.join(output_dir, f"out.{schema}.csv")
+        pd.DataFrame(rows).to_csv(p, index=False)
+        return [p]
+
+
+class _FakeBatchClient:
+    def __init__(self, dates):
+        self.batch = _FakeBatch(dates)
+
+
+def test_ingest_batch_writes_raw_and_manifest(tmp_path, monkeypatch):
+    monkeypatch.setenv("COTDATA_DATABENTO_RAW", str(tmp_path))
+    dates = pd.date_range("2020-01-01", periods=4)
+    from cotdata.providers.databento import ingest_batch
+
+    res = ingest_batch(symbols=["ES"], client=_FakeBatchClient(dates),
+                       end="2020-01-05", cold_start="2020-01-01")
+    assert res["ok"] and res["symbols"] == 1
+
+    man = json.loads((tmp_path / "ingest_manifest.json").read_text())
+    assert man["ES.n.0:ohlcv-1d"]["last_date"] == "2020-01-04"
+    assert man["ES.n.0:ohlcv-1d"]["batch"] is True
+    assert man["ES.n.0:statistics"]["last_date"] == "2020-01-04"
+
+    ohlcv = pd.read_parquet(tmp_path / "ohlcv" / "ES.n.0.parquet")
+    assert len(ohlcv) == 4 and ohlcv.index.name == "Date" and ohlcv.index.tz is None
+    stats = pd.read_parquet(tmp_path / "statistics" / "ES.n.0.parquet")
+    assert (stats["stat_type"] == 3).all()
+
+
+def test_reconcile_manifest_rebuilds_from_raw_store(tmp_path, monkeypatch):
+    monkeypatch.setenv("COTDATA_DATABENTO_RAW", str(tmp_path))
+    dates = pd.date_range("2020-01-01", periods=5)
+    ingest(symbols=["ES"], client=_FakeClient(_frames(dates)), end="2020-01-05", cold_start="2020-01-01")
+
+    # Simulate a run that wrote the raw parquets but never persisted the manifest.
+    _ingest_manifest_path().unlink()
+
+    recorded = reconcile_manifest()
+    assert recorded, "reconcile should record the on-disk raw tables"
+    man = json.loads(_ingest_manifest_path().read_text())
+    assert man["ES.n.0:ohlcv-1d"]["last_date"] == "2020-01-05"
+    assert man["ES.n.0:ohlcv-1d"]["n_rows"] == 5
+    assert man["ES.n.0:ohlcv-1d"]["reconciled"] is True
+    assert man["ES.n.0:statistics"]["last_date"] == "2020-01-05"
+
+    # With the manifest rebuilt, a re-ingest sees everything current -> no API calls.
+    client2 = _FakeClient(_frames(dates))
+    ingest(symbols=["ES"], client=client2, end="2020-01-05", cold_start="2020-01-01")
+    assert client2.calls == []