From fbc808ad6c422073e3e2cef8fbc6c14922a99e44 Mon Sep 17 00:00:00 2001 From: Matthew Spinola Date: Wed, 22 Jul 2026 21:58:31 -0400 Subject: [PATCH 1/4] Remove dead start_date param from databento.fetch_daily_ohlc MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every code path ignored it: a cold cache always backfilled from 2000-01-01 (clamped to the GLBX.MDP3 floor 2010-06-06), and a warm cache always resumed from last_date+1, regardless of what was passed. A caller trying to bound a fetch (e.g. 'just the last 3 months') silently got a full-history pull instead, at full Databento API cost, with no error or warning. Checked every caller before deciding how to fix it. grep across the workspace shows no one relies on it working: the function's only real internal caller (update_all_daily_prices) never passes it, both tests never pass it, and every OTHER fetch_daily_ohlc in the tree (npf.ml.labels) is an unrelated function that correctly forwards its own start_date into cotdata.get_prices — not this one. Removed rather than wired in. An explicit TypeError for a future caller beats another silent full-history surprise, and there is no current caller to write a correctness test against if the parameter were implemented instead. A genuinely bounded fetch, if ever needed, should be a new function with its own test proving the bound is honored. Found while building a bounded, one-off Databento divergence-check script in livebook, which deliberately avoided reusing this function for exactly this reason. 81 tests pass (5 apparent failures in test_norgate_provider.py are pre-existing and environment-dependent — COTDATA_STORE unset — confirmed via git stash, unrelated to this change). Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 12 ++++++++++++ src/cotdata/providers/databento.py | 19 ++++++++++++++++--- 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6bd736b..c513add 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,18 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). generalization markets. ([#24](https://github.com/mspinola/cotdata/pull/24)) ### Fixed +- **Removed the dead `start_date` parameter from `databento.fetch_daily_ohlc`** + (dormant provider). It was silently ignored on every code path — a cold cache + always backfilled from 2000-01-01 (clamped to the GLBX.MDP3 floor + 2010-06-06) 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. No caller in this workspace relied on it + working (the function's only real caller never passed it), so it is removed + rather than wired in: an explicit `TypeError` for anyone who passes it now + beats another silent full-history surprise. `fetch_daily_ohlc` always + maintains a from-inception, incrementally-updated cache; a genuinely bounded + fetch would need a new function with its own test proving the bound holds. - **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..bf8a7f0 100644 --- a/src/cotdata/providers/databento.py +++ b/src/cotdata/providers/databento.py @@ -114,11 +114,24 @@ 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", - force_refresh: bool = False, price_type: str = "close") -> pd.DataFrame: +def fetch_daily_ohlc(symbol: str, 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. + + Always maintains a FROM-INCEPTION cache (2000-01-01, clamped to the GLBX.MDP3 + floor 2010-06-06 on first fetch) and incrementally tops it up from the last + cached date on every call — there is no bounded/windowed fetch. A `start_date` + parameter used to sit here but was dead: every code path ignored it (cold cache + always started at 2000-01-01; warm cache always resumed at last_date+1), so a + caller passing it to bound a fetch silently got a full-history pull instead, at + full API cost. Removed rather than wired in, since nothing in this workspace + was relying on it working (grep confirmed the only real caller, + update_all_daily_prices, never passed it) — better an explicit TypeError for a + future caller than another silent full-history surprise. A genuinely bounded + fetch, if one is ever needed, should be a new function with its own test + proving the bound is honored, not this parameter reinstated.""" import databento as db display_sym = symbol _pt_suffix = "" if price_type == "close" else f"_{price_type}" From c87279edc21a8b666f1205d4e6bdb1e58f697c41 Mon Sep 17 00:00:00 2001 From: Matthew Spinola Date: Wed, 22 Jul 2026 22:36:21 -0400 Subject: [PATCH 2/4] Implement start_date instead of removing it, per review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Correctly pointed out that removal just relocates the problem — a future caller (including my own livebook divergence-check script, which had to write a separate bounded fetch specifically because this function couldn't do it) still needs a bounded fetch, and the test infrastructure to verify it correctly already exists in this file (the mock Historical client pattern). Cold cache (symbol's first-ever fetch): start_date now narrows the fetch floor (still clamped to the GLBX.MDP3 2010-06-06 floor), so a narrow first-time query is actually cheap — this is the real fix for the original API-cost bug. Warm cache (symbol already has history): start_date deliberately does NOT narrow the incremental fetch, which always resumes from the cache's own last_date+1. The cache is a single shared, from-inception series across every caller; letting a later, narrower start_date shrink what gets fetched would silently truncate a cache other callers rely on being complete. start_date still filters what's RETURNED in this case, just not what's fetched or persisted to disk. Both branches are tested against the mock Historical client, including the warm-cache case specifically (pre-seeded 3-day cache, narrower start_date requested, asserts the fetch call still starts from last_date+1 while the returned frame excludes the pre-start row AND the on-disk cache keeps the full series). 6 new tests, 8/8 in this file, 86/86 full suite (COTDATA_STORE set). Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 27 ++++---- src/cotdata/providers/databento.py | 71 ++++++++++++++------ tests/test_databento_provider.py | 104 +++++++++++++++++++++++++++++ 3 files changed, 168 insertions(+), 34 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c513add..3776e2d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,18 +25,21 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). generalization markets. ([#24](https://github.com/mspinola/cotdata/pull/24)) ### Fixed -- **Removed the dead `start_date` parameter from `databento.fetch_daily_ohlc`** - (dormant provider). It was silently ignored on every code path — a cold cache - always backfilled from 2000-01-01 (clamped to the GLBX.MDP3 floor - 2010-06-06) 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. No caller in this workspace relied on it - working (the function's only real caller never passed it), so it is removed - rather than wired in: an explicit `TypeError` for anyone who passes it now - beats another silent full-history surprise. `fetch_daily_ohlc` always - maintains a from-inception, incrementally-updated cache; a genuinely bounded - fetch would need a new function with its own test proving the bound holds. +- **`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 bf8a7f0..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,24 +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, force_refresh: bool = False, - price_type: str = "close") -> pd.DataFrame: +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. - Always maintains a FROM-INCEPTION cache (2000-01-01, clamped to the GLBX.MDP3 - floor 2010-06-06 on first fetch) and incrementally tops it up from the last - cached date on every call — there is no bounded/windowed fetch. A `start_date` - parameter used to sit here but was dead: every code path ignored it (cold cache - always started at 2000-01-01; warm cache always resumed at last_date+1), so a - caller passing it to bound a fetch silently got a full-history pull instead, at - full API cost. Removed rather than wired in, since nothing in this workspace - was relying on it working (grep confirmed the only real caller, - update_all_daily_prices, never passed it) — better an explicit TypeError for a - future caller than another silent full-history surprise. A genuinely bounded - fetch, if one is ever needed, should be a new function with its own test - proving the bound is honored, not this parameter reinstated.""" + 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}" @@ -148,24 +166,33 @@ def fetch_daily_ohlc(symbol: str, force_refresh: bool = False, 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") @@ -258,12 +285,12 @@ def fetch_daily_ohlc(symbol: str, force_refresh: bool = False, 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") From a534c93986387bb9c99c3d180116060ec6971dc8 Mon Sep 17 00:00:00 2001 From: Matt Spinola Date: Wed, 22 Jul 2026 22:54:12 -0400 Subject: [PATCH 3/4] Add ruff linting to CI Adds ruff configuration to pyproject.toml and a linting step to CI workflow before tests. Configuration: line-length=100, select=[E,F,W,I], target-version=py39. Co-Authored-By: Claude Haiku 4.5 --- .github/workflows/python-test.yml | 2 ++ pyproject.toml | 10 +++++++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/.github/workflows/python-test.yml b/.github/workflows/python-test.yml index 608ec55..c575b93 100644 --- a/.github/workflows/python-test.yml +++ b/.github/workflows/python-test.yml @@ -24,6 +24,8 @@ jobs: run: | python -m pip install --upgrade pip python -m pip install -e ".[dev]" + - name: Run linting + run: python -m ruff check src tests - name: Run tests run: | # Create a dummy store for the test environment to prevent COTDATA_STORE missing errors diff --git a/pyproject.toml b/pyproject.toml index 29abe13..2ef8951 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,7 +38,7 @@ norgate = ["norgatedata"] databento = ["databento>=0.30"] yahoo = ["yfinance"] -dev = ["pytest>=7"] +dev = ["pytest>=7", "ruff>=0.1.0"] [project.scripts] cotdata-update = "cotdata.update:main" @@ -48,3 +48,11 @@ where = ["src"] [tool.setuptools.package-data] cotdata = ["*.yaml"] + +[tool.ruff] +line-length = 100 +target-version = "py39" + +[tool.ruff.lint] +select = ["E", "F", "W", "I"] +ignore = ["E501"] From c995e6f7ed33202e962c5a97b66fbdff51ac1f28 Mon Sep 17 00:00:00 2001 From: Matt Spinola Date: Thu, 23 Jul 2026 10:15:32 -0400 Subject: [PATCH 4/4] =?UTF-8?q?Clear=20ruff=20findings;=20pin=20ruff=20?= =?UTF-8?q?=E2=86=92=20green=20the=20lint=20step?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 'Add ruff linting to CI' step (on this databento branch) ran but the code had 137 findings. Pinned ruff==0.15.22 and cleared them: whitespace, import sorting, unused imports, f-strings, and E713 not-in (all safe auto-fixes). The 3 remaining E402 are test imports after sys.modules mock injection (databento/norgate) — those must stay below the mock, so noqa'd at the line; one stray datetime import moved to the top. ruff check src tests clean; all 86 tests pass (with COTDATA_STORE set). Config unchanged (fleet parity: select E/F/W/I, ignore E501). Includes this branch's databento start_date work (fbc808a, c87279e). Co-Authored-By: Claude Opus 4.8 --- pyproject.toml | 2 +- src/cotdata/__init__.py | 6 +- src/cotdata/cot.py | 2 +- src/cotdata/providers/cftc_disagg.py | 13 ++-- src/cotdata/providers/cftc_tff.py | 13 ++-- src/cotdata/providers/norgate.py | 48 ++++++++------- src/cotdata/registry.py | 3 +- src/cotdata/status.py | 2 +- src/cotdata/store.py | 2 +- src/cotdata/update.py | 2 +- tests/test_adjustment.py | 2 +- tests/test_cftc_disagg_fidelity.py | 12 ++-- tests/test_cftc_reconciliation.py | 31 +++++----- tests/test_cftc_tff_reconciliation.py | 37 +++++------ tests/test_databento_provider.py | 2 +- tests/test_norgate_provider.py | 89 ++++++++++++++------------- tests/test_propadj.py | 4 +- tests/test_registry.py | 6 +- tests/test_status.py | 4 +- tests/test_store.py | 14 ++--- 20 files changed, 155 insertions(+), 139 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 2ef8951..e2b6356 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,7 +38,7 @@ norgate = ["norgatedata"] databento = ["databento>=0.30"] yahoo = ["yfinance"] -dev = ["pytest>=7", "ruff>=0.1.0"] +dev = ["pytest>=7", "ruff==0.15.22"] [project.scripts] cotdata-update = "cotdata.update:main" diff --git a/src/cotdata/__init__.py b/src/cotdata/__init__.py index 53e5bf9..03a46f9 100644 --- a/src/cotdata/__init__.py +++ b/src/cotdata/__init__.py @@ -1,8 +1,8 @@ """cotdata — canonical data layer (see README). Public consumer API.""" -from .prices import get_prices, roll_dates from .cot import get_cot -from .registry import symbol, all_symbols, REGISTRY, Symbol -from .store import load_manifest, schema_version, require_schema +from .prices import get_prices, roll_dates +from .registry import REGISTRY, Symbol, all_symbols, symbol +from .store import load_manifest, require_schema, schema_version __version__ = "0.1.0" __all__ = [ diff --git a/src/cotdata/cot.py b/src/cotdata/cot.py index d8dab5b..546a1a8 100644 --- a/src/cotdata/cot.py +++ b/src/cotdata/cot.py @@ -28,7 +28,7 @@ def get_cot(name: str, report: str = "legacy") -> pd.DataFrame: sym = REGISTRY.get(name) if sym is None: # allow lookup by primary CFTC code, not just symbol sym = next((s for s in REGISTRY.values() if s.cftc_code == name), None) - + read_fn = { "legacy": store.read_cot_legacy, "disagg": store.read_cot_disagg, diff --git a/src/cotdata/providers/cftc_disagg.py b/src/cotdata/providers/cftc_disagg.py index 94d14a6..e7963d1 100644 --- a/src/cotdata/providers/cftc_disagg.py +++ b/src/cotdata/providers/cftc_disagg.py @@ -5,7 +5,6 @@ the store via store.write_cot_disagg(). """ import datetime as dt -import io import zipfile from email.utils import parsedate_to_datetime from pathlib import Path @@ -60,10 +59,10 @@ def _parse_zip(zip_path: Path) -> pd.DataFrame: with zf.open(zf.namelist()[0]) as fh: # The .txt files are actually CSVs. low_memory=False prevents dtype warnings. df = pd.read_csv(fh, low_memory=False) - + # Strip any trailing whitespace from column names BEFORE accessing them df.columns = df.columns.str.strip() - + # CFTC sometimes changes the date column name in Disaggregated reports if REPORT_DATE not in df.columns: date_cols = [c for c in df.columns if "Report_Date" in c] @@ -71,16 +70,16 @@ def _parse_zip(zip_path: Path) -> pd.DataFrame: df.rename(columns={date_cols[0]: REPORT_DATE}, inplace=True) else: print(f"AVAILABLE COLUMNS: {list(df.columns)}") - + # Coerce the key schema columns to match the Legacy schema format df[CONTRACT_CODE] = df[CONTRACT_CODE].apply(_standardize_code) df[REPORT_DATE] = pd.to_datetime(df[REPORT_DATE]).dt.tz_localize(None) - + # Parquet cannot serialize mixed-type object columns (e.g. Traders_Tot_Old contains ints and strings) for col in df.select_dtypes(include=['object']).columns: if col not in [CONTRACT_CODE, REPORT_DATE]: df[col] = df[col].astype(str) - + return df @@ -105,7 +104,7 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> dict: want = set(code_to_sym.keys()) frames = [] - + # The CFTC bundles 2006-2016 in a single historical file if first_year <= 2016: url = "https://www.cftc.gov/files/dea/history/fut_disagg_txt_hist_2006_2016.zip" diff --git a/src/cotdata/providers/cftc_tff.py b/src/cotdata/providers/cftc_tff.py index fd3ab23..35ebc7d 100644 --- a/src/cotdata/providers/cftc_tff.py +++ b/src/cotdata/providers/cftc_tff.py @@ -5,7 +5,6 @@ the store via store.write_cot_tff(). """ import datetime as dt -import io import zipfile from email.utils import parsedate_to_datetime from pathlib import Path @@ -60,10 +59,10 @@ def _parse_zip(zip_path: Path) -> pd.DataFrame: with zf.open(zf.namelist()[0]) as fh: # The .txt files are actually CSVs. low_memory=False prevents dtype warnings. df = pd.read_csv(fh, low_memory=False) - + # Strip any trailing whitespace from column names BEFORE accessing them df.columns = df.columns.str.strip() - + # CFTC sometimes changes the date column name in TFF reports if REPORT_DATE not in df.columns: date_cols = [c for c in df.columns if "Report_Date" in c] @@ -71,16 +70,16 @@ def _parse_zip(zip_path: Path) -> pd.DataFrame: df.rename(columns={date_cols[0]: REPORT_DATE}, inplace=True) else: print(f"AVAILABLE COLUMNS: {list(df.columns)}") - + # Coerce the key schema columns to match the Legacy schema format df[CONTRACT_CODE] = df[CONTRACT_CODE].apply(_standardize_code) df[REPORT_DATE] = pd.to_datetime(df[REPORT_DATE]).dt.tz_localize(None) - + # Parquet cannot serialize mixed-type object columns (e.g. Traders_Tot_Old contains ints and strings) for col in df.select_dtypes(include=['object']).columns: if col not in [CONTRACT_CODE, REPORT_DATE]: df[col] = df[col].astype(str) - + return df @@ -105,7 +104,7 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> dict: want = set(code_to_sym.keys()) frames = [] - + # The CFTC bundles 2006-2016 in a single historical file if first_year <= 2016: url = "https://www.cftc.gov/files/dea/history/fin_fut_txt_2006_2016.zip" diff --git a/src/cotdata/providers/norgate.py b/src/cotdata/providers/norgate.py index bbc78f2..26d214d 100644 --- a/src/cotdata/providers/norgate.py +++ b/src/cotdata/providers/norgate.py @@ -11,13 +11,14 @@ from __future__ import annotations # PEP 604 unions (dict | None) on Python 3.9 import datetime as dt -import pandas as pd -import numpy as np import re from concurrent.futures import ThreadPoolExecutor, as_completed -from ..registry import REGISTRY, all_symbols +import numpy as np +import pandas as pd + from .. import store +from ..registry import REGISTRY, all_symbols CCB_SUFFIX = "_CCB" # Norgate "Continuous Contract Back-adjusted" @@ -67,14 +68,14 @@ def _reconstruct_volume(internal_symbol: str, continuous_df: pd.DataFrame, adjus if len(valid_dates) > 0: # Recompute trailing 60 days to catch late data & bridge partial failures last_date = valid_dates.max() - pd.Timedelta(days=60) - + # Base Norgate symbol (e.g. "&ES" -> "ES") base_sym = REGISTRY[internal_symbol].norgate.lstrip("&").split("_")[0] - + # 2. Find needed individual contracts all_futures = norgatedata.database_symbols('Futures') pattern = re.compile(rf"^{re.escape(base_sym)}-(\d{{4}})([FGHJKMNQUVXZ])$") - + needed_contracts = [] for sym in all_futures: m = pattern.match(sym) @@ -83,7 +84,7 @@ def _reconstruct_volume(internal_symbol: str, continuous_df: pd.DataFrame, adjus expiry_date = pd.Timestamp(year=year, month=month, day=1) + pd.DateOffset(months=1) if expiry_date >= last_date: needed_contracts.append(sym) - + # 3. Fallback: if no individual contracts (e.g., crypto, ICE softs) if not needed_contracts: res = continuous_df.copy() @@ -100,12 +101,12 @@ def _reconstruct_volume(internal_symbol: str, continuous_df: pd.DataFrame, adjus with ThreadPoolExecutor(max_workers=10) as pool: futs = { pool.submit( - norgatedata.price_timeseries, - c, - padding_setting=norgatedata.PaddingType.NONE, - timeseriesformat="pandas-dataframe", + norgatedata.price_timeseries, + c, + padding_setting=norgatedata.PaddingType.NONE, + timeseriesformat="pandas-dataframe", start_date=last_date.strftime("%Y-%m-%d") - ): c + ): c for c in needed_contracts } for f in as_completed(futs): @@ -130,10 +131,10 @@ def _reconstruct_volume(internal_symbol: str, continuous_df: pd.DataFrame, adjus res["Volume_Reconstructed"] = res["Volume"] res["Volume_Source"] = "raw" return res - + all_indiv = pd.concat(frames, ignore_index=True) all_indiv['Date'] = pd.to_datetime(all_indiv['Date']).dt.tz_localize(None).dt.normalize() - + # 5. Contract identification: First / Second = the two HIGHEST-VOLUME contracts # trading that day, NOT the two nearest by expiry. Products with serial months # around a bi-monthly liquid cycle (e.g. GC, SI) carry almost no volume in the @@ -155,9 +156,9 @@ def get_expiry(sym): order = np.argsort(-rank_key, axis=1, kind='stable') compressed = np.take_along_axis(vol_array, order, axis=1) names_arr = np.array(pivot.columns) - + rec_df = pd.DataFrame(index=pivot.index) - + num_cols = compressed.shape[1] if num_cols > 0: rec_df['FirstVolume'] = compressed[:, 0] @@ -165,18 +166,18 @@ def get_expiry(sym): else: rec_df['FirstVolume'] = np.nan rec_df['FirstContract'] = '' - + if num_cols > 1: rec_df['SecondVolume'] = compressed[:, 1] rec_df['SecondContract'] = np.where(np.isnan(compressed[:, 1]), '', names_arr[order[:, 1]]) else: rec_df['SecondVolume'] = np.nan rec_df['SecondContract'] = '' - + rec_df['Volume_Reconstructed'] = rec_df['FirstVolume'].fillna(0) + rec_df['SecondVolume'].fillna(0) rec_df.loc[rec_df['FirstVolume'].isna() & rec_df['SecondVolume'].isna(), 'Volume_Reconstructed'] = np.nan rec_df['Volume_Source'] = "reconstructed" - + # 6. Merge the newly reconstructed subset into the existing history for col in ["FirstVolume", "SecondVolume", "FirstContract", "SecondContract", "Volume_Reconstructed", "Volume_Source"]: if col not in res.columns: @@ -184,9 +185,9 @@ def get_expiry(sym): res[col] = existing_df[col] else: res[col] = "" if col in ("FirstContract", "SecondContract", "Volume_Source") else np.nan - + res.update(rec_df) - + common_idx = res.index.intersection(rec_df.index) for col in ["FirstContract", "SecondContract"]: res.loc[common_idx, col] = rec_df.loc[common_idx, col] @@ -194,7 +195,7 @@ def get_expiry(sym): mask = res['Volume_Reconstructed'].isna() res.loc[mask, 'Volume_Reconstructed'] = res.loc[mask, 'Volume'] res.loc[mask, 'Volume_Source'] = "raw" - + return res @@ -204,7 +205,7 @@ def fetch(internal_symbol: str, adjustment: str = "backadj", start: str = "1970- ng_sym = REGISTRY[internal_symbol].norgate if adjustment == "backadj": ng_sym += CCB_SUFFIX - + df = norgatedata.price_timeseries( ng_sym, padding_setting=norgatedata.PaddingType.NONE, @@ -330,6 +331,7 @@ def update(symbols=None, full: bool = False) -> None: logic change so old rows are recomputed under the new algorithm. """ import time + from .. import status _require_norgate_service() # abort cleanly if NDU is down (see helper docstring) diff --git a/src/cotdata/registry.py b/src/cotdata/registry.py index 7a13147..3ec4aae 100644 --- a/src/cotdata/registry.py +++ b/src/cotdata/registry.py @@ -17,11 +17,12 @@ unless a symbol overrides it explicitly. """ import os -import yaml from dataclasses import dataclass from pathlib import Path from typing import Dict, List, Optional +import yaml + @dataclass(frozen=True) class Symbol: diff --git a/src/cotdata/status.py b/src/cotdata/status.py index 4b7273b..ccb35a5 100644 --- a/src/cotdata/status.py +++ b/src/cotdata/status.py @@ -9,7 +9,7 @@ import os from typing import Optional -from . import store, config +from . import config, store STATUS_FILENAME = "status.json" diff --git a/src/cotdata/store.py b/src/cotdata/store.py index 023c98d..39aa34b 100644 --- a/src/cotdata/store.py +++ b/src/cotdata/store.py @@ -1,9 +1,9 @@ """Canonical store I/O: atomic Parquet writes + a manifest. The store is the contract between producers (write) and consumers (read).""" +import datetime as dt import json import os import tempfile -import datetime as dt from pathlib import Path import pandas as pd diff --git a/src/cotdata/update.py b/src/cotdata/update.py index 5f6de15..7d26a81 100644 --- a/src/cotdata/update.py +++ b/src/cotdata/update.py @@ -49,7 +49,7 @@ def main() -> None: return if args.reconcile: - from . import store, status + from . import status, store pruned = store.reconcile_manifest() if not pruned: print("manifest reconcile: nothing to prune (all entries have files).") diff --git a/tests/test_adjustment.py b/tests/test_adjustment.py index d2ce3b7..bb7ab9b 100644 --- a/tests/test_adjustment.py +++ b/tests/test_adjustment.py @@ -9,8 +9,8 @@ Run on Windows: python test_adjustment.py """ -import numpy as np import pandas as pd + try: import norgatedata except ImportError: diff --git a/tests/test_cftc_disagg_fidelity.py b/tests/test_cftc_disagg_fidelity.py index 5a50eaa..ae2f685 100644 --- a/tests/test_cftc_disagg_fidelity.py +++ b/tests/test_cftc_disagg_fidelity.py @@ -1,5 +1,7 @@ import zipfile -from cotdata.providers.cftc_disagg import _parse_zip, CONTRACT_CODE, REPORT_DATE + +from cotdata.providers.cftc_disagg import CONTRACT_CODE, REPORT_DATE, _parse_zip + def test_disagg_fidelity_preserves_columns(tmp_path): """ @@ -14,16 +16,16 @@ def test_disagg_fidelity_preserves_columns(tmp_path): zip_path = tmp_path / "fut_disagg_txt_2021.zip" with zipfile.ZipFile(zip_path, "w") as zf: zf.writestr("f_year.txt", csv_content) - + df = _parse_zip(zip_path) - + # 9 columns in the CSV -> 9 columns in the parsed DataFrame assert len(df.columns) == 9, "Lossless parsing must preserve all columns from the zip" - + # Check that crucial non-legacy columns survived assert "Traders_Tot_All" in df.columns assert "Swap__Positions_Long_All" in df.columns - + # Check that coercion worked assert df[CONTRACT_CODE].iloc[0] == "002602", "Contract code must be 6-digit zero padded" assert df[REPORT_DATE].iloc[0].year == 2021, "Report date must be parsed to datetime" diff --git a/tests/test_cftc_reconciliation.py b/tests/test_cftc_reconciliation.py index 9924048..9475cd7 100644 --- a/tests/test_cftc_reconciliation.py +++ b/tests/test_cftc_reconciliation.py @@ -1,22 +1,23 @@ import pandas as pd -import pytest + from cotdata import cot + def test_reconciliation_structural_identities(monkeypatch): """ Validates that the Disaggregated schema correctly sums into the Legacy schema groupings for the exact same report date. - + Legacy Commercial = Disagg Producer/Merchant + Swap Dealer Legacy NonCommercial = Disagg Money Manager + Other Reportable """ - + # Mock Legacy store returning 1 week of data legacy_df = pd.DataFrame({ "Comm_Positions_Long_All": [1000], "NonComm_Positions_Long_All": [500], }, index=pd.DatetimeIndex(["2021-01-05"])) - + # Mock Disaggregated store returning 1 week of data disagg_df = pd.DataFrame({ "Prod_Merc_Positions_Long_All": [600], @@ -24,31 +25,31 @@ def test_reconciliation_structural_identities(monkeypatch): "M_Money_Positions_Long_All": [400], "Other_Rept_Positions_Long_All": [100], # 400 + 100 = 500 (Legacy NonComm) }, index=pd.DatetimeIndex(["2021-01-05"])) - + def mock_read_cot_legacy(name): return legacy_df - + def mock_read_cot_disagg(name): return disagg_df - + monkeypatch.setattr("cotdata.store.read_cot_legacy", mock_read_cot_legacy) monkeypatch.setattr("cotdata.store.read_cot_disagg", mock_read_cot_disagg) - + leg = cot.get_cot("DUMMY", report="legacy") dis = cot.get_cot("DUMMY", report="disagg") - + # Assert structural reconciliation on the Long side c_derived = dis["Prod_Merc_Positions_Long_All"] + dis["Swap__Positions_Long_All"] nc_derived = dis["M_Money_Positions_Long_All"] + dis["Other_Rept_Positions_Long_All"] - + pd.testing.assert_series_equal( - c_derived, - leg["Comm_Positions_Long_All"], + c_derived, + leg["Comm_Positions_Long_All"], check_names=False ) - + pd.testing.assert_series_equal( - nc_derived, - leg["NonComm_Positions_Long_All"], + nc_derived, + leg["NonComm_Positions_Long_All"], check_names=False ) diff --git a/tests/test_cftc_tff_reconciliation.py b/tests/test_cftc_tff_reconciliation.py index 26f2095..14874c8 100644 --- a/tests/test_cftc_tff_reconciliation.py +++ b/tests/test_cftc_tff_reconciliation.py @@ -1,13 +1,14 @@ import pandas as pd -import pytest + from cotdata import cot + def test_tff_reconciliation_structural_identities(monkeypatch): """ Validates that the TFF schema correctly sums into the Legacy schema groupings for the exact same report date, and that Open Interest matches. """ - + # Mock Legacy store returning 1 week of data legacy_df = pd.DataFrame({ "Open_Interest_All": [5000], @@ -15,7 +16,7 @@ def test_tff_reconciliation_structural_identities(monkeypatch): "NonComm_Positions_Long_All": [1500], "NonRept_Positions_Long_All": [1000] }, index=pd.DatetimeIndex(["2021-01-05"])) - + # Mock TFF store returning 1 week of data tff_df = pd.DataFrame({ "Open_Interest_All": [5000], @@ -25,40 +26,40 @@ def test_tff_reconciliation_structural_identities(monkeypatch): "Other_Rept_Positions_Long_All": [500], "NonRept_Positions_Long_All": [1000] }, index=pd.DatetimeIndex(["2021-01-05"])) - + def mock_read_cot_legacy(name): return legacy_df - + def mock_read_cot_tff(name): return tff_df - + monkeypatch.setattr("cotdata.store.read_cot_legacy", mock_read_cot_legacy) monkeypatch.setattr("cotdata.store.read_cot_tff", mock_read_cot_tff) - + leg = cot.get_cot("DUMMY", report="legacy") tff = cot.get_cot("DUMMY", report="tff") - + # Assert Open Interest matches exactly pd.testing.assert_series_equal( - tff["Open_Interest_All"], - leg["Open_Interest_All"], + tff["Open_Interest_All"], + leg["Open_Interest_All"], check_names=False ) - + # Assert structural reconciliation on the Long side: # In financial futures, Dealer + Asset_Mgr roughly approximate Commercials, # while Lev_Money + Other_Rept roughly approximate Non-Commercials. # The total sum of all reportable longs + non-reportables must equal total OI. total_longs = ( - tff["Dealer_Positions_Long_All"] + - tff["Asset_Mgr_Positions_Long_All"] + - tff["Lev_Money_Positions_Long_All"] + - tff["Other_Rept_Positions_Long_All"] + + tff["Dealer_Positions_Long_All"] + + tff["Asset_Mgr_Positions_Long_All"] + + tff["Lev_Money_Positions_Long_All"] + + tff["Other_Rept_Positions_Long_All"] + tff["NonRept_Positions_Long_All"] ) - + pd.testing.assert_series_equal( - total_longs, - leg["Open_Interest_All"], + total_longs, + leg["Open_Interest_All"], check_names=False ) diff --git a/tests/test_databento_provider.py b/tests/test_databento_provider.py index dee1ba4..e8403c7 100644 --- a/tests/test_databento_provider.py +++ b/tests/test_databento_provider.py @@ -15,7 +15,7 @@ mock_databento = types.ModuleType("databento") sys.modules["databento"] = mock_databento -from cotdata.providers import databento as dbprov +from cotdata.providers import databento as dbprov # noqa: E402,I001 (import after sys.modules mock injection above) def _client(ohlcv_df, stats_df): diff --git a/tests/test_norgate_provider.py b/tests/test_norgate_provider.py index f71b80a..677020a 100644 --- a/tests/test_norgate_provider.py +++ b/tests/test_norgate_provider.py @@ -1,18 +1,21 @@ -import pandas as pd -import numpy as np -import pytest -from unittest import mock - # Mock norgatedata so we can import and test norgate on any OS +import datetime as _dt import sys import types +from unittest import mock + +import numpy as np +import pandas as pd +import pytest + mock_norgatedata = types.ModuleType("norgatedata") mock_norgatedata.PaddingType = mock.Mock() mock_norgatedata.PaddingType.NONE = "NONE" mock_norgatedata.status = mock.Mock(return_value=True) # NDU reachable (preflight) sys.modules["norgatedata"] = mock_norgatedata -from cotdata.providers import norgate +from cotdata.providers import norgate # noqa: E402 (import after sys.modules mock injection above) + @mock.patch("cotdata.providers.norgate.store.read_prices") @mock.patch("norgatedata.database_symbols", create=True) @@ -22,26 +25,26 @@ def test_norgate_update_fetches_both_adjustments(mock_price_ts, mock_write_price """Verify that update() fetches both the backadj and unadj series for a symbol.""" mock_db_symbols.return_value = [] mock_read_prices.return_value = pd.DataFrame() - + # Setup mock returns mock_df = pd.DataFrame({ "Open": [100.0], "High": [101.0], "Low": [99.0], "Close": [100.5], "Volume": [1000], "Open Interest": [5000], "Delivery Month": [202609] }, index=pd.DatetimeIndex(["2026-07-01"])) - + mock_price_ts.return_value = mock_df - + # Mock all_symbols to just return a dummy registry entry for "ES" mock_symbol = mock.Mock() mock_symbol.internal = "ES" - + # We must patch REGISTRY and all_symbols so it uses our mock with mock.patch("cotdata.providers.norgate.all_symbols", return_value=[mock_symbol]), \ mock.patch.dict("cotdata.providers.norgate.REGISTRY", {"ES": mock.Mock(norgate="&ES")}): - + # Run update norgate.update(symbols=["ES"]) - + # Verify norgatedata API was called twice with correct raw symbols assert mock_price_ts.call_count == 2 calls = mock_price_ts.call_args_list @@ -49,7 +52,7 @@ def test_norgate_update_fetches_both_adjustments(mock_price_ts, mock_write_price assert calls[0][0][0] == "&ES_CCB" # Call 2: unadj ("&ES") assert calls[1][0][0] == "&ES" - + # Verify store.write_prices was called twice with correct adjustment flags assert mock_write_prices.call_count == 2 write_calls = mock_write_prices.call_args_list @@ -66,13 +69,13 @@ def test_norgate_update_fetches_both_adjustments(mock_price_ts, mock_write_price @mock.patch("norgatedata.price_timeseries", create=True) def test_volume_reconstruction(mock_price_ts, mock_db_symbols, mock_read_prices, mock_write_prices): """Verify that _reconstruct_volume correctly appends additive columns without modifying default Volume.""" - + # Mock continuous dataframe mock_continuous = pd.DataFrame({ "Open": [100.0], "High": [101.0], "Low": [99.0], "Close": [100.5], "Volume": [1000], "Open Interest": [5000], "Delivery Month": [202609] }, index=pd.DatetimeIndex(["2026-07-01"])) - + # Mock individual contracts mock_indiv_H = pd.DataFrame({ "Date": [pd.Timestamp("2026-07-01")], @@ -84,42 +87,42 @@ def test_volume_reconstruction(mock_price_ts, mock_db_symbols, mock_read_prices, "Volume": [400], "Open Interest": [2000] }) - + def mock_ts_side_effect(sym, **kwargs): - if sym.endswith("CCB") or not "-" in sym: + if sym.endswith("CCB") or "-" not in sym: return mock_continuous.copy() if sym == "ES-2026H": return mock_indiv_H.copy() if sym == "ES-2026M": return mock_indiv_M.copy() return pd.DataFrame() - + mock_price_ts.side_effect = mock_ts_side_effect mock_db_symbols.return_value = ["ES-2026H", "ES-2026M", "ES-2025Z"] - + # Mock existing prices to trigger full backfill mock_read_prices.return_value = pd.DataFrame() - + mock_symbol = mock.Mock() mock_symbol.internal = "ES" - + with mock.patch("cotdata.providers.norgate.all_symbols", return_value=[mock_symbol]), \ mock.patch.dict("cotdata.providers.norgate.REGISTRY", {"ES": mock.Mock(norgate="&ES")}): - + norgate.update(symbols=["ES"]) - + # Verify the written dataframe has the additive columns and default Volume is untouched write_call = mock_write_prices.call_args_list[0] written_df = write_call[0][2] - + assert "FirstVolume" in written_df.columns assert "SecondVolume" in written_df.columns assert "Volume_Reconstructed" in written_df.columns assert "Volume_Source" in written_df.columns - + # Default Volume should be UNTOUCHED (1000) assert written_df["Volume"].iloc[0] == 1000 - + # FirstVolume (ES-2026H) + SecondVolume (ES-2026M) = 1000 assert written_df["FirstVolume"].iloc[0] == 600 assert written_df["SecondVolume"].iloc[0] == 400 @@ -181,7 +184,7 @@ def mock_ts_side_effect(sym, **kwargs): @mock.patch("norgatedata.price_timeseries", create=True) def test_volume_reconstruction_incremental(mock_price_ts, mock_db_symbols, mock_read_prices, mock_write_prices): """Verify that _reconstruct_volume preserves old Volume_Source during an incremental run.""" - + # Existing df has an old date with a "raw" fallback, and a slightly newer one with "reconstructed" mock_existing = pd.DataFrame({ "Volume": [500, 800], @@ -192,15 +195,15 @@ def test_volume_reconstruction_incremental(mock_price_ts, mock_db_symbols, mock_ "SecondContract": ["", "ES-2026M"], "Volume_Source": ["raw", "reconstructed"] }, index=pd.DatetimeIndex(["2020-01-01", "2026-06-01"])) - + mock_read_prices.return_value = mock_existing.copy() - + # New continuous dataframe has the old dates + a new date mock_continuous = pd.DataFrame({ "Open": [10, 10, 10], "High": [10, 10, 10], "Low": [10, 10, 10], "Close": [10, 10, 10], "Volume": [500, 800, 1000], "Open Interest": [0, 0, 0], "Delivery Month": [0, 0, 0] }, index=pd.DatetimeIndex(["2020-01-01", "2026-06-01", "2026-07-01"])) - + # The new date gets fetched. The trailing 60 days from 2026-06-01 is 2026-04-02. # We will just return some mock individual contracts. mock_indiv_U = pd.DataFrame({ @@ -211,37 +214,37 @@ def test_volume_reconstruction_incremental(mock_price_ts, mock_db_symbols, mock_ "Date": [pd.Timestamp("2026-07-01")], "Volume": [400], "Open Interest": [0] }) - + def mock_ts_side_effect(sym, **kwargs): - if sym.endswith("CCB") or not "-" in sym: + if sym.endswith("CCB") or "-" not in sym: return mock_continuous.copy() if sym == "ES-2026U": return mock_indiv_U.copy() if sym == "ES-2026Z": return mock_indiv_Z.copy() return pd.DataFrame() - + mock_price_ts.side_effect = mock_ts_side_effect mock_db_symbols.return_value = ["ES-2026U", "ES-2026Z"] - + mock_symbol = mock.Mock() mock_symbol.internal = "ES" - + with mock.patch("cotdata.providers.norgate.all_symbols", return_value=[mock_symbol]), \ mock.patch.dict("cotdata.providers.norgate.REGISTRY", {"ES": mock.Mock(norgate="&ES")}): - + norgate.update(symbols=["ES"]) - + write_call = mock_write_prices.call_args_list[0] written_df = write_call[0][2] - + # Verify the 2020-01-01 row is still "raw" assert written_df.loc["2020-01-01", "Volume_Source"] == "raw" assert written_df.loc["2020-01-01", "Volume_Reconstructed"] == 500 - + # Verify the 2026-06-01 row is still "reconstructed" assert written_df.loc["2026-06-01", "Volume_Source"] == "reconstructed" - + # Verify the newly fetched 2026-07-01 row is computed correctly assert written_df.loc["2026-07-01", "Volume_Source"] == "reconstructed" assert written_df.loc["2026-07-01", "Volume_Reconstructed"] == 1000 @@ -414,7 +417,8 @@ def test_covered_filter_drops_none_norgate(): assert norgate._norgate_covered(["ES", "MME"]) == ["ES"] -import datetime as _dt + + def test_finals_ready_pure_logic(): from cotdata.providers.norgate import _finals_ready now = _dt.datetime(2026, 7, 15, 21, 30) # 9:30pm local @@ -434,8 +438,9 @@ def test_finals_ready_pure_logic(): def test_finals_ready_handles_tz_aware_times(): """norgatedata returns tz-aware datetimes (e.g. -04:00); comparing them against a naive cutoff must not raise, and must evaluate by local wall-clock.""" - from cotdata.providers.norgate import _finals_ready import datetime as d + + from cotdata.providers.norgate import _finals_ready et = d.timezone(d.timedelta(hours=-4)) now = d.datetime(2026, 7, 15, 21, 30) # 9:30pm naive local after = d.datetime(2026, 7, 15, 20, 56, tzinfo=et) # aware, after 20:55 diff --git a/tests/test_propadj.py b/tests/test_propadj.py index 94a48f7..38b120f 100644 --- a/tests/test_propadj.py +++ b/tests/test_propadj.py @@ -123,7 +123,7 @@ def test_falls_back_to_offset_jumps_without_delivery_month(store_env): def test_empty_when_either_series_missing(store_env): - from cotdata import store, get_prices + from cotdata import get_prices, store U, _ = _milk_like() store.write_prices("DC", "unadj", U, source="test") # no backadj written assert get_prices("DC", "propadj").empty @@ -131,7 +131,7 @@ def test_empty_when_either_series_missing(store_env): def test_reconstructed_volume_view_still_works_on_propadj(store_env): """The volume view composes with the derived adjustment.""" - from cotdata import store, get_prices + from cotdata import get_prices, store U, B = _milk_like() U["Volume_Reconstructed"] = [15, 14, 10, 18, 20, 22, 25, 24, 26] U["Volume_Source"] = ["reconstructed"] * 9 diff --git a/tests/test_registry.py b/tests/test_registry.py index 9791713..99e5eeb 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -4,7 +4,11 @@ import pytest from cotdata.registry import ( - Symbol, symbol, all_symbols, by_asset_class, hist_code_scales, load_registry, + all_symbols, + by_asset_class, + hist_code_scales, + load_registry, + symbol, ) diff --git a/tests/test_status.py b/tests/test_status.py index 80a9fcc..aab603d 100644 --- a/tests/test_status.py +++ b/tests/test_status.py @@ -86,8 +86,10 @@ def test_build_status_doc_flat_map_and_domains(): def test_write_status_file_roundtrip(tmp_path, monkeypatch): monkeypatch.setenv("COTDATA_STORE", str(tmp_path)) - from cotdata import store, status as st import json + + from cotdata import status as st + from cotdata import store # seed the store manifest via a real write idx = __import__("pandas").date_range("2026-07-10", periods=3, freq="D", name="Date") df = __import__("pandas").DataFrame({"Open": [1, 2, 3], "High": [1, 2, 3], "Low": [1, 2, 3], diff --git a/tests/test_store.py b/tests/test_store.py index be85698..c81743b 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -21,7 +21,7 @@ def _sample(): def test_prices_roundtrip_and_manifest(store_env): - from cotdata import store, get_prices, load_manifest + from cotdata import get_prices, load_manifest, store store.write_prices("ES", "backadj", _sample(), source="test") df = get_prices("ES", "backadj") @@ -64,7 +64,7 @@ def test_reconcile_noop_when_clean(store_env): def test_roll_dates_from_delivery_month(store_env): - from cotdata import store, roll_dates + from cotdata import roll_dates, store store.write_prices("ES", "backadj", _sample(), source="test") rolls = roll_dates("ES", "backadj") # first bar + the delivery-month change on day 4 @@ -89,7 +89,7 @@ def _sample_reconstructed(): def test_default_volume_view_is_byte_identical(store_env): """volume='front' (default) must not change the pre-v2 output shape.""" - from cotdata import store, get_prices + from cotdata import get_prices, store store.write_prices("ES", "backadj", _sample_reconstructed(), source="test") df = get_prices("ES", "backadj") # default volume='front' @@ -103,7 +103,7 @@ def test_default_volume_view_is_byte_identical(store_env): def test_reconstructed_volume_view(store_env): """volume='reconstructed' swaps Volume in, keeps per-row raw fallback, and surfaces Volume_Source for audit.""" - from cotdata import store, get_prices + from cotdata import get_prices, store store.write_prices("ES", "backadj", _sample_reconstructed(), source="test") df = get_prices("ES", "backadj", volume="reconstructed") @@ -118,7 +118,7 @@ def test_reconstructed_volume_view(store_env): def test_reconstructed_view_falls_back_on_pre_v2_store(store_env): """A store written before reconstruction existed → reconstructed view returns front-month volume labelled 'raw', never NaN.""" - from cotdata import store, get_prices + from cotdata import get_prices, store store.write_prices("ES", "backadj", _sample(), source="test") # no recon cols df = get_prices("ES", "backadj", volume="reconstructed") @@ -127,7 +127,7 @@ def test_reconstructed_view_falls_back_on_pre_v2_store(store_env): def test_invalid_volume_arg_raises(store_env): - from cotdata import store, get_prices + from cotdata import get_prices, store store.write_prices("ES", "backadj", _sample(), source="test") with pytest.raises(ValueError): get_prices("ES", "backadj", volume="bogus") @@ -166,8 +166,8 @@ def test_upsert_metadata_on_empty_store_writes_all(store_env): def test_schema_version_and_require_schema(store_env): - from cotdata import store, schema_version, require_schema import cotdata.config as cfg + from cotdata import require_schema, schema_version, store # Empty store → no manifest yet → load_manifest defaults to config.SCHEMA_VERSION assert schema_version() == cfg.SCHEMA_VERSION