From b8b99ae8360bf391e0fd3adcbe4b3c79a50c0fa5 Mon Sep 17 00:00:00 2001 From: Matt Spinola Date: Tue, 14 Jul 2026 09:52:29 -0400 Subject: [PATCH 1/2] feat: add additive norgate volume reconstruction with point-in-time logic --- README.md | 10 +- scripts/reconcile_volume.py | 115 +++++++++++++++++++++++ src/cotdata/providers/norgate.py | 152 ++++++++++++++++++++++++++++++- tests/test_norgate_provider.py | 148 +++++++++++++++++++++++++++++- 4 files changed, 420 insertions(+), 5 deletions(-) create mode 100644 scripts/reconcile_volume.py diff --git a/README.md b/README.md index a1c13fa..7f0247f 100644 --- a/README.md +++ b/README.md @@ -176,9 +176,13 @@ The primary source for price history (Norgate Data). Indexed by tz-naive `Date`. | `Open` | float | Opening price. | | `High` | float | High price. | | `Low` | float | Low price. | -| `Close` | float | Exchange settlement close. | -| `Volume` | float | Trading volume. | -| `Open Interest` | float | Total open interest. | +| `Close` | float | Settlement Close price. | +| `Volume` | float | Continuous contract trading volume (front-month only). | +| `Open Interest` | float | Continuous contract open interest. | +| `Volume_Reconstructed` | float | True market volume (sum of First and Second contract). Note: systematically higher than raw `Volume`, not a drop-in replacement. | +| `Volume_Source` | string | `reconstructed` if First+Second available, `raw` fallback if not. | +| `FirstVolume` / `SecondVolume` | float | Trading volume of the specific first and second expiring contracts. | +| `FirstContract` / `SecondContract` | string | Contract names for the first and second expirations (e.g., `ES-2024H`). | | `Delivery Month` | float | Expiration month of the active contract (e.g. `202609`). Used to detect contract rolls. | ### Contract Specifications (`metadata/contract_specs.parquet`) diff --git a/scripts/reconcile_volume.py b/scripts/reconcile_volume.py new file mode 100644 index 0000000..a02cc5f --- /dev/null +++ b/scripts/reconcile_volume.py @@ -0,0 +1,115 @@ +import argparse +import pandas as pd +import numpy as np + +from cotdata.providers import norgate +from cotdata import store + +def get_roll_windows(df: pd.DataFrame, window_days: int = 5) -> pd.DataFrame: + """Find dates where the contract rolled (Delivery Month changed) and return a mask of those dates +/- window_days.""" + if "Delivery Month" not in df.columns: + return pd.Series(False, index=df.index) + + dm = df["Delivery Month"] + roll = dm.ne(dm.shift()) & dm.shift().notna() + + # Expand window + roll_dates = df.index[roll] + mask = pd.Series(False, index=df.index) + for rd in roll_dates: + start = rd - pd.Timedelta(days=window_days) + end = rd + pd.Timedelta(days=window_days) + mask.loc[start:end] = True + + return mask + +def verify_symbol(symbol: str): + print(f"\n{'='*50}\nReconciling {symbol}\n{'='*50}") + + print(f"1. Fetching & Reconstructing {symbol} via norgate.py ...") + norgate.update(symbols=[symbol]) + + print(f"2. Reading updated dataframe for {symbol} ...") + df = store.read_prices(symbol, "backadj") + + if df.empty: + print(f"Error: No data found for {symbol}.") + return + + # Check Columns + expected_cols = ["Volume", "Volume_Reconstructed", "FirstVolume", "SecondVolume", "Volume_Source", "FirstContract", "SecondContract"] + for col in expected_cols: + if col not in df.columns: + print(f"❌ Missing expected column: {col}") + return + + print("✅ All expected additive columns are present.") + + # Check Fallback + source = df["Volume_Source"].iloc[-1] + print(f"Volume Source: {source}") + + if source == "raw": + print(f"✅ Fallback applied correctly. Volume_Reconstructed == Volume for all rows: {(df['Volume_Reconstructed'] == df['Volume']).all()}") + return + + # Check bounds + print("3. Running Sanity Checks...") + + # a. First + Second = Reconstructed + valid = df.dropna(subset=["FirstVolume", "SecondVolume"]) + reconstructed_match = np.isclose( + valid["FirstVolume"] + valid["SecondVolume"], + valid["Volume_Reconstructed"] + ).all() + print(f"{'✅' if reconstructed_match else '❌'} FirstVolume + SecondVolume == Volume_Reconstructed") + + # b. Reconstructed >= Volume (raw) + # The true combined volume should be roughly >= the single front-month volume + # Some minor exceptions can happen due to Norgate raw volume inclusions, but generally true. + vol_diff = (valid["Volume_Reconstructed"] >= valid["Volume"] * 0.95).mean() + print(f"{'✅' if vol_diff > 0.99 else '❌'} Volume_Reconstructed is >= Raw Volume (in {vol_diff*100:.1f}% of days)") + + # c. No NaNs introduced into default Volume + nans_in_raw = df["Volume"].isna().sum() + print(f"{'✅' if nans_in_raw == 0 else '❌'} Default 'Volume' has 0 NaNs ({nans_in_raw} found)") + + print("\n4. Roll Window Drop-off Analysis") + # Compare average volume during roll windows vs non-roll windows + roll_mask = get_roll_windows(df, window_days=5) + + if roll_mask.sum() == 0: + print("No roll windows detected.") + return + + roll_raw = df.loc[roll_mask, "Volume"].mean() + nonroll_raw = df.loc[~roll_mask, "Volume"].mean() + + roll_rec = df.loc[roll_mask, "Volume_Reconstructed"].mean() + nonroll_rec = df.loc[~roll_mask, "Volume_Reconstructed"].mean() + + raw_drop = (nonroll_raw - roll_raw) / nonroll_raw * 100 if nonroll_raw else 0 + rec_drop = (nonroll_rec - roll_rec) / nonroll_rec * 100 if nonroll_rec else 0 + + print(f" Raw Volume:") + print(f" Non-Roll Avg: {nonroll_raw:,.0f}") + print(f" Roll Avg: {roll_raw:,.0f}") + print(f" Drop-off: {raw_drop:.1f}%") + + print(f" Reconstructed Volume:") + print(f" Non-Roll Avg: {nonroll_rec:,.0f}") + print(f" Roll Avg: {roll_rec:,.0f}") + print(f" Drop-off: {rec_drop:.1f}%") + + if rec_drop < raw_drop: + print("✅ Reconstructed volume successfully smoothed the roll drop-off!") + else: + print("❌ Reconstructed volume did not improve the roll drop-off.") + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Reconcile Volume Reconstruction") + parser.add_argument("--symbols", nargs="+", default=["ES", "BTC"], help="Internal symbols to test") + args = parser.parse_args() + + for sym in args.symbols: + verify_symbol(sym) diff --git a/src/cotdata/providers/norgate.py b/src/cotdata/providers/norgate.py index 38a4736..b2620e7 100644 --- a/src/cotdata/providers/norgate.py +++ b/src/cotdata/providers/norgate.py @@ -9,6 +9,9 @@ stitched out). A close-based stop needs the gap-free series → we fetch _CCB. """ import pandas as pd +import numpy as np +import re +from concurrent.futures import ThreadPoolExecutor, as_completed from ..registry import REGISTRY, all_symbols from .. import store @@ -25,6 +28,148 @@ "Delivery Month": "Delivery Month", # kept → exact roll detection downstream } +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: + """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. + """ + 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: + 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 + last_date = valid_dates.max() - pd.Timedelta(days=60) + + # Base Norgate symbol (e.g. "&ES" -> "ES") + base_sym = REGISTRY[internal_symbol].norgate.lstrip("&").split("_")[0] + + # 2. Find needed individual contracts + all_futures = norgatedata.database_symbols('Futures') + pattern = re.compile(rf"^{re.escape(base_sym)}-(\d{{4}})([FGHJKMNQUVXZ])$") + + needed_contracts = [] + for sym in all_futures: + m = pattern.match(sym) + if m: + year, month = int(m.group(1)), MONTH_CODES[m.group(2)] + expiry_date = pd.Timestamp(year=year, month=month, day=1) + pd.DateOffset(months=1) + if expiry_date >= last_date: + needed_contracts.append(sym) + + # 3. Fallback: if no individual contracts (e.g., crypto, ICE softs) + if not needed_contracts: + res = continuous_df.copy() + res["FirstVolume"] = np.nan + res["SecondVolume"] = np.nan + res["FirstContract"] = "" + res["SecondContract"] = "" + res["Volume_Reconstructed"] = res["Volume"] + res["Volume_Source"] = "raw" + return res + + # 4. Fetch OHLCV for needed contracts + frames = [] + with ThreadPoolExecutor(max_workers=10) as pool: + futs = { + pool.submit( + norgatedata.price_timeseries, + c, + padding_setting=norgatedata.PaddingType.NONE, + timeseriesformat="pandas-dataframe", + start_date=last_date.strftime("%Y-%m-%d") + ): c + for c in needed_contracts + } + for f in as_completed(futs): + c = futs[f] + try: + df_c = f.result() + if "Date" not in df_c.columns: + df_c = df_c.reset_index() + if not df_c.empty: + df_c["Symbol"] = c + frames.append(df_c[["Date", "Volume", "Symbol"]]) + except Exception as e: + print(f" ⚠️ Failed to fetch individual contract {c}: {e}") + + res = continuous_df.copy() + + if not frames: + res["FirstVolume"] = np.nan + res["SecondVolume"] = np.nan + res["FirstContract"] = "" + res["SecondContract"] = "" + res["Volume_Reconstructed"] = res["Volume"] + res["Volume_Source"] = "raw" + return res + + all_indiv = pd.concat(frames, ignore_index=True) + all_indiv['Date'] = pd.to_datetime(all_indiv['Date']).dt.tz_localize(None).dt.normalize() + + # 5. Point-in-Time Contract Identification + pivot = all_indiv.pivot(index="Date", columns="Symbol", values="Volume") + + def get_expiry(sym): + m = pattern.match(sym) + return pd.Timestamp(year=int(m.group(1)), month=MONTH_CODES[m.group(2)], day=1) + + sorted_cols = sorted(pivot.columns, key=get_expiry) + pivot = pivot[sorted_cols] + + vol_array = pivot.values + sort_key = np.isnan(vol_array).astype(int) + order = np.argsort(sort_key, axis=1, kind='stable') + compressed = np.take_along_axis(vol_array, order, axis=1) + names_arr = np.array(pivot.columns) + + rec_df = pd.DataFrame(index=pivot.index) + + num_cols = compressed.shape[1] + if num_cols > 0: + rec_df['FirstVolume'] = compressed[:, 0] + rec_df['FirstContract'] = np.where(np.isnan(compressed[:, 0]), '', names_arr[order[:, 0]]) + else: + rec_df['FirstVolume'] = np.nan + rec_df['FirstContract'] = '' + + if num_cols > 1: + rec_df['SecondVolume'] = compressed[:, 1] + rec_df['SecondContract'] = np.where(np.isnan(compressed[:, 1]), '', names_arr[order[:, 1]]) + else: + rec_df['SecondVolume'] = np.nan + rec_df['SecondContract'] = '' + + rec_df['Volume_Reconstructed'] = rec_df['FirstVolume'].fillna(0) + rec_df['SecondVolume'].fillna(0) + rec_df.loc[rec_df['FirstVolume'].isna() & rec_df['SecondVolume'].isna(), 'Volume_Reconstructed'] = np.nan + rec_df['Volume_Source'] = "reconstructed" + + # 6. Merge the newly reconstructed subset into the existing history + for col in ["FirstVolume", "SecondVolume", "FirstContract", "SecondContract", "Volume_Reconstructed", "Volume_Source"]: + if col not in res.columns: + if col in existing_df.columns: + res[col] = existing_df[col] + else: + res[col] = "" if col in ("FirstContract", "SecondContract", "Volume_Source") else np.nan + + res.update(rec_df) + + common_idx = res.index.intersection(rec_df.index) + for col in ["FirstContract", "SecondContract"]: + res.loc[common_idx, col] = rec_df.loc[common_idx, col] + + mask = res['Volume_Reconstructed'].isna() + res.loc[mask, 'Volume_Reconstructed'] = res.loc[mask, 'Volume'] + res.loc[mask, 'Volume_Source'] = "raw" + + return res + def fetch(internal_symbol: str, adjustment: str = "backadj", start: str = "1970-01-01") -> pd.DataFrame: """Fetch Norgate continuous bars: settlement close.""" @@ -76,10 +221,15 @@ def update(symbols=None) -> None: # 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") + + # 3. Volume Reconstruction (Additive) + out_backadj = _reconstruct_volume(sym, out_backadj, "backadj") + out_unadj = _reconstruct_volume(sym, out_unadj, "unadj") + + store.write_prices(sym, "backadj", out_backadj, source="norgate") store.write_prices(sym, "unadj", out_unadj, source="norgate") print(f"{sym:5s}: {len(out_backadj):6d} backadj, {len(out_unadj):6d} unadj -> store") diff --git a/tests/test_norgate_provider.py b/tests/test_norgate_provider.py index 5acd650..2c35ed1 100644 --- a/tests/test_norgate_provider.py +++ b/tests/test_norgate_provider.py @@ -1,4 +1,5 @@ import pandas as pd +import numpy as np from unittest import mock # Mock norgatedata so we can import and test norgate on any OS @@ -11,10 +12,14 @@ from cotdata.providers import norgate +@mock.patch("cotdata.providers.norgate.store.read_prices") +@mock.patch("norgatedata.database_symbols", create=True) @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): +def test_norgate_update_fetches_both_adjustments(mock_price_ts, mock_write_prices, mock_db_symbols, mock_read_prices): """Verify that update() fetches both the backadj and unadj series for a symbol.""" + mock_db_symbols.return_value = [] + mock_read_prices.return_value = pd.DataFrame() # Setup mock returns mock_df = pd.DataFrame({ @@ -52,3 +57,144 @@ def test_norgate_update_fetches_both_adjustments(mock_price_ts, mock_write_price # 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" + +@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_volume_reconstruction(mock_price_ts, mock_db_symbols, mock_read_prices, mock_write_prices): + """Verify that _reconstruct_volume correctly appends additive columns without modifying default Volume.""" + + # Mock continuous dataframe + mock_continuous = 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 individual contracts + mock_indiv_H = pd.DataFrame({ + "Date": [pd.Timestamp("2026-07-01")], + "Volume": [600], + "Open Interest": [3000] + }) + mock_indiv_M = pd.DataFrame({ + "Date": [pd.Timestamp("2026-07-01")], + "Volume": [400], + "Open Interest": [2000] + }) + + def mock_ts_side_effect(sym, **kwargs): + if sym.endswith("CCB") or not "-" in sym: + return mock_continuous.copy() + if sym == "ES-2026H": + return mock_indiv_H.copy() + if sym == "ES-2026M": + return mock_indiv_M.copy() + return pd.DataFrame() + + mock_price_ts.side_effect = mock_ts_side_effect + mock_db_symbols.return_value = ["ES-2026H", "ES-2026M", "ES-2025Z"] + + # Mock existing prices to trigger full backfill + mock_read_prices.return_value = pd.DataFrame() + + 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"]) + + # Verify the written dataframe has the additive columns and default Volume is untouched + write_call = mock_write_prices.call_args_list[0] + written_df = write_call[0][2] + + assert "FirstVolume" in written_df.columns + assert "SecondVolume" in written_df.columns + assert "Volume_Reconstructed" in written_df.columns + assert "Volume_Source" in written_df.columns + + # Default Volume should be UNTOUCHED (1000) + assert written_df["Volume"].iloc[0] == 1000 + + # FirstVolume (ES-2026H) + SecondVolume (ES-2026M) = 1000 + assert written_df["FirstVolume"].iloc[0] == 600 + assert written_df["SecondVolume"].iloc[0] == 400 + assert written_df["Volume_Reconstructed"].iloc[0] == 1000 + assert written_df["Volume_Source"].iloc[0] == "reconstructed" + assert written_df["FirstContract"].iloc[0] == "ES-2026H" + assert written_df["SecondContract"].iloc[0] == "ES-2026M" + + +@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_volume_reconstruction_incremental(mock_price_ts, mock_db_symbols, mock_read_prices, mock_write_prices): + """Verify that _reconstruct_volume preserves old Volume_Source during an incremental run.""" + + # Existing df has an old date with a "raw" fallback, and a slightly newer one with "reconstructed" + mock_existing = pd.DataFrame({ + "Volume": [500, 800], + "Volume_Reconstructed": [500, 800], + "FirstVolume": [np.nan, 500], + "SecondVolume": [np.nan, 300], + "FirstContract": ["", "ES-2026H"], + "SecondContract": ["", "ES-2026M"], + "Volume_Source": ["raw", "reconstructed"] + }, index=pd.DatetimeIndex(["2020-01-01", "2026-06-01"])) + + mock_read_prices.return_value = mock_existing.copy() + + # New continuous dataframe has the old dates + a new date + mock_continuous = pd.DataFrame({ + "Open": [10, 10, 10], "High": [10, 10, 10], "Low": [10, 10, 10], "Close": [10, 10, 10], + "Volume": [500, 800, 1000], "Open Interest": [0, 0, 0], "Delivery Month": [0, 0, 0] + }, index=pd.DatetimeIndex(["2020-01-01", "2026-06-01", "2026-07-01"])) + + # The new date gets fetched. The trailing 60 days from 2026-06-01 is 2026-04-02. + # We will just return some mock individual contracts. + mock_indiv_U = pd.DataFrame({ + "Date": [pd.Timestamp("2026-07-01")], + "Volume": [600], "Open Interest": [0] + }) + mock_indiv_Z = pd.DataFrame({ + "Date": [pd.Timestamp("2026-07-01")], + "Volume": [400], "Open Interest": [0] + }) + + 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_U.copy() + if sym == "ES-2026Z": + return mock_indiv_Z.copy() + return pd.DataFrame() + + mock_price_ts.side_effect = mock_ts_side_effect + mock_db_symbols.return_value = ["ES-2026U", "ES-2026Z"] + + 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"]) + + write_call = mock_write_prices.call_args_list[0] + written_df = write_call[0][2] + + # Verify the 2020-01-01 row is still "raw" + assert written_df.loc["2020-01-01", "Volume_Source"] == "raw" + assert written_df.loc["2020-01-01", "Volume_Reconstructed"] == 500 + + # Verify the 2026-06-01 row is still "reconstructed" + assert written_df.loc["2026-06-01", "Volume_Source"] == "reconstructed" + + # Verify the newly fetched 2026-07-01 row is computed correctly + 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" From 4fc8d0e4376b8c68b168624bbbb7722485e0278f Mon Sep 17 00:00:00 2001 From: Matt Spinola Date: Tue, 14 Jul 2026 11:45:19 -0400 Subject: [PATCH 2/2] feat: promote reconstructed volume via schema v2 (intent view + accessors) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deliberate, versioned promotion of the additive reconstruction columns to a usable default — kept separate from the additive commit so the default-changing step is visible, not silent. - get_prices(volume="front"|"reconstructed"): "front" (default) is byte-identical to the pre-v2 API; "reconstructed" serves reconstructed-with-per-row-raw-fallback in Volume plus a Volume_Source audit column. The fallback policy lives once, in the data layer, instead of being re-derived at every consumer call site. - schema_version()/require_schema(): expose the manifest schema so consumers key cache invalidation on a real token instead of a column-presence heuristic. - SCHEMA_VERSION 1 -> 2. Store is not actually v2 until a full producer pass re-writes it; see docs/plan_promote_reconstructed_volume.md for rollout order. - README schema note; tests for the view, fallback, and schema guard (31 passed). Co-Authored-By: Claude Opus 4.8 --- README.md | 4 + docs/plan_promote_reconstructed_volume.md | 190 ++++++++++++++++++++++ src/cotdata/__init__.py | 5 +- src/cotdata/config.py | 6 +- src/cotdata/prices.py | 40 ++++- src/cotdata/store.py | 21 +++ tests/test_store.py | 70 ++++++++ 7 files changed, 330 insertions(+), 6 deletions(-) create mode 100644 docs/plan_promote_reconstructed_volume.md diff --git a/README.md b/README.md index 7f0247f..9638f39 100644 --- a/README.md +++ b/README.md @@ -170,6 +170,10 @@ The canonical store uses standard Parquet files. When loaded into a pandas DataF ### Price Data (`prices/{symbol}_{adjustment}.parquet`) 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. +**Reading reconstructed volume:** the reconstruction columns below are internal storage. Consumers should not read `Volume_Reconstructed` directly — call `get_prices(symbol, volume="reconstructed")` and the `Volume` column is served as reconstructed-with-per-row-raw-fallback, plus a `Volume_Source` column for audit. The default `volume="front"` returns the front-month series unchanged (byte-identical to the pre-v2 API). See `docs/plan_promote_reconstructed_volume.md`. + +**Schema versioning:** `schema_version` in `manifest.json` records the on-disk data version (v2 = reconstructed volume promoted). Consumers key cache invalidation on `cotdata.schema_version()` and can guard with `cotdata.require_schema(min_version)`. + | Column | Type | Description | |--------|------|-------------| | `Date` | DatetimeIndex | Trading day (tz-naive, normalized to midnight). | diff --git a/docs/plan_promote_reconstructed_volume.md b/docs/plan_promote_reconstructed_volume.md new file mode 100644 index 0000000..74c69ea --- /dev/null +++ b/docs/plan_promote_reconstructed_volume.md @@ -0,0 +1,190 @@ +# Plan — Promote reconstructed volume to default (versioned, deliberate) + +## Why this is its own step + +The additive change (`c8dfa02`) added `Volume_Reconstructed`, `Volume_Source`, +`First/SecondVolume`, `First/SecondContract` to the store **without touching the +canonical `Volume`/`Open Interest` columns**. That step is intentionally inert +downstream: the new columns are dropped before any consumer sees them. + +There is exactly **one** live gate: `cotdata/prices.py` → `_COLS` whitelist, +returns `df[keep]`. Every live price/volume read in both consumers goes through +`cotdata.get_prices`. (`cot-analyzer/src/core/norgate_reader.py` has its own +`_PIPELINE_COLS` whitelist and a duplicate `roll_dates`, but it has **zero live +callers** — it is dead code and should be deleted, not taught the new columns.) + +"Promote to default" is the opposite kind of move: it changes the number that +existing consumers already read. Folding it into the additive commit would be a +**silent semantic change** — the failure mode we called out in the V1 review. +The CotIndexer cache-staleness guards key on **column presence, not value**, so +swapping what `Volume` means under the same name would not invalidate a single +cache. Stale volume-derived features (Build Ratio `|ΔOI|/volume`, Speculation +Ratio `volume/OI`) would persist with no signal that anything moved. + +So: promotion is a separate, versioned release with an explicit cache bust and a +downstream note. This document is the runbook for that release. + +## Current state (facts to build on) + +- `config.SCHEMA_VERSION = 1`. `_touch_manifest` stamps it into `manifest.json` + on every write. **No consumer branches on it** — `load_manifest()` exposes it + but nothing validates it. A version bump is inert until something reads it. +- Reconstructed columns exist in both `*_backadj.parquet` and `*_unadj.parquet` + for every symbol that has individual-contract history; `Volume_Source` marks + `reconstructed` vs `raw` fallback per row. +- **Actual volume surface is thin** (audited across both consumers): + - cot-analyzer: `CotIndexer.py:405` reads `Volume` only as an OI-missing + fallback (5% of volume proxy flow + a flow cap). No Build/Speculation Ratio + is actually computed. Fetches via `cotdata.get_prices`. + - pardo: `Volume` is in `xgboost_grinder` `exclude_cols` (deliberately **not** + an ML feature); used by the zipline adapter (needs a column named `volume`, + indifferent to raw vs reconstructed) and a news-outlier diagnostic. PIT + feature builders don't touch volume. Fetches via `cotdata.get_prices`. +- **Neither consumer needs the reconstructed number physically in the `Volume` + column** → open question #1 is resolved: **1B** (keep raw, expose an intent + view). See "Consumer updates" below. + +## Goal + +Make reconstructed volume the value that volume-derived features consume, as an +explicit `schema_version 1 → 2` release: caches rebuild, the change is visible in +the manifest and a changelog note, and rollback is a one-line revert. + +## Guardrails + +1. **Never overwrite the meaning of `Volume` in place.** Keep raw front-month + `Volume` as-is (audit trail, sizing, `Volume_Source='raw'` reconciliation). + Promotion = flip *which column the pipeline reads*, not mutate `Volume`. +2. **The version bump must be load-bearing** before it ships — a consumer has to + read `schema_version` and change behavior, or bumping it is theater. +3. **Cache key must include `schema_version`** so this promotion and every future + one auto-invalidate instead of relying on humans to remember. +4. One PR per repo, cotdata first (producer/API), cot-analyzer second (consumer), + never interleaved. + +## Step 1 — Promotion surface — RESOLVED: 1B (intent view) + +Audited both consumers (see "Current state"): **neither requires the reconstructed +value physically in the `Volume` column**, so 1A is not forced. + +- **1A — swap the `Volume` column's contents** to reconstructed. Maximum + blast radius, hardest to audit (raw number gone from the default view). Rejected. +- **1B — keep both columns; cotdata owns an intent view.** cotdata exposes + `Volume_Reconstructed` + `Volume_Source`, plus a `get_prices(..., volume=...)` + parameter that populates the returned `Volume` column with the chosen semantics + (`"front"` = raw, default; `"reconstructed"` = reconstructed-with-per-row-raw- + fallback). Consumers express *intent*; the fallback logic lives once, in the + data layer, not re-derived at every call site. Raw is never mutated → reversible. + **Chosen.** + +### Data-layer ownership (do these in cotdata, not downstream) + +The staged rollout surfaced logic that was about to be pushed into consumers: + +1. **Volume source-selection is a cotdata concern.** Without the `volume=` view, + every consumer would re-implement `row.get('Volume_Reconstructed', Volume)` + fallback (CotIndexer:405, zipline adapter, news diagnostic, …). Own it once. +2. **Schema/version is a cotdata contract.** Add `cotdata.schema_version()` and + `require_schema(min_version)` so consumers key caches on a real token instead + of cot-analyzer's column-presence heuristic. (Optionally a per-symbol + `data_version` from the manifest's `last_date`/`n_rows`/`updated_at`.) +3. **Delete the dead `norgate_reader.py`** in cot-analyzer (zero live callers) — + a drifted second copy of read/normalize/whitelist/`roll_dates`. Remove it so + there is a single schema-aware reader, rather than teaching it the new columns. + +## Step 2 — Version bump + make it load-bearing (cotdata PR) + +1. `config.SCHEMA_VERSION = 2`. +2. Add a short migration note constant/table describing what v2 means: + "prices carry reconstructed volume; consumers may read `Volume_Reconstructed` + / `Volume_Source`." +3. **Backfill correctly.** `_touch_manifest` writes the *current* `SCHEMA_VERSION` + on any write, so a partial producer run would stamp `2` onto a store whose + other parquet files are still v1-shaped. Either (a) run a full producer pass + so every prices entry is rewritten under v2 before the bump lands, or (b) move + the version stamp so per-entry schema is tracked, not just a global int. + Prefer (a) for this release — it's one `norgate.update()` over all symbols. +4. Expose the columns + intent view through the API (`prices.py`): add + `Volume_Reconstructed` and `Volume_Source` to `_COLS`; add the + `volume="front"|"reconstructed"` param (default `"front"` = today's behavior, + so the API addition is itself non-breaking). Add `schema_version()` / + `require_schema(min_version=2)` helpers (export from `cotdata.__init__`). +5. Tests: `get_prices` returns the new columns; `volume="reconstructed"` puts the + reconstructed value in `Volume` and preserves `Volume_Source`; default still + returns raw front-month; a v1 manifest raises/warns via `require_schema`; + reconciliation (`scripts/reconcile_volume.py`) still green. + +## Step 3 — Consumer updates + +### cot-analyzer PR +The two cache layers (startup `try_load_from_cache`, freshness at +`CotIndexer.py:606`) key on column presence, so they will NOT self-invalidate on +a value change. Wire the version in explicitly: + +1. Switch the flow engine to reconstructed volume: `CotIndexer.py:374` fetch + becomes `cotdata.get_prices(symbol, adjustment='backadj', volume='reconstructed', start=...)`. + The OI-fallback read at `:405` then transparently uses reconstructed volume. +2. Read `cotdata.schema_version()` (or `require_schema`) at cache-build time and + **fold it into the cache key / cache-file name** — forces a rebuild on this + promotion and every future one, automatically. +3. One-time: bust on-disk cache artifacts built under v1 (delete/rename cache dir + or a `--rebuild` flag) so the first post-deploy run recomputes from + reconstructed volume. +4. **Delete `src/core/norgate_reader.py`** (dead — zero live callers). Confirm no + imports break. +5. Regression check: recompute affected outputs before/after on a couple of + symbols; deltas should localize to roll windows / OI-missing rows, not + everywhere — a diff everywhere means something else moved. + +### pardo PR — capability-only, zero behaviour change +pardo is **indifferent** to the promotion: its read shapes (OHLC + front volume) +are unchanged by the additive+opt-in v2, so it must not be forced onto +reconstructed volume and must not be gated. + +1. **No ML change** — `Volume` is already in `xgboost_grinder` `exclude_cols`; + PIT feature builders don't use volume. (Extended `exclude_cols` with the + reconstruction column names as a defensive leak-guard.) +2. **Do NOT** hard-`require_schema(2)` at the data-loader — pardo reads unchanged + shapes; a gate would block the honest-null pipeline against a change it doesn't + consume. +3. **Do NOT** silently switch the zipline backtest or `news_failure_trigger` + diagnostic to reconstructed volume — zipline's volume-based slippage/commission + models would change results, and the news outlier would flag different days. + Both stay on front volume. +4. Capability only: `fetch_daily_ohlc(..., volume=...)` now passes through to + `get_prices`, default `'front'`. pardo *can* opt a specific call into + reconstructed volume later, deliberately, without a silent global change. + +## Step 4 — Note it to downstream + +- cotdata `README.md` schema table already documents the columns; add a + **CHANGELOG / schema-version note**: "v2 — reconstructed volume promoted; + consumers should read `Volume_Reconstructed` and honor `Volume_Source`." +- One message to downstream owners: what changed, which features rebuild, how to + force a rebuild (`schema_version` now in the cache key), and the rollback. +- Update `docs/keenan_coverage_map.md` / any analysis notes that assumed raw + front-month volume in Build Ratio, since that input has changed. + +## Step 5 — Rollout, verify, rollback + +- **Order:** cotdata PR (Step 2) → full producer pass → cot-analyzer PR (Step 3) + → pardo PR. Never ship a consumer flip before the store is uniformly v2. +- **Verify:** manifest shows `schema_version: 2` for all prices entries; + `get_prices` returns the columns; indexer rebuilds on first run; feature deltas + localized to roll windows; `Volume_Source` distribution sane (mostly + `reconstructed`, `raw` only for crypto/ICE-soft symbols with no individual + contracts). +- **Rollback:** revert the consumer flip (features read `Volume` again) — raw + column was never mutated, so this is clean. `SCHEMA_VERSION` can stay at 2 + (columns are additive); only the *consumption* is reverted. + +## Open questions + +1. ~~Does any consumer need reconstructed volume *in* the `Volume` column?~~ + **Resolved: no.** Both audited; 1B (intent view) chosen. +2. Per-entry schema tracking vs global int in the manifest — worth doing now + (Step 2.3b) or defer? Global int + full-pass is fine for this release. +3. Should `unadj` volume be promoted too, or only `backadj`? Reconstruction is + adjustment-independent (volume isn't adjusted), so both carry the columns; the + live consumers all read `adjustment='backadj'`, so `backadj` is the one that + matters — `unadj` can stay reconstructed-in-store but no consumer flips to it. diff --git a/src/cotdata/__init__.py b/src/cotdata/__init__.py index df14aa6..53e5bf9 100644 --- a/src/cotdata/__init__.py +++ b/src/cotdata/__init__.py @@ -2,10 +2,11 @@ from .prices import get_prices, roll_dates from .cot import get_cot from .registry import symbol, all_symbols, REGISTRY, Symbol -from .store import load_manifest +from .store import load_manifest, schema_version, require_schema __version__ = "0.1.0" __all__ = [ "get_prices", "roll_dates", "get_cot", - "symbol", "all_symbols", "REGISTRY", "Symbol", "load_manifest", + "symbol", "all_symbols", "REGISTRY", "Symbol", + "load_manifest", "schema_version", "require_schema", ] diff --git a/src/cotdata/config.py b/src/cotdata/config.py index a84412b..f301fda 100644 --- a/src/cotdata/config.py +++ b/src/cotdata/config.py @@ -2,7 +2,11 @@ import os from pathlib import Path -SCHEMA_VERSION = 1 +# v2 — reconstructed volume promoted: prices carry Volume_Reconstructed / +# Volume_Source, and get_prices(volume="reconstructed") serves them. The store +# does not actually carry v2 shape until a full producer pass re-writes it; see +# docs/plan_promote_reconstructed_volume.md for the rollout order. +SCHEMA_VERSION = 2 def store_root() -> Path: diff --git a/src/cotdata/prices.py b/src/cotdata/prices.py index 4c938d6..8bf9140 100644 --- a/src/cotdata/prices.py +++ b/src/cotdata/prices.py @@ -10,14 +10,30 @@ def get_prices(symbol: str, adjustment: str = "backadj", - start: Optional[str] = None) -> pd.DataFrame: + start: Optional[str] = None, + volume: str = "front") -> pd.DataFrame: """Daily bars for `symbol`. adjustment: 'backadj' (signals + stops — settlement close, gap-free rolls, shape-preserving) or 'unadj' (absolute price / point-value sizing). + + volume: which series the `Volume` column carries — + 'front' : continuous front-month volume as Norgate reports it + (default — output is byte-identical to the pre-v2 API). + 'reconstructed' : true market volume (first + second expiring contract, + with per-row fall-back to front-month where individual + contracts are unavailable). The intent view: the fall-back + policy lives here, in the data layer, so consumers ask for + what they want instead of re-deriving it. Adds a + 'Volume_Source' column ('reconstructed' / 'raw') for audit. + Returns Open/High/Low/Close/Volume/Open Interest indexed by tz-naive Date - (plus 'Delivery Month' if the producer carried it), or empty if absent. + (plus 'Delivery Month' if the producer carried it, plus 'Volume_Source' when + volume='reconstructed'), or empty if absent. """ + if volume not in ("front", "reconstructed"): + raise ValueError(f"volume must be 'front' or 'reconstructed', got {volume!r}") + df = store.read_prices(symbol, adjustment) if df.empty: return df @@ -29,7 +45,25 @@ def get_prices(symbol: str, adjustment: str = "backadj", for c in _COLS: if c not in df.columns: df[c] = float("nan") - keep = _COLS + (["Delivery Month"] if "Delivery Month" in df.columns else []) + + keep = list(_COLS) + if volume == "reconstructed": + # Swap in reconstructed volume with a per-row raw fall-back. The producer + # already writes Volume_Reconstructed==Volume on 'raw' rows, so reading it + # is fall-back-safe where the column exists; guard for older/partial data + # (pre-reconstruction parquet, or a stray NaN) by filling from front-month. + if "Volume_Reconstructed" in df.columns: + df["Volume"] = df["Volume_Reconstructed"].where( + df["Volume_Reconstructed"].notna(), df["Volume"]) + df["Volume_Source"] = ( + df["Volume_Source"] if "Volume_Source" in df.columns else "reconstructed") + else: + # Store predates reconstruction → everything is front-month. + df["Volume_Source"] = "raw" + keep = keep + ["Volume_Source"] + + if "Delivery Month" in df.columns: + keep = keep + ["Delivery Month"] return df[keep].copy() diff --git a/src/cotdata/store.py b/src/cotdata/store.py index ccbc73b..ff3fd4a 100644 --- a/src/cotdata/store.py +++ b/src/cotdata/store.py @@ -89,6 +89,27 @@ def load_manifest() -> dict: return {"schema_version": config.SCHEMA_VERSION, "metadata": {}, "prices": {}, "cot_legacy": {}, "cot_disagg": {}, "cot_tff": {}} +def schema_version() -> int: + """Schema version recorded in the *store's* manifest — the version of the data + on disk, which is NOT the same as config.SCHEMA_VERSION (the library's target) + until a producer pass has re-written the store. Consumers key cache + invalidation on this so a schema bump forces a rebuild.""" + return int(load_manifest().get("schema_version", 0)) + + +def require_schema(min_version: int) -> None: + """Fail fast if the store predates a schema the caller depends on. Lets a + consumer refuse to run against a stale store rather than silently read the + old shape.""" + v = schema_version() + if v < min_version: + raise RuntimeError( + f"cotdata store schema_version={v} < required {min_version}. " + f"Re-run the producer (e.g. norgate.update) to migrate the store — " + f"see docs/plan_promote_reconstructed_volume.md." + ) + + def _touch_manifest(kind: str, name: str, df: pd.DataFrame, source: str) -> None: m = load_manifest() last = None diff --git a/tests/test_store.py b/tests/test_store.py index c8f8e52..b9c42b3 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -46,3 +46,73 @@ def test_roll_dates_from_delivery_month(store_env): def test_missing_symbol_returns_empty(store_env): from cotdata import get_prices assert get_prices("ZZ", "backadj").empty + + +def _sample_reconstructed(): + """Sample carrying the v2 reconstruction columns, with one 'raw' fallback row + (row 2) where no individual contracts were available.""" + df = _sample() + df["Volume_Reconstructed"] = [15, 14, 10, 18, 20] # row idx 2 == front-month (raw) + df["Volume_Source"] = ["reconstructed", "reconstructed", "raw", + "reconstructed", "reconstructed"] + return df + + +def test_default_volume_view_is_byte_identical(store_env): + """volume='front' (default) must not change the pre-v2 output shape.""" + from cotdata import store, get_prices + store.write_prices("ES", "backadj", _sample_reconstructed(), source="test") + + df = get_prices("ES", "backadj") # default volume='front' + assert list(df.columns) == ["Open", "High", "Low", "Close", "Volume", + "Open Interest", "Delivery Month"] + assert "Volume_Source" not in df.columns + assert "Volume_Reconstructed" not in df.columns + assert df["Volume"].tolist() == [10] * 5 # untouched front-month + + +def test_reconstructed_volume_view(store_env): + """volume='reconstructed' swaps Volume in, keeps per-row raw fallback, and + surfaces Volume_Source for audit.""" + from cotdata import store, get_prices + store.write_prices("ES", "backadj", _sample_reconstructed(), source="test") + + df = get_prices("ES", "backadj", volume="reconstructed") + assert "Volume_Source" in df.columns + assert "Volume_Reconstructed" not in df.columns # internal — not leaked + # reconstructed values flow into Volume; the 'raw' row keeps front-month (10) + assert df["Volume"].tolist() == [15, 14, 10, 18, 20] + assert df["Volume_Source"].tolist() == ["reconstructed", "reconstructed", + "raw", "reconstructed", "reconstructed"] + + +def test_reconstructed_view_falls_back_on_pre_v2_store(store_env): + """A store written before reconstruction existed → reconstructed view returns + front-month volume labelled 'raw', never NaN.""" + from cotdata import store, get_prices + store.write_prices("ES", "backadj", _sample(), source="test") # no recon cols + + df = get_prices("ES", "backadj", volume="reconstructed") + assert df["Volume"].tolist() == [10] * 5 + assert (df["Volume_Source"] == "raw").all() + + +def test_invalid_volume_arg_raises(store_env): + from cotdata import store, get_prices + store.write_prices("ES", "backadj", _sample(), source="test") + with pytest.raises(ValueError): + get_prices("ES", "backadj", volume="bogus") + + +def test_schema_version_and_require_schema(store_env): + from cotdata import store, schema_version, require_schema + import cotdata.config as cfg + + # Empty store → no manifest yet → load_manifest defaults to config.SCHEMA_VERSION + assert schema_version() == cfg.SCHEMA_VERSION + + store.write_prices("ES", "backadj", _sample(), source="test") + assert schema_version() == cfg.SCHEMA_VERSION # stamped by the write + require_schema(cfg.SCHEMA_VERSION) # satisfied → no raise + with pytest.raises(RuntimeError): + require_schema(cfg.SCHEMA_VERSION + 1) # store too old