OpenFin/backend/services/institutional_scheduler.py

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