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/ 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)