Warrior_EA/research/altdata/cot.py

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