""" CFTC Commitment of Traders (COT) weekly fetcher. Data from CFTC Socrata public API — no API key required. Two modes: - fetch_cot_report(): legacy institutional_reports format (used by institutional_scheduler) - fetch_all_cot(): flat list of per-commodity/asset COT entries (used by specialist desks / cot_data table) """ import logging import math from datetime import datetime, timedelta from typing import Any, Dict, List, Optional import requests logger = logging.getLogger(__name__) SOCRATA_URL = "https://publicreporting.cftc.gov/resource/6dca-aqww.json" COT_INSTRUMENTS = { "088691": {"name": "Gold", "category": "metals", "signal_field": "signal_metals"}, "084691": {"name": "Silver", "category": "metals", "signal_field": "signal_metals"}, "085692": {"name": "Copper", "category": "metals", "signal_field": "signal_metals"}, "067651": {"name": "Crude Oil (WTI)", "category": "energy", "signal_field": "signal_energy"}, "023651": {"name": "Natural Gas", "category": "energy", "signal_field": "signal_energy"}, "13874+": {"name": "S&P 500", "category": "equities", "signal_field": "signal_indices"}, "099741": {"name": "EUR/USD", "category": "forex", "signal_field": "signal_forex"}, } _SIGNAL_PRIORITY = {"bullish": 3, "bearish": 3, "neutral": 0} def _fetch_cot_history(contract_code: str, weeks: int = 56) -> List[Dict]: cutoff = (datetime.utcnow() - timedelta(weeks=weeks)).strftime("%Y-%m-%dT00:00:00.000") params = { "$where": f"cftc_contract_market_code='{contract_code}' AND report_date_as_yyyy_mm_dd>='{cutoff}'", "$order": "report_date_as_yyyy_mm_dd DESC", "$limit": weeks + 5, "$select": ( "report_date_as_yyyy_mm_dd," "noncomm_positions_long_all,noncomm_positions_short_all," "comm_positions_long_all,comm_positions_short_all," "open_interest_all" ), } try: resp = requests.get(SOCRATA_URL, params=params, timeout=25) resp.raise_for_status() return resp.json() except Exception as e: logger.warning(f"[COT] Failed to fetch {contract_code}: {e}") return [] def _compute_net_and_zscore(rows: List[Dict]) -> Optional[Dict]: if not rows: return None nets = [] for row in rows: try: longs = float(row.get("noncomm_positions_long_all") or 0) shorts = float(row.get("noncomm_positions_short_all") or 0) nets.append(longs - shorts) except (TypeError, ValueError): continue if not nets: return None current_net = nets[0] latest_date = rows[0].get("report_date_as_yyyy_mm_dd", "")[:10] history_52w = nets[:52] if len(history_52w) >= 4: mean = sum(history_52w) / len(history_52w) variance = sum((x - mean) ** 2 for x in history_52w) / len(history_52w) std = math.sqrt(variance) if variance > 0 else 1.0 z_score = (current_net - mean) / std else: mean = current_net z_score = 0.0 week_change = current_net - nets[1] if len(nets) > 1 else 0.0 return { "report_date": latest_date, "net_positioning": round(current_net), "week_change": round(week_change), "z_score": round(z_score, 2), "mean_52w": round(mean), "history_count": len(history_52w), } def _signal_from_zscore(z: float) -> str: if z >= 1.5: return "bullish" elif z <= -1.5: return "bearish" return "neutral" def fetch_cot_report() -> Optional[Dict]: """Fetch latest COT data for all tracked instruments and return structured report dict.""" results: Dict[str, Any] = {} report_dates = [] for code, meta in COT_INSTRUMENTS.items(): rows = _fetch_cot_history(code) analysis = _compute_net_and_zscore(rows) if analysis: results[meta["name"]] = { **analysis, "category": meta["category"], "signal_field": meta["signal_field"], "signal": _signal_from_zscore(analysis["z_score"]), } if analysis["report_date"]: report_dates.append(analysis["report_date"]) if not results: logger.warning("[COT] No results returned from CFTC API") return None report_date = max(report_dates) if report_dates else datetime.utcnow().strftime("%Y-%m-%d") signals = { "signal_energy": "neutral", "signal_metals": "neutral", "signal_indices": "neutral", "signal_forex": "neutral", } key_points: List[str] = [] extremes: List[str] = [] for name, data in results.items(): z = data["z_score"] net_k = round(data["net_positioning"] / 1000, 1) change_k = round(data["week_change"] / 1000, 1) kp = f"{name}: net {net_k:+.0f}k contracts (z-score {z:+.2f})" if abs(z) >= 1.5: kp += " — EXTREME" extremes.append(name) elif abs(z) >= 0.8: kp += " — notable" if abs(change_k) >= 5: dir_str = "+" if change_k >= 0 else "" kp += f", WoW {dir_str}{change_k:.0f}k" key_points.append(kp) sf = data["signal_field"] new_sig = data["signal"] if _SIGNAL_PRIORITY.get(new_sig, 0) > _SIGNAL_PRIORITY.get(signals.get(sf, "neutral"), 0): signals[sf] = new_sig implications: List[str] = [] for name, data in results.items(): z = data["z_score"] if z >= 1.5: implications.append( f"Extreme long spec positioning in {name} — crowded trade, watch for reversal" ) elif z <= -1.5: implications.append( f"Extreme short spec positioning in {name} — short squeeze risk" ) elif z >= 1.0: implications.append(f"Bullish spec momentum building in {name}") elif z <= -1.0: implications.append(f"Bearish spec momentum in {name}") if not implications: implications = ["COT positioning broadly neutral — no directional extreme this week"] importance = 3 if extremes else 2 ai_summary = ( f"CFTC COT weekly ({report_date}). " f"Tracked {len(results)} instruments. " + (f"Extreme readings: {', '.join(extremes)}. " if extremes else "") + f"Energy={signals['signal_energy']}, Metals={signals['signal_metals']}, " f"Indices={signals['signal_indices']}, Forex={signals['signal_forex']}." ) return { "report_type": "cot", "report_date": report_date, "title": f"CFTC COT Weekly — {report_date}", "source": "CFTC Socrata API", "importance": importance, "category": "multi", "raw_data": results, "key_points": key_points, "trading_implications": " | ".join(implications), **signals, "ai_summary": ai_summary, } # ── Specialist Desks flat COT feed (cot_data table) ────────────────────────── _DISAGG_URL = "https://publicreporting.cftc.gov/resource/72hh-3qpy.json" # disaggregated commodities _LEGACY_FIN_URL = "https://publicreporting.cftc.gov/resource/gpe5-46if.json" # financial/forex _DISAGG_MARKETS = [ ("CRUDE OIL, LIGHT SWEET - NYMEX", "WTI Crude", "energy"), ("NATURAL GAS - NYMEX", "Natural Gas", "energy"), ("GOLD - COMEX", "Gold", "metals"), ("SILVER - COMEX", "Silver", "metals"), ("COPPER- #1 - COMEX", "Copper", "metals"), ("CORN - CBOT", "Corn", "agri"), ("WHEAT - CBOT", "Wheat", "agri"), ("SOYBEANS - CBOT", "Soybeans", "agri"), ("SOYBEAN OIL - CBOT", "Soybean Oil", "agri"), ("COCOA - ICE", "Cocoa", "agri"), ("COFFEE C - ICE", "Coffee", "agri"), ] _FIN_MARKETS = [ ("EURO FX - CME", "Euro (EUR/USD)", "forex"), ("JAPANESE YEN - CME", "Japanese Yen", "forex"), ("BRITISH POUND STERLING - CME", "British Pound", "forex"), ("SWISS FRANC - CME", "Swiss Franc", "forex"), ("U.S. DOLLAR INDEX - ICE FUTURES U.S.", "DXY Index", "forex"), ("30-DAY FEDERAL FUNDS - CBOT", "Fed Funds", "bonds"), ("U.S. TREASURY BONDS - CBOT", "US T-Bonds", "bonds"), ("10-YEAR U.S. TREASURY NOTES - CBOT", "10Y T-Notes", "bonds"), ] def _parse_int_flat(v) -> int: try: return int(str(v).replace(",", "").strip()) except Exception: return 0 def _fetch_disaggregated_flat() -> List[Dict[str, Any]]: """Fetch commodity COT (disaggregated) — MM long/short positions.""" results = [] market_names = [m[0] for m in _DISAGG_MARKETS] name_map = {m[0]: (m[1], m[2]) for m in _DISAGG_MARKETS} quoted = ", ".join(f"'{n}'" for n in market_names) params = { "$select": "market_and_exchange_names,report_date_as_yyyy_mm_dd,m_money_positions_long_all,m_money_positions_short_all,open_interest_all,change_in_m_money_long_all,change_in_m_money_short_all", "$where": f"market_and_exchange_names in ({quoted})", "$order": "report_date_as_yyyy_mm_dd DESC", "$limit": str(len(market_names) * 2), } try: resp = requests.get(_DISAGG_URL, params=params, timeout=20) resp.raise_for_status() rows = resp.json() except Exception as e: logger.error(f"COT disaggregated fetch failed: {e}") return [] seen: set = set() for row in rows: mkt = row.get("market_and_exchange_names", "") if mkt not in name_map or mkt in seen: continue seen.add(mkt) label, ac = name_map[mkt] mm_long = _parse_int_flat(row.get("m_money_positions_long_all", 0)) mm_short = _parse_int_flat(row.get("m_money_positions_short_all", 0)) oi = _parse_int_flat(row.get("open_interest_all", 0)) net = mm_long - mm_short chg_long = _parse_int_flat(row.get("change_in_m_money_long_all", 0)) chg_short = _parse_int_flat(row.get("change_in_m_money_short_all", 0)) change_net = chg_long - chg_short net_pct_oi = round(net / oi * 100, 2) if oi else 0.0 results.append({ "market_name": mkt, "commodity": label, "asset_class": ac, "report_date": row.get("report_date_as_yyyy_mm_dd", "")[:10], "mm_long": mm_long, "mm_short": mm_short, "open_interest": oi, "net_position": net, "net_pct_oi": net_pct_oi, "change_net": change_net, }) return results def _fetch_financial_flat() -> List[Dict[str, Any]]: """Fetch financial/forex COT (legacy) — Non-commercial long/short.""" results = [] market_names = [m[0] for m in _FIN_MARKETS] name_map = {m[0]: (m[1], m[2]) for m in _FIN_MARKETS} quoted = ", ".join(f"'{n}'" for n in market_names) params = { "$select": "market_and_exchange_names,report_date_as_yyyy_mm_dd,noncomm_positions_long_all,noncomm_positions_short_all,open_interest_all,change_in_noncomm_long_all,change_in_noncomm_short_all", "$where": f"market_and_exchange_names in ({quoted})", "$order": "report_date_as_yyyy_mm_dd DESC", "$limit": str(len(market_names) * 2), } try: resp = requests.get(_LEGACY_FIN_URL, params=params, timeout=20) resp.raise_for_status() rows = resp.json() except Exception as e: logger.error(f"COT financial fetch failed: {e}") return [] seen: set = set() for row in rows: mkt = row.get("market_and_exchange_names", "") if mkt not in name_map or mkt in seen: continue seen.add(mkt) label, ac = name_map[mkt] nc_long = _parse_int_flat(row.get("noncomm_positions_long_all", 0)) nc_short = _parse_int_flat(row.get("noncomm_positions_short_all", 0)) oi = _parse_int_flat(row.get("open_interest_all", 0)) net = nc_long - nc_short chg_long = _parse_int_flat(row.get("change_in_noncomm_long_all", 0)) chg_short = _parse_int_flat(row.get("change_in_noncomm_short_all", 0)) change_net = chg_long - chg_short net_pct_oi = round(net / oi * 100, 2) if oi else 0.0 results.append({ "market_name": mkt, "commodity": label, "asset_class": ac, "report_date": row.get("report_date_as_yyyy_mm_dd", "")[:10], "mm_long": nc_long, "mm_short": nc_short, "open_interest": oi, "net_position": net, "net_pct_oi": net_pct_oi, "change_net": change_net, }) return results def fetch_all_cot() -> List[Dict[str, Any]]: """Fetch both disaggregated (commodities) and financial (forex/bonds) COT data. Returns a flat list of per-market dicts suitable for save_cot_data(). """ logger.info("Fetching COT data from CFTC (specialist desks feed)...") all_data: List[Dict[str, Any]] = [] all_data.extend(_fetch_disaggregated_flat()) all_data.extend(_fetch_financial_flat()) logger.info(f"COT: fetched {len(all_data)} markets") return all_data