feat: institutional reports — CFTC COT + EIA petroleum weekly

- New institutional_reports table (DB) with importance, signals per asset class, key points, absorption tracking
- cot_fetcher.py: CFTC Socrata API (6dca-aqww), 7 instruments (Gold/Silver/Copper/WTI/NatGas/SP500/EURUSD), net positioning + 52-week z-score
- eia_fetcher.py: EIA API v2, 4 series (crude/Cushing/gasoline/distillates), WoW surprise detection
- institutional.py router: GET /reports, GET /reports/{id}, POST /refresh, GET /stats
- institutional_scheduler.py: weekly auto-fetch (COT Saturdays, EIA Wednesday afternoons)
- ai_analyzer.py: build_institutional_block() + institutional_block param injected into AI scoring prompt
- auto_cycle.py: inject institutional block into suggestion + scoring, absorption tracking via keyword overlap after each cycle commentary
- InstitutionalReports.tsx: full page with filter bar (type/category/importance/period), cards with key point bullets, EXTREME alerts highlighted, signal badges, absorption badge, trading implications, expandable detail

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

backend/main.py

@@ -4,6 +4,7 @@ from routers import market_data, geopolitical, options, backtest, ai, portfolio,
 from routers import logs as logs_router
 from routers import var as var_router
 from routers import reports as reports_router
+from routers import institutional as institutional_router
 from services.database import init_db, get_config, cleanup_stale_running_cycles
 import os
 import logging
@@ -80,6 +81,9 @@ def startup():
     from services.var_scheduler import start_var_scheduler, start_pnl_scheduler
     start_var_scheduler()
     start_pnl_scheduler()
+    # Start institutional reports weekly scheduler (COT Fridays, EIA Wednesdays)
+    from services.institutional_scheduler import start_institutional_scheduler
+    start_institutional_scheduler()
 
 
 @app.on_event("shutdown")
@@ -89,6 +93,8 @@ def shutdown():
     from services.var_scheduler import stop_var_scheduler, stop_pnl_scheduler
     stop_var_scheduler()
     stop_pnl_scheduler()
+    from services.institutional_scheduler import stop_institutional_scheduler
+    stop_institutional_scheduler()
 
 
 app.include_router(market_data.router)
@@ -110,6 +116,7 @@ app.include_router(risk_router.router)
 app.include_router(logs_router.router)
 app.include_router(var_router.router)
 app.include_router(reports_router.router)
+app.include_router(institutional_router.router)
 
 
 @app.get("/")

backend/routers/institutional.py

