""" 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 - Earnings / VX curve / Central bank RSS / Sentiment: fetch daily (after 8am UTC) Checks once per hour; only fetches when the day matches and report not yet fetched today. """ 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: now = datetime.utcnow() return now.weekday() == 5 # Saturday def _should_fetch_eia() -> bool: now = datetime.utcnow() return now.weekday() == 2 and now.hour >= 16 # Wednesday ≥16:00 UTC def _should_fetch_daily() -> bool: """Earnings, VX, central bank, sentiment — fetch once per day after 8am UTC.""" return datetime.utcnow().hour >= 8 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 _fetch_and_save(label: str, fetch_fn, *args): """Generic helper: call fetch_fn(*args), save result if not None.""" try: from routers.institutional import save_institutional_report report = fetch_fn(*args) if report: save_institutional_report(report) logger.info(f"[InstitutionalScheduler] {label} saved: {report['report_date']}") else: logger.info(f"[InstitutionalScheduler] {label} returned no data") except Exception as e: logger.warning(f"[InstitutionalScheduler] {label} fetch failed: {e}") def _run_loop(): logger.info("[InstitutionalScheduler] Started") while not _stop_event.is_set(): try: today = datetime.utcnow().strftime("%Y-%m-%d") # ── Weekly: COT (Saturday) ────────────────────────────────────── if _should_fetch_cot() and _last_fetch_date("cot") != today: logger.info("[InstitutionalScheduler] Fetching COT...") from services.cot_fetcher import fetch_cot_report _fetch_and_save("COT", fetch_cot_report) # ── Weekly: EIA (Wednesday ≥16:00 UTC) ───────────────────────── if _should_fetch_eia() and _last_fetch_date("eia") != today: logger.info("[InstitutionalScheduler] Fetching EIA...") try: from services.database import get_config from services.eia_fetcher import fetch_eia_report key = get_config("eia_api_key") or "" if key: _fetch_and_save("EIA", fetch_eia_report, key) else: logger.info("[InstitutionalScheduler] EIA skipped — no API key configured") except Exception as e: logger.warning(f"[InstitutionalScheduler] EIA fetch failed: {e}") # ── Daily: Earnings, VX curve, Central banks, Sentiment ───────── if _should_fetch_daily(): if _last_fetch_date("earnings") != today: logger.info("[InstitutionalScheduler] Fetching Earnings...") from services.earnings_fetcher import fetch_earnings_report _fetch_and_save("Earnings", fetch_earnings_report) if _last_fetch_date("vx_curve") != today: logger.info("[InstitutionalScheduler] Fetching VX term structure...") from services.vx_fetcher import fetch_vx_report _fetch_and_save("VX", fetch_vx_report) if _last_fetch_date("central_bank") != today: logger.info("[InstitutionalScheduler] Fetching Central Bank RSS...") from services.central_bank_fetcher import fetch_central_bank_reports _fetch_and_save("CentralBank", fetch_central_bank_reports) if _last_fetch_date("sentiment") != today: logger.info("[InstitutionalScheduler] Fetching Sentiment...") from services.sentiment_fetcher import fetch_sentiment_report _fetch_and_save("Sentiment", fetch_sentiment_report) 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)