OpenFin/backend/services/cot_fetcher.py

"""
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,
    }