Implement start_date in databento.fetch_daily_ohlc (was silently dead)

CHANGELOG.md

@@ -25,6 +25,21 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   generalization markets. ([#24](https://github.com/mspinola/cotdata/pull/24))
 
 ### Fixed
+- **`databento.fetch_daily_ohlc`'s `start_date` parameter now actually does
+  something** (dormant provider). It was previously silently ignored on every
+  code path — a cold cache always backfilled from 2000-01-01 and a warm cache
+  always resumed from `last_date + 1 day` regardless of what was passed — so a
+  caller trying to bound a fetch (e.g. "just the last 3 months") got a
+  full-history pull instead, at full Databento API cost, with no error or
+  warning. Now: on a **cold** cache (a symbol's first-ever fetch) `start_date`
+  narrows the fetch floor, so a narrow first-time query is actually cheap. On
+  a **warm** cache it does *not* narrow the incremental fetch — the top-up
+  always resumes from the cache's own `last_date + 1`, so a later, narrower
+  `start_date` can never silently truncate a cache other callers already rely
+  on being complete — but the *returned* frame is still filtered to
+  `>= start_date` either way. The on-disk cache always persists the full
+  series regardless of `start_date`; only the fetch cost (cold cache) and the
+  return value (always) are affected.
 - **Fail fast when the Norgate service (NDU) is unreachable** — the producer now
   probes `norgatedata.status()` before fetching and aborts with a clear error and
   a non-zero exit. Previously norgatedata retried each call 10x then called bare

src/cotdata/providers/databento.py

@@ -22,6 +22,7 @@
 import time
 import warnings
 from pathlib import Path
+from typing import Optional
 
 import pandas as pd
 
@@ -114,11 +115,41 @@ def run_batch_backfill(symbols: list) -> None:
                 df_clean.to_parquet(_cache_dir() / f"{internal_sym}_daily.parquet")
 
 
-def fetch_daily_ohlc(symbol: str, start_date: str = "2000-01-01",
+def fetch_daily_ohlc(symbol: str, start_date: Optional[str] = None,
                      force_refresh: bool = False, price_type: str = "close") -> pd.DataFrame:
     """Daily OHLC + Open Interest via Databento (GLBX.MDP3 ohlcv-1d + statistics),
     append-only cache; yfinance fallback for _DATABENTO_UNSUPPORTED. price_type
-    'settlement' pulls stat_type 3 dated by ts_ref."""
+    'settlement' pulls stat_type 3 dated by ts_ref.
+
+    The cache is always a single, from-inception, append-only series per
+    symbol+price_type, shared across every caller — that is what makes repeated
+    calls cheap. `start_date` therefore means two different things depending on
+    whether it is the FIRST call for a symbol or not, and both are deliberate:
+
+    * **Cold cache (symbol never fetched before):** `start_date` clamps the fetch
+      floor (still no lower than the GLBX.MDP3 history floor 2010-06-06), so a
+      narrow first-time query — e.g. "just the last 3 months" — actually costs a
+      narrow API pull, not a from-2000 one.
+    * **Warm cache (symbol already has some history):** `start_date` does NOT
+      change what gets fetched — the top-up always resumes from the cache's
+      `last_date + 1 day`, exactly as before. Letting a later, narrower
+      `start_date` shrink the fetch would silently truncate a cache other
+      callers already rely on being complete. `start_date` still filters what is
+      RETURNED to this call, just not what is fetched or persisted.
+
+    In both cases the returned frame never contains rows before `start_date` — the
+    filter is applied uniformly to whatever the cache ends up holding. `None`
+    (the default) is unbounded: full cached history, unfiltered, unchanged
+    behaviour from before this parameter existed.
+
+    One consequence of the "warm cache never shrinks its floor" rule: if a symbol's
+    cache was FIRST populated by a narrow `start_date` (e.g. "since 2024"), a later
+    call asking for `start_date="2010-01-01"` will NOT backfill 2010-2023 — the
+    cache only ever grows forward. Not a concern for the one real caller today
+    (`update_all_daily_prices`, which never passes `start_date` and always wants
+    full history), but worth knowing before relying on this for a symbol multiple
+    call sites touch with different windows.
+    """
     import databento as db
     display_sym = symbol
     _pt_suffix = "" if price_type == "close" else f"_{price_type}"
@@ -135,24 +166,33 @@ def fetch_daily_ohlc(symbol: str, start_date: str = "2000-01-01",
         except Exception as e:  # noqa: BLE001
             logger.warning("Failed to read cache for %s: %s", display_sym, e)
 
+    def _filtered(df: pd.DataFrame) -> pd.DataFrame:
+        if start_date and not df.empty:
+            return df[df.index >= pd.Timestamp(start_date)]
+        return df
+
     if local_df.empty:
-        fetch_start = "2000-01-01"
+        # Cold cache: start_date narrows the fetch floor (still API-cost-bounded
+        # by the GLBX floor below), so a first-time narrow query is actually cheap.
+        fetch_start = start_date if start_date else "2000-01-01"
     else:
+        # Warm cache: start_date does NOT narrow this — see the docstring. It only
+        # affects the return-value filter below.
         last_date = local_df.index.max()
         today = pd.Timestamp.now().normalize()
         if last_date >= today - pd.Timedelta(days=1):
-            return local_df
+            return _filtered(local_df)
         if "--fast" in sys.argv:
-            return local_df
+            return _filtered(local_df)
         now = time.time()
         if not force_refresh and (now - _API_LAST_CHECKED.get(symbol, 0) < 3600):
-            return local_df
+            return _filtered(local_df)
         _API_LAST_CHECKED[symbol] = now
         fetch_start = (last_date + pd.Timedelta(days=1)).strftime("%Y-%m-%d")
 
     fetch_end = (pd.Timestamp.now() - pd.Timedelta(days=1)).strftime("%Y-%m-%d")
     if pd.Timestamp(fetch_start) >= pd.Timestamp(fetch_end):
-        return local_df
+        return _filtered(local_df)
 
     new_df = pd.DataFrame()
     db_key = os.environ.get("DATABENTO_API_KEY")
@@ -245,12 +285,12 @@ def fetch_daily_ohlc(symbol: str, start_date: str = "2000-01-01",
     if not local_df.empty and not new_df.empty:
         new_df = new_df[new_df.index > local_df.index.max()]
     if new_df.empty:
-        return local_df
+        return _filtered(local_df)
     new_df = new_df.dropna(subset=["Close"])
     combined = pd.concat([local_df, new_df]) if not local_df.empty else new_df
     combined = combined[~combined.index.duplicated(keep="last")].sort_index()
-    combined.to_parquet(cache_path)
-    return combined
+    combined.to_parquet(cache_path)          # cache always persists the FULL series
+    return _filtered(combined)               # start_date only shapes what's returned
 
 
 def update_all_daily_prices(force_refresh: bool = False) -> None:

tests/test_databento_provider.py

@@ -87,3 +87,107 @@ def test_settlement_overrides_close_dated_by_ts_ref(tmp_path, monkeypatch, ohlcv
 
     assert df.loc["2020-01-02", "Close"] == 101.5            # settle, not ohlcv 100.0
     assert df.loc["2020-01-02", "Open Interest"] == 5000.0
+
+
+# ── start_date ──────────────────────────────────────────────────────────────
+# start_date means two different things depending on cache state (see the
+# fetch_daily_ohlc docstring): it narrows the FETCH on a cold cache (cost benefit —
+# the whole reason this was broken), but only narrows the RETURN on a warm cache
+# (correctness — never lets a later, narrower caller truncate what other callers
+# already rely on being cached). Each get is exercised separately below.
+
+def test_cold_cache_start_date_narrows_the_fetch_floor(tmp_path, monkeypatch, ohlcv, stats):
+    monkeypatch.setenv("COTDATA_STORE", str(tmp_path))
+    monkeypatch.setenv("DATABENTO_API_KEY", "test-key")
+    client = _client(ohlcv, stats)
+    mock_databento.Historical = mock.Mock(return_value=client)
+
+    dbprov.fetch_daily_ohlc("ES", start_date="2019-01-01", price_type="close")
+
+    ohlcv_call = next(c for c in client.timeseries.get_range.call_args_list
+                      if c.kwargs.get("schema") == "ohlcv-1d")
+    assert ohlcv_call.kwargs["start"] == "2019-01-01"        # NOT "2000-01-01"
+
+
+def test_cold_cache_start_date_is_still_clamped_to_the_glbx_floor(tmp_path, monkeypatch,
+                                                                   ohlcv, stats):
+    monkeypatch.setenv("COTDATA_STORE", str(tmp_path))
+    monkeypatch.setenv("DATABENTO_API_KEY", "test-key")
+    client = _client(ohlcv, stats)
+    mock_databento.Historical = mock.Mock(return_value=client)
+
+    dbprov.fetch_daily_ohlc("ES", start_date="1990-01-01", price_type="close")
+
+    ohlcv_call = next(c for c in client.timeseries.get_range.call_args_list
+                      if c.kwargs.get("schema") == "ohlcv-1d")
+    assert ohlcv_call.kwargs["start"] == "2010-06-06"         # clamped, not 1990
+
+
+def test_returned_frame_excludes_rows_before_start_date(tmp_path, monkeypatch, ohlcv, stats):
+    monkeypatch.setenv("COTDATA_STORE", str(tmp_path))
+    monkeypatch.setenv("DATABENTO_API_KEY", "test-key")
+    mock_databento.Historical = mock.Mock(return_value=_client(ohlcv, stats))
+
+    df = dbprov.fetch_daily_ohlc("ES", start_date="2020-01-03", price_type="close")
+
+    assert "2020-01-02" not in df.index.strftime("%Y-%m-%d")  # before start_date
+
+
+def test_no_start_date_is_unbounded_default(tmp_path, monkeypatch, ohlcv, stats):
+    """The default (None) is unchanged from before this parameter existed."""
+    monkeypatch.setenv("COTDATA_STORE", str(tmp_path))
+    monkeypatch.setenv("DATABENTO_API_KEY", "test-key")
+    client = _client(ohlcv, stats)
+    mock_databento.Historical = mock.Mock(return_value=client)
+
+    df = dbprov.fetch_daily_ohlc("ES", price_type="close")
+
+    ohlcv_call = next(c for c in client.timeseries.get_range.call_args_list
+                      if c.kwargs.get("schema") == "ohlcv-1d")
+    assert ohlcv_call.kwargs["start"] == "2010-06-06"    # 2000-01-01 clamped to GLBX floor
+    assert "2020-01-02" in df.index.strftime("%Y-%m-%d")
+
+
+def test_warm_cache_start_date_does_not_narrow_the_fetch_but_still_filters_the_return(
+        tmp_path, monkeypatch):
+    """The case the docstring warns about: a cache already holding 06-01..06-03 must
+    keep being topped up from its own last_date, unaffected by a later, narrower
+    start_date — but the RETURN must still respect it, excluding 06-01."""
+    monkeypatch.setenv("COTDATA_STORE", str(tmp_path))
+    monkeypatch.setenv("DATABENTO_API_KEY", "test-key")
+    monkeypatch.setattr(dbprov, "_API_LAST_CHECKED", {})   # fresh throttle state
+
+    cache_dir = tmp_path / "_cache" / "databento"
+    cache_dir.mkdir(parents=True)
+    existing = pd.DataFrame(
+        {"Open": [10.0, 11.0, 12.0], "High": [10.5, 11.5, 12.5],
+         "Low": [9.5, 10.5, 11.5], "Close": [10.2, 11.2, 12.2],
+         "Volume": [100, 100, 100], "Open Interest": [500.0, 500.0, 500.0]},
+        index=pd.DatetimeIndex(["2019-06-01", "2019-06-02", "2019-06-03"], name="Date"))
+    existing.to_parquet(cache_dir / "ES_daily.parquet")
+
+    new_ohlcv = pd.DataFrame(
+        {"open": [13.0], "high": [13.5], "low": [12.5], "close": [13.2], "volume": [100]},
+        index=pd.DatetimeIndex(["2019-06-04"], name="ts_event"))
+    new_stats = pd.DataFrame(
+        {"stat_type": [9], "price": [0.0], "quantity": [600.0],
+         "ts_ref": pd.to_datetime(["2019-06-04"])},
+        index=pd.DatetimeIndex(["2019-06-04"], name="ts_event"))
+    client = _client(new_ohlcv, new_stats)
+    mock_databento.Historical = mock.Mock(return_value=client)
+
+    df = dbprov.fetch_daily_ohlc("ES", start_date="2019-06-02", force_refresh=False,
+                                 price_type="close")
+
+    ohlcv_call = next(c for c in client.timeseries.get_range.call_args_list
+                      if c.kwargs.get("schema") == "ohlcv-1d")
+    assert ohlcv_call.kwargs["start"] == "2019-06-04"        # resumed from last_date+1,
+                                                              # NOT narrowed to start_date
+    dates = set(df.index.strftime("%Y-%m-%d"))
+    assert "2019-06-01" not in dates                         # excluded: before start_date
+    assert {"2019-06-02", "2019-06-03", "2019-06-04"} <= dates
+
+    # the ON-DISK cache still holds the full series — start_date shaped the
+    # return, not what was persisted.
+    on_disk = pd.read_parquet(cache_dir / "ES_daily.parquet")
+    assert "2019-06-01" in on_disk.index.strftime("%Y-%m-%d")