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 = False202122def _handle(sig, frame):23 global _stop24 _stop = True252627def schedule_tick(log) -> None:28 """Enqueue recurring jobs (dedupe keys make this idempotent). Times in America/New_York."""29 now = datetime.now(ET)30 day = now.date().isoformat()31 # vendor daily update lands ~23:30–03:00 ET → poll from 02:00 every 30 min (incremental skips when vendor unchanged)32 if now.weekday() < 6 and (now.hour >= 2 or now.hour < 1):33 queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "day"}, priority=20,34 dedupe_key=f"incremental:stock:day:{day}:{now.hour}:{now.minute // 30}")35 if now.weekday() == 5 and now.hour == 6: # Saturday 06:00 ET — week archive repairs any missed day36 queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "week", "force": True}, priority=25,37 dedupe_key=f"incremental:stock:week:{day}")38 if now.weekday() == 6 and now.hour == 7:39 queue.enqueue("metadata", None, {"types": ["stock", "etf"]}, priority=30, dedupe_key=f"metadata:{day}")40 queue.enqueue("compaction", "market_bars.stock.1m.raw", {}, priority=60, dedupe_key=f"compaction:1m:{day}")41 queue.enqueue("compaction", "market_bars.stock.5m.raw", {}, priority=61, dedupe_key=f"compaction:5m:{day}")42 if now.hour == 5 and now.minute < 30:43 queue.enqueue("fundamentals", "fundamentals.financials", {"tracked": True}, priority=50, dedupe_key=f"fundamentals:{day}")44 if now.hour == 4 and now.minute < 30 and get_settings().backup_target:45 queue.enqueue("backup", None, {}, priority=70, dedupe_key=f"backup:{day}")46 if now.hour == 6 and now.minute < 30:47 queue.enqueue("quality", "market_bars.stock.1m.raw", {}, priority=80, dedupe_key=f"quality:{day}")48 queue.enqueue("gc", None, {}, priority=90, dedupe_key=f"gc:{day}")495051def main(once: bool = False, kinds: list[str] | None = None, with_scheduler: bool = True) -> None:52 configure_logging()53 log = get_logger("worker")54 signal.signal(signal.SIGTERM, _handle)55 signal.signal(signal.SIGINT, _handle)56 lake = Lake()57 requeued = queue.requeue_stale()58 if requeued:59 log.warning("worker.requeued_stale", n=requeued)60 last_tick = 0.061 with WriterLock(lake.root):62 log.info("worker.start", lake=str(lake.root), free_gb=round(lake.free_gb(), 1))63 while not _stop:64 if with_scheduler and time.monotonic() - last_tick > 300:65 try:66 schedule_tick(log)67 except Exception as e:68 log.error("scheduler.error", error=str(e))69 last_tick = time.monotonic()70 job = queue.claim_next(kinds)71 if job is None:72 if once:73 break74 time.sleep(5)75 continue76 jl = log.bind(job_id=str(job["id"]), kind=job["kind"], dataset=job.get("dataset_id"), attempt=job["attempts"])77 jl.info("job.start")78 t0 = time.monotonic()79 try:80 fn = REGISTRY[job["kind"]]81 result = fn(job, lake, jl)82 queue.finish(job, True, {**(result or {}), "seconds": round(time.monotonic() - t0, 1)})83 jl.info("job.done", seconds=round(time.monotonic() - t0, 1))84 except Exception as e:85 err = f"{type(e).__name__}: {e}\n{traceback.format_exc()[-1500:]}"86 queue.finish(job, False, error=err)87 jl.error("job.failed", error=str(e)[:300], seconds=round(time.monotonic() - t0, 1))88 if once:89 break90 log.info("worker.stop")91