fix: upsert scoped Norgate metadata instead of replacing the whole table

src/cotdata/providers/norgate.py

@@ -375,13 +375,20 @@ def get_symbol_metadata(internal_symbol: str) -> dict | None:
 
 
 def update_metadata(symbols=None) -> None:
-    """Fetch and write contract specifications (metadata) to the store."""
+    """Fetch and write contract specifications (metadata) to the store.
+
+    A scoped run (`symbols` given) UPSERTS by Symbol into the existing
+    contract_specs table — rows for markets NOT in the request are preserved
+    (contract specs share one table, so a plain write would drop them). With no
+    `symbols`, regenerate the full registry and replace the table.
+    """
     import concurrent.futures
+    scoped = symbols is not None
     syms = symbols or [s.internal for s in all_symbols()]
-    
+
     print(f"Fetching metadata for {len(syms)} symbols...")
     metadata_rows = []
-    
+
     with concurrent.futures.ThreadPoolExecutor(max_workers=10) as pool:
         futs = {pool.submit(get_symbol_metadata, s): s for s in syms}
         for f in concurrent.futures.as_completed(futs):
@@ -393,7 +400,11 @@ def update_metadata(symbols=None) -> None:
         df = pd.DataFrame(metadata_rows)
         # Ensure consistent column ordering and sorting
         df = df.sort_values("Symbol").reset_index(drop=True)
-        store.write_metadata(df, source="norgate")
-        print(f"Successfully wrote metadata for {len(df)} symbols -> store")
+        if scoped:
+            store.upsert_metadata(df, source="norgate")
+            print(f"Upserted metadata for {len(df)} symbols -> store (unlisted markets preserved)")
+        else:
+            store.write_metadata(df, source="norgate")
+            print(f"Successfully wrote metadata for {len(df)} symbols -> store")
     else:
         print("No metadata fetched.")

src/cotdata/store.py

@@ -26,11 +26,30 @@ def _atomic_write_parquet(df: pd.DataFrame, path: Path) -> None:
 
 
 # ── Metadata ──────────────────────────────────────────────────────────────
+# Unlike prices/COT (one parquet per symbol), contract specs live in ONE table
+# keyed by Symbol. So a *scoped* refresh must upsert (see upsert_metadata) — a
+# plain write_metadata would replace the whole table and drop unlisted markets.
 def write_metadata(df: pd.DataFrame, source: str = "norgate") -> None:
     _atomic_write_parquet(df, config.metadata_dir() / "contract_specs.parquet")
     _touch_manifest("metadata", "contract_specs", df, source)
 
 
+def upsert_metadata(df: pd.DataFrame, source: str = "norgate") -> None:
+    """Merge `df` into the existing contract_specs table by ``Symbol``: rows for
+    symbols present in `df` are replaced/added; rows for symbols NOT in `df` are
+    kept. Use for a scoped (subset-of-symbols) refresh so it never drops the
+    contract specs of markets that weren't in the request. Use write_metadata to
+    replace the whole table (a full registry regeneration)."""
+    existing = read_metadata()
+    if not existing.empty and "Symbol" in existing.columns:
+        keep = existing[~existing["Symbol"].isin(df["Symbol"])]
+        merged = pd.concat([keep, df], ignore_index=True)
+    else:
+        merged = df
+    merged = merged.sort_values("Symbol").reset_index(drop=True)
+    write_metadata(merged, source=source)
+
+
 def read_metadata() -> pd.DataFrame:
     p = config.metadata_dir() / "contract_specs.parquet"
     return pd.read_parquet(p) if p.exists() else pd.DataFrame()

tests/test_norgate_provider.py

@@ -292,6 +292,52 @@ def mock_ts_side_effect(sym, **kwargs):
     assert all(s == "1970-01-01" for s in indiv_starts), indiv_starts
 
 
