diff --git a/backend/main.py b/backend/main.py index d230b96..9465a8a 100644 --- a/backend/main.py +++ b/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("/") diff --git a/backend/routers/institutional.py b/backend/routers/institutional.py new file mode 100644 index 0000000..3cb4550 --- /dev/null +++ b/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() diff --git a/backend/services/ai_analyzer.py b/backend/services/ai_analyzer.py index 7eb9d3f..ca89844 100644 --- a/backend/services/ai_analyzer.py +++ b/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]], diff --git a/backend/services/auto_cycle.py b/backend/services/auto_cycle.py index cedb841..36f49b7 100644 --- a/backend/services/auto_cycle.py +++ b/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( diff --git a/backend/services/cot_fetcher.py b/backend/services/cot_fetcher.py new file mode 100644 index 0000000..a6f6e02 --- /dev/null +++ b/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, + } diff --git a/backend/services/database.py b/backend/services/database.py index eb9f855..59de684 100644 --- a/backend/services/database.py +++ b/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)") diff --git a/backend/services/eia_fetcher.py b/backend/services/eia_fetcher.py new file mode 100644 index 0000000..faf58da --- /dev/null +++ b/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, + } diff --git a/backend/services/institutional_scheduler.py b/backend/services/institutional_scheduler.py new file mode 100644 index 0000000..c13f768 --- /dev/null +++ b/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) diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 996debd..68a9176 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -19,6 +19,7 @@ import RiskDashboard from './pages/RiskDashboard' import SystemLogs from './pages/SystemLogs' import VaRAnalysis from './pages/VaRAnalysis' import PositionHistory from './pages/PositionHistory' +import InstitutionalReports from './pages/InstitutionalReports' import { useCycleWatcher } from './hooks/useApi' function GlobalWatcher() { @@ -53,6 +54,7 @@ export default function App() { } /> } /> } /> + } /> diff --git a/frontend/src/components/layout/Sidebar.tsx b/frontend/src/components/layout/Sidebar.tsx index 26cc0b3..6a93692 100644 --- a/frontend/src/components/layout/Sidebar.tsx +++ b/frontend/src/components/layout/Sidebar.tsx @@ -1,7 +1,7 @@ import { NavLink } from 'react-router-dom' import { LayoutDashboard, Globe, BarChart2, FlaskConical, - History, Calendar, TrendingUp, Zap, DollarSign, Settings, BrainCircuit, Activity, BookOpen, FileBarChart, Brain, ShieldAlert, Microscope, ScrollText, Gauge, GitCompare + History, Calendar, TrendingUp, Zap, DollarSign, Settings, BrainCircuit, Activity, BookOpen, FileBarChart, Brain, ShieldAlert, Microscope, ScrollText, Gauge, GitCompare, Building2 } from 'lucide-react' import { useGeoRiskScore, useAiStatus, usePortfolioSummary } from '../../hooks/useApi' import clsx from 'clsx' @@ -24,6 +24,7 @@ const nav = [ { to: '/position-history', icon: GitCompare, label: 'Position History' }, { to: '/backtest', icon: History, label: 'Backtest' }, { to: '/calendar', icon: Calendar, label: 'Calendar' }, + { to: '/institutional', icon: Building2, label: 'Inst. Reports' }, { to: '/logs', icon: ScrollText, label: 'System Logs' }, { to: '/config', icon: Settings, label: 'Configuration' }, ] diff --git a/frontend/src/hooks/useApi.ts b/frontend/src/hooks/useApi.ts index a27d65c..682ceee 100644 --- a/frontend/src/hooks/useApi.ts +++ b/frontend/src/hooks/useApi.ts @@ -914,3 +914,32 @@ export const useRemoveWatchlistTicker = () => useMutation({ mutationFn: (ticker: string) => api.delete(`/options-vol/watchlist-tickers/${encodeURIComponent(ticker)}`).then(r => r.data), }) + + +// ── Institutional Reports ───────────────────────────────────────────────────── + +export const useInstitutionalReports = (params: { + report_type?: string + category?: string + importance?: number + days?: number +}) => + useQuery({ + queryKey: ['institutional-reports', params], + queryFn: () => + api.get('/institutional/reports', { params }).then(r => r.data), + refetchInterval: 3_600_000, + }) + +export const useInstitutionalStats = () => + useQuery({ + queryKey: ['institutional-stats'], + queryFn: () => api.get('/institutional/stats').then(r => r.data), + staleTime: 300_000, + }) + +export const useRefreshInstitutionalReports = () => + useMutation({ + mutationFn: (report_type?: string) => + api.post('/institutional/refresh', null, { params: report_type ? { report_type } : {} }).then(r => r.data), + }) diff --git a/frontend/src/pages/InstitutionalReports.tsx b/frontend/src/pages/InstitutionalReports.tsx new file mode 100644 index 0000000..15818e8 --- /dev/null +++ b/frontend/src/pages/InstitutionalReports.tsx @@ -0,0 +1,366 @@ +import { useState } from 'react' +import { useInstitutionalReports, useInstitutionalStats, useRefreshInstitutionalReports } from '../hooks/useApi' +import { useQueryClient } from '@tanstack/react-query' +import { RefreshCw, TrendingUp, TrendingDown, Minus, AlertTriangle, Star, CheckCircle, Clock } from 'lucide-react' +import clsx from 'clsx' +import { format } from 'date-fns' + +interface InstitutionalReport { + id: number + report_type: string + report_date: string + fetch_date: string + title: string + source: string + importance: number + category: string + key_points: string[] + trading_implications: string + signal_energy: string + signal_metals: string + signal_indices: string + signal_forex: string + ai_summary: string + absorbed_score: number | null + injected_in_cycle_id: string | null +} + +const SIGNAL_CONFIG: Record = { + bullish: { label: 'Bullish', color: 'text-emerald-400 bg-emerald-900/30 border-emerald-700/40', icon: }, + bearish: { label: 'Bearish', color: 'text-red-400 bg-red-900/30 border-red-700/40', icon: }, + neutral: { label: 'Neutral', color: 'text-slate-400 bg-slate-800/50 border-slate-700/40', icon: }, +} + +const ASSET_CLASSES = ['energy', 'metals', 'indices', 'forex'] as const + +function SignalBadge({ signal, label }: { signal: string; label: string }) { + const cfg = SIGNAL_CONFIG[signal] ?? SIGNAL_CONFIG.neutral + return ( + + {cfg.icon} + {label}: + {cfg.label} + + ) +} + +function ImportanceStars({ n }: { n: number }) { + return ( + + {[1, 2, 3].map(i => ( + + ))} + + ) +} + +function AbsorptionBadge({ score }: { score: number | null }) { + if (score === null || score === undefined) { + return ( + + + Not absorbed + + ) + } + const pct = Math.round(score * 100) + const isAbsorbed = pct >= 30 + return ( + + + {isAbsorbed ? `Absorbed ${pct}%` : `Low ${pct}%`} + + ) +} + +function ReportCard({ report }: { report: InstitutionalReport }) { + const [expanded, setExpanded] = useState(false) + const isCot = report.report_type === 'cot' + const typeColor = isCot ? 'bg-purple-900/30 text-purple-300 border-purple-700/40' : 'bg-orange-900/30 text-orange-300 border-orange-700/40' + + return ( +
+ {/* Header */} +
+
+
+ + {report.report_type} + + + + {report.report_date} + + +
+
{report.title}
+
{report.source}
+
+ +
+ + {/* Signal badges row */} +
+ + + + +
+ + {/* Key points preview (always visible) */} + {report.key_points?.length > 0 && ( +
+
    + {(expanded ? report.key_points : report.key_points.slice(0, 3)).map((kp, i) => { + const isExtreme = kp.includes('EXTREME') + return ( +
  • + {isExtreme + ? + : • + } + {kp} +
    • + ) + })} + {!expanded && report.key_points.length > 3 && ( +
  • +{report.key_points.length - 3} more…
    • + )} +
+
+ )} + + {/* Expanded: trading implications + AI summary */} + {expanded && ( +
+ {report.trading_implications && ( +
+
+ Trading Implications +
+
+ {report.trading_implications.split(' | ').map((impl, i) => ( +

{impl}

+ ))} +
+
+ )} + {report.ai_summary && ( +
+
+ AI Summary +
+

{report.ai_summary}

+
+ )} + {report.injected_in_cycle_id && ( +
+ Injected in cycle: {report.injected_in_cycle_id.slice(0, 16)} +
+ )} +
+ )} +
+ ) +} + +export default function InstitutionalReports() { + const qc = useQueryClient() + const [filterType, setFilterType] = useState('') + const [filterCategory, setFilterCategory] = useState('') + const [filterImportance, setFilterImportance] = useState(0) + const [filterDays, setFilterDays] = useState(90) + + const params = { + ...(filterType ? { report_type: filterType } : {}), + ...(filterCategory ? { category: filterCategory } : {}), + ...(filterImportance ? { importance: filterImportance } : {}), + days: filterDays, + } + + const { data: reports = [], isLoading, isFetching } = useInstitutionalReports(params) + const { data: stats } = useInstitutionalStats() + const refresh = useRefreshInstitutionalReports() + + const handleRefresh = (type?: string) => { + refresh.mutate(type, { + onSuccess: () => { + qc.invalidateQueries({ queryKey: ['institutional-reports'] }) + qc.invalidateQueries({ queryKey: ['institutional-stats'] }) + }, + }) + } + + return ( +
+ {/* Header */} +
+
+

Institutional Reports

+

+ CFTC COT (weekly) + EIA Petroleum (weekly) — positioning & supply signals +

+
+
+ + +
+
+ + {/* Stats bar */} + {stats && ( +
+ {[ + { label: 'Total reports', value: stats.total }, + { label: 'COT reports', value: stats.by_type?.cot ?? 0 }, + { label: 'EIA reports', value: stats.by_type?.eia ?? 0 }, + { label: 'Absorbed (>30%)', value: stats.absorbed_count }, + ].map(({ label, value }) => ( +
+
{value}
+
{label}
+
+ ))} +
+ )} + + {/* Latest dates */} + {stats && (stats.latest_cot || stats.latest_eia) && ( +
+ {stats.latest_cot && Latest COT: {stats.latest_cot}} + {stats.latest_eia && Latest EIA: {stats.latest_eia}} +
+ )} + + {/* Filters */} +
+
+ Type: +
+ {['', 'cot', 'eia'].map(t => ( + + ))} +
+
+ +
+ Category: +
+ {['', 'energy', 'metals', 'equities', 'forex', 'multi'].map(c => ( + + ))} +
+
+ +
+ Min importance: +
+ {[0, 1, 2, 3].map(n => ( + + ))} +
+
+ +
+ Period: + +
+
+ + {/* Report list */} + {isLoading || isFetching ? ( +
+ + Loading reports… +
+ ) : reports.length === 0 ? ( +
+
📋
+
No institutional reports yet
+

+ Click Refresh COT to fetch the latest CFTC positioning data, + or configure your EIA API key in Configuration to enable petroleum reports. +

+
+ ) : ( +
+
{reports.length} report{reports.length > 1 ? 's' : ''}
+ {(reports as InstitutionalReport[]).map(r => ( + + ))} +
+ )} +
+ ) +}