1"""End-of-day option chains: one vendor archive per (year, quarter) or `period=month`; one vendor file per underlying.2Layout: options.eod/base/underlying=<U>/period=<YYYYqN|latest>/part-*.parquet. The monthly archive replaces the3`latest` partition and drops rows already covered by a completed quarter archive (no overlap in the live set)."""4from __future__ import annotations56from datetime import UTC, date, datetime7from typing import Any89import pyarrow as pa10import pyarrow.compute as pc11from findata_core.db import connection, transaction12from findata_market_firstrate import (13 FirstRateMarketConnector,14 FRDNoData,15 iter_archive_files,16 parse_options_file,17 ticker_from_filename,18)19from findata_schema.connector import FetchRequest2021from findata_ingest import queue22from findata_ingest.lake import Lake23from findata_ingest.tasks.common import record_raw_archive2425DATASET = "options.eod"262728def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:29 p = job.get("params") or {}30 run_id = job.get("run_id")31 if p.get("period"):32 params = {"endpoint": "data_file", "type": "options", "period": p["period"]}33 period_label = "latest"34 else:35 params = {"endpoint": "data_file", "type": "options", "year": str(p["year"]), "quarter": p["quarter"]}36 period_label = f"{p['year']}{p['quarter']}"37 conn_fr = FirstRateMarketConnector()38 ck = queue.get_checkpoint(f"job:{job['id']}", "state") or {}39 done = set(ck.get("done") or [])40 if ck.get("raw_path") and lake.abs(ck["raw_path"]).exists():41 from findata_schema.connector import FetchResult42 res = FetchResult(FetchRequest(DATASET, params), lake.abs(ck["raw_path"]), ck["sha256"], ck["bytes"], datetime.fromisoformat(ck["fetched_at"]), None)43 log.info("options.resume", done=len(done))44 else:45 queue.progress(job["id"], {"stage": "download", "period": period_label})46 try:47 res = conn_fr.fetch_range(FetchRequest(DATASET, params), lake.raw_dir("firstrate", DATASET))48 except FRDNoData as e:49 return {"skipped": True, "reason": str(e)[:300]}50 with transaction() as conn:51 record_raw_archive(conn, "firstrate", DATASET, params, lake.rel(res.path), res.sha256, res.bytes, res.file_count, res.vendor_last_update, run_id)52 queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": lake.rel(res.path), "sha256": res.sha256, "bytes": res.bytes,53 "fetched_at": res.fetched_at.isoformat(), "done": []})54 log.info("options.downloaded", bytes=res.bytes, files=res.file_count, period=period_label)55 # rows already covered by completed quarter archives are excluded from `latest`56 cutoff: date | None = None57 if period_label == "latest":58 with connection() as conn:59 r = conn.execute("SELECT max(date_max) AS d FROM lake_files WHERE dataset_id=%s AND version_removed IS NULL AND path NOT LIKE %s",60 (DATASET, "%/period=latest/%")).fetchone()61 cutoff = r["d"]62 ingested_at = datetime.now(UTC)63 written, retire = [], []64 stats = {"underlyings": 0, "rows": 0, "raw_rows": 0, "files": 0, "skipped_files": 0}65 with connection() as conn:66 old_latest = {r["symbol_min"]: r["path"] for r in conn.execute("SELECT symbol_min, path FROM lake_files WHERE dataset_id=%s AND version_removed IS NULL AND path LIKE %s",67 (DATASET, f"%/period={period_label}/%")).fetchall()}68 batch_rows = 069 for name, data in iter_archive_files(res.path):70 und = ticker_from_filename(name)71 if und in done or und.startswith("_"):72 continue73 try:74 table, raw_rows = parse_options_file(data, und, "firstrate", ingested_at)75 except Exception as e:76 stats["skipped_files"] += 177 log.warning("options.skip_file", file=name, error=str(e)[:120])78 continue79 if cutoff is not None and table.num_rows:80 table = table.filter(pc.greater(table.column("trade_date"), pa.scalar(cutoff, pa.date32())))81 stats["raw_rows"] += raw_rows82 if table.num_rows == 0:83 done.add(und)84 continue85 written.append(lake.write_table(DATASET, table, "base", {"underlying": und, "period": period_label}, sort=False))86 if und in old_latest:87 retire.append(old_latest[und])88 stats["underlyings"] += 189 stats["rows"] += table.num_rows90 done.add(und)91 batch_rows += table.num_rows92 if len(written) >= 1000: # publish in batches so a crash loses little work93 lake.publish(DATASET, written, retire, run_id, note=f"options {period_label} batch")94 stats["files"] += len(written)95 written, retire = [], []96 queue.set_checkpoint(f"job:{job['id']}", "state", {**ck, "raw_path": lake.rel(res.path), "sha256": res.sha256, "bytes": res.bytes,97 "fetched_at": res.fetched_at.isoformat(), "done": sorted(done)})98 queue.progress(job["id"], {"stage": "publish", **stats})99 if written:100 lake.publish(DATASET, written, retire, run_id, note=f"options {period_label}")101 stats["files"] += len(written)102 with transaction() as conn:103 conn.execute("UPDATE raw_archives SET status='published' WHERE sha256=%s", (res.sha256,))104 conn.execute("UPDATE datasets SET status='active', updated_at=now() WHERE id=%s AND status='planned'", (DATASET,))105 queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": lake.rel(res.path), "sha256": res.sha256, "bytes": res.bytes,106 "fetched_at": res.fetched_at.isoformat(), "done": sorted(done)})107 log.info("options.done", period=period_label, **stats)108 return {"period": period_label, **stats}109