Read bars from marketdata, not cotdata (ADR-0007 step 4)

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

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]

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}")

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

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:

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

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

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)