OpenFin/backend/services/cot_fetcher.py

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