Cot tff

README.md

@@ -53,6 +53,8 @@ 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_legacy/{code}.parquet` — weekly CFTC Legacy positioning.
+- `cot_disagg/{code}.parquet` — weekly CFTC Disaggregated positioning.
+- `cot_tff/{code}.parquet` — weekly CFTC Traders in Financial Futures positioning.
 - `manifest.json` — per-table `last_date`, `n_rows`, `source`, `updated_at`, `schema_version`.
 
 ## Consumer
@@ -63,6 +65,7 @@ df = cotdata.get_prices("ES", adjustment="backadj")   # USE THIS FOR SIGNALS + S
 sz = cotdata.get_prices("ES", adjustment="unadj")     # USE FOR POSITION SIZING / POINT VALUE
 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)
+cot_tff = cotdata.get_cot("ES", report="tff")         # USE FOR TRADER COUNTS (FINANCIALS)
 ```
 
 Set `COTDATA_STORE` to the synced store directory. 
@@ -75,6 +78,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-legacy                # CFTC Legacy (cross-platform)
 COTDATA_STORE=/store  cotdata-update --cot-disagg                # CFTC Disaggregated (cross-platform)
+COTDATA_STORE=/store  cotdata-update --cot-tff                   # CFTC Traders in Financial Futures (cross-platform)
+COTDATA_STORE=/store  cotdata-update --cot-all                   # Update all CFTC COT pipelines
 ```
 
 Schedule nightly (prices, after the Norgate Data Updater) and weekly (COT Friday releases).
@@ -121,6 +126,25 @@ This diagnostic script tests:
 2. **Subscription Access**: Validates that your Norgate subscription is active and has the required CME futures data package enabled.
 3. **Roll Gap Validation**: Mathematically proves whether your Norgate Data Updater is configured globally to return back-adjusted or unadjusted continuous contracts. It hunts for artificial calendar-spread gaps at contract roll dates to ensure you are receiving gap-free, continuous data, which is absolutely vital for accurate stop-loss modeling.
 
+## COT Formats Explained
+
+The CFTC publishes positioning data in three distinct formats. `cotdata` manages all three to ensure complete market coverage and the deepest possible historical backtesting.
+
+1. **Legacy Format (1986–Present)**
+   - **Scope:** All markets.
+   - **Categories:** Divides traders broadly into **Commercial** (hedgers) and **Non-Commercial** (large speculators). 
+   - **Use Case:** This is the original format. While its broad categories make it less precise for modern analysis, it is the only format that provides data prior to 2006, making it essential for long-term historical backtesting.
+
+2. **Disaggregated Format (DIS) (2006–Present)**
+   - **Scope:** Physical commodities only (Agriculture, Energy, Metals).
+   - **Categories:** Splits traders into four granular groups: **Producer/Merchant** (classic hedgers), **Swap Dealers** (financial intermediaries), **Managed Money** (hedge funds / CTAs), and **Other Reportables**.
+   - **Use Case:** Provides a much clearer view of the "Smart Money" (Managed Money) in commodity markets.
+
+3. **Traders in Financial Futures (TFF) (2006–Present)**
+   - **Scope:** Financial markets only (Equities, Rates, Currencies).
+   - **Categories:** The financial counterpart to Disaggregated. Splits traders into: **Dealer/Intermediary** (sell-side), **Asset Manager** (pension/mutual funds), **Leveraged Funds** (hedge funds / CTAs), and **Other Reportables**.
+   - **Use Case:** The definitive source for tracking speculative flow (Leveraged Funds) in financial markets.
+
 ## Data Schemas
 
 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.
@@ -140,7 +164,7 @@ The primary source for price history (Norgate Data). Indexed by tz-naive `Date`.
 | `Delivery Month` | float | Expiration month of the active contract (e.g. `202609`). Used to detect contract rolls. |
 
 ### 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`.
