Feat/norgate metadata

README.md

@@ -168,7 +168,7 @@ The CFTC publishes positioning data in three distinct formats. `cotdata` manages
 The canonical store uses standard Parquet files. When loaded into a pandas DataFrame (e.g., via `pd.read_parquet()`), they conform to the following schemas.
 
 ### Price Data (`prices/{symbol}_{adjustment}.parquet`)
-The primary source for price history (Norgate Data). Indexed by tz-naive `Date`.
+The primary source for price history (Norgate Data). Indexed by tz-naive `Date`. The pipeline automatically downloads both the back-adjusted (`backadj`) series for signals/stops and the unadjusted (`unadj`) series for true transaction cost modeling.
 
 | Column | Type | Description |
 |--------|------|-------------|

src/cotdata/cot.py

@@ -39,12 +39,12 @@ def get_cot(name: str, report: str = "legacy") -> pd.DataFrame:
 
     if sym is None or not sym.cftc_code:
         return read_fn(name)
-    primary = read_fn(sym.cftc_code)
+    primary = read_fn(f"{sym.internal}_{sym.cftc_code}")
     if not sym.hist_codes:
         return primary
     frames = [primary]                    # primary first → wins de-duplication on overlaps
     for hc, scale in hist_code_scales(sym.hist_codes):
-        h = read_fn(hc)
+        h = read_fn(f"{sym.internal}_{hc}")
         if h.empty:
             continue
         h = h.copy()

src/cotdata/providers/cftc.py

@@ -104,13 +104,18 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
     Rebuilds the complete per-code table each run (parse is cheap; downloads are
     cached and skip when unchanged). Incremental append is a future optimization.
     """
+    code_to_sym = {}
+    for s in all_symbols():
+        if s.cftc_code:
+            code_to_sym[s.cftc_code] = s.internal
+        for hc, _ in hist_code_scales(s.hist_codes):
+            code_to_sym[hc] = s.internal
+
     last_year = last_year or dt.date.today().year
     if codes:
         want = set(codes)
     else:
-        want = {s.cftc_code for s in all_symbols() if s.cftc_code}
-        for s in all_symbols():      # predecessor codes (migrated-contract history)
-            want.update(code for code, _ in hist_code_scales(s.hist_codes))
+        want = set(code_to_sym.keys())
 
     frames = []
     for year in range(first_year, last_year + 1):
@@ -133,5 +138,7 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
             print(f"{code}: no rows")
             continue
         sub = sub.sort_values(REPORT_DATE).set_index(REPORT_DATE)
-        store.write_cot_legacy(code, sub, source="cftc")
-        print(f"{code}: {len(sub):5d} weeks (legacy) -> store")
+        sym_name = code_to_sym.get(code)
+        file_name = f"{sym_name}_{code}" if sym_name else code
+        store.write_cot_legacy(file_name, sub, source="cftc")
+        print(f"{file_name}: {len(sub):5d} weeks (legacy) -> store")

src/cotdata/providers/cftc_disagg.py

@@ -90,13 +90,18 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
     codes: iterable of CFTC codes; default = all registry codes.
     Rebuilds the complete per-code table each run.
     """
+    code_to_sym = {}
+    for s in all_symbols():
+        if s.cftc_code:
+            code_to_sym[s.cftc_code] = s.internal
+        for hc, _ in hist_code_scales(s.hist_codes):
+            code_to_sym[hc] = s.internal
+
     last_year = last_year or dt.date.today().year
     if codes:
         want = set(codes)
     else:
-        want = {s.cftc_code for s in all_symbols() if s.cftc_code}
-        for s in all_symbols():      # predecessor codes (migrated-contract history)
-            want.update(code for code, _ in hist_code_scales(s.hist_codes))
+        want = set(code_to_sym.keys())
 
     frames = []
 
@@ -142,5 +147,7 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
 
         # Index by report date (DatetimeIndex → manifest last_date)
         sub = sub.sort_values(REPORT_DATE).set_index(REPORT_DATE)
-        store.write_cot_disagg(code, sub, source="cftc_disagg")
-        print(f"{code}: {len(sub):5d} weeks (disagg) -> store")
+        sym_name = code_to_sym.get(code)
+        file_name = f"{sym_name}_{code}" if sym_name else code
+        store.write_cot_disagg(file_name, sub, source="cftc_disagg")
+        print(f"{file_name}: {len(sub):5d} weeks (disagg) -> store")

src/cotdata/providers/cftc_tff.py

