feat: add --full flag to force a complete reconstructed-volume rebuild

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")

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)

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