diff --git a/CHANGELOG.md b/CHANGELOG.md index 6bd736b..3776e2d 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/cotdata/providers/databento.py b/src/cotdata/providers/databento.py index b2b8ca9..caf3559 100644 --- a/src/cotdata/providers/databento.py +++ b/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: diff --git a/tests/test_databento_provider.py b/tests/test_databento_provider.py index 711145a..dee1ba4 100644 --- a/tests/test_databento_provider.py +++ b/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")