"""Single ingest worker: claims jobs from Postgres, holds the lake writer lock, runs the task registry. Also hosts the scheduler loop (idempotent enqueue with dedupe keys) so there is exactly one writer process.""" from __future__ import annotations import signal import time import traceback from datetime import datetime from zoneinfo import ZoneInfo from findata_core.config import get_settings from findata_core.logging import configure_logging, get_logger from findata_ingest import queue from findata_ingest.lake import Lake, WriterLock from findata_ingest.tasks import REGISTRY ET = ZoneInfo("America/New_York") _stop = False BULK_KINDS = ["backfill", "options", "fundamentals"] PRIMARY_KINDS = ["metadata", "incremental", "derive", "compaction", "backup", "quality", "gc"] def _handle(sig, frame): global _stop _stop = True def schedule_tick(log) -> None: """Enqueue recurring jobs (dedupe keys make this idempotent). Times in America/New_York.""" now = datetime.now(ET) day = now.date().isoformat() stale = queue.requeue_stale() if stale: log.warning("scheduler.requeued_stale", n=stale) # vendor daily update lands ~23:30–03:00 ET → poll from 02:00 every 30 min (incremental skips when vendor unchanged) if now.weekday() < 6 and (now.hour >= 2 or now.hour < 1): queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "day"}, priority=20, dedupe_key=f"incremental:stock:day:{day}:{now.hour}:{now.minute // 30}", rerun_succeeded=False) if now.weekday() == 5 and now.hour == 6: # Saturday 06:00 ET — week archive repairs any missed day queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "week", "force": True}, priority=25, dedupe_key=f"incremental:stock:week:{day}", rerun_succeeded=False) if now.weekday() == 6 and now.hour == 7: queue.enqueue("metadata", None, {"types": ["stock", "etf"]}, priority=30, dedupe_key=f"metadata:{day}", rerun_succeeded=False) queue.enqueue("compaction", "market_bars.stock.1m.raw", {}, priority=60, dedupe_key=f"compaction:1m:{day}", rerun_succeeded=False) queue.enqueue("compaction", "market_bars.stock.5m.raw", {}, priority=61, dedupe_key=f"compaction:5m:{day}", rerun_succeeded=False) # other raw 1-minute feeds and daily bars (vendor publishes daily; crypto 7/7) if now.hour >= 2 or now.hour < 1: for t in ("index", "fx", "crypto"): if now.weekday() < 6 or t == "crypto": queue.enqueue("incremental", f"market_bars.{t}.1m.raw", {"type": t, "period": "day", "timeframe": "1min"}, priority=26, dedupe_key=f"incremental:{t}:day:{day}:{now.hour}:{now.minute // 30}", rerun_succeeded=False) if now.weekday() < 6: for t in ("stock", "etf"): queue.enqueue("incremental", f"market_bars.{t}.1d.raw", {"type": t, "period": "day", "timeframe": "1day", "adjustment": "UNADJUSTED"}, priority=27, dedupe_key=f"incremental:{t}:1day:{day}:{now.hour}:{now.minute // 30}", rerun_succeeded=False) queue.enqueue("incremental", "market_bars.etf.1m.raw", {"type": "etf", "period": "day"}, priority=21, dedupe_key=f"incremental:etf:day:{day}:{now.hour}:{now.minute // 30}", rerun_succeeded=False) if now.weekday() == 5 and now.hour == 6: # weekly: futures continuous (week), individual contracts update, adjusted stock/etf histories (full_update) for tf in ("1min", "1day", "5min", "30min", "1hour"): for adj in ("contin_UNadj", "contin_adj_ratio", "contin_adj_absolute"): queue.enqueue("incremental", None, {"type": "futures", "period": "week", "timeframe": tf, "adjustment": adj}, priority=36, dedupe_key=f"incremental:futures:week:{tf}:{adj}:{day}", rerun_succeeded=False) queue.enqueue("backfill", None, {"type": "futures", "endpoint": "futures_contract", "contract_files": "update", "timeframe": tf, "contracts": True}, priority=37, dedupe_key=f"contracts:update:{tf}:{day}", rerun_succeeded=False) for t in ("stock", "etf"): for adj in ("adj_split", "adj_splitdiv"): queue.enqueue("backfill", None, {"type": t, "endpoint": "full_update", "period": "week", "timeframe": tf, "adjustment": adj}, priority=38, dedupe_key=f"full_update:{t}:{tf}:{adj}:{day}", rerun_succeeded=False) for t in ("index", "fx", "crypto"): if tf != "1min": queue.enqueue("incremental", None, {"type": t, "period": "week", "timeframe": tf}, priority=39, dedupe_key=f"incremental:{t}:week:{tf}:{day}", rerun_succeeded=False) queue.enqueue("backfill", "market_bars.stock.1m.raw", {"type": "stock", "endpoint": "delisted_data_file", "update": "week", "timeframe": "1min", "adjustment": "UNADJUSTED"}, priority=38, dedupe_key=f"delisted:week:{day}", rerun_succeeded=False) if now.day == 2 and now.hour == 8: queue.enqueue("options", "options.eod", {"period": "month"}, priority=55, dedupe_key=f"options:month:{now.year}-{now.month:02d}", rerun_succeeded=False) if now.hour == 5 and now.minute < 30: queue.enqueue("fundamentals", "fundamentals.financials", {"tracked": True}, priority=50, dedupe_key=f"fundamentals:{day}", rerun_succeeded=False) if now.hour == 4 and now.minute < 30 and get_settings().backup_target: queue.enqueue("backup", None, {}, priority=70, dedupe_key=f"backup:{day}", rerun_succeeded=False) if now.hour == 6 and now.minute < 30: queue.enqueue("quality", "market_bars.stock.1m.raw", {}, priority=80, dedupe_key=f"quality:{day}", rerun_succeeded=False) queue.enqueue("gc", None, {}, priority=90, dedupe_key=f"gc:{day}", rerun_succeeded=False) def main(once: bool = False, kinds: list[str] | None = None, with_scheduler: bool = True, role: str = "primary") -> None: """role=primary: holds the lake lock, runs the scheduler and the serialized kinds; role=bulk: one of N parallel workers for backfill/options/fundamentals (publication is serialized by Postgres row locks).""" configure_logging() log = get_logger("worker") signal.signal(signal.SIGTERM, _handle) signal.signal(signal.SIGINT, _handle) signal.signal(signal.SIGHUP, _handle) # pty hang-up when run under sshd (see infra/scripts/run-under-sshd.sh) lake = Lake() if role == "bulk": _loop(lake, log, once, kinds or BULK_KINDS, False) return requeued = queue.requeue_stale() if requeued: log.warning("worker.requeued_stale", n=requeued) if kinds is None: kinds = PRIMARY_KINDS if get_settings().bulk_workers > 0 else None with WriterLock(lake.root): _loop(lake, log, once, kinds, with_scheduler) log.info("worker.stop") def _loop(lake: Lake, log, once: bool, kinds: list[str] | None, with_scheduler: bool) -> None: last_tick = 0.0 if True: log.info("worker.start", lake=str(lake.root), cold=str(lake.cold), free_gb=round(lake.free_gb(), 1), kinds=kinds) while not _stop: if with_scheduler and time.monotonic() - last_tick > 300: try: schedule_tick(log) except Exception as e: log.error("scheduler.error", error=str(e)) last_tick = time.monotonic() job = queue.claim_next(kinds) if job is None: if once: break time.sleep(5) continue jl = log.bind(job_id=str(job["id"]), kind=job["kind"], dataset=job.get("dataset_id"), attempt=job["attempts"]) jl.info("job.start") t0 = time.monotonic() try: fn = REGISTRY[job["kind"]] result = fn(job, lake, jl) queue.finish(job, True, {**(result or {}), "seconds": round(time.monotonic() - t0, 1)}) jl.info("job.done", seconds=round(time.monotonic() - t0, 1)) except Exception as e: err = f"{type(e).__name__}: {e}\n{traceback.format_exc()[-1500:]}" queue.finish(job, False, error=err) jl.error("job.failed", error=str(e)[:300], seconds=round(time.monotonic() - t0, 1)) if once: break