Spaces:
Sleeping
Sleeping
File size: 6,559 Bytes
511a00c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | """
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)
|