1"""Single ingest worker: claims jobs from Postgres, holds the lake writer lock, runs the task registry.2Also hosts the scheduler loop (idempotent enqueue with dedupe keys) so there is exactly one writer process."""3from __future__ import annotations45import signal6import time7import traceback8from datetime import datetime9from zoneinfo import ZoneInfo1011from findata_core.config import get_settings12from findata_core.logging import configure_logging, get_logger1314from findata_ingest import queue15from findata_ingest.lake import Lake, WriterLock16from findata_ingest.tasks import REGISTRY1718ET = ZoneInfo("America/New_York")19_stop = False20BULK_KINDS = ["backfill", "options", "fundamentals"]21PRIMARY_KINDS = ["metadata", "incremental", "derive", "compaction", "backup", "quality", "gc"]222324def _handle(sig, frame):25 global _stop26 _stop = True272829def schedule_tick(log) -> None:30 """Enqueue recurring jobs (dedupe keys make this idempotent). Times in America/New_York."""31 now = datetime.now(ET)32 day = now.date().isoformat()33 # vendor daily update lands ~23:30–03:00 ET → poll from 02:00 every 30 min (incremental skips when vendor unchanged)34 if now.weekday() < 6 and (now.hour >= 2 or now.hour < 1):35 queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "day"}, priority=20,36 dedupe_key=f"incremental:stock:day:{day}:{now.hour}:{now.minute // 30}")37 if now.weekday() == 5 and now.hour == 6: # Saturday 06:00 ET — week archive repairs any missed day38 queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "week", "force": True}, priority=25,39 dedupe_key=f"incremental:stock:week:{day}")40 if now.weekday() == 6 and now.hour == 7:41 queue.enqueue("metadata", None, {"types": ["stock", "etf"]}, priority=30, dedupe_key=f"metadata:{day}")42 queue.enqueue("compaction", "market_bars.stock.1m.raw", {}, priority=60, dedupe_key=f"compaction:1m:{day}")43 queue.enqueue("compaction", "market_bars.stock.5m.raw", {}, priority=61, dedupe_key=f"compaction:5m:{day}")44 # other raw 1-minute feeds and daily bars (vendor publishes daily; crypto 7/7)45 if now.hour >= 2 or now.hour < 1:46 for t in ("index", "fx", "crypto"):47 if now.weekday() < 6 or t == "crypto":48 queue.enqueue("incremental", f"market_bars.{t}.1m.raw", {"type": t, "period": "day", "timeframe": "1min"}, priority=26,49 dedupe_key=f"incremental:{t}:day:{day}:{now.hour}:{now.minute // 30}")50 if now.weekday() < 6:51 for t in ("stock", "etf"):52 queue.enqueue("incremental", f"market_bars.{t}.1d.raw", {"type": t, "period": "day", "timeframe": "1day", "adjustment": "UNADJUSTED"}, priority=27,53 dedupe_key=f"incremental:{t}:1day:{day}:{now.hour}:{now.minute // 30}")54 queue.enqueue("incremental", "market_bars.etf.1m.raw", {"type": "etf", "period": "day"}, priority=21,55 dedupe_key=f"incremental:etf:day:{day}:{now.hour}:{now.minute // 30}")56 if now.weekday() == 5 and now.hour == 6:57 # weekly: futures continuous (week), individual contracts update, adjusted stock/etf histories (full_update)58 for tf in ("1min", "1day", "5min", "30min", "1hour"):59 for adj in ("contin_UNadj", "contin_adj_ratio", "contin_adj_absolute"):60 queue.enqueue("incremental", None, {"type": "futures", "period": "week", "timeframe": tf, "adjustment": adj}, priority=36,61 dedupe_key=f"incremental:futures:week:{tf}:{adj}:{day}")62 queue.enqueue("backfill", None, {"type": "futures", "endpoint": "futures_contract", "contract_files": "update", "timeframe": tf, "contracts": True},63 priority=37, dedupe_key=f"contracts:update:{tf}:{day}")64 for t in ("stock", "etf"):65 for adj in ("adj_split", "adj_splitdiv"):66 queue.enqueue("backfill", None, {"type": t, "endpoint": "full_update", "period": "week", "timeframe": tf, "adjustment": adj},67 priority=38, dedupe_key=f"full_update:{t}:{tf}:{adj}:{day}")68 for t in ("index", "fx", "crypto"):69 if tf != "1min":70 queue.enqueue("incremental", None, {"type": t, "period": "week", "timeframe": tf}, priority=39, dedupe_key=f"incremental:{t}:week:{tf}:{day}")71 queue.enqueue("backfill", "market_bars.stock.1m.raw", {"type": "stock", "endpoint": "delisted_data_file", "update": "week", "timeframe": "1min", "adjustment": "UNADJUSTED"},72 priority=38, dedupe_key=f"delisted:week:{day}")73 if now.day == 2 and now.hour == 8:74 queue.enqueue("options", "options.eod", {"period": "month"}, priority=55, dedupe_key=f"options:month:{now.year}-{now.month:02d}")75 if now.hour == 5 and now.minute < 30:76 queue.enqueue("fundamentals", "fundamentals.financials", {"tracked": True}, priority=50, dedupe_key=f"fundamentals:{day}")77 if now.hour == 4 and now.minute < 30 and get_settings().backup_target:78 queue.enqueue("backup", None, {}, priority=70, dedupe_key=f"backup:{day}")79 if now.hour == 6 and now.minute < 30:80 queue.enqueue("quality", "market_bars.stock.1m.raw", {}, priority=80, dedupe_key=f"quality:{day}")81 queue.enqueue("gc", None, {}, priority=90, dedupe_key=f"gc:{day}")828384def main(once: bool = False, kinds: list[str] | None = None, with_scheduler: bool = True, role: str = "primary") -> None:85 """role=primary: holds the lake lock, runs the scheduler and the serialized kinds; role=bulk: one of N86 parallel workers for backfill/options/fundamentals (publication is serialized by Postgres row locks)."""87 configure_logging()88 log = get_logger("worker")89 signal.signal(signal.SIGTERM, _handle)90 signal.signal(signal.SIGINT, _handle)91 signal.signal(signal.SIGHUP, _handle) # pty hang-up when run under sshd (see infra/scripts/run-under-sshd.sh)92 lake = Lake()93 if role == "bulk":94 _loop(lake, log, once, kinds or BULK_KINDS, False)95 return96 requeued = queue.requeue_stale()97 if requeued:98 log.warning("worker.requeued_stale", n=requeued)99 if kinds is None:100 kinds = PRIMARY_KINDS if get_settings().bulk_workers > 0 else None101 with WriterLock(lake.root):102 _loop(lake, log, once, kinds, with_scheduler)103 log.info("worker.stop")104105106def _loop(lake: Lake, log, once: bool, kinds: list[str] | None, with_scheduler: bool) -> None:107 last_tick = 0.0108 if True:109 log.info("worker.start", lake=str(lake.root), cold=str(lake.cold), free_gb=round(lake.free_gb(), 1), kinds=kinds)110 while not _stop:111 if with_scheduler and time.monotonic() - last_tick > 300:112 try:113 schedule_tick(log)114 except Exception as e:115 log.error("scheduler.error", error=str(e))116 last_tick = time.monotonic()117 job = queue.claim_next(kinds)118 if job is None:119 if once:120 break121 time.sleep(5)122 continue123 jl = log.bind(job_id=str(job["id"]), kind=job["kind"], dataset=job.get("dataset_id"), attempt=job["attempts"])124 jl.info("job.start")125 t0 = time.monotonic()126 try:127 fn = REGISTRY[job["kind"]]128 result = fn(job, lake, jl)129 queue.finish(job, True, {**(result or {}), "seconds": round(time.monotonic() - t0, 1)})130 jl.info("job.done", seconds=round(time.monotonic() - t0, 1))131 except Exception as e:132 err = f"{type(e).__name__}: {e}\n{traceback.format_exc()[-1500:]}"133 queue.finish(job, False, error=err)134 jl.error("job.failed", error=str(e)[:300], seconds=round(time.monotonic() - t0, 1))135 if once:136 break137