SPB Git
spboucher.ai

spb/findata

Public

scheduler: daily dedupe keys fire once (quality/gc were re-enqueued every tick); backup scope documented

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Simon-Pierre Boucher committed (Nov 2, 2026) parent 8d88a53 Browse files

3 changed files +27 −23

3 changed files
Unified Split
modified docs/OPERATIONS.md +1 −0
@@ -38,6 +38,7 @@ Jobs can also be enqueued/retried/cancelled from the console (Ingestion page) or
38 38 Remote-first: edit on the node, `pm2 restart findata-api findata-worker`, then `git commit && git push origin main` (origin = spbgit `findata.git`).
39 39
40 40 ## Backups, RPO / RTO
41 +**Scope**: the daily backup covers the **hot root** (`~/findata-lake`: stock/ETF 1m + 1d raw, derived 5m, manifests, exports) and the metadata database. The **cold root** (`/Volumes/ssd/findata`: raw vendor archives, adjusted/coarse/futures/index/FX/crypto/options datasets, ≈ 1 TB) is **not** backed up — it is fully re-downloadable from the vendor while the subscription is active (archive sha256 recorded in `raw_archives`) and the catalog restore recreates the exact file list. First full-catalog backup 2026-11-02 04:04 → 04:41 (36 min, pg_dump 2.1 GB incl. 18 M fundamentals facts).
41 42 - Daily 04:00 ET: `pg_dump | zstd` (14 kept) + `rsync -a --delete` of `~/findata-lake` (minus `_tmp`, `_retired`, `exports`) → `M4M64a:~/backups/findata/{lake,pgdump}`; remote file count compared with local (`verified`).
42 43 - **RPO**: 24 h for metadata and curated data (raw archives are re-downloadable from the vendor while the subscription is active; sha256 kept). **RTO**: ≈ 30 min for metadata (restore dump, start API) + lake copy time over LAN (≈ 1 Gbit/s → 100 GB in 15–20 min).
43 44 - Restore drill: `findata restore-drill --dest /tmp/findata-restore-drill` restores the latest dump into a scratch database, copies one dataset back and verifies every Parquet sha256 against the manifest. Record the output in `docs/STATUS.md`.
modified services/ingest/src/findata_ingest/queue.py +8 −5
@@ -17,14 +17,17 @@ WORKER_ID = f"{socket.gethostname()}:{random.randint(1000, 9999)}"
17 17
18 18 def enqueue(kind: str, dataset_id: str | None = None, params: dict[str, Any] | None = None, priority: int = 50,
19 19 dedupe_key: str | None = None, max_attempts: int = 5, created_by: str = "system",
20 − run_at: datetime | None = None) -> str | None:
21 − """Insert a job; returns id, or None if a job with the same dedupe_key is already queued/running."""
20 + run_at: datetime | None = None, rerun_succeeded: bool = True) -> str | None:
21 + """Insert a job; returns id, or None if a job with the same dedupe_key is already queued/running.
22 + With rerun_succeeded=False (scheduler), a key whose job already succeeded is not re-enqueued either —
23 + this is what makes daily keys like `quality:2026-11-02` fire once per day."""
22 24 with transaction() as conn:
23 25 if dedupe_key:
24 − r = conn.execute("SELECT id FROM ingest_jobs WHERE dedupe_key=%s", (dedupe_key,)).fetchone()
26 + r = conn.execute("SELECT id, status FROM ingest_jobs WHERE dedupe_key=%s", (dedupe_key,)).fetchone()
25 27 if r:
26 − st = conn.execute("SELECT status FROM ingest_jobs WHERE id=%s", (r["id"],)).fetchone()["status"]
27 − if st in ("queued", "running"):
28 + if r["status"] in ("queued", "running"):
29 + return None
30 + if r["status"] == "succeeded" and not rerun_succeeded:
28 31 return None
29 32 conn.execute("UPDATE ingest_jobs SET dedupe_key=NULL WHERE id=%s", (r["id"],))
30 33 r = conn.execute(
modified services/ingest/src/findata_ingest/worker.py +18 −18
@@ -36,52 +36,52 @@ def schedule_tick(log) -> None:
36 36 # vendor daily update lands ~23:30–03:00 ET → poll from 02:00 every 30 min (incremental skips when vendor unchanged)
37 37 if now.weekday() < 6 and (now.hour >= 2 or now.hour < 1):
38 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}")
39 + dedupe_key=f"incremental:stock:day:{day}:{now.hour}:{now.minute // 30}", rerun_succeeded=False)
40 40 if now.weekday() == 5 and now.hour == 6: # Saturday 06:00 ET — week archive repairs any missed day
41 41 queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "week", "force": True}, priority=25,
42 − dedupe_key=f"incremental:stock:week:{day}")
42 + dedupe_key=f"incremental:stock:week:{day}", rerun_succeeded=False)
43 43 if now.weekday() == 6 and now.hour == 7:
44 − queue.enqueue("metadata", None, {"types": ["stock", "etf"]}, priority=30, dedupe_key=f"metadata:{day}")
45 − queue.enqueue("compaction", "market_bars.stock.1m.raw", {}, priority=60, dedupe_key=f"compaction:1m:{day}")
46 − queue.enqueue("compaction", "market_bars.stock.5m.raw", {}, priority=61, dedupe_key=f"compaction:5m:{day}")
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 47 # other raw 1-minute feeds and daily bars (vendor publishes daily; crypto 7/7)
48 48 if now.hour >= 2 or now.hour < 1:
49 49 for t in ("index", "fx", "crypto"):
50 50 if now.weekday() < 6 or t == "crypto":
51 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}")
52 + dedupe_key=f"incremental:{t}:day:{day}:{now.hour}:{now.minute // 30}", rerun_succeeded=False)
53 53 if now.weekday() < 6:
54 54 for t in ("stock", "etf"):
55 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}")
56 + dedupe_key=f"incremental:{t}:1day:{day}:{now.hour}:{now.minute // 30}", rerun_succeeded=False)
57 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}")
58 + dedupe_key=f"incremental:etf:day:{day}:{now.hour}:{now.minute // 30}", rerun_succeeded=False)
59 59 if now.weekday() == 5 and now.hour == 6:
60 60 # weekly: futures continuous (week), individual contracts update, adjusted stock/etf histories (full_update)
61 61 for tf in ("1min", "1day", "5min", "30min", "1hour"):
62 62 for adj in ("contin_UNadj", "contin_adj_ratio", "contin_adj_absolute"):
63 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}")
64 + dedupe_key=f"incremental:futures:week:{tf}:{adj}:{day}", rerun_succeeded=False)
65 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}")
66 + priority=37, dedupe_key=f"contracts:update:{tf}:{day}", rerun_succeeded=False)
67 67 for t in ("stock", "etf"):
68 68 for adj in ("adj_split", "adj_splitdiv"):
69 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}")
70 + priority=38, dedupe_key=f"full_update:{t}:{tf}:{adj}:{day}", rerun_succeeded=False)
71 71 for t in ("index", "fx", "crypto"):
72 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}")
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 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}")
75 + priority=38, dedupe_key=f"delisted:week:{day}", rerun_succeeded=False)
76 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}")
77 + queue.enqueue("options", "options.eod", {"period": "month"}, priority=55, dedupe_key=f"options:month:{now.year}-{now.month:02d}", rerun_succeeded=False)
78 78 if now.hour == 5 and now.minute < 30:
79 − queue.enqueue("fundamentals", "fundamentals.financials", {"tracked": True}, priority=50, dedupe_key=f"fundamentals:{day}")
79 + queue.enqueue("fundamentals", "fundamentals.financials", {"tracked": True}, priority=50, dedupe_key=f"fundamentals:{day}", rerun_succeeded=False)
80 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}")
81 + queue.enqueue("backup", None, {}, priority=70, dedupe_key=f"backup:{day}", rerun_succeeded=False)
82 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}")
84 − queue.enqueue("gc", None, {}, priority=90, dedupe_key=f"gc:{day}")
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)
85 85
86 86
87 87 def main(once: bool = False, kinds: list[str] | None = None, with_scheduler: bool = True, role: str = "primary") -> None:
88 88
Menu

Theme
↑↓ navigate ↵ open esc close / or ⌘K anywhere