From 1290ec0c97e0f10845ca788d4cf90932268d76b4 Mon Sep 17 00:00:00 2001 From: "Matt S." Date: Sun, 9 Aug 2026 02:25:32 +0000 Subject: [PATCH 1/2] Read bars from marketdata, not cotdata (ADR-0007 step 4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three price call sites move from cotdata.get_prices to marketdata.get_bars, all of them adjustment="backadj", which is what ADR-0007 measured. Tier names and the returned frame are unchanged, so the reads themselves are a rename. COT positioning still comes from cotdata; only bars moved. The part that is not a rename is the cache marker, and it would have failed silently. CotIndexer busts its parquet caches when the upstream store schema moves, because the per-symbol guards key on column presence and cannot see a value-only change. The example that guard was written for is named in its own comment: reconstructed volume being promoted, which was a PRICE schema bump. So after this repoint, watching cotdata alone leaves the exact case the guard exists for uncovered — stale metrics computed against a superseded bar schema, with no error anywhere. The marker now records both store versions, as separate keys rather than a combined number: they are independent counters, and a max would hide a bump in whichever store happens to sit lower. A marker written before the split has no marketdata version, reads as 0, and busts once — correct rather than merely tolerated, since the price source moved underneath those caches. Verified the wider frame is safe rather than assuming it: marketdata's futures path returns the stored columns without cotdata's keep-list filter, and all three consumers select columns explicitly, so the extra reconstruction columns pass through untouched. 238 tests pass against real cotdata and marketdata installs, ruff clean. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Jo4iovRfc2fzE9MwcLp7r2 --- pyproject.toml | 3 +- src/cotmetrics/CotIndexer.py | 107 +++++++++++++++++++++------------ src/cotmetrics/market_data.py | 3 +- src/cotmetrics/options_data.py | 4 +- src/cotmetrics/signals.py | 8 ++- tests/test_categories.py | 52 ++++++++++++++++ tests/test_signals.py | 4 +- 7 files changed, 135 insertions(+), 46 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ecd0abd..5af16bb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,7 +25,8 @@ dependencies = [ "pytz", "requests>=2.28", # legacy CFTC ETL (cotmetrics.etl) "xlrd>=2.0", # legacy CFTC dea_fut_xls are .xls - "cotdata>=0.1.0", # shared data layer; editable in the workspace: -e ../cotdata + "cotdata>=0.1.0", # COT positioning; editable in the workspace: -e ../cotdata + "marketdata>=0.1.0", # daily bars (ADR-0007 moved them out of cotdata): -e ../marketdata ] [project.optional-dependencies] diff --git a/src/cotmetrics/CotIndexer.py b/src/cotmetrics/CotIndexer.py index 1d2b38a..b1fb8a1 100644 --- a/src/cotmetrics/CotIndexer.py +++ b/src/cotmetrics/CotIndexer.py @@ -1,4 +1,5 @@ import copy +import importlib import json import os import threading @@ -211,9 +212,15 @@ def _load_raw_cot(self, columns=None) -> pd.DataFrame: @staticmethod def _cache_schema_marker_path() -> str: - """Sidecar recording the cotdata schema_version and the cotmetrics - METRICS_CACHE_VERSION the caches were built under. A sidecar (not a df - column) so it never leaks into metrics/ML features.""" + """Sidecar recording the schema versions of BOTH upstream stores plus the + cotmetrics METRICS_CACHE_VERSION the caches were built under. A sidecar (not + a df column) so it never leaks into metrics/ML features. + + Two stores since ADR-0007: cotdata for COT positioning, marketdata for bars. + The filename is unchanged so existing markers still parse; a marker written + before the split has no marketdata version, reads as 0, and busts once — + which is the correct outcome, because the price source moved underneath it. + """ return os.path.join(const.CACHE_DIR, "_cotdata_schema.json") @classmethod @@ -232,6 +239,13 @@ def _read_cache_schema(cls) -> int: except (TypeError, ValueError): return 0 + @classmethod + def _read_cache_marketdata_schema(cls) -> int: + try: + return int(cls._read_cache_marker().get("marketdata_schema_version", 0)) + except (TypeError, ValueError): + return 0 + @classmethod def _read_cache_metrics_version(cls) -> int: try: @@ -241,20 +255,30 @@ def _read_cache_metrics_version(cls) -> int: @classmethod def _stamp_cache_schema(cls) -> None: - """Record the cotdata schema_version and our METRICS_CACHE_VERSION next to - the parquet caches, so both an upstream store move and an internal - metrics-logic change bust the caches.""" + """Record both upstream store schema versions and our METRICS_CACHE_VERSION + next to the parquet caches, so a move in either store — or an internal + metrics-logic change — busts the caches. + + The two versions are kept as SEPARATE keys rather than combined. They are + independent counters, and collapsing them (a max, say) would hide a bump in + whichever store happens to sit at the lower number. + """ marker = {"metrics_version": int(const.METRICS_CACHE_VERSION)} - try: - import cotdata - marker["schema_version"] = int(cotdata.schema_version()) - except Exception as e: - # Preserve any previously recorded store schema rather than zeroing it - # (which would force a rebuild on every boot without cotdata). - prev = cls._read_cache_schema() - if prev: - marker["schema_version"] = prev - utils.cot_logger.warning(f"_stamp_cache_schema: cotdata schema unavailable: {e}") + # Preserve a previously recorded version rather than zeroing it when a store + # is unavailable, which would otherwise force a rebuild on every boot. + for key, mod_name, prev_read in ( + ("schema_version", "cotdata", cls._read_cache_schema), + ("marketdata_schema_version", "marketdata", + cls._read_cache_marketdata_schema)): + try: + mod = importlib.import_module(mod_name) + marker[key] = int(mod.schema_version()) + except Exception as e: + prev = prev_read() + if prev: + marker[key] = prev + utils.cot_logger.warning( + f"_stamp_cache_schema: {mod_name} schema unavailable: {e}") try: os.makedirs(const.CACHE_DIR, exist_ok=True) with open(cls._cache_schema_marker_path(), "w") as f: @@ -283,20 +307,26 @@ def try_load_from_cache(self) -> bool: f"— rebuilding all caches.") return False - # Bust all caches when the cotdata store schema moved (e.g. reconstructed - # volume promoted). The per-symbol guards below key on column *presence*, - # so they can't see a value-only change like front→reconstructed volume; - # the schema marker can. - try: - import cotdata - store_schema = int(cotdata.schema_version()) - if self._read_cache_schema() < store_schema: - utils.cot_logger.info( - f"try_load_from_cache: cache schema {self._read_cache_schema()} " - f"< cotdata schema {store_schema} — rebuilding all caches.") - return False - except Exception as e: - utils.cot_logger.warning(f"try_load_from_cache: schema check skipped: {e}") + # Bust all caches when EITHER upstream store's schema moved. The per-symbol + # guards below key on column *presence*, so they cannot see a value-only + # change like front→reconstructed volume; a schema marker can. + # + # Both stores are checked, and that is the point rather than tidiness. The + # example this guard was written for — reconstructed volume promoted — was a + # PRICE schema bump, and prices moved to marketdata under ADR-0007. Watching + # cotdata alone would leave the case it exists for uncovered. + for mod_name, cached in (("cotdata", self._read_cache_schema), + ("marketdata", self._read_cache_marketdata_schema)): + try: + store_schema = int(importlib.import_module(mod_name).schema_version()) + if cached() < store_schema: + utils.cot_logger.info( + f"try_load_from_cache: cache {mod_name} schema {cached()} " + f"< store schema {store_schema} — rebuilding all caches.") + return False + except Exception as e: + utils.cot_logger.warning( + f"try_load_from_cache: {mod_name} schema check skipped: {e}") self.years[-1] @@ -707,16 +737,19 @@ def retrieve_report_date_closing_prices(self, instrument, years, force_refresh=F instrument.df[col] = fallback_df[col] return fallback_df - # The per-instrument cache above missed, so read the bars from the cotdata store. - # This is a local parquet read, not a fetch: cotdata.get_prices never goes to the - # network. Say so, because a message about downloading sends anyone debugging a - # slow or failing boot looking for a network problem that cannot exist. + # The per-instrument cache above missed, so read the bars from the marketdata + # store. This is a local parquet read, not a fetch: marketdata.get_bars never goes + # to the network. Say so, because a message about downloading sends anyone + # debugging a slow or failing boot looking for a network problem that cannot exist. + # + # Bars moved out of cotdata under ADR-0007, which makes cotdata CFTC positioning + # only. COT reads in this file still go to cotdata; only prices moved. if price_data is None: - print(f"Reading {symbol} prices from the cotdata store...") + print(f"Reading {symbol} prices from the marketdata store...") try: - import cotdata + import marketdata start_date = f"{years[0]}-01-01" - price_data = cotdata.get_prices(symbol, adjustment='backadj', start=start_date) + price_data = marketdata.get_bars(symbol, 'backadj', start=start_date) except Exception as e: print(f"Error reading prices for {symbol} from the store: {e}") utils.cot_logger.error(f"Error reading prices for {symbol} from the store: {e}") diff --git a/src/cotmetrics/market_data.py b/src/cotmetrics/market_data.py index d2edadb..36e643a 100644 --- a/src/cotmetrics/market_data.py +++ b/src/cotmetrics/market_data.py @@ -2,7 +2,8 @@ The Databento daily-price fetch that used to live here has moved to the dormant cotdata provider (cotdata/providers/databento.py). All live price reads now go -through `cotdata.get_prices` (Norgate-backed store). Only the params.yaml-derived +through `marketdata.get_bars` (Norgate-backed store; bars moved out of cotdata under +ADR-0007). Only the params.yaml-derived `_SYMBOL_TO_NAME` map remains here — used by core.options_data for max-pain. """ import yaml diff --git a/src/cotmetrics/options_data.py b/src/cotmetrics/options_data.py index 97b316d..a0a9bc4 100644 --- a/src/cotmetrics/options_data.py +++ b/src/cotmetrics/options_data.py @@ -354,7 +354,7 @@ def update_all_daily_options(): """ Iterates over all supported instruments and fetches the daily options Max Pain snapshot. """ - import cotdata + import marketdata import cotmetrics.utils as utils from cotmetrics.market_data import _SYMBOL_TO_NAME @@ -363,7 +363,7 @@ def update_all_daily_options(): for symbol in _SYMBOL_TO_NAME.keys(): try: # Fetch the latest prices to provide the live price for scaling the proxy ETF - price_df = cotdata.get_prices(symbol, adjustment="backadj") + price_df = marketdata.get_bars(symbol, "backadj") if price_df is not None and not price_df.empty: live_price = price_df['Close'].iloc[-1] else: diff --git a/src/cotmetrics/signals.py b/src/cotmetrics/signals.py index 35be560..e67900b 100644 --- a/src/cotmetrics/signals.py +++ b/src/cotmetrics/signals.py @@ -1,6 +1,6 @@ import types -import cotdata +import marketdata import numpy as np import pandas as pd @@ -1024,8 +1024,10 @@ def compute_weekly_rejection_scores(symbol: str, cot_dates: pd.DatetimeIndex, fo feature cannot peek past the entry the way the old post-cutoff window did — see ``prior_week_rejection_window``. """ - # force_refresh is now a no-op: prices come from the cotdata store (producer-updated). - daily_df = cotdata.get_prices(symbol, adjustment="backadj") + # force_refresh is now a no-op: prices come from the marketdata store + # (producer-updated). Moved off cotdata by ADR-0007, which makes cotdata CFTC + # positioning only; the tier name and the returned frame are unchanged. + daily_df = marketdata.get_bars(symbol, "backadj") if daily_df is None or daily_df.empty: return pd.DataFrame() diff --git a/tests/test_categories.py b/tests/test_categories.py index 0021674..7470e97 100644 --- a/tests/test_categories.py +++ b/tests/test_categories.py @@ -276,3 +276,55 @@ def code_only(fn): assert "LEV_LONG_PSIZE_IDX" not in src, fn.__name__ assert "MM_LONG_POS_XLS" in src, fn.__name__ assert "LEV_LONG_POS_XLS" in src, fn.__name__ + + +def test_cache_marker_watches_both_upstream_stores(tmp_path, monkeypatch): + """ADR-0007 split the upstream: COT stays in cotdata, bars moved to marketdata. + + The cache-busting marker was written for a PRICE schema bump — reconstructed + volume being promoted — and prices are no longer in cotdata. Watching cotdata + alone would leave uncovered the exact case the guard exists for, and the failure + is silent: stale cached metrics computed off a superseded bar schema, with no + error anywhere. + + The two versions are kept as separate keys on purpose. Collapsing them into one + number (a max, say) would hide a bump in whichever store sits lower. + """ + import cotmetrics.constants as const + from cotmetrics.CotIndexer import CotIndexer + + # Both roots must be set for either version to be readable: each store raises on + # an unset root rather than defaulting to somewhere nobody looks. + monkeypatch.setenv("COTDATA_STORE", str(tmp_path / "cot")) + monkeypatch.setenv("MARKETDATA_STORE", str(tmp_path / "bars")) + import marketdata.store as md_store + md_store.stamp_flags() # give the bar store a manifest to report + + monkeypatch.setattr(const, "CACHE_DIR", str(tmp_path / "cache")) + CotIndexer._stamp_cache_schema() + + marker = CotIndexer._read_cache_marker() + assert "schema_version" in marker, "cotdata's store version is not recorded" + assert "marketdata_schema_version" in marker, ( + "marketdata's store version is not recorded, so a bar schema bump would " + "not bust the caches computed from it") + assert CotIndexer._read_cache_marketdata_schema() >= 1 + + +def test_a_marker_predating_the_split_busts_once(tmp_path, monkeypatch): + """Backward compatibility, and the right kind of it. A marker written before the + split records no marketdata version; that reads as 0, which is below any real + store and therefore forces exactly one rebuild. Correct rather than merely + tolerated — the price source moved underneath those caches.""" + import json + + import cotmetrics.constants as const + from cotmetrics.CotIndexer import CotIndexer + + monkeypatch.setattr(const, "CACHE_DIR", str(tmp_path)) + with open(CotIndexer._cache_schema_marker_path(), "w") as f: + json.dump({"metrics_version": const.METRICS_CACHE_VERSION, + "schema_version": 99}, f) # old marker: cotdata only + + assert CotIndexer._read_cache_schema() == 99 + assert CotIndexer._read_cache_marketdata_schema() == 0 diff --git a/tests/test_signals.py b/tests/test_signals.py index 488f3bc..6f236af 100644 --- a/tests/test_signals.py +++ b/tests/test_signals.py @@ -160,11 +160,11 @@ def test_compute_weekly_rejection_scores_ignores_post_cutoff_bars(monkeypatch): cot_date = dates[5] cot_index = pd.DatetimeIndex([cot_date]) - monkeypatch.setattr(signals.cotdata, "get_prices", + monkeypatch.setattr(signals.marketdata, "get_bars", lambda *a, **k: _rejection_ohlc(dates, monster_post=False)) calm = signals.compute_weekly_rejection_scores("TEST", cot_index) - monkeypatch.setattr(signals.cotdata, "get_prices", + monkeypatch.setattr(signals.marketdata, "get_bars", lambda *a, **k: _rejection_ohlc(dates, monster_post=True)) monster = signals.compute_weekly_rejection_scores("TEST", cot_index) From 61ff46ff7064869a60e12dbaf6735294eb3ed6e3 Mon Sep 17 00:00:00 2001 From: "Matt S." Date: Sun, 9 Aug 2026 02:28:54 +0000 Subject: [PATCH 2/2] CI: check out marketdata, since it is not on PyPI Every job went red at install. This repo resolves its internal siblings by checking them out and installing editable, and marketdata has no PyPI release, so declaring the dependency without adding the checkout makes `pip install -e .` fail before a single test runs. A dummy MARKETDATA_STORE joins the dummy COTDATA_STORE for the same reason the first one exists: neither package defaults a missing store root to somewhere plausible, both raise by name, and there are two roots now. The local run passed because the sibling was already installed in the venv. CI builds the environment from scratch, which is the only place a missing distribution shows up. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Jo4iovRfc2fzE9MwcLp7r2 --- .github/workflows/python-test.yml | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/.github/workflows/python-test.yml b/.github/workflows/python-test.yml index 8aa2247..ccaf311 100644 --- a/.github/workflows/python-test.yml +++ b/.github/workflows/python-test.yml @@ -26,6 +26,15 @@ jobs: repository: mspinola/cotdata path: cotdata + - name: Check out marketdata (daily bars — resolves the `marketdata` dep) + # ADR-0007 moved bars out of cotdata. marketdata is not on PyPI, so the + # declared dependency resolves only from this sibling checkout; without it + # `pip install -e .` fails before a single test runs. + uses: actions/checkout@v4 + with: + repository: mspinola/marketdata + path: marketdata + - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v5 with: @@ -35,7 +44,7 @@ jobs: working-directory: cotmetrics run: | python -m pip install --upgrade pip - python -m pip install -e ../cotdata -e .[options,scheduler,dev] + python -m pip install -e ../cotdata -e ../marketdata -e .[options,scheduler,dev] - name: Verify internal dependency floors # cotdata is installed editable from ../ at whatever HEAD was checked out, so @@ -51,9 +60,12 @@ jobs: - name: Run tests working-directory: cotmetrics env: - # Dummy store so cotdata's COTDATA_STORE guard doesn't error at import. + # Dummy stores so each package's store guard doesn't error at import. + # Two of them since ADR-0007: positioning and bars are separate roots, and + # neither package defaults a missing root to somewhere plausible. COTDATA_STORE: /tmp/cotdata_store + MARKETDATA_STORE: /tmp/marketdata_store COTMETRICS_CACHE: /tmp/cotmetrics_cache run: | - mkdir -p /tmp/cotdata_store /tmp/cotmetrics_cache + mkdir -p /tmp/cotdata_store /tmp/marketdata_store /tmp/cotmetrics_cache pytest tests/