@@ -90,13 +90,18 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
     codes: iterable of CFTC codes; default = all registry codes.
     Rebuilds the complete per-code table each run.
     """
+    code_to_sym = {}
+    for s in all_symbols():
+        if s.cftc_code:
+            code_to_sym[s.cftc_code] = s.internal
+        for hc, _ in hist_code_scales(s.hist_codes):
+            code_to_sym[hc] = s.internal
+
     last_year = last_year or dt.date.today().year
     if codes:
         want = set(codes)
     else:
-        want = {s.cftc_code for s in all_symbols() if s.cftc_code}
-        for s in all_symbols():      # predecessor codes (migrated-contract history)
-            want.update(code for code, _ in hist_code_scales(s.hist_codes))
+        want = set(code_to_sym.keys())
 
     frames = []
 
@@ -142,5 +147,7 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
 
         # Index by report date (DatetimeIndex → manifest last_date)
         sub = sub.sort_values(REPORT_DATE).set_index(REPORT_DATE)
-        store.write_cot_tff(code, sub, source="cftc_tff")
-        print(f"{code}: {len(sub):5d} weeks (tff) -> store")
+        sym_name = code_to_sym.get(code)
+        file_name = f"{sym_name}_{code}" if sym_name else code
+        store.write_cot_tff(file_name, sub, source="cftc_tff")
+        print(f"{file_name}: {len(sub):5d} weeks (tff) -> store")

src/cotdata/providers/norgate.py

@@ -26,10 +26,13 @@
 }
 
 
-def fetch(internal_symbol: str, start: str = "1970-01-01") -> pd.DataFrame:
-    """Fetch Norgate back-adjusted continuous bars: settlement close, gap-free."""
+def fetch(internal_symbol: str, adjustment: str = "backadj", start: str = "1970-01-01") -> pd.DataFrame:
+    """Fetch Norgate continuous bars: settlement close."""
     import norgatedata  # imported lazily; only present on the Windows producer
-    ng_sym = REGISTRY[internal_symbol].norgate + CCB_SUFFIX  # "&ES" → "&ES_CCB"
+    ng_sym = REGISTRY[internal_symbol].norgate
+    if adjustment == "backadj":
+        ng_sym += CCB_SUFFIX
+
     df = norgatedata.price_timeseries(
         ng_sym,
         padding_setting=norgatedata.PaddingType.NONE,
@@ -66,14 +69,20 @@ def _check_roll_gaps(internal_symbol: str, df: pd.DataFrame) -> bool:
 
 
 def update(symbols=None) -> None:
-    """Fetch + write to the store for the given internal symbols (backadj only)."""
+    """Fetch + write to the store for the given internal symbols (backadj and unadj)."""
     syms = symbols or [s.internal for s in all_symbols()]
     for sym in syms:
         try:
-            out = fetch(sym)
-            _check_roll_gaps(sym, out)  # sanity: warn (don't block) if it looks unadjusted
-            store.write_prices(sym, "backadj", out, source="norgate")
-            print(f"{sym:5s}: {len(out):6d} bars -> store")
+            # 1. Back-Adjusted
+            out_backadj = fetch(sym, adjustment="backadj")
+            _check_roll_gaps(sym, out_backadj)  # sanity: warn if backadj looks unadjusted
+            store.write_prices(sym, "backadj", out_backadj, source="norgate")
+
+            # 2. Unadjusted (Raw calendar spreads)
+            out_unadj = fetch(sym, adjustment="unadj")
+            store.write_prices(sym, "unadj", out_unadj, source="norgate")
+
+            print(f"{sym:5s}: {len(out_backadj):6d} backadj, {len(out_unadj):6d} unadj -> store")
         except Exception as e:  # noqa: BLE001
             print(f"{sym:5s}: FAILED — {e}")
 

tests/test_norgate_provider.py

@@ -0,0 +1,54 @@
+import pandas as pd
+from unittest import mock
+
+# Mock norgatedata so we can import and test norgate on any OS
+import sys
+import types
+mock_norgatedata = types.ModuleType("norgatedata")
+mock_norgatedata.PaddingType = mock.Mock()
+mock_norgatedata.PaddingType.NONE = "NONE"
+sys.modules["norgatedata"] = mock_norgatedata
+
+from cotdata.providers import norgate
+
+@mock.patch("cotdata.providers.norgate.store.write_prices")
+@mock.patch("norgatedata.price_timeseries", create=True)
+def test_norgate_update_fetches_both_adjustments(mock_price_ts, mock_write_prices):
+    """Verify that update() fetches both the backadj and unadj series for a symbol."""
+
+    # Setup mock returns
+    mock_df = pd.DataFrame({
+        "Open": [100.0], "High": [101.0], "Low": [99.0], "Close": [100.5],
+        "Volume": [1000], "Open Interest": [5000], "Delivery Month": [202609]
+    }, index=pd.DatetimeIndex(["2026-07-01"]))
+
+    mock_price_ts.return_value = mock_df
+
+    # Mock all_symbols to just return a dummy registry entry for "ES"
+    mock_symbol = mock.Mock()
+    mock_symbol.internal = "ES"
+
+    # We must patch REGISTRY and all_symbols so it uses our mock
+    with mock.patch("cotdata.providers.norgate.all_symbols", return_value=[mock_symbol]), \
+         mock.patch.dict("cotdata.providers.norgate.REGISTRY", {"ES": mock.Mock(norgate="&ES")}):
+
+        # Run update
+        norgate.update(symbols=["ES"])
+
+        # Verify norgatedata API was called twice with correct raw symbols
+        assert mock_price_ts.call_count == 2
+        calls = mock_price_ts.call_args_list
+        # Call 1: backadj ("&ES_CCB")
+        assert calls[0][0][0] == "&ES_CCB"
+        # Call 2: unadj ("&ES")
+        assert calls[1][0][0] == "&ES"
+
+        # Verify store.write_prices was called twice with correct adjustment flags
+        assert mock_write_prices.call_count == 2
+        write_calls = mock_write_prices.call_args_list
+        # Call 1: store.write_prices("ES", "backadj", out_backadj, source="norgate")
+        assert write_calls[0][0][0] == "ES"
+        assert write_calls[0][0][1] == "backadj"
+        # Call 2: store.write_prices("ES", "unadj", out_unadj, source="norgate")
+        assert write_calls[1][0][0] == "ES"
+        assert write_calls[1][0][1] == "unadj"