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 stale = queue.requeue_stale()34 if stale:35 log.warning("scheduler.requeued_stale", n=stale)36 # vendor daily update lands ~23:30–03:00 ET → poll from 02:00 every 30 min (incremental skips when vendor unchanged)37 if now.weekday() < 6 and (now.hour >= 2 or now.hour < 1):38 queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "day"}, priority=20,39 dedupe_key=f"incremental:stock:day:{day}:{now.hour}:{now.minute // 30}", rerun_succeeded=False)40 if now.weekday() == 5 and now.hour == 6: # Saturday 06:00 ET — week archive repairs any missed day41 queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "week", "force": True}, priority=25,42 dedupe_key=f"incremental:stock:week:{day}", rerun_succeeded=False)43 if now.weekday() == 6 and now.hour == 7:44 queue.enqueue("metadata", None, {"types": ["stock", "etf"]}, priority=30, dedupe_key=f"metadata:{day}", rerun_succeeded=False)45 queue.enqueue("compaction", "market_bars.stock.1m.raw", {}, priority=60, dedupe_key=f"compaction:1m:{day}", rerun_succeeded=False)46 queue.enqueue("compaction", "market_bars.stock.5m.raw", {}, priority=61, dedupe_key=f"compaction:5m:{day}", rerun_succeeded=False)47 # other raw 1-minute feeds and daily bars (vendor publishes daily; crypto 7/7)48 if now.hour >= 2 or now.hour < 1:49 for t in ("index", "fx", "crypto"):50 if now.weekday() < 6 or t == "crypto":51 queue.enqueue("incremental", f"market_bars.{t}.1m.raw", {"type": t, "period": "day", "timeframe": "1min"}, priority=26,52 dedupe_key=f"incremental:{t}:day:{day}:{now.hour}:{now.minute // 30}", rerun_succeeded=False)53 if now.weekday() < 6:54 for t in ("stock", "etf"):55 queue.enqueue("incremental", f"market_bars.{t}.1d.raw", {"type": t, "period": "day", "timeframe": "1day", "adjustment": "UNADJUSTED"}, priority=27,56 dedupe_key=f"incremental:{t}:1day:{day}:{now.hour}:{now.minute // 30}", rerun_succeeded=False)57 queue.enqueue("incremental", "market_bars.etf.1m.raw", {"type": "etf", "period": "day"}, priority=21,58 dedupe_key=f"incremental:etf:day:{day}:{now.hour}:{now.minute // 30}", rerun_succeeded=False)59 if now.weekday() == 5 and now.hour == 6:60 # weekly: futures continuous (week), individual contracts update, adjusted stock/etf histories (full_update)61 for tf in ("1min", "1day", "5min", "30min", "1hour"):62 for adj in ("contin_UNadj", "contin_adj_ratio", "contin_adj_absolute"):63 queue.enqueue("incremental", None, {"type": "futures", "period": "week", "timeframe": tf, "adjustment": adj}, priority=36,64 dedupe_key=f"incremental:futures:week:{tf}:{adj}:{day}", rerun_succeeded=False)65 queue.enqueue("backfill", None, {"type": "futures", "endpoint": "futures_contract", "contract_files": "update", "timeframe": tf, "contracts": True},66 priority=37, dedupe_key=f"contracts:update:{tf}:{day}", rerun_succeeded=False)67 for t in ("stock", "etf"):68 for adj in ("adj_split", "adj_splitdiv"):69 queue.enqueue("backfill", None, {"type": t, "endpoint": "full_update", "period": "week", "timeframe": tf, "adjustment": adj},70 priority=38, dedupe_key=f"full_update:{t}:{tf}:{adj}:{day}", rerun_succeeded=False)71 for t in ("index", "fx", "crypto"):72 if tf != "1min":73 queue.enqueue("incremental", None, {"type": t, "period": "week", "timeframe": tf}, priority=39, dedupe_key=f"incremental:{t}:week:{tf}:{day}", rerun_succeeded=False)74 queue.enqueue("backfill", "market_bars.stock.1m.raw", {"type": "stock", "endpoint": "delisted_data_file", "update": "week", "timeframe": "1min", "adjustment": "UNADJUSTED"},75 priority=38, dedupe_key=f"delisted:week:{day}", rerun_succeeded=False)76 if now.day == 2 and now.hour == 8:77 queue.enqueue("options", "options.eod", {"period": "month"}, priority=55, dedupe_key=f"options:month:{now.year}-{now.month:02d}", rerun_succeeded=False)78 if now.hour == 5 and now.minute < 30:79 queue.enqueue("fundamentals", "fundamentals.financials", {"tracked": True}, priority=50, dedupe_key=f"fundamentals:{day}", rerun_succeeded=False)80 if now.hour == 4 and now.minute < 30 and get_settings().backup_target:81 queue.enqueue("backup", None, {}, priority=70, dedupe_key=f"backup:{day}", rerun_succeeded=False)82 if now.hour == 6 and now.minute < 30:83 queue.enqueue("quality", "market_bars.stock.1m.raw", {}, priority=80, dedupe_key=f"quality:{day}", rerun_succeeded=False)84 queue.enqueue("gc", None, {}, priority=90, dedupe_key=f"gc:{day}", rerun_succeeded=False)858687def main(once: bool = False, kinds: list[str] | None = None, with_scheduler: bool = True, role: str = "primary") -> None:88 """role=primary: holds the lake lock, runs the scheduler and the serialized kinds; role=bulk: one of N89 parallel workers for backfill/options/fundamentals (publication is serialized by Postgres row locks)."""90 configure_logging()91 log = get_logger("worker")92 signal.signal(signal.SIGTERM, _handle)93 signal.signal(signal.SIGINT, _handle)94 signal.signal(signal.SIGHUP, _handle) # pty hang-up when run under sshd (see infra/scripts/run-under-sshd.sh)95 lake = Lake()96 if role == "bulk":97 _loop(lake, log, once, kinds or BULK_KINDS, False)98 return99 requeued = queue.requeue_stale()100 if requeued:101 log.warning("worker.requeued_stale", n=requeued)102 if kinds is None:103 kinds = PRIMARY_KINDS if get_settings().bulk_workers > 0 else None104 with WriterLock(lake.root):105 _loop(lake, log, once, kinds, with_scheduler)106 log.info("worker.stop")107108109def _loop(lake: Lake, log, once: bool, kinds: list[str] | None, with_scheduler: bool) -> None:110 last_tick = 0.0111 if True:112 log.info("worker.start", lake=str(lake.root), cold=str(lake.cold), free_gb=round(lake.free_gb(), 1), kinds=kinds)113 while not _stop:114 if with_scheduler and time.monotonic() - last_tick > 300:115 try:116 schedule_tick(log)117 except Exception as e:118 log.error("scheduler.error", error=str(e))119 last_tick = time.monotonic()120 job = queue.claim_next(kinds)121 if job is None:122 if once:123 break124 time.sleep(5)125 continue126 jl = log.bind(job_id=str(job["id"]), kind=job["kind"], dataset=job.get("dataset_id"), attempt=job["attempts"])127 jl.info("job.start")128 t0 = time.monotonic()129 try:130 fn = REGISTRY[job["kind"]]131 result = fn(job, lake, jl)132 queue.finish(job, True, {**(result or {}), "seconds": round(time.monotonic() - t0, 1)})133 jl.info("job.done", seconds=round(time.monotonic() - t0, 1))134 except Exception as e:135 err = f"{type(e).__name__}: {e}\n{traceback.format_exc()[-1500:]}"136 queue.finish(job, False, error=err)137 jl.error("job.failed", error=str(e)[:300], seconds=round(time.monotonic() - t0, 1))138 if once:139 break140