@@ -0,0 +1,222 @@
+"""
+Institutional reports router — CFTC COT + EIA petroleum weekly.
+GET  /api/institutional/reports        — list with filters
+GET  /api/institutional/reports/{id}   — single report detail
+POST /api/institutional/refresh        — trigger fetch
+GET  /api/institutional/stats          — counts + latest dates
+"""
+import json
+import logging
+from datetime import datetime, timedelta
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, HTTPException, Query
+
+from services.database import get_conn, get_config
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/institutional", tags=["institutional"])
+
+
+# ── DB helpers ─────────────────────────────────────────────────────────────────
+
+def _row_to_dict(row) -> Dict:
+    d = dict(row)
+    for field in ("raw_data_json", "key_points_json"):
+        raw = d.pop(field, None) or ("[]" if "key_points" in field else "{}")
+        try:
+            d[field.replace("_json", "")] = json.loads(raw)
+        except Exception:
+            d[field.replace("_json", "")] = [] if "key_points" in field else {}
+    return d
+
+
+def save_institutional_report(report: Dict) -> Optional[int]:
+    """Insert or update an institutional report. Returns row id."""
+    conn = get_conn()
+    try:
+        existing = conn.execute(
+            "SELECT id FROM institutional_reports WHERE report_type=? AND report_date=?",
+            (report["report_type"], report["report_date"]),
+        ).fetchone()
+
+        raw_json = json.dumps(report.get("raw_data", {}), ensure_ascii=False)
+        kp_json  = json.dumps(report.get("key_points", []), ensure_ascii=False)
+
+        if existing:
+            conn.execute(
+                """UPDATE institutional_reports SET
+                   fetch_date=datetime('now'), raw_data_json=?, key_points_json=?,
+                   trading_implications=?, signal_energy=?, signal_metals=?,
+                   signal_indices=?, signal_forex=?, ai_summary=?, importance=?,
+                   title=?
+                   WHERE id=?""",
+                (
+                    raw_json, kp_json,
+                    report.get("trading_implications", ""),
+                    report.get("signal_energy", "neutral"),
+                    report.get("signal_metals", "neutral"),
+                    report.get("signal_indices", "neutral"),
+                    report.get("signal_forex", "neutral"),
+                    report.get("ai_summary", ""),
+                    report.get("importance", 2),
+                    report.get("title", ""),
+                    existing["id"],
+                ),
+            )
+            conn.commit()
+            return existing["id"]
+
+        cursor = conn.execute(
+            """INSERT INTO institutional_reports
+               (report_type, report_date, title, source, importance, category,
+                raw_data_json, key_points_json, trading_implications,
+                signal_energy, signal_metals, signal_indices, signal_forex, ai_summary)
+               VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
+            (
+                report["report_type"],
+                report["report_date"],
+                report.get("title", ""),
+                report.get("source", ""),
+                report.get("importance", 2),
+                report.get("category", "multi"),
+                raw_json, kp_json,
+                report.get("trading_implications", ""),
+                report.get("signal_energy", "neutral"),
+                report.get("signal_metals", "neutral"),
+                report.get("signal_indices", "neutral"),
+                report.get("signal_forex", "neutral"),
+                report.get("ai_summary", ""),
+            ),
+        )
+        conn.commit()
+        return cursor.lastrowid
+    finally:
+        conn.close()
+
+
+# ── Routes ─────────────────────────────────────────────────────────────────────
+
+@router.get("/reports")
+def list_reports(
+    report_type: Optional[str] = Query(None),
+    category:    Optional[str] = Query(None),
+    importance:  Optional[int] = Query(None),
+    days:        int           = Query(90),
+    limit:       int           = Query(50),
+):
+    conn = get_conn()
+    try:
+        conditions = ["1=1"]
+        params: List[Any] = []
+
+        if report_type:
+            conditions.append("report_type = ?")
+            params.append(report_type)
+        if category:
+            conditions.append("category = ?")
+            params.append(category)
+        if importance:
+            conditions.append("importance >= ?")
+            params.append(importance)
+        if days:
+            cutoff = (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%d")
+            conditions.append("report_date >= ?")
+            params.append(cutoff)
+
+        where = " AND ".join(conditions)
+        rows = conn.execute(
+            f"SELECT * FROM institutional_reports WHERE {where} "
+            f"ORDER BY report_date DESC, importance DESC LIMIT ?",
+            params + [limit],
+        ).fetchall()
+        return [_row_to_dict(r) for r in rows]
+    finally:
+        conn.close()
+
+
+@router.get("/reports/{report_id}")
+def get_report(report_id: int):
+    conn = get_conn()
+    try:
+        row = conn.execute(
+            "SELECT * FROM institutional_reports WHERE id=?", (report_id,)
+        ).fetchone()
+        if not row:
+            raise HTTPException(status_code=404, detail="Report not found")
+        return _row_to_dict(row)
+    finally:
+        conn.close()
+
+
+@router.post("/refresh")
+def refresh_reports(report_type: Optional[str] = Query(None)):
+    """Trigger a live fetch of COT and/or EIA reports."""
+    results: Dict[str, str] = {}
+
+    if not report_type or report_type == "cot":
+        try:
+            from services.cot_fetcher import fetch_cot_report
+            report = fetch_cot_report()
+            if report:
+                save_institutional_report(report)
+                results["cot"] = "ok"
+                logger.info(f"[Institutional] COT report saved: {report['report_date']}")
+            else:
+                results["cot"] = "no_data"
+        except Exception as e:
+            logger.warning(f"[Institutional] COT refresh failed: {e}")
+            results["cot"] = f"error: {str(e)[:100]}"
+
+    if not report_type or report_type == "eia":
+        try:
+            eia_key = get_config("eia_api_key") or ""
+            if eia_key:
+                from services.eia_fetcher import fetch_eia_report
+                report = fetch_eia_report(eia_key)
+                if report:
+                    save_institutional_report(report)
+                    results["eia"] = "ok"
+                    logger.info(f"[Institutional] EIA report saved: {report['report_date']}")
+                else:
+                    results["eia"] = "no_data"
+            else:
+                results["eia"] = "no_api_key"
+        except Exception as e:
+            logger.warning(f"[Institutional] EIA refresh failed: {e}")
+            results["eia"] = f"error: {str(e)[:100]}"
+
+    return {"status": "done", "results": results}
+
+
+@router.get("/stats")
+def get_stats():
+    conn = get_conn()
+    try:
+        total    = conn.execute("SELECT COUNT(*) FROM institutional_reports").fetchone()[0]
+        by_type  = dict(conn.execute(
+            "SELECT report_type, COUNT(*) FROM institutional_reports GROUP BY report_type"
+        ).fetchall())
+        latest_cot = conn.execute(
+            "SELECT MAX(report_date) FROM institutional_reports WHERE report_type='cot'"
+        ).fetchone()[0]
+        latest_eia = conn.execute(
+            "SELECT MAX(report_date) FROM institutional_reports WHERE report_type='eia'"
+        ).fetchone()[0]
+        absorbed = conn.execute(
+            "SELECT COUNT(*) FROM institutional_reports "
+            "WHERE absorbed_score IS NOT NULL AND absorbed_score > 0.3"
+        ).fetchone()[0]
+        high_importance = conn.execute(
+            "SELECT COUNT(*) FROM institutional_reports WHERE importance=3"
+        ).fetchone()[0]
+        return {
+            "total": total,
+            "by_type": by_type,
+            "latest_cot": latest_cot,
+            "latest_eia": latest_eia,
+            "absorbed_count": absorbed,
+            "high_importance_count": high_importance,
+        }
+    finally:
+        conn.close()

backend/services/ai_analyzer.py

@@ -393,6 +393,7 @@ def score_patterns_with_context(
     fred_block: str = "",
     price_discovery_block: str = "",
     portfolio_context_block: str = "",
+    institutional_block: str = "",
     run_id: str = "",
 ) -> List[Dict[str, Any]]:
     """Score all patterns with rich context (news, prices, IV, risk clusters) using GPT-4o."""
@@ -616,6 +617,7 @@ Scoring instructions:
     _fred_sc_section = f"\n{fred_block}\n" if fred_block else ""
     _pd_sc_section = f"\n{price_discovery_block}\n" if price_discovery_block else ""
     _portfolio_sc_section = f"\n{portfolio_context_block}\n" if portfolio_context_block else ""
+    _inst_sc_section = f"\n{institutional_block}\n" if institutional_block else ""
 
     user = f"""GLOBAL CONTEXT:
 - Geopolitical risk score: {geo_score.get('score', 50)}/100 ({geo_score.get('level', 'medium')})
@@ -625,6 +627,7 @@ Scoring instructions:
 {_fred_sc_section}
 {_pd_sc_section}
 {_tech_sc_section}
+{_inst_sc_section}
 {_portfolio_sc_section}
 SCORING TEMPLATE:
 {scoring_template}
@@ -1194,6 +1197,51 @@ def _build_temporal_news_block(partitioned: Dict[str, List], cycle_meta: Dict) -
     return block
 
 
+def build_institutional_block(days: int = 7) -> str:
+    """Build a concise institutional reports block for injection into AI prompts."""
+    try:
+        from services.database import get_conn
+        conn = get_conn()
+        try:
+            from datetime import datetime as _dt, timedelta as _td
+            cutoff = (_dt.utcnow() - _td(days=days)).strftime("%Y-%m-%d")
+            rows = conn.execute(
+                "SELECT report_type, report_date, key_points_json, trading_implications, "
+                "signal_energy, signal_metals, signal_indices, signal_forex, importance "
+                "FROM institutional_reports WHERE report_date >= ? ORDER BY report_date DESC LIMIT 6",
+                (cutoff,),
+            ).fetchall()
+        finally:
+            conn.close()
+
+        if not rows:
+            return ""
+
+        import json as _json
+        lines = ["## INSTITUTIONAL REPORTS (CFTC COT + EIA — last 7 days)"]
+        for r in rows:
+            rtype = r["report_type"].upper()
+            rdate = r["report_date"]
+            importance_str = "★★★" if r["importance"] == 3 else "★★" if r["importance"] == 2 else "★"
+            signals = (
+                f"Energy={r['signal_energy']} | Metals={r['signal_metals']} | "
+                f"Indices={r['signal_indices']} | Forex={r['signal_forex']}"
+            )
+            lines.append(f"\n### {rtype} {rdate} {importance_str} — {signals}")
+            try:
+                kps = _json.loads(r["key_points_json"] or "[]")
+                for kp in kps[:4]:
+                    lines.append(f"  • {kp}")
+            except Exception:
+                pass
+            if r["trading_implications"]:
+                lines.append(f"  → Implications: {r['trading_implications'][:200]}")
+
+        return "\n".join(lines)
+    except Exception as _e:
+        return ""
+
+
 def suggest_patterns_from_market_context(
     news: List[Dict],
     quotes_by_class: Dict[str, List[Dict]],

backend/services/auto_cycle.py

@@ -14,7 +14,7 @@ Cycle steps:
 import logging
 import threading
 import uuid
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Any, Dict, List, Optional
 
 logger = logging.getLogger(__name__)
@@ -551,6 +551,16 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
             except Exception as _cbe:
                 logger.warning(f"[Cycle] Convergence block build failed (non-blocking): {_cbe}")
 
+            # ── Build institutional reports block ──────────────────────────
+            _institutional_block = ""
+            try:
+                from services.ai_analyzer import build_institutional_block
+                _institutional_block = build_institutional_block(days=7)
+                if _institutional_block:
+                    logger.info(f"[Cycle {run_id[:16]}] Institutional block injected")
+            except Exception as _ibe:
+                logger.warning(f"[Cycle] Institutional block failed (non-blocking): {_ibe}")
+
             suggestions = suggest_patterns_from_market_context(
                 news, quotes, calendar, macro_regime=macro_regime, geo_score=geo_score_obj,
                 portfolio_lessons=portfolio_lessons,
@@ -705,6 +715,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
                 fred_block=_fred_block,
                 price_discovery_block=_price_discovery_block,
                 portfolio_context_block=_portfolio_block,
+                institutional_block=_institutional_block,
                 run_id=run_id,
             )
             scored_with_id = [s for s in scored if s.get("pattern_id")]
@@ -991,6 +1002,56 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
         _current_status["last_run_at"] = datetime.utcnow().isoformat()
         logger.info(f"[Cycle {run_id[:16]}] Completed — {added_count} new patterns, {len(scored)} scored")
 
+        # ── Step 6.1: Institutional report absorption tracking ───────────────
+        try:
+            from services.database import get_conn as _get_conn
+            import json as _json
+            _commentary_text = ""
+            if commentary:
+                try:
+                    _c = _json.loads(commentary) if isinstance(commentary, str) else commentary
+                    _commentary_text = " ".join([
+                        str(_c.get("narrative", "")),
+                        str(_c.get("key_catalyst", "")),
+                        " ".join(str(x) for x in (_c.get("risks", []) or [])),
+                    ]).lower()
+                except Exception:
+                    _commentary_text = str(commentary).lower()
+
+            if _commentary_text:
+                _conn = _get_conn()
+                try:
+                    cutoff = (datetime.utcnow() - timedelta(days=14)).strftime("%Y-%m-%d")
+                    _inst_rows = _conn.execute(
+                        "SELECT id, key_points_json FROM institutional_reports "
+                        "WHERE report_date >= ? AND (absorbed_score IS NULL OR absorbed_score = 0)",
+                        (cutoff,),
+                    ).fetchall()
+                    for _ir in _inst_rows:
+                        try:
+                            _kps = _json.loads(_ir["key_points_json"] or "[]")
+                            _kp_words = set(
+                                w for kp in _kps for w in kp.lower().split()
+                                if len(w) > 4
+                            )
+                            if not _kp_words:
+                                continue
+                            _overlap = sum(1 for w in _kp_words if w in _commentary_text)
+                            _score = min(1.0, _overlap / max(len(_kp_words), 1))
+                            if _score > 0:
+                                _conn.execute(
+                                    "UPDATE institutional_reports SET absorbed_score=?, injected_in_cycle_id=? WHERE id=?",
+                                    (round(_score, 3), run_id, _ir["id"]),
+                                )
+                        except Exception:
+                            pass
+                    _conn.commit()
+                    logger.info(f"[Cycle {run_id[:16]}] Absorption tracked for {len(_inst_rows)} institutional reports")
+                finally:
+                    _conn.close()
+        except Exception as _abs_e:
+            logger.warning(f"[Cycle] Absorption tracking failed (non-blocking): {_abs_e}")
+
         # ── Step 6.5: Generate full cycle report ─────────────────────────────
         try:
             _cycle_report = _generate_cycle_report(

backend/services/cot_fetcher.py

@@ -0,0 +1,194 @@
+"""
+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,
+    }

backend/services/database.py

@@ -549,6 +549,33 @@ def init_db():
     except Exception:
         pass
 
+    c.execute("""CREATE TABLE IF NOT EXISTS institutional_reports (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        report_type TEXT NOT NULL,
+        report_date TEXT NOT NULL,
+        fetch_date TEXT NOT NULL DEFAULT (datetime('now')),
+        title TEXT NOT NULL,
+        source TEXT NOT NULL DEFAULT '',
+        importance INTEGER NOT NULL DEFAULT 2,
+        category TEXT NOT NULL DEFAULT 'multi',
+        raw_data_json TEXT DEFAULT '{}',
+        key_points_json TEXT DEFAULT '[]',
+        trading_implications TEXT DEFAULT '',
+        signal_energy TEXT DEFAULT 'neutral',
+        signal_metals TEXT DEFAULT 'neutral',
+        signal_indices TEXT DEFAULT 'neutral',
+        signal_forex TEXT DEFAULT 'neutral',
+        ai_summary TEXT DEFAULT '',
+        injected_in_cycle_id TEXT DEFAULT NULL,
+        absorbed_score REAL DEFAULT NULL,
+        UNIQUE(report_type, report_date)
+    )""")
+    try:
+        c.execute("CREATE INDEX IF NOT EXISTS idx_inst_type_date ON institutional_reports(report_type, report_date DESC)")
+        c.execute("CREATE INDEX IF NOT EXISTS idx_inst_importance ON institutional_reports(importance DESC, report_date DESC)")
+    except Exception:
+        pass
+
     try:
         c.execute("CREATE INDEX IF NOT EXISTS idx_kb_category ON knowledge_base(category, status)")
         c.execute("CREATE INDEX IF NOT EXISTS idx_rs_version ON reasoning_state(version DESC)")

backend/services/eia_fetcher.py

@@ -0,0 +1,165 @@
+"""
+EIA Petroleum Weekly Status Report fetcher.
+Uses EIA API v2 — free key stored in DB config as 'eia_api_key'.
+"""
+import logging
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+EIA_BASE_URL = "https://api.eia.gov/v2/petroleum/stoc/wstk/data/"
+
+EIA_SERIES: Dict[str, Dict[str, str]] = {
+    "WCRSTUS1": {"name": "US Crude Oil Stocks (ex-SPR)", "unit": "Mb"},
+    "WCUOK1A":  {"name": "Cushing, OK Crude Stocks",     "unit": "Mb"},
+    "WGTSTUS1": {"name": "US Total Gasoline Stocks",      "unit": "Mb"},
+    "WDISTUS1": {"name": "US Distillate Fuel Oil Stocks", "unit": "Mb"},
+}
+
+SURPRISE_THRESHOLD_MB = 2.0
+
+
+def _fetch_series(series_id: str, api_key: str, num_weeks: int = 8) -> List[Dict]:
+    params = {
+        "api_key": api_key,
+        "frequency": "weekly",
+        "data[]": "value",
+        "facets[series][]": series_id,
+        "sort[0][column]": "period",
+        "sort[0][direction]": "desc",
+        "length": num_weeks,
+        "offset": 0,
+    }
+    try:
+        resp = requests.get(EIA_BASE_URL, params=params, timeout=20)
+        resp.raise_for_status()
+        return resp.json().get("response", {}).get("data", [])
+    except Exception as e:
+        logger.warning(f"[EIA] Failed to fetch {series_id}: {e}")
+        return []
+
+
+def fetch_eia_report(api_key: str) -> Optional[Dict]:
+    """Fetch latest EIA petroleum weekly stocks. Returns None if key missing or all fetches fail."""
+    if not api_key:
+        logger.warning("[EIA] No API key — skipping")
+        return None
+
+    results: Dict[str, Any] = {}
+    report_dates: List[str] = []
+
+    for sid, meta in EIA_SERIES.items():
+        rows = _fetch_series(sid, api_key)
+        if not rows:
+            continue
+        current = rows[0]
+        prior = rows[1] if len(rows) > 1 else None
+        cur_val = float(current.get("value") or 0)
+        pri_val = float(prior.get("value") or 0) if prior else None
+        wow = round(cur_val - pri_val, 1) if pri_val is not None else None
+
+        results[sid] = {
+            "name": meta["name"],
+            "period": (current.get("period") or "")[:10],
+            "value_mb": round(cur_val, 1),
+            "prior_mb": round(pri_val, 1) if pri_val is not None else None,
+            "wow_change_mb": wow,
+        }
+        if current.get("period"):
+            report_dates.append(current["period"][:10])
+
+    if not results:
+        logger.warning("[EIA] No data returned")
+        return None
+
+    report_date = max(report_dates) if report_dates else datetime.utcnow().strftime("%Y-%m-%d")
+
+    key_points: List[str] = []
+    implications: List[str] = []
+    surprises: List[str] = []
+    signal_energy = "neutral"
+
+    crude = results.get("WCRSTUS1")
+    if crude:
+        wow_str = f"{crude['wow_change_mb']:+.1f} Mb" if crude["wow_change_mb"] is not None else "N/A"
+        kp = f"US crude oil stocks: {crude['value_mb']:.1f} Mb (WoW {wow_str})"
+        if crude["wow_change_mb"] is not None and abs(crude["wow_change_mb"]) >= SURPRISE_THRESHOLD_MB:
+            direction = "draw" if crude["wow_change_mb"] < 0 else "build"
+            kp += f" — significant {direction}"
+            surprises.append("crude")
+            signal_energy = "bullish" if crude["wow_change_mb"] < 0 else "bearish"
+        key_points.append(kp)
+
+    cushing = results.get("WCUOK1A")
+    if cushing:
+        wow_str = f"{cushing['wow_change_mb']:+.1f} Mb" if cushing["wow_change_mb"] is not None else "N/A"
+        key_points.append(f"Cushing, OK: {cushing['value_mb']:.1f} Mb (WoW {wow_str})")
+
+    gasoline = results.get("WGTSTUS1")
+    if gasoline:
+        wow_str = f"{gasoline['wow_change_mb']:+.1f} Mb" if gasoline["wow_change_mb"] is not None else "N/A"
+        kp = f"US gasoline stocks: {gasoline['value_mb']:.1f} Mb (WoW {wow_str})"
+        if gasoline["wow_change_mb"] is not None and abs(gasoline["wow_change_mb"]) >= SURPRISE_THRESHOLD_MB:
+            surprises.append("gasoline")
+        key_points.append(kp)
+
+    distillates = results.get("WDISTUS1")
+    if distillates:
+        wow_str = f"{distillates['wow_change_mb']:+.1f} Mb" if distillates["wow_change_mb"] is not None else "N/A"
+        kp = f"US distillate stocks: {distillates['value_mb']:.1f} Mb (WoW {wow_str})"
+        if distillates["wow_change_mb"] is not None and abs(distillates["wow_change_mb"]) >= SURPRISE_THRESHOLD_MB:
+            surprises.append("distillates")
+        key_points.append(kp)
+
+    crude_wow = (crude or {}).get("wow_change_mb")
+    if crude_wow is not None:
+        if crude_wow < -SURPRISE_THRESHOLD_MB:
+            implications.append(
+                f"Crude draw of {abs(crude_wow):.1f} Mb — bullish for WTI/Brent spot"
+            )
+        elif crude_wow > SURPRISE_THRESHOLD_MB:
+            implications.append(
+                f"Crude build of {crude_wow:.1f} Mb — bearish for WTI/Brent spot"
+            )
+
+    gasoline_wow = (gasoline or {}).get("wow_change_mb")
+    if gasoline_wow is not None and abs(gasoline_wow) >= SURPRISE_THRESHOLD_MB:
+        demand_str = "strong" if gasoline_wow < 0 else "weak"
+        implications.append(
+            f"Gasoline {'draw' if gasoline_wow < 0 else 'build'} ({gasoline_wow:+.1f} Mb) — {demand_str} driving demand"
+        )
+
+    if not implications:
+        implications = ["EIA petroleum stocks in line with seasonal norms — no major surprise"]
+
+    importance = 3 if len(surprises) >= 2 else 2 if len(surprises) == 1 else 1
+    crude_val_str = f"{crude['value_mb']:.1f}" if crude else "N/A"
+    crude_wow_str = f" (WoW {crude['wow_change_mb']:+.1f} Mb)" if crude and crude["wow_change_mb"] is not None else ""
+
+    ai_summary = (
+        f"EIA Petroleum Weekly ({report_date}). "
+        f"Crude: {crude_val_str} Mb{crude_wow_str}. "
+        f"Signal: {signal_energy.upper()}. "
+        + (f"Surprises: {', '.join(surprises)}. " if surprises else "")
+        + implications[0]
+    )
+
+    return {
+        "report_type": "eia",
+        "report_date": report_date,
+        "title": f"EIA Petroleum Weekly — {report_date}",
+        "source": "EIA API v2",
+        "importance": importance,
+        "category": "energy",
+        "raw_data": results,
+        "key_points": key_points,
+        "trading_implications": " | ".join(implications),
+        "signal_energy": signal_energy,
+        "signal_metals": "neutral",
+        "signal_indices": "neutral",
+        "signal_forex": "neutral",
+        "ai_summary": ai_summary,
+    }

backend/services/institutional_scheduler.py

@@ -0,0 +1,104 @@
+"""
+Weekly scheduler for institutional reports:
+- CFTC COT: released every Friday ~15:30 ET → fetch Saturday UTC
+- EIA Petroleum Weekly: released every Wednesday ~10:30 ET → fetch Wednesday afternoon UTC
+Checks once per hour; only fetches when the day matches and report not yet fetched this week.
+"""
+import logging
+import threading
+from datetime import datetime, timedelta
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+_stop_event = threading.Event()
+_thread: Optional[threading.Thread] = None
+_CHECK_INTERVAL_S = 3600  # check every hour
+
+
+def _should_fetch_cot() -> bool:
+    """Saturday UTC = day after COT release."""
+    now = datetime.utcnow()
+    return now.weekday() == 5  # Saturday
+
+
+def _should_fetch_eia() -> bool:
+    """Wednesday afternoon UTC."""
+    now = datetime.utcnow()
+    return now.weekday() == 2 and now.hour >= 16  # Wednesday ≥16:00 UTC
+
+
+def _last_fetch_date(report_type: str) -> Optional[str]:
+    try:
+        from services.database import get_conn
+        conn = get_conn()
+        try:
+            row = conn.execute(
+                "SELECT MAX(fetch_date) FROM institutional_reports WHERE report_type=?",
+                (report_type,),
+            ).fetchone()
+            return (row[0] or "")[:10] if row else None
+        finally:
+            conn.close()
+    except Exception:
+        return None
+
+
+def _run_loop():
+    logger.info("[InstitutionalScheduler] Started")
+    while not _stop_event.is_set():
+        try:
+            today = datetime.utcnow().strftime("%Y-%m-%d")
+
+            if _should_fetch_cot():
+                last = _last_fetch_date("cot")
+                if last != today:
+                    logger.info("[InstitutionalScheduler] Fetching COT...")
+                    try:
+                        from services.cot_fetcher import fetch_cot_report
+                        from routers.institutional import save_institutional_report
+                        report = fetch_cot_report()
+                        if report:
+                            save_institutional_report(report)
+                            logger.info(f"[InstitutionalScheduler] COT saved: {report['report_date']}")
+                    except Exception as e:
+                        logger.warning(f"[InstitutionalScheduler] COT fetch failed: {e}")
+
+            if _should_fetch_eia():
+                last = _last_fetch_date("eia")
+                if last != today:
+                    logger.info("[InstitutionalScheduler] Fetching EIA...")
+                    try:
+                        from services.database import get_config
+                        from services.eia_fetcher import fetch_eia_report
+                        from routers.institutional import save_institutional_report
+                        key = get_config("eia_api_key") or ""
+                        if key:
+                            report = fetch_eia_report(key)
+                            if report:
+                                save_institutional_report(report)
+                                logger.info(f"[InstitutionalScheduler] EIA saved: {report['report_date']}")
+                        else:
+                            logger.info("[InstitutionalScheduler] EIA skipped — no API key configured")
+                    except Exception as e:
+                        logger.warning(f"[InstitutionalScheduler] EIA fetch failed: {e}")
+
+        except Exception as e:
+            logger.warning(f"[InstitutionalScheduler] Loop error: {e}")
+
+        _stop_event.wait(_CHECK_INTERVAL_S)
+
+    logger.info("[InstitutionalScheduler] Stopped")
+
+
+def start_institutional_scheduler():
+    global _thread
+    _stop_event.clear()
+    _thread = threading.Thread(target=_run_loop, daemon=True, name="institutional-scheduler")
+    _thread.start()
+
+
+def stop_institutional_scheduler():
+    _stop_event.set()
+    if _thread and _thread.is_alive():
+        _thread.join(timeout=5)