""" CFTC Commitment of Traders (COT) weekly fetcher. Data from CFTC Socrata public API — no API key required. """ 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, }