Docs hygiene, uv.lock, test coverage, and Norgate producer reliability

CHANGELOG.md

@@ -16,4 +16,31 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   R-multiples unusable; `propadj` yields a strictly-positive DC series
   (4.68–25.01) over the full 1997–2026 history. Recommended for low-priced,
   long-history contracts. Derived from already-stored series — no producer
-  re-run or schema change required. ([#23](https://github.com/mspinola/cotdata/pull/23))
+  re-run or schema change required.
+  ([#23](https://github.com/mspinola/cotdata/pull/23), [#26](https://github.com/mspinola/cotdata/pull/26))
+- **Yahoo Finance price provider** (`cotdata-update --prices-yahoo`) — a
+  cross-platform, research-grade price source for registry symbols carrying a
+  `yahoo` ticker, so markets Norgate/databento don't cover can still be priced
+  off ETF proxies. Adds the MSCI EM (MME→EEM) and EAFE (MFS→EFA) held-out
+  generalization markets. ([#24](https://github.com/mspinola/cotdata/pull/24))
+
+### Fixed
+- **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
+  `sys.exit()`, which exits **0** (a scheduled run looked "successful" while
+  writing nothing and never retried) and raised `SystemExit` past the per-symbol
+  handler, killing the run on the first symbol.
+- **Never persist all-null contract-spec rows** — if Norgate returns nothing for
+  every spec field of a covered symbol (a transient failure), `--metadata` now
+  skips it with a warning instead of writing a null row or, on a scoped upsert,
+  overwriting good existing specs with nulls.
+- **Skip Yahoo-only markets in the Norgate producer** — MME/MFS have no Norgate
+  continuous series, so `--prices`/`--metadata` were erroring on `&MME_CCB` /
+  `&MFS_CCB` and silently writing all-null contract-spec rows. They are now
+  marked `norgate: null` in the registry and skipped by the Norgate producer.
+  ([#27](https://github.com/mspinola/cotdata/pull/27))
+- **Scoped metadata refresh no longer drops other markets** — a
+  `--metadata --symbols …` run now UPSERTs by `Symbol` into `contract_specs`
+  instead of replacing the whole table, so specs for markets outside the request
+  survive. ([#25](https://github.com/mspinola/cotdata/pull/25))

README.md

@@ -304,7 +304,7 @@ The canonical store uses standard Parquet files. Loaded with `pd.read_parquet()`
 ### Price Data (`prices/{symbol}_{adjustment}.parquet`)
 Primary price history (Norgate Data), indexed by tz-naive `Date`. The pipeline downloads both the back-adjusted (`backadj`) series for signals/stops and the unadjusted (`unadj`) series for true transaction-cost modeling.
 
-**Reading reconstructed volume:** the reconstruction columns below are internal storage. Consumers should not read `Volume_Reconstructed` directly — call `get_prices(symbol, volume="reconstructed")` and the `Volume` column is served as reconstructed-with-per-row-raw-fallback, plus a `Volume_Source` column for audit. The default `volume="front"` returns the front-month series unchanged (byte-identical to the pre-v2 API). See `docs/plan_promote_reconstructed_volume.md`.
+**Reading reconstructed volume:** the reconstruction columns below are internal storage. Consumers should not read `Volume_Reconstructed` directly — call `get_prices(symbol, volume="reconstructed")` and the `Volume` column is served as reconstructed-with-per-row-raw-fallback, plus a `Volume_Source` column for audit. The default `volume="front"` returns the front-month series unchanged (byte-identical to the pre-v2 API).
 
 **Schema versioning:** `schema_version` in `manifest.json` records the on-disk data version (v2 = reconstructed volume promoted). Consumers key cache invalidation on `cotdata.schema_version()` and can guard with `cotdata.require_schema(min_version)`.
 

src/cotdata/config.py

@@ -6,7 +6,7 @@
 # Volume_Source, and get_prices(volume="reconstructed") serves them. The store
 # was migrated by a full producer pass (2026-07-14) and now carries v2 shape;
 # schema_version() reflects the on-disk manifest, so a fresh/partial store can
-# still read <2. See docs/plan_promote_reconstructed_volume.md for the rollout.
+# still read <2.
 SCHEMA_VERSION = 2
 
 

src/cotdata/providers/norgate.py

@@ -40,6 +40,12 @@
 
 MONTH_CODES = {'F': 1, 'G': 2, 'H': 3, 'J': 4, 'K': 5, 'M': 6, 'N': 7, 'Q': 8, 'U': 9, 'V': 10, 'X': 11, 'Z': 12}
 
+# Contract-spec fields fetched per symbol (everything in a metadata row except the
+# Symbol/Norgate_Symbol identifiers). If Norgate returns nothing for ALL of these,
+# the row is junk — skip it rather than persist an all-null spec row.
+_SPEC_FIELDS = ("Name", "Exchange", "Group", "Contract Size", "Tick Size",
+                "Tick Value", "Point Value", "Currency", "Margin")
+
 
 def _reconstruct_volume(internal_symbol: str, continuous_df: pd.DataFrame, adjustment: str,
                         full: bool = False) -> pd.DataFrame:
@@ -292,6 +298,30 @@ def _norgate_covered(symbols):
     return covered
 
 
+def _require_norgate_service() -> None:
+    """Fail fast, with a clear message and a normal exception, if the Norgate Data
+    Updater (NDU) service isn't reachable — BEFORE any fetch.
+
+    Why this matters: norgatedata retries each data call 10x and then calls bare
+    ``sys.exit()``, which (a) exits with code 0, so a scheduled producer run looks
+    "successful" while writing nothing and never triggers the scheduler's retry,
+    and (b) raises SystemExit — not caught by the per-symbol ``except Exception`` —
+    so the whole run dies on the first symbol. ``norgatedata.status()`` is a safe
+    probe (haltonerror=False, maxretries=1 → returns False instead of exiting)."""
+    import norgatedata
+    try:
+        reachable = bool(norgatedata.status())
+    except BaseException:  # noqa: BLE001 — never let the probe itself take us down
+        reachable = False
+    if not reachable:
+        raise RuntimeError(
+            "Norgate Data service is not reachable — is the Norgate Data Updater "
+            "(NDU) running and authenticated? cotdata prices/metadata are produced "
+            "on Windows with NDU running. Aborting before fetch (non-zero exit so a "
+            "scheduler retries)."
+        )
+
+
 def update(symbols=None, full: bool = False) -> None:
     """Fetch + write to the store for the given internal symbols (backadj and unadj).
 
@@ -302,6 +332,7 @@ def update(symbols=None, full: bool = False) -> None:
     import time
     from .. import status
 
+    _require_norgate_service()  # abort cleanly if NDU is down (see helper docstring)
     syms = _norgate_covered(symbols)
     prior = store.load_manifest().get("prices", {})  # to report per-symbol date deltas
     t0 = time.time()
@@ -401,6 +432,7 @@ def update_metadata(symbols=None) -> None:
     """
     import concurrent.futures
     scoped = symbols is not None
+    _require_norgate_service()  # abort cleanly if NDU is down (see helper docstring)
     syms = _norgate_covered(symbols)
 
     print(f"Fetching metadata for {len(syms)} symbols...")
@@ -410,8 +442,16 @@ def update_metadata(symbols=None) -> None:
         futs = {pool.submit(get_symbol_metadata, s): s for s in syms}
         for f in concurrent.futures.as_completed(futs):
             result = f.result()
-            if result:
-                metadata_rows.append(result)
+            if not result:
+                continue
+            # A covered symbol whose specs all came back None is a transient Norgate
+            # failure, not real data — skip rather than persist a null row (and, on a
+            # scoped upsert, rather than overwrite good existing specs with nulls).
+            if all(result.get(k) is None for k in _SPEC_FIELDS):
+                print(f"  ⚠️  {result.get('Symbol')}: all specs empty (Norgate "
+                      f"returned nothing) — skipping to avoid a null row")
+                continue
+            metadata_rows.append(result)
 
     if metadata_rows:
         df = pd.DataFrame(metadata_rows)

src/cotdata/store.py

@@ -124,8 +124,7 @@ def require_schema(min_version: int) -> None:
     if v < min_version:
         raise RuntimeError(
             f"cotdata store schema_version={v} < required {min_version}. "
-            f"Re-run the producer (e.g. norgate.update) to migrate the store — "
-            f"see docs/plan_promote_reconstructed_volume.md."
+            f"Re-run the producer (e.g. norgate.update) to migrate the store."
         )
 
 

tests/test_cot.py

@@ -0,0 +1,76 @@
+"""Tests for the consumer COT read API (cotdata.cot.get_cot): report dispatch,
+symbol/code lookup, and the predecessor-code stitching (scale + de-dup) that
+splices an instrument's earlier CFTC listing behind its current one."""
+from unittest import mock
+
+import pandas as pd
+import pytest
+
+from cotdata import cot
+from cotdata.registry import REGISTRY
+
+_CODE_COL = "CFTC_Contract_Market_Code"
+
+
+def _df(dates, oi, code):
+    return pd.DataFrame(
+        {"Open_Interest_All": oi, _CODE_COL: [code] * len(dates)},
+        index=pd.DatetimeIndex(dates),
+    )
+
+
+def test_unknown_report_raises():
+    with pytest.raises(ValueError, match="Unknown report type"):
+        cot.get_cot("ES", report="bogus")
+
+
+def test_simple_symbol_returns_primary_by_symbol_code_key():
+    """A symbol with no hist_codes (ES) reads exactly `<internal>_<cftc_code>` from
+    the requested report and returns it unchanged."""
+    primary = _df(["2020-01-01"], [10], REGISTRY["ES"].cftc_code)
+    with mock.patch("cotdata.cot.store.read_cot_legacy", return_value=primary) as m:
+        out = cot.get_cot("ES")
+    m.assert_called_once_with(f"ES_{REGISTRY['ES'].cftc_code}")
+    pd.testing.assert_frame_equal(out, primary)
+
+
+def test_report_dispatch_uses_matching_store_reader():
+    """report='disagg' must read from the disagg store, not legacy."""
+    with mock.patch("cotdata.cot.store.read_cot_disagg",
+                    return_value=_df(["2020-01-01"], [1], REGISTRY["GC"].cftc_code)) as m:
+        cot.get_cot("GC", report="disagg")
+    m.assert_called_once_with(f"GC_{REGISTRY['GC'].cftc_code}")
+
+
+def test_lookup_by_primary_cftc_code_resolves_symbol():
+    """Passing the CFTC code instead of the internal symbol resolves to the symbol
+    and reads the same `<internal>_<code>` key."""
+    primary = _df(["2020-01-01"], [10], REGISTRY["ES"].cftc_code)
+    with mock.patch("cotdata.cot.store.read_cot_legacy", return_value=primary) as m:
+        cot.get_cot(REGISTRY["ES"].cftc_code)          # "13874A"
+    m.assert_called_once_with(f"ES_{REGISTRY['ES'].cftc_code}")
+
+
+def test_hist_code_stitching_scales_and_dedups_keeping_primary():
+    """LBR splices predecessor 058643 (scale 4.0) behind primary 058644:
+      - predecessor numeric columns are scaled by 4.0 (a contract-size bridge),
+      - predecessor rows are re-stamped with the primary code,
+      - overlapping dates keep the PRIMARY row (concat primary-first, dedup first),
+      - the result is sorted ascending."""
+    lbr = REGISTRY["LBR"]
+    primary_key = f"LBR_{lbr.cftc_code}"            # LBR_058644
+    hist_key = "LBR_058643"
+
+    primary = _df(["2020-01-08", "2020-01-15"], [100, 200], lbr.cftc_code)
+    hist = _df(["2020-01-01", "2020-01-08"], [10, 999], "058643")   # overlaps 01-08
+
+    def reader(key):
+        return {primary_key: primary, hist_key: hist}.get(key, pd.DataFrame())
+
+    with mock.patch("cotdata.cot.store.read_cot_legacy", side_effect=reader):
+        out = cot.get_cot("LBR")
+
+    assert list(out.index) == list(pd.DatetimeIndex(["2020-01-01", "2020-01-08", "2020-01-15"]))
+    assert out.loc["2020-01-01", "Open_Interest_All"] == 40      # 10 * 4.0 (scaled predecessor)
+    assert out.loc["2020-01-08", "Open_Interest_All"] == 100     # primary wins the overlap (not 999*4)
+    assert (out[_CODE_COL] == lbr.cftc_code).all()               # predecessor re-stamped to primary code

tests/test_databento_provider.py

@@ -0,0 +1,89 @@
+"""Smoke tests for the DORMANT Databento provider. It isn't in the live EOD path,
+but it carries the hard-won statistics logic (Open Interest = stat_type 9;
+settlement = stat_type 3, NOT 7=LOWEST_OFFER, dated by ts_ref) that the intraday
+work will reuse — so lock that parsing here. `databento` is an optional extra, so
+mock the SDK (as test_norgate_provider mocks norgatedata) to run on any machine."""
+import sys
+import types
+from unittest import mock
+
+import pandas as pd
+import pytest
+
+# databento is optional (the [databento] extra); force a mock module so the lazy
+# `import databento as db` inside the provider resolves without the SDK installed.
+mock_databento = types.ModuleType("databento")
+sys.modules["databento"] = mock_databento
+
+from cotdata.providers import databento as dbprov
+
+
+def _client(ohlcv_df, stats_df):
+    """A fake databento Historical whose timeseries.get_range returns the OHLCV
+    frame for schema 'ohlcv-1d' and the statistics frame otherwise."""
+    client = mock.Mock()
+
+    def get_range(**kwargs):
+        res = mock.Mock()
+        res.to_df.return_value = ohlcv_df if kwargs.get("schema") == "ohlcv-1d" else stats_df
+        return res
+
+    client.timeseries.get_range.side_effect = get_range
+    return client
+
+
+@pytest.fixture
+def ohlcv():
+    idx = pd.DatetimeIndex(["2020-01-02"], name="ts_event")
+    return pd.DataFrame(
+        {"open": [99.0], "high": [101.0], "low": [98.0], "close": [100.0], "volume": [1234]},
+        index=idx,
+    )
+
+
+@pytest.fixture
+def stats():
+    # OI (stat_type 9) is disseminated for the session date; the settlement
+    # (stat_type 3) for the 01-02 session is disseminated the NEXT morning
+    # (ts_event 01-03) but carries ts_ref 01-02 — the session it applies to.
+    idx = pd.DatetimeIndex(["2020-01-02", "2020-01-03"], name="ts_event")
+    return pd.DataFrame(
+        {
+            "stat_type": [9, 3],
+            "price": [0.0, 101.5],
+            "quantity": [5000.0, 0.0],
+            "ts_ref": pd.to_datetime(["2020-01-02", "2020-01-02"]),
+        },
+        index=idx,
+    )
+
+
+def test_provider_imports_without_sdk():
+    """The provider module imports and exposes its entry points even though
+    `databento` isn't a hard dependency (import is lazy, behind the extra)."""
+    assert callable(dbprov.fetch_daily_ohlc)
+    assert callable(dbprov.update_all_daily_prices)
+
+
+def test_fetch_extracts_open_interest_from_stat_type_9(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", price_type="close")
+
+    assert df.loc["2020-01-02", "Close"] == 100.0            # ohlcv close, untouched
+    assert df.loc["2020-01-02", "Open Interest"] == 5000.0   # from stat_type 9
+
+
+def test_settlement_overrides_close_dated_by_ts_ref(tmp_path, monkeypatch, ohlcv, stats):
+    """price_type='settlement' replaces Close with stat_type 3's price, joined by
+    ts_ref (the session), so the next-morning-disseminated settle lands on 01-02."""
+    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", price_type="settlement")
+
+    assert df.loc["2020-01-02", "Close"] == 101.5            # settle, not ohlcv 100.0
+    assert df.loc["2020-01-02", "Open Interest"] == 5000.0

tests/test_norgate_provider.py

@@ -1,5 +1,6 @@
 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
@@ -8,6 +9,7 @@
 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
@@ -338,6 +340,45 @@ def test_full_update_metadata_replaces_table(tmp_path, monkeypatch):
     assert set(store.read_metadata()["Symbol"]) == {"ES", "NQ"}  # OLD gone
 
 
+def test_update_aborts_fast_when_ndu_unreachable(monkeypatch):
+    """When NDU is down, update() must raise BEFORE any fetch — not fall into
+    norgatedata's 10x-retry + bare sys.exit() (which exits 0 and defeats scheduler
+    retry). norgatedata.status() returning False is the trip wire."""
+    monkeypatch.setattr(mock_norgatedata, "status", mock.Mock(return_value=False))
+    with pytest.raises(RuntimeError, match="Norgate Data service is not reachable"):
+        norgate.update(symbols=["ES"])
+
+
+def test_update_metadata_aborts_fast_when_ndu_unreachable(monkeypatch):
+    monkeypatch.setattr(mock_norgatedata, "status", mock.Mock(return_value=False))
+    with pytest.raises(RuntimeError, match="Norgate Data service is not reachable"):
+        norgate.update_metadata(symbols=["ES"])
+
+
+def test_metadata_skips_all_null_spec_rows(tmp_path, monkeypatch):
+    """A COVERED symbol whose specs all come back None (a transient Norgate failure,
+    not the MME/MFS no-coverage case) must be skipped — never written as a null row,
+    and on a scoped upsert never used to overwrite good existing specs."""
+    monkeypatch.setenv("COTDATA_STORE", str(tmp_path))
+    from cotdata import store
+
+    sym_es = mock.Mock(internal="ES", norgate="&ES")
+    sym_nq = mock.Mock(internal="NQ", norgate="&NQ")
+
+    def fake_meta(sym):
+        base = {"Symbol": sym, "Norgate_Symbol": f"&{sym}_CCB"}
+        if sym == "NQ":                                   # all specs empty → junk
+            return {**base, **{k: None for k in norgate._SPEC_FIELDS}}
+        return {**base, **{k: None for k in norgate._SPEC_FIELDS}, "Tick Size": 0.25}
+
+    with mock.patch("cotdata.providers.norgate.all_symbols", return_value=[sym_es, sym_nq]), \
+         mock.patch.dict("cotdata.providers.norgate.REGISTRY", {"ES": sym_es, "NQ": sym_nq}), \
+         mock.patch("cotdata.providers.norgate.get_symbol_metadata", side_effect=fake_meta):
+        norgate.update_metadata()  # full run
+
+    assert set(store.read_metadata()["Symbol"]) == {"ES"}   # NQ null row skipped
+
+
 def test_metadata_skips_symbols_without_norgate_coverage(tmp_path, monkeypatch):
     """Yahoo-only markets (registry norgate=None, e.g. MME/MFS) must be skipped by
     the Norgate metadata producer — never fetched, never written as null rows. The