diff --git a/.github/workflows/python-test.yml b/.github/workflows/python-test.yml new file mode 100644 index 0000000..608ec55 --- /dev/null +++ b/.github/workflows/python-test.yml @@ -0,0 +1,32 @@ +name: Python package + +on: + push: + branches: [ "main" ] + pull_request: + branches: [ "main" ] + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"] + + steps: + - uses: actions/checkout@v4 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install dependencies + run: | + python -m pip install --upgrade pip + python -m pip install -e ".[dev]" + - name: Run tests + run: | + # Create a dummy store for the test environment to prevent COTDATA_STORE missing errors + mkdir -p /tmp/cotdata_store + export COTDATA_STORE=/tmp/cotdata_store + pytest tests/ diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..2e8e106 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,56 @@ +# Contributing to cotdata + +Thank you for your interest in contributing to `cotdata`! This project provides the canonical data layer for quantitative analysis by fetching CFTC COT and Norgate price data into a shared, offline, cross-platform parquet store. + +## How to Contribute + +1. **Reporting Bugs**: Open an issue and provide as much detail as possible, including your OS (Windows vs Mac is critical here due to Norgate Data), Python version, and clear steps to reproduce the issue. +2. **Suggesting Enhancements**: Feel free to open an issue or start a Discussion on GitHub. If you're adding support for a new data provider, please propose the architecture first. +3. **Submitting Pull Requests**: + - Fork the repository and create your feature branch (`git checkout -b feature/amazing-feature`). + - Write clear, documented code. + - Run the local test suite (see below) before submitting your PR. + - Open the PR against the `main` branch. + +## Development Setup + +This project uses `uv` (or standard `pip` environments) for dependency management. + +1. **Clone the repository:** + ```bash + git clone https://github.com/your-username/cotdata.git + cd cotdata + ``` + +2. **Set up a virtual environment and install dev dependencies:** + ```bash + uv venv + uv pip install -e ".[dev]" + ``` + *Note: If you are on Windows and intend to work on the Norgate integration, you should install the Norgate extras: `uv pip install -e ".[dev,norgate]"`.* + +3. **Set the temporary datastore:** + For local development, you should point the store to a local temporary folder to avoid overwriting your live data: + ```bash + export COTDATA_STORE=/tmp/cotdata_test_store # Mac/Linux + # OR + $env:COTDATA_STORE = "C:\temp\cotdata_test_store" # Windows + ``` + +## Running Tests + +We use `pytest` for all unit and integration tests. + +```bash +pytest tests/ +``` + +### Important Note on Norgate Testing +The Norgate integration (`src/cotdata/providers/norgate.py` and `tests/test_adjustment.py`) **only runs on Windows** because it requires the proprietary Norgate Data Updater application to be installed, running, and authenticated locally. +- If you submit a PR on a Mac or Linux machine, only the generic CFTC parsing tests will run locally. +- The CI pipeline will attempt to run cross-platform tests where possible, but cannot test live Norgate API calls. Please be extremely careful when modifying Norgate adapter logic. + +## Code Style + +- We prefer standard PEP8 formatting. +- Try to keep the file sizes of the generated parquet data as small as possible. The `cotdata` pipeline aggressively drops irrelevant columns (like CFTC concentration ratios) to ensure the data lake remains highly performant for downstream pandas/XGBoost models. If you are adding a new data feature, only retain the columns necessary for quantitative analysis. diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..81f989b --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Matt S. + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/OPEN_SOURCE_TODO.md b/OPEN_SOURCE_TODO.md new file mode 100644 index 0000000..b47b5a2 --- /dev/null +++ b/OPEN_SOURCE_TODO.md @@ -0,0 +1,21 @@ +# Remaining Open Source Tasks + +This checklist tracks the remaining tasks to get `cotdata` fully ready for public release and PyPI publication. + +## Code & Repository Updates +- [ ] **Update `pyproject.toml` Metadata**: Add the required fields to publish to PyPI. + - `authors` list (e.g., `[{name = "Matt S.", email = "..."}]`) + - `urls` block (e.g., `Homepage`, `Repository`, `Issues`) + - `classifiers` (e.g., `License :: OSI Approved :: MIT License`, `Programming Language :: Python :: 3`) +- [ ] **Update `README.md` Installation Instructions**: Add a generic `pip install cotdata` and PyPI link, moving the local `uv` workspace instructions to an "Advanced/Development" section. +- [ ] **Add Badges to `README.md`**: Add markdown shields at the top of the README for CI Status, PyPI Version, and MIT License. +- [ ] **Create Issue Templates**: Add `.github/ISSUE_TEMPLATE/bug_report.md` and `feature_request.md` (make sure they ask the user to specify their OS since Norgate requires Windows). +- [ ] **Create Pull Request Template**: Add `.github/PULL_REQUEST_TEMPLATE.md` to ensure PRs reference related issues and confirm tests have passed. + +## GitHub Settings Configuration +*(To be done in the repository's Settings tab on github.com)* +- [ ] **Branch Protection Rules**: Protect `main`. Require PRs before merging and require the "test" GitHub Action status check to pass. +- [ ] **Security & Analysis**: Enable Dependabot alerts and security updates for dependencies. +- [ ] **Enable Discussions**: Turn on GitHub Discussions in settings to field general usage/data questions away from the Issue tracker. +- [ ] **Repository Details**: Add a project description, website link, and tags/topics (e.g., `algotrading`, `cot-data`, `cftc`, `norgatedata`) to the repository's about section. +- [ ] **Create First Release**: Tag `v0.1.0` and use GitHub Releases to generate the release notes once ready to publish. diff --git a/README.md b/README.md index 21cbe24..13448d7 100644 --- a/README.md +++ b/README.md @@ -8,10 +8,10 @@ Canonical data layer for the COT / futures-strategy stack. It exists so that qua │ writes │ ▼ ▼ ┌───────────────────────────────────────┐ - │ CANONICAL STORE ($COTDATA_STORE) │ - │ prices/*.parquet cot/*.parquet │ ← synced (rsync / Dropbox / S3) - │ manifest.json (the contract) │ - └───────────────────────────────────────┘ + │ CANONICAL STORE ($COTDATA_STORE) │ + │ prices/*.parquet cot_legacy/*.parquet │ ← synced (rsync / Dropbox / S3) + │ cot_disagg/*.parquet manifest.json │ + └────────────────────────────────────────────────┘ ▲ ▲ reads (offline, cross-platform) ┌──────────────┴───┐ ┌────┴──────────────┐ │ COT analyzer │ │ analysis toolset │ both: import cotdata @@ -52,7 +52,7 @@ Swapping a vendor is a producer-only change. - `prices/{symbol}_{adjustment}.parquet` — Open/High/Low/Close/Volume/Open Interest, tz-naive `Date` index. `adjustment` ∈ {`backadj`, `unadj`}. Close = exchange settlement. -- `cot/{code}.parquet` — weekly CFTC positioning. +- `cot_legacy/{code}.parquet` — weekly CFTC Legacy positioning. - `manifest.json` — per-table `last_date`, `n_rows`, `source`, `updated_at`, `schema_version`. ## Consumer @@ -61,7 +61,8 @@ Swapping a vendor is a producer-only change. import cotdata df = cotdata.get_prices("ES", adjustment="backadj") # USE THIS FOR SIGNALS + STOPS sz = cotdata.get_prices("ES", adjustment="unadj") # USE FOR POSITION SIZING / POINT VALUE -cot = cotdata.get_cot("ES") +cot_legacy = cotdata.get_cot("ES", report="legacy") # USE FOR COMM/NON-COMM +cot_disagg = cotdata.get_cot("ES", report="disagg") # USE FOR TRADER COUNTS (MM/SD/OR) ``` Set `COTDATA_STORE` to the synced store directory. @@ -72,7 +73,8 @@ Set `COTDATA_STORE` to the synced store directory. ``` COTDATA_STORE=/store cotdata-update --prices --symbols ES NQ # Norgate (Windows) -COTDATA_STORE=/store cotdata-update --cot # CFTC (cross-platform) +COTDATA_STORE=/store cotdata-update --cot-legacy # CFTC Legacy (cross-platform) +COTDATA_STORE=/store cotdata-update --cot-disagg # CFTC Disaggregated (cross-platform) ``` Schedule nightly (prices, after the Norgate Data Updater) and weekly (COT Friday releases). @@ -137,8 +139,8 @@ The primary source for price history (Norgate Data). Indexed by tz-naive `Date`. | `Open Interest` | float | Total open interest. | | `Delivery Month` | float | Expiration month of the active contract (e.g. `202609`). Used to detect contract rolls. | -### COT Data (`cot/{code}.parquet`) -The primary source for positioning data (CFTC Legacy Futures Report). Indexed by tz-naive `Report_Date_as_MM_DD_YYYY`. +### COT Legacy Data (`cot_legacy/{code}.parquet`) +The primary source for Legacy positioning data (CFTC Legacy Futures Report). Indexed by tz-naive `Report_Date_as_MM_DD_YYYY`. > [!NOTE] > **Legacy Reports**: The Legacy reports are broken down by exchange. These reports have a futures only report and a combined futures and options report. Legacy reports break down the reportable open interest positions into two classifications: non-commercial and commercial traders. The `cotdata` pipeline strictly downloads the **Futures-only** reports (located at `https://www.cftc.gov/files/dea/history/dea_fut_xls_{YEAR}.zip`). @@ -158,3 +160,9 @@ The primary source for positioning data (CFTC Legacy Futures Report). Indexed by | `NonComm_Positions_Short_All` | float | Non-Commercial (Large Speculator) Short positions. | | `NonRept_Positions_Long_All` | float | Non-Reportable (Small Speculator) Long positions. | | `NonRept_Positions_Short_All` | float | Non-Reportable (Small Speculator) Short positions. | + +### COT Disaggregated Data (`cot_disagg/{code}.parquet`) +The primary source for entity-specific positioning and trader counts (CFTC Disaggregated Futures-Only Report, dating back to 2006). Indexed by tz-naive `Report_Date_as_MM_DD_YYYY`. + +> [!NOTE] +> **Lossless Image**: Unlike the Legacy schema which filters down to 10 specific columns, the Disaggregated parquets are a **lossless image** of the source CFTC `txt` files. They contain all granular entity groups (Money Manager, Swap Dealer, Producer/Merchant, Other Reportable) and their respective `Traders_*` counts (e.g., `Traders_Tot_All`, `Traders_M_Money_Long_All`). This is the required store for computing Position Size and Clustering metrics. diff --git a/docs/spec_disaggregated_ingest.md b/docs/spec_disaggregated_ingest.md new file mode 100644 index 0000000..ce2ca2b --- /dev/null +++ b/docs/spec_disaggregated_ingest.md @@ -0,0 +1,28 @@ +# Spec: Disaggregated Futures-Only COT Ingest + +This document dictates the canonical data layer extraction for the Disaggregated Futures-Only CFTC report. + +## 1. Why Disaggregated? +The existing Legacy report (`cot/`) provides Commercial and NonCommercial positional breakdowns. However, it does not include trader counts, nor does it split out Swap Dealers (SD) or Money Managers (MM). +To compute Keenan's Positioning Trio (Concentration, Clustering, and Position Size), we must ingest the Disaggregated report, which includes `Traders_*` columns natively. + +## 2. Ingest Architecture +This pipeline operates strictly independently from the Legacy pipeline. +* **Producer Module**: `src/cotdata/providers/cftc_disagg.py` +* **Source Files**: `fut_disagg_txt_{year}.zip` (TXT/CSV is significantly faster to parse and natively supported by `pandas.read_csv` without the heavy `xlrd` dependency required by Legacy `.xls`). +* **First Available Year**: 2006 +* **Fidelity**: Unlike Legacy, the Disaggregated pipeline is **near-lossless**. It preserves every `Traders_*` column directly from the zip without sub-setting, only normalizing `Report_Date_as_MM_DD_YYYY` to datetimes and padding `CFTC_Contract_Market_Code` to 6-digits. +* **Store Namespace**: To prevent collision, it writes to a new `$COTDATA_STORE/cot_disagg/` subdirectory and is tracked under a `"cot_disagg"` key in `manifest.json`. + +## 3. Public API +The consumer API remains `cotdata.get_cot(name)`, but now accepts a `report` argument: +```python +df_legacy = cotdata.get_cot("ES", report="legacy") # existing behavior +df_disagg = cotdata.get_cot("ES", report="disagg") # new lossless disaggregated DataFrame +``` +Predecessor contract stitching (`hist_codes`) applies identically to both reports. + +## 4. Reconciliation +Because Disaggregated is parallel to Legacy, we validate it by checking that the summed entities match the Legacy groups: +* `Money_Manager` + `Other_Reportable` ≈ `NonCommercial` (Large Speculators) +* `Producer_Merchant` + `Swap_Dealer` ≈ `Commercial` (Hedgers) diff --git a/src/cotdata/config.py b/src/cotdata/config.py index b1cbbb5..956624c 100644 --- a/src/cotdata/config.py +++ b/src/cotdata/config.py @@ -19,8 +19,12 @@ def prices_dir() -> Path: return store_root() / "prices" -def cot_dir() -> Path: - return store_root() / "cot" +def cot_legacy_dir() -> Path: + return store_root() / "cot_legacy" + + +def cot_disagg_dir() -> Path: + return store_root() / "cot_disagg" def manifest_path() -> Path: diff --git a/src/cotdata/cot.py b/src/cotdata/cot.py index 0e0e06d..b155fe0 100644 --- a/src/cotdata/cot.py +++ b/src/cotdata/cot.py @@ -12,7 +12,7 @@ _CODE_COL = "CFTC_Contract_Market_Code" # Legacy schema contract-code column -def get_cot(name: str) -> pd.DataFrame: +def get_cot(name: str, report: str = "legacy") -> pd.DataFrame: """Weekly COT for an internal symbol OR a raw CFTC code. Empty if absent. If the resolved symbol declares hist_codes (predecessor exchange listings of @@ -24,14 +24,17 @@ def get_cot(name: str) -> pd.DataFrame: sym = REGISTRY.get(name) if sym is None: # allow lookup by primary CFTC code, not just symbol sym = next((s for s in REGISTRY.values() if s.cftc_code == name), None) + + read_fn = store.read_cot_disagg if report == "disagg" else store.read_cot_legacy + if sym is None or not sym.cftc_code: - return store.read_cot(name) - primary = store.read_cot(sym.cftc_code) + return read_fn(name) + primary = read_fn(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 = store.read_cot(hc) + h = read_fn(hc) if h.empty: continue h = h.copy() diff --git a/src/cotdata/providers/cftc.py b/src/cotdata/providers/cftc.py index bb47d78..a22fb65 100644 --- a/src/cotdata/providers/cftc.py +++ b/src/cotdata/providers/cftc.py @@ -40,7 +40,7 @@ def _cache_dir() -> Path: - d = config.store_root() / "_cache" / "cot" + d = config.store_root() / "_cache" / "cot_legacy" d.mkdir(parents=True, exist_ok=True) return d @@ -116,8 +116,6 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None: if sub.empty: print(f"{code}: no rows") continue - # Index by report date (DatetimeIndex → manifest last_date), keeping the - # Legacy column name so a consumer's reset_index() gets the exact schema. sub = sub.sort_values(REPORT_DATE).set_index(REPORT_DATE) - store.write_cot(code, sub, source="cftc") + store.write_cot_legacy(code, sub, source="cftc") print(f"{code}: {len(sub):5d} weeks -> store") diff --git a/src/cotdata/providers/cftc_disagg.py b/src/cotdata/providers/cftc_disagg.py new file mode 100644 index 0000000..25f73e0 --- /dev/null +++ b/src/cotdata/providers/cftc_disagg.py @@ -0,0 +1,146 @@ +"""CFTC COT Disaggregated Futures-Only producer — cross-platform. + +Downloads fut_disagg_txt_{year}.zip, parses losslessly (preserving all Traders_* +and detailed entity columns), and writes per-code weekly positioning tables to +the store via store.write_cot_disagg(). +""" +import datetime as dt +import io +import zipfile +from email.utils import parsedate_to_datetime +from pathlib import Path + +import pandas as pd +import requests + +from .. import config, store +from ..registry import all_symbols, hist_code_scales + +URL_PREFIX = "https://www.cftc.gov/files/dea/history/fut_disagg_txt_" +FIRST_YEAR = 2006 # Disaggregated futures history start + +# CFTC TXT column names we must coerce/standardize (others pass through losslessly) +REPORT_DATE = "Report_Date_as_MM_DD_YYYY" +CONTRACT_CODE = "CFTC_Contract_Market_Code" + + +def _cache_dir() -> Path: + d = config.store_root() / "_cache" / "cot_disagg" + d.mkdir(parents=True, exist_ok=True) + return d + + +def _standardize_code(val) -> str: + """CFTC contract codes → 6-digit zero-padded string (matches CotSymbolCodeMap).""" + s = str(val).strip() + return s.zfill(6) if s.isdigit() else s + + +def _download_url(url: str, filename: str): + """Download a zip to the cache; skip if the server copy isn't newer.""" + zip_path = _cache_dir() / filename + try: + if zip_path.exists(): + head = requests.head(url, timeout=30) + server_mtime = head.headers.get("Last-Modified") + if server_mtime and zip_path.stat().st_mtime >= parsedate_to_datetime(server_mtime).timestamp(): + return zip_path # up to date + r = requests.get(url, timeout=180) + r.raise_for_status() + zip_path.write_bytes(r.content) + return zip_path + except Exception as e: # noqa: BLE001 + print(f" {filename} (disagg): download failed — {e}") + return zip_path if zip_path.exists() else None + + +def _parse_zip(zip_path: Path) -> pd.DataFrame: + """Extract the .txt (CSV) from a year zip → full lossless DataFrame.""" + with zipfile.ZipFile(zip_path) as zf: + with zf.open(zf.namelist()[0]) as fh: + # The .txt files are actually CSVs. low_memory=False prevents dtype warnings. + df = pd.read_csv(fh, low_memory=False) + + # Strip any trailing whitespace from column names BEFORE accessing them + df.columns = df.columns.str.strip() + + # CFTC sometimes changes the date column name in Disaggregated reports + if REPORT_DATE not in df.columns: + date_cols = [c for c in df.columns if "Report_Date" in c] + if date_cols: + df.rename(columns={date_cols[0]: REPORT_DATE}, inplace=True) + else: + print(f"AVAILABLE COLUMNS: {list(df.columns)}") + + # Coerce the key schema columns to match the Legacy schema format + df[CONTRACT_CODE] = df[CONTRACT_CODE].apply(_standardize_code) + df[REPORT_DATE] = pd.to_datetime(df[REPORT_DATE]).dt.tz_localize(None) + + # Parquet cannot serialize mixed-type object columns (e.g. Traders_Tot_Old contains ints and strings) + for col in df.select_dtypes(include=['object']).columns: + if col not in [CONTRACT_CODE, REPORT_DATE]: + df[col] = df[col].astype(str) + + return df + + +def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None: + """Download + parse CFTC Disaggregated futures COT; write full per-code history. + + codes: iterable of CFTC codes; default = all registry codes. + Rebuilds the complete per-code table each run. + """ + 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)) + + frames = [] + + # The CFTC bundles 2006-2016 in a single historical file + if first_year <= 2016: + url = "https://www.cftc.gov/files/dea/history/fut_disagg_txt_hist_2006_2016.zip" + zp = _download_url(url, "fut_disagg_txt_hist_2006_2016.zip") + if zp: + try: + df = _parse_zip(zp) + # Filter down to the requested year range so we don't accidentally pull earlier + # than first_year if the user explicitly wanted a shorter window. + df = df[(df[REPORT_DATE].dt.year >= first_year) & (df[REPORT_DATE].dt.year <= 2016)] + frames.append(df) + except Exception as e: + print(f" hist_2006_2016 (disagg): parse failed — {e}") + + # Fetch individual years for 2017+ (or first_year if it's > 2016) + for year in range(max(2017, first_year), last_year + 1): + zp = _download_url(f"{URL_PREFIX}{year}.zip", f"fut_disagg_txt_{year}.zip") + if not zp: + continue + try: + frames.append(_parse_zip(zp)) + except Exception as e: # noqa: BLE001 + print(f" {year} (disagg): parse failed — {e}") + + if not frames: + print("cftc_disagg: no data parsed") + return + + allrows = pd.concat(frames, ignore_index=True) + + # Parquet cannot serialize mixed-type object columns after concat (due to NaNs) + for col in allrows.select_dtypes(include=['object']).columns: + if col not in [CONTRACT_CODE, REPORT_DATE]: + allrows[col] = allrows[col].astype(str) + + for code in sorted(want): + sub = allrows[allrows[CONTRACT_CODE] == code].copy() + if sub.empty: + continue + + # 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") diff --git a/src/cotdata/store.py b/src/cotdata/store.py index 3e4dbff..94df35d 100644 --- a/src/cotdata/store.py +++ b/src/cotdata/store.py @@ -36,23 +36,35 @@ def read_prices(symbol: str, adjustment: str) -> pd.DataFrame: return pd.read_parquet(p) if p.exists() else pd.DataFrame() -# ── COT ─────────────────────────────────────────────────────────────────── -def write_cot(name: str, df: pd.DataFrame, source: str) -> None: - _atomic_write_parquet(df, config.cot_dir() / f"{name}.parquet") - _touch_manifest("cot", name, df, source) +# ── COT Legacy ──────────────────────────────────────────────────────────── +def write_cot_legacy(name: str, df: pd.DataFrame, source: str) -> None: + _atomic_write_parquet(df, config.cot_legacy_dir() / f"{name}.parquet") + _touch_manifest("cot_legacy", name, df, source) -def read_cot(name: str) -> pd.DataFrame: - p = config.cot_dir() / f"{name}.parquet" +def read_cot_legacy(name: str) -> pd.DataFrame: + p = config.cot_legacy_dir() / f"{name}.parquet" return pd.read_parquet(p) if p.exists() else pd.DataFrame() +# ── COT Disaggregated ───────────────────────────────────────────────────── +def write_cot_disagg(name: str, df: pd.DataFrame, source: str) -> None: + _atomic_write_parquet(df, config.cot_disagg_dir() / f"{name}.parquet") + _touch_manifest("cot_disagg", name, df, source) + + +def read_cot_disagg(name: str) -> pd.DataFrame: + p = config.cot_disagg_dir() / f"{name}.parquet" + return pd.read_parquet(p) if p.exists() else pd.DataFrame() + + + # ── Manifest ────────────────────────────────────────────────────────────── def load_manifest() -> dict: p = config.manifest_path() if p.exists(): return json.loads(p.read_text()) - return {"schema_version": config.SCHEMA_VERSION, "prices": {}, "cot": {}} + return {"schema_version": config.SCHEMA_VERSION, "prices": {}, "cot_legacy": {}, "cot_disagg": {}} def _touch_manifest(kind: str, name: str, df: pd.DataFrame, source: str) -> None: diff --git a/src/cotdata/update.py b/src/cotdata/update.py index a5fe902..43d3bef 100644 --- a/src/cotdata/update.py +++ b/src/cotdata/update.py @@ -10,20 +10,24 @@ def main() -> None: p = argparse.ArgumentParser(description="cotdata producer — fetch sources into the store.") p.add_argument("--prices", action="store_true", help="Update Norgate price bars (Windows).") - p.add_argument("--cot", action="store_true", help="Update CFTC COT (cross-platform).") + p.add_argument("--cot-legacy", action="store_true", help="Update CFTC COT Legacy (cross-platform).") + p.add_argument("--cot-disagg", action="store_true", help="Update CFTC COT Disaggregated Futures-Only (cross-platform).") p.add_argument("--symbols", nargs="+", default=None, help="Internal symbols; default = all in registry.") args = p.parse_args() config.store_root() # fail fast if COTDATA_STORE unset - if not (args.prices or args.cot): - p.error("nothing to do — pass --prices and/or --cot") + if not (args.prices or args.cot_legacy or args.cot_disagg): + p.error("nothing to do — pass --prices, --cot-legacy, and/or --cot-disagg") if args.prices: from .providers import norgate norgate.update(symbols=args.symbols) - if args.cot: + if args.cot_legacy: from .providers import cftc cftc.update() + if args.cot_disagg: + from .providers import cftc_disagg + cftc_disagg.update() if __name__ == "__main__": diff --git a/tests/test_cftc_disagg_fidelity.py b/tests/test_cftc_disagg_fidelity.py new file mode 100644 index 0000000..5a50eaa --- /dev/null +++ b/tests/test_cftc_disagg_fidelity.py @@ -0,0 +1,29 @@ +import zipfile +from cotdata.providers.cftc_disagg import _parse_zip, CONTRACT_CODE, REPORT_DATE + +def test_disagg_fidelity_preserves_columns(tmp_path): + """ + Fidelity test: Unlike the legacy provider which filters to a strict 10-column + subset, the disaggregated provider must be a lossless image of the CSV, + preserving all granular entity classes and Traders_* counts. + """ + # A dummy CSV mimicking the CFTC Disaggregated txt format + csv_content = b"""Market_and_Exchange_Names,Report_Date_as_MM_DD_YYYY,CFTC_Contract_Market_Code,Open_Interest_All,Prod_Merc_Positions_Long_All,Traders_Tot_All,Swap__Positions_Long_All,M_Money_Positions_Long_All,Other_Rept_Positions_Long_All +"CORN - CHICAGO BOARD OF TRADE",2021-01-05,002602,1000,200,50,150,300,100 +""" + zip_path = tmp_path / "fut_disagg_txt_2021.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("f_year.txt", csv_content) + + df = _parse_zip(zip_path) + + # 9 columns in the CSV -> 9 columns in the parsed DataFrame + assert len(df.columns) == 9, "Lossless parsing must preserve all columns from the zip" + + # Check that crucial non-legacy columns survived + assert "Traders_Tot_All" in df.columns + assert "Swap__Positions_Long_All" in df.columns + + # Check that coercion worked + assert df[CONTRACT_CODE].iloc[0] == "002602", "Contract code must be 6-digit zero padded" + assert df[REPORT_DATE].iloc[0].year == 2021, "Report date must be parsed to datetime" diff --git a/tests/test_cftc_reconciliation.py b/tests/test_cftc_reconciliation.py new file mode 100644 index 0000000..9924048 --- /dev/null +++ b/tests/test_cftc_reconciliation.py @@ -0,0 +1,54 @@ +import pandas as pd +import pytest +from cotdata import cot + +def test_reconciliation_structural_identities(monkeypatch): + """ + Validates that the Disaggregated schema correctly sums into the Legacy schema + groupings for the exact same report date. + + Legacy Commercial = Disagg Producer/Merchant + Swap Dealer + Legacy NonCommercial = Disagg Money Manager + Other Reportable + """ + + # Mock Legacy store returning 1 week of data + legacy_df = pd.DataFrame({ + "Comm_Positions_Long_All": [1000], + "NonComm_Positions_Long_All": [500], + }, index=pd.DatetimeIndex(["2021-01-05"])) + + # Mock Disaggregated store returning 1 week of data + disagg_df = pd.DataFrame({ + "Prod_Merc_Positions_Long_All": [600], + "Swap__Positions_Long_All": [400], # 600 + 400 = 1000 (Legacy Comm) + "M_Money_Positions_Long_All": [400], + "Other_Rept_Positions_Long_All": [100], # 400 + 100 = 500 (Legacy NonComm) + }, index=pd.DatetimeIndex(["2021-01-05"])) + + def mock_read_cot_legacy(name): + return legacy_df + + def mock_read_cot_disagg(name): + return disagg_df + + monkeypatch.setattr("cotdata.store.read_cot_legacy", mock_read_cot_legacy) + monkeypatch.setattr("cotdata.store.read_cot_disagg", mock_read_cot_disagg) + + leg = cot.get_cot("DUMMY", report="legacy") + dis = cot.get_cot("DUMMY", report="disagg") + + # Assert structural reconciliation on the Long side + c_derived = dis["Prod_Merc_Positions_Long_All"] + dis["Swap__Positions_Long_All"] + nc_derived = dis["M_Money_Positions_Long_All"] + dis["Other_Rept_Positions_Long_All"] + + pd.testing.assert_series_equal( + c_derived, + leg["Comm_Positions_Long_All"], + check_names=False + ) + + pd.testing.assert_series_equal( + nc_derived, + leg["NonComm_Positions_Long_All"], + check_names=False + )