"""CFTC Commitments of Traders collector - no API key needed. Two report families cover every instrument this project trades: TFF (Traders in Financial Futures, fut_fin_txt_YYYY.zip): E-mini S&P 500, currency futures (EUR/JPY/CAD/GBP), VIX, Bitcoin. Categories: Dealer / Asset Manager / Leveraged Funds / Other / Nonreportable. Disaggregated (fut_disagg_txt_YYYY.zip): Gold (COMEX), WTI crude (NYMEX). Categories: Producer / Swap Dealer / Managed Money / Other / Nonreportable. History: 2010+ pulled by default (2006 earliest for both families). PUBLICATION LAG (the lookahead rule): report is as-of TUESDAY, released FRIDAY 15:30 ET. We stamp `published` = Saturday 00:00 UTC after that Friday - conservative under both DST regimes, and for D1 bars it means the data is usable from the following Monday, exactly as it would have been live. SIGN CONVENTION: currency futures are quoted FX/USD, so for USDJPY and USDCAD (USD base) a net-long yen future is a net-SHORT USDJPY view. The raw CSVs keep the futures' own sign; flip at the feature-derivation step, not here. Usage: python -m altdata.cot # download + rebuild per-symbol CSVs python -m altdata.cot --start 2006 """ import argparse import datetime as dt import io import re import pandas as pd from .common import DATA_ROOT, RAW, cached_download, ensure_dirs, read_zip_member CFTC = "https://www.cftc.gov/files/dea/history" # symbol -> (report family, regex on Market_and_Exchange_Names) MARKETS = { "SP500": ("fin", r"^E-MINI S&P 500"), "VIX": ("fin", r"^VIX FUTURES"), "BTCUSD": ("fin", r"BITCOIN"), "EURUSD": ("fin", r"^EURO FX -"), "USDJPY": ("fin", r"^JAPANESE YEN"), "USDCAD": ("fin", r"^CANADIAN DOLLAR"), "GBPUSD": ("fin", r"^BRITISH POUND"), "XAUUSD": ("disagg", r"^GOLD -"), "XTIUSD": ("disagg", r"^(CRUDE OIL, LIGHT SWEET|WTI-PHYSICAL)"), } # columns kept from either family (present = kept; absent = ignored) KEEP = [ "Open_Interest_All", # TFF "Dealer_Positions_Long_All", "Dealer_Positions_Short_All", "Asset_Mgr_Positions_Long_All", "Asset_Mgr_Positions_Short_All", "Lev_Money_Positions_Long_All", "Lev_Money_Positions_Short_All", "Other_Rept_Positions_Long_All", "Other_Rept_Positions_Short_All", "NonRept_Positions_Long_All", "NonRept_Positions_Short_All", # Disaggregated "Prod_Merc_Positions_Long_All", "Prod_Merc_Positions_Short_All", "Swap_Positions_Long_All", "Swap__Positions_Short_All", "M_Money_Positions_Long_All", "M_Money_Positions_Short_All", ] FAMILY_FILE = {"fin": "fut_fin_txt_{y}.zip", "disagg": "fut_disagg_txt_{y}.zip"} def _published(report_date: pd.Timestamp) -> pd.Timestamp: """Saturday 00:00 UTC after the Friday release that follows a Tuesday report.""" days_to_sat = (5 - report_date.weekday()) % 7 or 7 # next Saturday, strictly after return (report_date + pd.Timedelta(days=days_to_sat)).normalize() def load_family(family: str, years: range) -> pd.DataFrame: frames = [] for y in years: url = f"{CFTC}/{FAMILY_FILE[family].format(y=y)}" dest = RAW / "cot" / FAMILY_FILE[family].format(y=y) try: path = cached_download(url, dest, refresh=(y == years[-1])) except RuntimeError as e: print(f" {family} {y}: unavailable ({e})") continue _, blob = read_zip_member(path) df = pd.read_csv(io.BytesIO(blob), low_memory=False) frames.append(df) out = pd.concat(frames, ignore_index=True) out.columns = [c.strip() for c in out.columns] return out def extract(df: pd.DataFrame, pattern: str) -> pd.DataFrame: name_col = "Market_and_Exchange_Names" hit = df[df[name_col].str.contains(pattern, regex=True, na=False)].copy() matched = sorted(hit[name_col].str.strip().unique()) print(f" matched contract names: {matched}") observed = pd.Series(pd.NaT, index=hit.index) if "Report_Date_as_YYYY-MM-DD" in hit.columns: observed = pd.to_datetime(hit["Report_Date_as_YYYY-MM-DD"], errors="coerce") if "Report_Date_as_MM_DD_YYYY" in hit.columns: # column name lies in some vintages: contains ISO dates fallback = pd.to_datetime(hit["Report_Date_as_MM_DD_YYYY"], format="mixed", errors="coerce") observed = observed.fillna(fallback) hit["observed"] = observed bad = hit["observed"].isna().sum() if bad: print(f" !! dropping {bad} rows with unparseable report dates") hit = hit.dropna(subset=["observed"]) keep = [c for c in KEEP if c in hit.columns] out = hit[["observed", name_col] + keep].rename(columns={name_col: "contract"}) for c in keep: out[c] = pd.to_numeric(out[c], errors="coerce") # multiple contract variants (e.g. exchange renames) -> keep the row with the # largest open interest per report date, so the series never double-counts out = (out.sort_values(["observed", "Open_Interest_All"], ascending=[True, False]) .drop_duplicates("observed", keep="first") .sort_values("observed").reset_index(drop=True)) out["published"] = out["observed"].map(_published) return out def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--start", type=int, default=2010) args = ap.parse_args() ensure_dirs() this_year = dt.date.today().year years = range(args.start, this_year + 1) out_dir = DATA_ROOT / "cot" out_dir.mkdir(parents=True, exist_ok=True) for family in ("fin", "disagg"): print(f"family {family}: downloading {years.start}-{this_year}") df = load_family(family, years) for sym, (fam, pattern) in MARKETS.items(): if fam != family: continue print(f" {sym}:") series = extract(df, pattern) if series.empty: print(f" !! no rows matched - check pattern {pattern!r}") continue dest = out_dir / f"{sym}_cot.csv" series.to_csv(dest, index=False) print(f" {len(series)} weekly rows {series['observed'].min():%Y-%m-%d} " f"-> {series['observed'].max():%Y-%m-%d} => {dest.name}") if __name__ == "__main__": main()