""" 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) ────────────────────────── # All markets fetched from legacy endpoint via stable contract codes. # noncomm_positions = non-commercial (speculative) positioning. _COT_MARKETS = { "067651": ("WTI Crude", "energy"), "023651": ("Natural Gas", "energy"), "088691": ("Gold", "metals"), "084691": ("Silver", "metals"), "085692": ("Copper", "metals"), "002602": ("Corn", "agri"), "001602": ("Wheat", "agri"), "005602": ("Soybeans", "agri"), "007601": ("Soybean Oil", "agri"), "073732": ("Cocoa", "agri"), "083731": ("Coffee", "agri"), "099741": ("Euro (EUR/USD)", "forex"), "097741": ("Japanese Yen", "forex"), "096742": ("British Pound", "forex"), "092741": ("Swiss Franc", "forex"), "098662": ("DXY Index", "forex"), "045601": ("Fed Funds", "bonds"), "020601": ("US T-Bonds", "bonds"), "043602": ("10Y T-Notes", "bonds"), } def _parse_int_flat(v) -> int: try: return int(str(v).replace(",", "").strip()) except Exception: return 0 def fetch_all_cot() -> List[Dict[str, Any]]: """Fetch COT non-commercial positioning for all tracked markets via contract codes. Uses the legacy CFTC Socrata endpoint which covers all asset classes with stable contract codes and noncomm_positions_long/short_all fields. Returns a flat list of per-market dicts suitable for save_cot_data(). """ logger.info("Fetching COT data from CFTC (specialist desks feed)...") codes = list(_COT_MARKETS.keys()) codes_clause = ", ".join(f"'{c}'" for c in codes) params = { "$select": ( "cftc_contract_market_code,market_and_exchange_names," "report_date_as_yyyy_mm_dd,open_interest_all," "noncomm_positions_long_all,noncomm_positions_short_all," "change_in_noncomm_long_all,change_in_noncomm_short_all" ), "$where": f"cftc_contract_market_code in ({codes_clause})", "$order": "report_date_as_yyyy_mm_dd DESC", "$limit": str(len(codes) * 3), } try: resp = requests.get(SOCRATA_URL, params=params, timeout=25) resp.raise_for_status() rows = resp.json() except Exception as e: logger.error(f"COT fetch_all_cot failed: {e}") return [] seen: set = set() results: List[Dict[str, Any]] = [] for row in rows: code = row.get("cftc_contract_market_code", "") if code not in _COT_MARKETS or code in seen: continue seen.add(code) label, ac = _COT_MARKETS[code] mkt = row.get("market_and_exchange_names", "") 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, }) logger.info(f"COT: fetched {len(results)}/{len(codes)} markets") return results