diff --git a/src/cotdata/providers/norgate.py b/src/cotdata/providers/norgate.py index ef8b08a..53cf2f3 100644 --- a/src/cotdata/providers/norgate.py +++ b/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.") diff --git a/src/cotdata/store.py b/src/cotdata/store.py index f012b42..c104a53 100644 --- a/src/cotdata/store.py +++ b/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() diff --git a/tests/test_norgate_provider.py b/tests/test_norgate_provider.py index da5eca2..a328292 100644 --- a/tests/test_norgate_provider.py +++ b/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 diff --git a/tests/test_store.py b/tests/test_store.py index fb8c2f6..be85698 100644 --- a/tests/test_store.py +++ b/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