diff --git a/README.md b/README.md index 468110b..a1c13fa 100644 --- a/README.md +++ b/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 | |--------|------|-------------| diff --git a/src/cotdata/cot.py b/src/cotdata/cot.py index 8d6c0e3..d8dab5b 100644 --- a/src/cotdata/cot.py +++ b/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() diff --git a/src/cotdata/providers/cftc.py b/src/cotdata/providers/cftc.py index ba18006..72c1d29 100644 --- a/src/cotdata/providers/cftc.py +++ b/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") diff --git a/src/cotdata/providers/cftc_disagg.py b/src/cotdata/providers/cftc_disagg.py index 25f73e0..5d913ec 100644 --- a/src/cotdata/providers/cftc_disagg.py +++ b/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") diff --git a/src/cotdata/providers/cftc_tff.py b/src/cotdata/providers/cftc_tff.py index 5f9b26c..c83873c 100644 --- a/src/cotdata/providers/cftc_tff.py +++ b/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") diff --git a/src/cotdata/providers/norgate.py b/src/cotdata/providers/norgate.py index 62f36d1..38a4736 100644 --- a/src/cotdata/providers/norgate.py +++ b/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}") diff --git a/tests/test_norgate_provider.py b/tests/test_norgate_provider.py new file mode 100644 index 0000000..5acd650 --- /dev/null +++ b/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"