+def test_scoped_update_metadata_upserts_preserving_others(tmp_path, monkeypatch):
+    """`update_metadata(symbols=[...])` must UPSERT by Symbol into the existing
+    contract_specs — the data-loss regression where a 5-symbol run replaced the
+    whole 42-market table. Untouched markets survive; requested ones are refreshed.
+    Exercises the real store round-trip through a tmp COTDATA_STORE."""
+    monkeypatch.setenv("COTDATA_STORE", str(tmp_path))
+    from cotdata import store
+
+    # Pre-existing full table (stand-in for the 42 markets already on disk)
+    store.write_metadata(
+        pd.DataFrame({"Symbol": ["ES", "NQ", "DC"], "Tick Size": [0.25, 0.25, 0.01]}),
+        source="seed",
+    )
+
+    def fake_meta(sym):
+        return {"Symbol": sym, "Tick Size": 99.0}  # sentinel refreshed value
+
+    with mock.patch("cotdata.providers.norgate.get_symbol_metadata", side_effect=fake_meta):
+        norgate.update_metadata(symbols=["DC"])
+
+    df = store.read_metadata().set_index("Symbol")
+    assert set(df.index) == {"ES", "NQ", "DC"}   # ES/NQ preserved — not dropped
+    assert df.loc["DC", "Tick Size"] == 99.0      # DC refreshed
+    assert df.loc["ES", "Tick Size"] == 0.25      # untouched market unchanged
+
+
+def test_full_update_metadata_replaces_table(tmp_path, monkeypatch):
+    """`update_metadata()` with no symbols regenerates the whole registry and
+    REPLACES the table (drops symbols no longer produced)."""
+    monkeypatch.setenv("COTDATA_STORE", str(tmp_path))
+    from cotdata import store
+
+    store.write_metadata(
+        pd.DataFrame({"Symbol": ["OLD"], "Tick Size": [1.0]}), source="seed",
+    )
+
+    sym_a = mock.Mock(internal="ES")
+    sym_b = mock.Mock(internal="NQ")
+    with mock.patch("cotdata.providers.norgate.all_symbols", return_value=[sym_a, sym_b]), \
+         mock.patch("cotdata.providers.norgate.get_symbol_metadata",
+                    side_effect=lambda s: {"Symbol": s, "Tick Size": 1.0}):
+        norgate.update_metadata()  # no symbols → full replace
+
+    assert set(store.read_metadata()["Symbol"]) == {"ES", "NQ"}  # OLD gone
+
+
 import datetime as _dt
 def test_finals_ready_pure_logic():
     from cotdata.providers.norgate import _finals_ready

tests/test_store.py

@@ -133,6 +133,38 @@ def test_invalid_volume_arg_raises(store_env):
         get_prices("ES", "backadj", volume="bogus")
 
 
+def _specs(symbols, tick=1.0):
+    """Minimal contract_specs frame (Symbol-keyed, RangeIndex — as norgate writes)."""
+    return pd.DataFrame({"Symbol": symbols, "Tick Size": [tick] * len(symbols)})
+
+
+def test_upsert_metadata_preserves_unlisted_symbols(store_env):
+    """A scoped upsert must keep rows for symbols not in the incoming frame — the
+    data-loss regression: writing 5 symbols must not drop the other 42."""
+    from cotdata import store
+
+    store.write_metadata(_specs(["ES", "NQ", "DC", "ZO", "KE"]), source="test")
+
+    # Scoped refresh of only DC + a brand-new symbol
+    store.upsert_metadata(_specs(["DC", "EMD"], tick=9.9), source="norgate")
+
+    df = store.read_metadata()
+    assert set(df["Symbol"]) == {"ES", "NQ", "DC", "ZO", "KE", "EMD"}  # none dropped
+    # DC replaced with the new value; untouched markets keep their original
+    assert df.set_index("Symbol").loc["DC", "Tick Size"] == 9.9
+    assert df.set_index("Symbol").loc["ES", "Tick Size"] == 1.0
+    assert df.set_index("Symbol").loc["EMD", "Tick Size"] == 9.9  # new row added
+    # manifest reflects the union, not just the 2 upserted rows
+    assert store.load_manifest()["metadata"]["contract_specs"]["n_rows"] == 6
+
+
+def test_upsert_metadata_on_empty_store_writes_all(store_env):
+    """Upsert against an empty store behaves like a plain write."""
+    from cotdata import store
+    store.upsert_metadata(_specs(["ES", "NQ"]), source="norgate")
+    assert set(store.read_metadata()["Symbol"]) == {"ES", "NQ"}
+
+
 def test_schema_version_and_require_schema(store_env):
     from cotdata import store, schema_version, require_schema
     import cotdata.config as cfg