From 405490834ec58b1e81edc72bb902c7ca87a35647 Mon Sep 17 00:00:00 2001 From: Matt Spinola Date: Tue, 14 Jul 2026 18:55:31 -0400 Subject: [PATCH] feat: add --full flag to force a complete reconstructed-volume rebuild MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reconstruction fetch is incremental: when the store already carries Volume_Reconstructed, it only recomputes a trailing 60-day window. That's right for nightly data updates but wrong after a reconstruction *logic* change — the by-volume-rank fix — which needs every historical row recomputed. A plain `cotdata-update --prices` would leave old rows (e.g. GC's understated history) on the previous algorithm. Add full=True to norgate.update / _reconstruct_volume and a --full CLI flag that resets the incremental bound to epoch, forcing a from-scratch rebuild. Default behavior (incremental) is unchanged. Test: full=True issues the individual-contract fetch with start_date=1970-01-01 even when the store already has recent Volume_Reconstructed. Operational note: after the by-rank fix, the store must be rebuilt with `cotdata-update --prices --full` on the Windows/norgatedata producer before consumers see corrected volume. Co-Authored-By: Claude Opus 4.8 --- src/cotdata/providers/norgate.py | 28 ++++++++++++------- src/cotdata/update.py | 5 +++- tests/test_norgate_provider.py | 46 ++++++++++++++++++++++++++++++++ 3 files changed, 69 insertions(+), 10 deletions(-) diff --git a/src/cotdata/providers/norgate.py b/src/cotdata/providers/norgate.py index 12b1636..fca2eb1 100644 --- a/src/cotdata/providers/norgate.py +++ b/src/cotdata/providers/norgate.py @@ -33,17 +33,22 @@ 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} -def _reconstruct_volume(internal_symbol: str, continuous_df: pd.DataFrame, adjustment: str) -> pd.DataFrame: +def _reconstruct_volume(internal_symbol: str, continuous_df: pd.DataFrame, adjustment: str, + full: bool = False) -> pd.DataFrame: """Calculate FirstVolume, SecondVolume, and Volume_Reconstructed. Uses incremental fetching based on the last successful Volume_Reconstructed date. Returns continuous_df with the additive columns attached. + + full=True recomputes the ENTIRE history from scratch, ignoring the incremental + window. Needed when the reconstruction *logic* changes (not just new data): the + trailing-60-day window would otherwise leave old rows on the previous algorithm. """ import norgatedata - + # 1. Gap-aware incremental bounds existing_df = store.read_prices(internal_symbol, adjustment) last_date = pd.Timestamp("1970-01-01") - if "Volume_Reconstructed" in existing_df.columns: + if not full and "Volume_Reconstructed" in existing_df.columns: valid_dates = existing_df.dropna(subset=["Volume_Reconstructed"]).index if len(valid_dates) > 0: # Recompute trailing 60 days to catch late data & bridge partial failures @@ -221,21 +226,26 @@ def _check_roll_gaps(internal_symbol: str, df: pd.DataFrame) -> bool: return False -def update(symbols=None) -> None: - """Fetch + write to the store for the given internal symbols (backadj and unadj).""" +def update(symbols=None, full: bool = False) -> None: + """Fetch + write to the store for the given internal symbols (backadj and unadj). + + full=True forces a complete rebuild of the reconstructed-volume columns rather + than the trailing-60-day incremental update — use it after a reconstruction + logic change so old rows are recomputed under the new algorithm. + """ syms = symbols or [s.internal for s in all_symbols()] for sym in syms: try: # 1. Back-Adjusted out_backadj = fetch(sym, adjustment="backadj") _check_roll_gaps(sym, out_backadj) # sanity: warn if backadj looks unadjusted - + # 2. Unadjusted (Raw calendar spreads) out_unadj = fetch(sym, adjustment="unadj") - + # 3. Volume Reconstruction (Additive) - out_backadj = _reconstruct_volume(sym, out_backadj, "backadj") - out_unadj = _reconstruct_volume(sym, out_unadj, "unadj") + out_backadj = _reconstruct_volume(sym, out_backadj, "backadj", full=full) + out_unadj = _reconstruct_volume(sym, out_unadj, "unadj", full=full) store.write_prices(sym, "backadj", out_backadj, source="norgate") store.write_prices(sym, "unadj", out_unadj, source="norgate") diff --git a/src/cotdata/update.py b/src/cotdata/update.py index 4073e68..d80bece 100644 --- a/src/cotdata/update.py +++ b/src/cotdata/update.py @@ -16,6 +16,9 @@ def main() -> None: p.add_argument("--cot-tff", action="store_true", help="Update Traders in Financial Futures (TFF) COT (cross-platform).") p.add_argument("--cot-all", action="store_true", help="Update all COT pipelines (Legacy, Disagg, TFF).") p.add_argument("--symbols", nargs="+", default=None, help="Internal symbols; default = all in registry.") + p.add_argument("--full", action="store_true", + help="Full rebuild of reconstructed volume (ignore the incremental " + "60-day window). Use after a reconstruction-logic change.") args = p.parse_args() config.store_root() # fail fast if COTDATA_STORE unset @@ -27,7 +30,7 @@ def main() -> None: from .providers import norgate if args.prices: - norgate.update(symbols=args.symbols) + norgate.update(symbols=args.symbols, full=args.full) if args.metadata: norgate.update_metadata(symbols=args.symbols) diff --git a/tests/test_norgate_provider.py b/tests/test_norgate_provider.py index 6605173..eaf4c0d 100644 --- a/tests/test_norgate_provider.py +++ b/tests/test_norgate_provider.py @@ -244,3 +244,49 @@ def mock_ts_side_effect(sym, **kwargs): assert written_df.loc["2026-07-01", "Volume_Source"] == "reconstructed" assert written_df.loc["2026-07-01", "Volume_Reconstructed"] == 1000 assert written_df.loc["2026-07-01", "FirstContract"] == "ES-2026U" + + +@mock.patch("cotdata.providers.norgate.store.write_prices") +@mock.patch("cotdata.providers.norgate.store.read_prices") +@mock.patch("norgatedata.database_symbols", create=True) +@mock.patch("norgatedata.price_timeseries", create=True) +def test_full_rebuild_bypasses_incremental_window(mock_price_ts, mock_db_symbols, mock_read_prices, mock_write_prices): + """update(full=True) must recompute from epoch, ignoring the trailing-60-day + window — even when the store already carries recent Volume_Reconstructed. The + individual-contract fetch should be issued with start_date=1970-01-01.""" + mock_existing = pd.DataFrame({ + "Volume": [800], + "Volume_Reconstructed": [800], + "FirstVolume": [500], "SecondVolume": [300], + "FirstContract": ["ES-2026H"], "SecondContract": ["ES-2026M"], + "Volume_Source": ["reconstructed"], + }, index=pd.DatetimeIndex(["2026-06-01"])) + mock_read_prices.return_value = mock_existing.copy() + + mock_continuous = pd.DataFrame({ + "Open": [10, 10], "High": [10, 10], "Low": [10, 10], "Close": [10, 10], + "Volume": [800, 1000], "Open Interest": [0, 0], "Delivery Month": [0, 0], + }, index=pd.DatetimeIndex(["2026-06-01", "2026-07-01"])) + mock_indiv = pd.DataFrame({"Date": [pd.Timestamp("2026-07-01")], "Volume": [1000]}) + + def mock_ts_side_effect(sym, **kwargs): + if sym.endswith("CCB") or "-" not in sym: + return mock_continuous.copy() + if sym == "ES-2026U": + return mock_indiv.copy() + return pd.DataFrame() + + mock_price_ts.side_effect = mock_ts_side_effect + mock_db_symbols.return_value = ["ES-2026U"] + + mock_symbol = mock.Mock() + mock_symbol.internal = "ES" + with mock.patch("cotdata.providers.norgate.all_symbols", return_value=[mock_symbol]), \ + mock.patch.dict("cotdata.providers.norgate.REGISTRY", {"ES": mock.Mock(norgate="&ES")}): + norgate.update(symbols=["ES"], full=True) + + # Every individual-contract fetch (sym containing '-') must start from epoch. + indiv_starts = [c.kwargs.get("start_date") for c in mock_price_ts.call_args_list + if "-" in c.args[0]] + assert indiv_starts, "expected at least one individual-contract fetch" + assert all(s == "1970-01-01" for s in indiv_starts), indiv_starts