""" CFTC COT report parser — downloads and parses Commitments of Traders data. Tracks Gold, EUR, GBP, JPY positioning. """ import asyncio import io import logging import zipfile from datetime import datetime import aiohttp import pandas as pd from database import db as database logger = logging.getLogger("gap_system.data.cot") COT_URL = "https://www.cftc.gov/dea/newcot/deacom.txt" COT_ZIP_URL = f"https://www.cftc.gov/files/dea/history/deacom_txt_{datetime.now().year}.zip" # Market codes in COT report MARKET_CODES = { "GOLD": {"name": "GOLD", "asset": "XAUUSD", "search": "GOLD"}, "EURO FX": {"name": "EURO", "asset": "EURUSD", "search": "EURO FX"}, "BP": {"name": "GBP", "asset": "GBPUSD", "search": "BRITISH POUND"}, "JY": {"name": "JPY", "asset": "USDJPY", "search": "JAPANESE YEN"}, } # Historical data for percentile calculations _historical_positions: dict[str, list[float]] = {} async def fetch_cot_report() -> list[dict]: """Fetch and parse latest COT report.""" results = [] try: # User-Agent is strictly required to bypass CFTC 403 Forbidden blocks headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} async with aiohttp.ClientSession(headers=headers) as session: async with session.get(COT_URL, timeout=30) as resp: if resp.status != 200: logger.warning("COT report fetch returned %d, trying zip", resp.status) return await _fetch_cot_from_zip(session) text = await resp.text() # deacom.txt has no header row. We must use header=None and index dynamically. df = pd.read_csv(io.StringIO(text), header=None, dtype=str) for search_term, info in MARKET_CODES.items(): # Find rows matching this market using column index 0 ("Market_and_Exchange_Names") mask = df[0].str.contains(info["search"], case=False, na=False) market_df = df[mask] if market_df.empty: logger.warning("No COT data found for %s", info["name"]) continue # Get latest row row = market_df.iloc[-1] # In legacy deacom.txt format: # 0: Market Name, 1: Date YYMMDD, 8: Open Interest # 9: NonComm Long, 10: NonComm Short # 19: Change NonComm Long, 20: Change NonComm Short try: nc_long = float(row.iloc[9]) if pd.notna(row.iloc[9]) else 0.0 nc_short = float(row.iloc[10]) if pd.notna(row.iloc[10]) else 0.0 net_position = nc_long - nc_short nc_long_chg = float(row.iloc[19]) if pd.notna(row.iloc[19]) else 0.0 nc_short_chg = float(row.iloc[20]) if pd.notna(row.iloc[20]) else 0.0 week_change = nc_long_chg - nc_short_chg except (ValueError, IndexError): logger.warning("COT layout structural error for %s. CFTC may have modified index locations.", info["name"]) continue # Track historical for percentile key = info["asset"] _historical_positions.setdefault(key, []) _historical_positions[key].append(net_position) # Extreme positioning flag (>90th percentile) extreme_flag = False hist = _historical_positions[key] if len(hist) >= 10: sorted_hist = sorted(hist) p90 = sorted_hist[int(len(sorted_hist) * 0.9)] p10 = sorted_hist[int(len(sorted_hist) * 0.1)] extreme_flag = net_position > p90 or net_position < p10 # Determine bias if net_position > 0: bias = "BULLISH" elif net_position < 0: bias = "BEARISH" else: bias = "NEUTRAL" report_date = str(row.iloc[1]) result = { "asset": info["asset"], "net_long": nc_long, "net_short": nc_short, "net_position": net_position, "week_change": week_change, "extreme_flag": extreme_flag, "bias": bias, "report_date": report_date, "institutional_bias": bias, # alias for aggregator } results.append(result) await database.insert_cot( asset=info["asset"], net_long=nc_long, net_short=nc_short, net_position=net_position, week_change=week_change, extreme_flag=extreme_flag, bias=bias, report_date=report_date, ) logger.info( "COT %s: net=%+.0f chg=%+.0f %s%s", info["asset"], net_position, week_change, bias, " ⚠EXTREME" if extreme_flag else "", ) except Exception as e: logger.error("COT report fetch error: %s", e) return results async def _fetch_cot_from_zip(session: aiohttp.ClientSession) -> list[dict]: """Fallback: fetch COT from zip archive.""" try: async with session.get(COT_ZIP_URL, timeout=60) as resp: if resp.status != 200: logger.error("COT zip fetch failed: %d", resp.status) return [] data = await resp.read() with zipfile.ZipFile(io.BytesIO(data)) as zf: names = zf.namelist() if not names: return [] with zf.open(names[0]) as f: df = pd.read_csv(f) logger.info("Loaded COT from zip with %d rows", len(df)) return [] # simplified — use same parsing as above except Exception as e: logger.error("COT zip fallback error: %s", e) return [] def format_cot_digest(cot_data: list[dict]) -> str: """Format COT data as a readable digest for agents.""" if not cot_data: return "No COT data available." lines = ["=== COT INSTITUTIONAL POSITIONING (CFTC) ==="] for d in cot_data: direction = "▲" if d["week_change"] > 0 else "▼" if d["week_change"] < 0 else "━" extreme = " ⚠ EXTREME POSITIONING" if d["extreme_flag"] else "" lines.append( f" {d['asset']}: Net {d['net_position']:+,.0f} " f"({d['bias']}) {direction} chg {d['week_change']:+,.0f}{extreme}" ) return "\n".join(lines)