+The primary source for Legacy positioning data (CFTC Legacy Futures Report). **History starts in 1986.** 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`).
@@ -162,7 +186,16 @@ The primary source for Legacy positioning data (CFTC Legacy Futures Report). Ind
 | `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`.
+The primary source for entity-specific positioning and trader counts (CFTC Disaggregated Futures-Only Report). **History starts in 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.
+
+### COT Traders in Financial Futures (TFF) Data (`cot_tff/{code}.parquet`)
+The primary source for entity-specific positioning and trader counts for Financial markets (CFTC Traders in Financial Futures Futures-Only Report). **History starts in 2006.** Indexed by tz-naive `Report_Date_as_MM_DD_YYYY`.
+
+> [!NOTE]
+> **Financials Counterpart**: TFF is the exact counterpart to Disaggregated reports, used exclusively for financial markets (Equities, FX, Rates) which do not have Disaggregated reports.
+
+> [!NOTE]
+> **Lossless Image**: Like Disaggregated, TFF parquets are a **lossless image** of the source CFTC `txt` files. They contain the financial entity groups (`Dealer`, `Asset_Mgr`, `Lev_Money`, `Other_Rept`) and their respective `Traders_*` counts. This is the required store for computing Position Size and Clustering metrics for financial assets.

src/cotdata/config.py

@@ -27,5 +27,9 @@ def cot_disagg_dir() -> Path:
     return store_root() / "cot_disagg"
 
 
+def cot_tff_dir() -> Path:
+    return store_root() / "cot_tff"
+
+
 def manifest_path() -> Path:
     return store_root() / "manifest.json"

src/cotdata/cot.py

@@ -13,19 +13,29 @@
 
 
 def get_cot(name: str, report: str = "legacy") -> pd.DataFrame:
-    """Weekly COT for an internal symbol OR a raw CFTC code. Empty if absent.
+    """Read a symbol's weekly COT history with predecessor code stitching.
 
-    If the resolved symbol declares hist_codes (predecessor exchange listings of
-    the same contract), they're stitched in chronologically to fill gaps the
-    primary code doesn't cover. The primary code wins on overlapping report dates,
-    and the stitched-in rows are relabelled to the primary code so the series
-    presents as a single contract to code-keyed consumers (e.g. CotIndexer).
+    name: the internal pipeline symbol, e.g. 'ES' or 'GC'.
+    report: 'legacy' (default), 'disagg' (commodity futures only), or 'tff' (financial futures only).
+
+    Returns:
+    DataFrame indexed by date. The columns depend on the `report` requested:
+      - legacy: Open_Interest_All, NonComm_Positions_Long_All, Comm_Positions_Long_All, etc.
+      - disagg: Open_Interest_All, Prod_Merc_Positions_Long_All, M_Money_Positions_Long_All, etc.
+      - tff: Open_Interest_All, Dealer_Positions_Long_All, Lev_Money_Positions_Long_All, etc.
+    If no data exists, returns an empty 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
+    read_fn = {
+        "legacy": store.read_cot_legacy,
+        "disagg": store.read_cot_disagg,
+        "tff": store.read_cot_tff,
+    }.get(report)
+    if not read_fn:
+        raise ValueError(f"Unknown report type: {report}. Expected 'legacy', 'disagg', or 'tff'")
 
     if sym is None or not sym.cftc_code:
         return read_fn(name)

src/cotdata/providers/cftc.py

@@ -51,10 +51,18 @@ def _standardize_code(val) -> str:
     return s.zfill(6) if s.isdigit() else s
 
 
-def _download_year(year: int):
-    """Download a year's zip to the cache; skip if the server copy isn't newer."""
-    url = f"{URL_PREFIX}{year}.zip"
-    zip_path = _cache_dir() / f"dea_fut_xls_{year}.zip"
+def _download_year(year: int) -> Path | None:
+    # CFTC changed their naming convention in 2004.
+    # 1986 - 2003: deafut_xls_{year}.zip
+    # 2004 - Present: dea_fut_xls_{year}.zip
+    if year < 2004:
+        url = f"https://www.cftc.gov/files/dea/history/deafut_xls_{year}.zip"
+        filename = f"deafut_xls_{year}.zip"
+    else:
+        url = f"{URL_PREFIX}{year}.zip"
+        filename = f"dea_fut_xls_{year}.zip"
+
+    zip_path = _cache_dir() / filename
     try:
         if zip_path.exists():
             head = requests.head(url, timeout=30)
@@ -118,4 +126,4 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
             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 -> store")
+        print(f"{code}: {len(sub):5d} weeks (legacy) -> store")

src/cotdata/providers/cftc_tff.py

@@ -0,0 +1,146 @@
+"""CFTC COT Traders in Financial Futures (TFF) producer — cross-platform.
+
+Downloads fut_fin_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_tff().
+"""
+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_fin_txt_"
+FIRST_YEAR = 2006  # TFF 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_tff"
+    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} (tff): 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 TFF 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 TFF 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/fin_fut_txt_2006_2016.zip"
+        zp = _download_url(url, "fin_fut_txt_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 (tff): 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_fin_txt_{year}.zip")
+        if not zp:
+            continue
+        try:
+            frames.append(_parse_zip(zp))
+        except Exception as e:  # noqa: BLE001
+            print(f"  {year} (tff): parse failed — {e}")
+
+    if not frames:
+        print("cftc_tff: 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_tff(code, sub, source="cftc_tff")
+        print(f"{code}: {len(sub):5d} weeks (tff) -> store")

src/cotdata/registry.py

@@ -29,6 +29,7 @@ class Symbol:
     norgate: str                  # Norgate continuous symbol, e.g. "&ES"
     asset_class: str
     is_equity: bool
+    report_type: str = "disagg"   # "tff" for financials, "disagg" for commodities
     cftc_code: Optional[str] = None
     # Predecessor CFTC codes from earlier exchange/contract listings of the SAME
     # instrument, stitched in chronologically behind cftc_code by get_cot. Each
@@ -102,6 +103,8 @@ def load_registry(yaml_path=None) -> Dict[str, Symbol]:
                 # Derived from the class so the two can't drift; an explicit
                 # is_equity in the YAML still wins if a symbol ever needs it.
                 is_equity=attrs.get("is_equity", asset_class == "Equities"),
+                # Financials use TFF (Traders in Financial Futures), Commodities use Disaggregated
+                report_type=attrs.get("report_type", "tff" if asset_class in ("Equities", "FX", "Rates") else "disagg"),
                 cftc_code=attrs["cftc_code"],
                 hist_codes=_coerce_hist_codes(attrs.get("hist_codes")),
             )

src/cotdata/store.py

@@ -58,13 +58,24 @@ def read_cot_disagg(name: str) -> pd.DataFrame:
     return pd.read_parquet(p) if p.exists() else pd.DataFrame()
 
 
+# ── COT TFF (Traders in Financial Futures) ────────────────────────────────
+def write_cot_tff(name: str, df: pd.DataFrame, source: str) -> None:
+    _atomic_write_parquet(df, config.cot_tff_dir() / f"{name}.parquet")
+    _touch_manifest("cot_tff", name, df, source)
+
+
+def read_cot_tff(name: str) -> pd.DataFrame:
+    p = config.cot_tff_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_legacy": {}, "cot_disagg": {}}
+    return {"schema_version": config.SCHEMA_VERSION, "prices": {}, "cot_legacy": {}, "cot_disagg": {}, "cot_tff": {}}
 
 
 def _touch_manifest(kind: str, name: str, df: pd.DataFrame, source: str) -> None:

src/cotdata/update.py

@@ -12,23 +12,31 @@ def main() -> None:
     p.add_argument("--prices", action="store_true", help="Update Norgate price bars (Windows).")
     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("--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.")
     args = p.parse_args()
 
     config.store_root()  # fail fast if COTDATA_STORE unset
-    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 not (args.prices or args.cot_legacy or args.cot_disagg or args.cot_tff or args.cot_all):
+        p.error("nothing to do — pass --prices, --cot-legacy, --cot-disagg, --cot-tff, or --cot-all")
 
     if args.prices:
         from .providers import norgate
         norgate.update(symbols=args.symbols)
-    if args.cot_legacy:
+
+    if args.cot_legacy or args.cot_all:
         from .providers import cftc
         cftc.update()
-    if args.cot_disagg:
+
+    if args.cot_disagg or args.cot_all:
         from .providers import cftc_disagg
         cftc_disagg.update()
-
+
+    if args.cot_tff or args.cot_all:
+        from .providers import cftc_tff
+        cftc_tff.update()
 
 if __name__ == "__main__":
     main()