full FirstRate catalog: cold lake root (external SSD), 57 new datasets, index/contracts/options parsers, options task, bulk workers, complete download plan (798 jobs), extended scheduler

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Simon-Pierre Boucher committed yesterday (Nov 2, 2026) parent 9a819d1 Browse files

24 changed files +945 −99

modified connectors/market/src/findata_market_firstrate/__init__.py +8 −2

		`@@ -1,3 +1,9 @@`
1		−from findata_market_firstrate.archive import iter_archive_files, parse_bar_file, ticker_from_filename # noqa: F401
2		−from findata_market_firstrate.client import FRDClient, FRDError # noqa: F401
	1	+from findata_market_firstrate.archive import ( # noqa: F401
	2	+ iter_archive_files,
	3	+ parse_bar_file,
	4	+ parse_options_file,
	5	+ ticker_from_filename,
	6	+)
	7	+from findata_market_firstrate.catalog import VENDOR_TYPES, build_full_plan, dataset_id_for # noqa: F401
	8	+from findata_market_firstrate.client import FRDClient, FRDError, FRDNoData # noqa: F401
3	9	from findata_market_firstrate.connector import FirstRateMarketConnector # noqa: F401

modified connectors/market/src/findata_market_firstrate/archive.py +51 −4

		`@@ -22,8 +22,14 @@ VENDOR_ADJ = {"UNADJUSTED": Adjustment.raw, "adj_split": Adjustment.split, "adj_`
22	22	_TICKER_RE = re.compile(r"^([A-Za-z0-9.\-]+?)(?:[_-](?:full\|day\|week\|month\|1min\|5min\|30min\|1hour\|1day\|UNADJUSTED\|adj_split\|adj_splitdiv\|contin.))\.(txt\|csv)$", re.IGNORECASE)
23	23
24	24
25		−def ticker_from_filename(name: str) -> str:
	25	+_TF_SUFFIX_RE = re.compile(r"_(1min\|5min\|30min\|1hour\|1day)$", re.I)
	26	+
	27	+
	28	+def ticker_from_filename(name: str, contract: bool = False) -> str:
26	29	base = Path(name).name
	30	+ if contract: # individual futures contracts: 'A6_F25_1day.txt' -> 'A6_F25'
	31	+ stem = Path(base).stem
	32	+ return _TF_SUFFIX_RE.sub("", stem).upper()
27	33	m = _TICKER_RE.match(base)
28	34	if m:
29	35	return m.group(1).upper()
		`@@ -71,17 +77,25 @@ def parse_bar_file(data: bytes, symbol: str, asset_class: AssetClass, interval:`
71	77	df = df.with_columns((pl.col("d") + " " + pl.col("t")).alias("dt_raw")).drop("d", "t")
72	78	fmt = "%Y%m%d %H:%M:%S"
73	79	else:
74		− names = ["dt_raw", "open", "high", "low", "close", "volume"] + (["open_interest"] if ncols == 7 else [])
75		− if ncols not in (6, 7):
	80	+ if ncols == 5: # indices: no volume column
	81	+ names = ["dt_raw", "open", "high", "low", "close"]
	82	+ else:
	83	+ names = ["dt_raw", "open", "high", "low", "close", "volume"] + (["open_interest"] if ncols == 7 else [])
	84	+ if ncols not in (5, 6, 7):
76	85	raise ValueError(f"unexpected column count {ncols} in {symbol}")
77	86	df = pl.read_csv(buf, has_header=False, new_columns=names, infer_schema_length=0)
	87	+ if ncols == 5:
	88	+ df = df.with_columns(pl.lit("0").alias("volume"))
78	89	fmt = "%Y-%m-%d %H:%M:%S" if interval != Interval.d1 else None
79	90	for c in ("open", "high", "low", "close", "volume"):
80	91	df = df.with_columns(pl.col(c).str.strip_chars().cast(pl.Float64, strict=False))
81	92	if interval == Interval.d1:
82	93	local = pl.col("dt_raw").str.strip_chars().str.slice(0, 10).str.to_date("%Y-%m-%d", strict=False).cast(pl.Datetime("us"))
83	94	else:
84		− local = pl.col("dt_raw").str.strip_chars().str.to_datetime(fmt, strict=False, time_unit="us")
	95	+ local = pl.coalesce(
	96	+ pl.col("dt_raw").str.strip_chars().str.to_datetime(fmt, strict=False, time_unit="us"),
	97	+ pl.col("dt_raw").str.strip_chars().str.to_datetime("%Y-%m-%d %H:%M", strict=False, time_unit="us"),
	98	+ )
85	99	df = df.with_columns(local.alias("local_naive"))
86	100	df = df.filter(pl.col("local_naive").is_not_null() & pl.col("open").is_not_null() & pl.col("close").is_not_null())
87	101	df = df.with_columns(
		`@@ -128,3 +142,36 @@ def parse_meta_pairs(data: bytes) -> list[tuple[str, float]]:`
128	142	except ValueError:
129	143	continue
130	144	return out
	145	+
	146	+
	147	+def parse_options_file(data: bytes, underlying: str, source_id: str, ingested_at: datetime) -> tuple[pa.Table, int]:
	148	+ """Vendor option-chain CSV (16 columns, no header): Trade Date, Strike, Expiry, Call/Put, Last, Bid, Ask, Bid IV,
	149	+ Ask IV, Open Interest, Volume, Delta, Gamma, Vega, Theta, Rho → OPTIONS_SCHEMA."""
	150	+ from findata_schema import OPTIONS_SCHEMA
	151	+ text = data.decode("utf-8", errors="replace").replace("\r\n", "\n").replace("\r", "\n")
	152	+ lines = [ln for ln in text.split("\n") if ln.strip()]
	153	+ if lines and lines[0].lower().startswith(("trade", "date")):
	154	+ lines = lines[1:]
	155	+ raw_rows = len(lines)
	156	+ empty = pa.table({f.name: pa.array([], type=f.type) for f in OPTIONS_SCHEMA}, schema=OPTIONS_SCHEMA)
	157	+ if raw_rows == 0:
	158	+ return empty, 0
	159	+ names = ["trade_date", "strike", "expiry", "call_put", "last_price", "bid", "ask", "bid_iv", "ask_iv", "open_interest",
	160	+ "volume", "delta", "gamma", "vega", "theta", "rho"]
	161	+ ncols = len(lines[0].split(","))
	162	+ if ncols != 16:
	163	+ raise ValueError(f"unexpected option-chain column count {ncols} for {underlying}")
	164	+ df = pl.read_csv("\n".join(lines).encode(), has_header=False, new_columns=names, infer_schema_length=0)
	165	+ num = [c for c in names if c not in ("trade_date", "expiry", "call_put")]
	166	+ df = df.with_columns([pl.col(c).str.strip_chars().cast(pl.Float64, strict=False) for c in num])
	167	+ df = df.with_columns(
	168	+ pl.col("trade_date").str.strip_chars().str.slice(0, 10).str.to_date("%Y-%m-%d", strict=False),
	169	+ pl.col("expiry").str.strip_chars().str.slice(0, 10).str.to_date("%Y-%m-%d", strict=False),
	170	+ pl.col("call_put").str.strip_chars().str.to_lowercase(),
	171	+ ).filter(pl.col("trade_date").is_not_null() & pl.col("expiry").is_not_null() & pl.col("strike").is_not_null())
	172	+ n = df.height
	173	+ out = df.with_columns(
	174	+ pl.lit(underlying).alias("underlying"), pl.lit(source_id).alias("source_id"),
	175	+ pl.lit(ingested_at).cast(pl.Datetime("us", "UTC")).alias("ingested_at"),
	176	+ ).select(OPTIONS_SCHEMA.names)
	177	+ return out.to_arrow().cast(OPTIONS_SCHEMA), raw_rows

added connectors/market/src/findata_market_firstrate/catalog.py +103 −0

		`@@ -0,0 +1,103 @@`
	1	+"""Vendor catalog → FinData dataset ids, and the complete download plan (every licensed archive)."""
	2	+from __future__ import annotations
	3	+
	4	+from datetime import date
	5	+from typing import Any
	6	+
	7	+VENDOR_TYPES = ("stock", "etf", "futures", "index", "fx", "crypto", "options")
	8	+TF_MAP = {"1min": "1m", "5min": "5m", "30min": "30m", "1hour": "1h", "1day": "1d"}
	9	+ADJ_MAP = {"UNADJUSTED": "raw", "adj_split": "split", "adj_splitdiv": "split_div",
	10	+ "contin_UNadj": "continuous_raw", "contin_adj_ratio": "continuous_ratio", "contin_adj_absolute": "continuous_absolute"}
	11	+TIMEFRAMES = ["1min", "1day", "5min", "30min", "1hour"]
	12	+LETTERS = [chr(c) for c in range(ord("A"), ord("Z") + 1)]
	13	+OPTIONS_FIRST_YEAR = 2010
	14	+
	15	+
	16	+def dataset_id_for(type_: str, timeframe: str, adjustment: str \| None, contracts: bool = False) -> str:
	17	+ tf = TF_MAP[timeframe]
	18	+ if type_ == "options":
	19	+ return "options.eod"
	20	+ if type_ == "futures":
	21	+ if contracts:
	22	+ return f"market_bars.futures_contracts.{tf}.raw"
	23	+ return f"market_bars.futures.{tf}.{ADJ_MAP[adjustment or 'contin_UNadj']}"
	24	+ if type_ in ("index", "fx", "crypto"):
	25	+ return f"market_bars.{type_}.{tf}.raw"
	26	+ return f"market_bars.{type_}.{tf}.{ADJ_MAP[adjustment or 'UNADJUSTED']}"
	27	+
	28	+
	29	+def _adjs(type_: str, tf: str) -> list[str]:
	30	+ if type_ in ("stock", "etf"):
	31	+ return (["UNADJUSTED"] if tf in ("1min", "1day") else []) + ["adj_split", "adj_splitdiv"]
	32	+ if type_ == "futures":
	33	+ return ["contin_UNadj", "contin_adj_ratio", "contin_adj_absolute"]
	34	+ return [None]
	35	+
	36	+
	37	+def build_full_plan(types: list[str] \| None = None, include_options: bool = True, include_delisted: bool = True,
	38	+ today: date \| None = None) -> list[dict[str, Any]]:
	39	+ """Return job specs {kind, dataset_id, params, priority, dedupe_key}. Hot datasets first."""
	40	+ today = today or date.today()
	41	+ types = types or list(VENDOR_TYPES)
	42	+ jobs: list[dict[str, Any]] = []
	43	+
	44	+ def add(kind: str, dsid: str, params: dict[str, Any], prio: int, key: str) -> None:
	45	+ jobs.append({"kind": kind, "dataset_id": dsid, "params": params, "priority": prio, "dedupe_key": key})
	46	+
	47	+ # priority bands: 10 hot raw stock/etf 1min, 14 raw 1day, 20 other asset classes, 30 delisted raw, 40 adjusted 1min,
	48	+ # 45 adjusted 1day, 50 coarse timeframes, 60 options
	49	+ for t in ("stock", "etf"):
	50	+ if t not in types:
	51	+ continue
	52	+ for tf in TIMEFRAMES:
	53	+ for adj in _adjs(t, tf):
	54	+ if adj == "UNADJUSTED":
	55	+ prio = 10 if tf == "1min" else 14
	56	+ elif tf == "1min":
	57	+ prio = 40
	58	+ elif tf == "1day":
	59	+ prio = 45
	60	+ else:
	61	+ prio = 50
	62	+ prio += 0 if t == "stock" else 1
	63	+ dsid = dataset_id_for(t, tf, adj)
	64	+ for L in LETTERS:
	65	+ add("backfill", dsid, {"type": t, "ticker_range": L, "timeframe": tf, "adjustment": adj, "period": "full"}, prio,
	66	+ f"backfill:{t}:{L}:{tf}:{adj}")
	67	+ if include_delisted and "stock" in types:
	68	+ for tf in TIMEFRAMES:
	69	+ for adj in _adjs("stock", tf):
	70	+ if adj == "UNADJUSTED" and tf != "1min":
	71	+ continue # vendor: delisted UNADJUSTED only at 1min
	72	+ dsid = dataset_id_for("stock", tf, adj)
	73	+ prio = 30 if adj == "UNADJUSTED" else (42 if tf == "1min" else 52)
	74	+ for arc in ("1", "2", "3", "4", "5"):
	75	+ add("backfill", dsid, {"type": "stock", "endpoint": "delisted_data_file", "archive_number": arc, "timeframe": tf, "adjustment": adj},
	76	+ prio, f"delisted:{arc}:{tf}:{adj}")
	77	+ add("backfill", dsid, {"type": "stock", "endpoint": "delisted_data_file", "update": "year", "timeframe": tf, "adjustment": adj},
	78	+ prio, f"delisted:year:{tf}:{adj}")
	79	+ for t in ("index", "fx", "crypto"):
	80	+ if t not in types:
	81	+ continue
	82	+ for tf in TIMEFRAMES:
	83	+ add("backfill", dataset_id_for(t, tf, None), {"type": t, "timeframe": tf, "period": "full"}, 20 if tf in ("1min", "1day") else 50,
	84	+ f"backfill:{t}:{tf}")
	85	+ if "futures" in types:
	86	+ for tf in TIMEFRAMES:
	87	+ for adj in _adjs("futures", tf):
	88	+ add("backfill", dataset_id_for("futures", tf, adj), {"type": "futures", "timeframe": tf, "adjustment": adj, "period": "full"},
	89	+ 22 if tf in ("1min", "1day") else 50, f"backfill:futures:{tf}:{adj}")
	90	+ for cf in ("archive", "update"):
	91	+ add("backfill", dataset_id_for("futures", tf, None, contracts=True),
	92	+ {"type": "futures", "endpoint": "futures_contract", "contract_files": cf, "timeframe": tf, "contracts": True},
	93	+ 24 if tf in ("1min", "1day") else 50, f"contracts:{cf}:{tf}")
	94	+ if include_options and "options" in types:
	95	+ cy, cq = today.year, (today.month - 1) // 3 + 1
	96	+ for y in range(OPTIONS_FIRST_YEAR, cy + 1):
	97	+ for q in range(1, 5):
	98	+ if y == cy and q >= cq:
	99	+ continue
	100	+ add("options", "options.eod", {"year": y, "quarter": f"q{q}"}, 60, f"options:{y}:q{q}")
	101	+ add("options", "options.eod", {"period": "month"}, 61, f"options:month:{cy}-{today.month:02d}")
	102	+ jobs.sort(key=lambda j: (j["priority"], j["dedupe_key"]))
	103	+ return jobs

modified connectors/market/src/findata_market_firstrate/client.py +10 −1

		`@@ -30,6 +30,10 @@ class FRDError(RuntimeError):`
30	30	pass
31	31
32	32
	33	+class FRDNoData(FRDError):
	34	+ """The vendor answered with a tiny text body instead of an archive (e.g. 'no update this period')."""
	35	+
	36	+
33	37	@dataclass
34	38	class Downloaded:
35	39	path: Path
		`@@ -120,8 +124,11 @@ class FRDClient:`
120	124	body = r.read()[:300]
121	125	raise FRDError(f"HTTP {r.status_code} downloading {endpoint}: {self._redact(body)}")
122	126	ctype = r.headers.get("content-type", "")
123		− if "text/html" in ctype:
	127	+ clen = int(r.headers.get("content-length") or 0)
	128	+ if "text/html" in ctype or (clen and clen < 512 and "zip" not in ctype):
124	129	body = r.read()[:300]
	130	+ if clen and clen < 512:
	131	+ raise FRDNoData(f"no archive available for {endpoint}: {self._redact(body.decode(errors='replace').strip())}")
125	132	raise FRDError(f"HTML instead of archive from {endpoint}: {self._redact(body)}")
126	133	with open(part, "wb") as f:
127	134	for chunk in r.iter_bytes(CHUNK):
		`@@ -134,6 +141,8 @@ class FRDClient:`
134	141	except (httpx.TransportError, FRDError, OSError) as e:
135	142	last = e
136	143	part.unlink(missing_ok=True)
	144	+ if isinstance(e, FRDNoData):
	145	+ raise
137	146	if isinstance(e, FRDError) and "HTTP 4" in str(e):
138	147	raise FRDError(self._redact(e)) from None
139	148	time.sleep(min(120, 5 * 2 ** attempt) + random.uniform(0, 2))

modified connectors/market/src/findata_market_firstrate/connector.py +19 −8

		`@@ -23,13 +23,13 @@ from findata_schema.connector import (`
23	23	)
24	24
25	25	from findata_market_firstrate.archive import (
26		− VENDOR_ADJ,
27	26	VENDOR_TF,
28	27	iter_archive_files,
29	28	parse_bar_file,
30	29	parse_meta_pairs,
31	30	ticker_from_filename,
32	31	)
	32	+from findata_market_firstrate.catalog import ADJ_MAP
33	33	from findata_market_firstrate.client import FRDClient
34	34
35	35	SOURCE_ID = "firstrate"
		`@@ -102,18 +102,20 @@ class FirstRateMarketConnector(Connector):`
102	102	p = result.request.params
103	103	asset = AssetClass(p.get("type", "stock"))
104	104	interval = VENDOR_TF[p.get("timeframe", "1min")]
105		− adjustment = VENDOR_ADJ.get(p.get("adjustment", "UNADJUSTED"))
106		− if adjustment is None:
107		− from findata_schema import Adjustment
108		− adjustment = Adjustment.continuous
	105	+ adj_vendor = p.get("adjustment") or ("contin_UNadj" if asset == AssetClass.futures and not p.get("contracts") else "UNADJUSTED")
	106	+ adjustment_value = ADJ_MAP.get(adj_vendor, "raw")
	107	+ contracts = bool(p.get("contracts"))
109	108	ingested_at = datetime.now(UTC)
110	109	only: set[str] \| None = set(kwargs.get("symbols") or []) or None
111	110	tz = SOURCE_TZ.get(asset.value, "America/New_York")
112	111	for name, data in iter_archive_files(result.path):
113		− symbol = ticker_from_filename(name)
	112	+ base = Path(name).name.lower()
	113	+ if base.startswith("_") or "readme" in base or base.endswith((".csv", ".txt")) is False:
	114	+ continue
	115	+ symbol = ticker_from_filename(name, contract=contracts)
114	116	if only and symbol not in only:
115	117	continue
116		− table, raw_rows = parse_bar_file(data, symbol, asset, interval, adjustment, SOURCE_ID, ingested_at, source_tz=tz)
	118	+ table, raw_rows = parse_bar_file(data, symbol, asset, interval, _Adj(adjustment_value), SOURCE_ID, ingested_at, source_tz=tz)
117	119	if table.num_rows == 0:
118	120	continue
119	121	sd = table.column("session_date").to_pylist()
		`@@ -123,7 +125,8 @@ class FirstRateMarketConnector(Connector):`
123	125	# ------------------------------------------------------------------ validate
124	126	def validate(self, batch: NormalizedBatch) -> ValidationReport:
125	127	interval = Interval(batch.table.column("interval")[0].as_py())
126		− table, anomalies = validate_bars(batch.table, interval)
	128	+ asset = batch.table.column("asset_class")[0].as_py()
	129	+ table, anomalies = validate_bars(batch.table, interval, asset_class=asset)
127	130	batch.table = table
128	131	flagged = sum(1 for f in table.column("quality_flags").to_pylist() if f)
129	132	errors = [a for a in anomalies if a["severity"] == "error"]
		`@@ -170,6 +173,14 @@ class FirstRateMarketConnector(Connector):`
170	173	return res, rows
171	174
172	175
	176	+class _Adj(str):
	177	+ """Adjustment label passthrough (dataset adjustment ids are richer than the Adjustment enum)."""
	178	+
	179	+ @property
	180	+ def value(self) -> str: # mimics Enum.value
	181	+ return str(self)
	182	+
	183	+
173	184	def _d(s: str \| None) -> date \| None:
174	185	if not s:
175	186	return None

modified infra/mld/findata.json +172 −7

		`@@ -6,14 +6,175 @@`
6	6	"health_path": "/api/v1/health",
7	7	"dir": "~/apps/findata",
8	8	"extra_paths": [],
9		− "sync_excludes": [".git/", ".DS_Store", ".venv/", "apps/web/node_modules/", "apps/web/.next/", "apps/web/out/", "data/", "logs/", "backups/", ".env", "__pycache__/", ".pytest_cache/", ".ruff_cache/", "tests/_tmp/"],
10		− "requires": { "runtimes": ["uv", "node", "pnpm", "pm2", "postgresql@17"], "ram_gb": 6, "disk_gb": 300, "ports": [8600] },
	9	+ "sync_excludes": [
	10	+ ".git/",
	11	+ ".DS_Store",
	12	+ ".venv/",
	13	+ "apps/web/node_modules/",
	14	+ "apps/web/.next/",
	15	+ "apps/web/out/",
	16	+ "data/",
	17	+ "logs/",
	18	+ "backups/",
	19	+ ".env",
	20	+ "__pycache__/",
	21	+ ".pytest_cache/",
	22	+ ".ruff_cache/",
	23	+ "tests/_tmp/"
	24	+ ],
	25	+ "requires": {
	26	+ "runtimes": [
	27	+ "uv",
	28	+ "node",
	29	+ "pnpm",
	30	+ "pm2",
	31	+ "postgresql@17"
	32	+ ],
	33	+ "ram_gb": 6,
	34	+ "disk_gb": 1500,
	35	+ "ports": [
	36	+ 8600
	37	+ ]
	38	+ },
11	39	"ram_mb_observed": 900,
12	40	"size_mb": 20,
13		− "placement": { "pin": "M3U96a", "prefer": "M3U96a", "avoid": ["M1M32"], "reason": "Storage node chosen in docs/DISCOVERY.md: 1 TB Apple NVMe (508 GB free, SMART verified, 5 GB/s), 32 cores / 96 GB for DuckDB, Postgres 17 + Redis already local, wg1 10.67.0.10. Lake at ~/findata-lake (outside the app dir)." },
	41	+ "placement": {
	42	+ "pin": "M3U96a",
	43	+ "prefer": "M3U96a",
	44	+ "avoid": [
	45	+ "M1M32"
	46	+ ],
	47	+ "reason": "Storage node chosen in docs/DISCOVERY.md: 1 TB Apple NVMe (508 GB free, SMART verified, 5 GB/s), 32 cores / 96 GB for DuckDB, Postgres 17 + Redis already local, wg1 10.67.0.10. Lake at ~/findata-lake (outside the app dir)."
	48	+ },
14	49	"processes": [
15		− { "name": "findata-api", "manager": "pm2", "script": "{{HOME}}/apps/findata/.venv/bin/python", "args": ["-m", "uvicorn", "findata_api.main:app", "--host", "0.0.0.0", "--port", "8600", "--workers", "1"], "interpreter": null, "cwd": "{{HOME}}/apps/findata", "env": {"FINDATA_ROOT": "{{HOME}}/apps/findata", "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"}, "cron_restart": null, "autorestart": true, "max_memory_restart": "6G" },
16		− { "name": "findata-worker", "manager": "pm2", "script": "{{HOME}}/apps/findata/.venv/bin/python", "args": ["-m", "findata_ingest.cli", "worker"], "interpreter": null, "cwd": "{{HOME}}/apps/findata", "env": {"FINDATA_ROOT": "{{HOME}}/apps/findata", "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"}, "cron_restart": null, "autorestart": true, "max_memory_restart": "24G" }
	50	+ {
	51	+ "name": "findata-api",
	52	+ "manager": "pm2",
	53	+ "script": "{{HOME}}/apps/findata/.venv/bin/python",
	54	+ "args": [
	55	+ "-m",
	56	+ "uvicorn",
	57	+ "findata_api.main:app",
	58	+ "--host",
	59	+ "0.0.0.0",
	60	+ "--port",
	61	+ "8600",
	62	+ "--workers",
	63	+ "1"
	64	+ ],
	65	+ "interpreter": null,
	66	+ "cwd": "{{HOME}}/apps/findata",
	67	+ "env": {
	68	+ "FINDATA_ROOT": "{{HOME}}/apps/findata",
	69	+ "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"
	70	+ },
	71	+ "cron_restart": null,
	72	+ "autorestart": true,
	73	+ "max_memory_restart": "6G"
	74	+ },
	75	+ {
	76	+ "name": "findata-worker",
	77	+ "manager": "pm2",
	78	+ "script": "{{HOME}}/apps/findata/.venv/bin/python",
	79	+ "args": [
	80	+ "-m",
	81	+ "findata_ingest.cli",
	82	+ "worker"
	83	+ ],
	84	+ "interpreter": null,
	85	+ "cwd": "{{HOME}}/apps/findata",
	86	+ "env": {
	87	+ "FINDATA_ROOT": "{{HOME}}/apps/findata",
	88	+ "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"
	89	+ },
	90	+ "cron_restart": null,
	91	+ "autorestart": true,
	92	+ "max_memory_restart": "24G"
	93	+ },
	94	+ {
	95	+ "name": "findata-bulk-1",
	96	+ "manager": "pm2",
	97	+ "script": "{{HOME}}/apps/findata/.venv/bin/python",
	98	+ "args": [
	99	+ "-m",
	100	+ "findata_ingest.cli",
	101	+ "worker",
	102	+ "--role",
	103	+ "bulk"
	104	+ ],
	105	+ "interpreter": null,
	106	+ "cwd": "{{HOME}}/apps/findata",
	107	+ "env": {
	108	+ "FINDATA_ROOT": "{{HOME}}/apps/findata",
	109	+ "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"
	110	+ },
	111	+ "cron_restart": null,
	112	+ "autorestart": true,
	113	+ "max_memory_restart": "12G"
	114	+ },
	115	+ {
	116	+ "name": "findata-bulk-2",
	117	+ "manager": "pm2",
	118	+ "script": "{{HOME}}/apps/findata/.venv/bin/python",
	119	+ "args": [
	120	+ "-m",
	121	+ "findata_ingest.cli",
	122	+ "worker",
	123	+ "--role",
	124	+ "bulk"
	125	+ ],
	126	+ "interpreter": null,
	127	+ "cwd": "{{HOME}}/apps/findata",
	128	+ "env": {
	129	+ "FINDATA_ROOT": "{{HOME}}/apps/findata",
	130	+ "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"
	131	+ },
	132	+ "cron_restart": null,
	133	+ "autorestart": true,
	134	+ "max_memory_restart": "12G"
	135	+ },
	136	+ {
	137	+ "name": "findata-bulk-3",
	138	+ "manager": "pm2",
	139	+ "script": "{{HOME}}/apps/findata/.venv/bin/python",
	140	+ "args": [
	141	+ "-m",
	142	+ "findata_ingest.cli",
	143	+ "worker",
	144	+ "--role",
	145	+ "bulk"
	146	+ ],
	147	+ "interpreter": null,
	148	+ "cwd": "{{HOME}}/apps/findata",
	149	+ "env": {
	150	+ "FINDATA_ROOT": "{{HOME}}/apps/findata",
	151	+ "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"
	152	+ },
	153	+ "cron_restart": null,
	154	+ "autorestart": true,
	155	+ "max_memory_restart": "12G"
	156	+ },
	157	+ {
	158	+ "name": "findata-bulk-4",
	159	+ "manager": "pm2",
	160	+ "script": "{{HOME}}/apps/findata/.venv/bin/python",
	161	+ "args": [
	162	+ "-m",
	163	+ "findata_ingest.cli",
	164	+ "worker",
	165	+ "--role",
	166	+ "bulk"
	167	+ ],
	168	+ "interpreter": null,
	169	+ "cwd": "{{HOME}}/apps/findata",
	170	+ "env": {
	171	+ "FINDATA_ROOT": "{{HOME}}/apps/findata",
	172	+ "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"
	173	+ },
	174	+ "cron_restart": null,
	175	+ "autorestart": true,
	176	+ "max_memory_restart": "12G"
	177	+ }
17	178	],
18	179	"ngrok": null,
19	180	"launchd": [],
		`@@ -29,6 +190,10 @@`
29	190	"post_start": []
30	191	},
31	192	"ka_repo": false,
32		− "tunnel": { "domain": "findata.spboucher.ai", "gateway": "BHS64", "note": "DNS GoDaddy A findata.spboucher.ai → 51.161.112.61 vérifié le 2026-11-01 avant déploiement ; route Caddy posée par mld deploy ; findata-api écoute 0.0.0.0:8600 avec garde de provenance (loopback, wg1 10.67.0.0/16, LAN) ; Caddy → wg1 10.67.0.10:8600" },
33		− "notes": "Monorepo uv (Python 3.13) : services/api (FastAPI :8600, sert aussi la console Next.js statique apps/web/out), services/ingest (worker unique = verrou d'écriture du lac + ordonnanceur), lac Parquet ~/findata-lake, Postgres 17 local (DB findata, rôle findata). Secrets uniquement dans ~/apps/findata/.env (600) ; copie ~/Desktop/Cluster/secrets/findata/.env. Sauvegarde quotidienne 04:00 ET → M4M64a:~/backups/findata (rsync + pg_dump zstd). CLI : .venv/bin/findata status\|discover\|ingest …"
	193	+ "tunnel": {
	194	+ "domain": "findata.spboucher.ai",
	195	+ "gateway": "BHS64",
	196	+ "note": "DNS GoDaddy A findata.spboucher.ai → 51.161.112.61 vérifié le 2026-11-01 avant déploiement ; route Caddy posée par mld deploy ; findata-api écoute 0.0.0.0:8600 avec garde de provenance (loopback, wg1 10.67.0.0/16, LAN) ; Caddy → wg1 10.67.0.10:8600"
	197	+ },
	198	+ "notes": "Monorepo uv (Python 3.13) : services/api (FastAPI :8600, sert aussi la console Next.js statique apps/web/out), services/ingest (worker unique = verrou d'écriture du lac + ordonnanceur), lac Parquet ~/findata-lake, Postgres 17 local (DB findata, rôle findata). Secrets uniquement dans ~/apps/findata/.env (600) ; copie ~/Desktop/Cluster/secrets/findata/.env. Sauvegarde quotidienne 04:00 ET → M4M64a:~/backups/findata (rsync + pg_dump zstd). CLI : .venv/bin/findata status\|discover\|ingest … \| 2026-11-01 soir : lac froid FINDATA_LAKE_COLD=/Volumes/ssd/findata (SSD externe 4 To, marqueur .findata-cold-root) pour les archives brutes et les datasets non « hot » ; 4 workers bulk (findata-bulk-1..4) pour backfill/options/fundamentals en parallèle ; téléchargement complet FirstRate lancé via `findata ingest backfill-all`."
34	199	}

modified packages/core/src/findata_core/config.py +11 −1

		`@@ -22,6 +22,12 @@ class Settings(BaseSettings):`
22	22	secret_key: SecretStr = SecretStr("dev-only-change-me-dev-only-change-me")
23	23	lake_soft_limit_gb: float = 250.0
24	24	lake_hard_min_free_gb: float = 60.0
	25	+ # second lake root on the external SSD (cold datasets + raw archives); must contain .findata-cold-root
	26	+ lake_cold: Path \| None = None
	27	+ lake_cold_hard_min_free_gb: float = 100.0
	28	+ hot_datasets: str = ("market_bars.stock.1m.raw,market_bars.etf.1m.raw,market_bars.stock.1d.raw,market_bars.etf.1d.raw,"
	29	+ "market_bars.stock.5m.raw,market_bars.etf.5m.raw")
	30	+ bulk_workers: int = 4
25	31	backup_target: str = ""
26	32	# vendor + optional providers (no FINDATA_ prefix: read explicitly)
27	33	frd_userid: SecretStr = Field(default=SecretStr(""), validation_alias="FRD_USERID")
		`@@ -41,9 +47,13 @@ class Settings(BaseSettings):`
41	47	def is_production(self) -> bool:
42	48	return self.env == "production"
43	49
	50	+ @property
	51	+ def hot_dataset_ids(self) -> set[str]:
	52	+ return {x.strip() for x in self.hot_datasets.split(",") if x.strip()}
	53	+
44	54	@property
45	55	def raw_dir(self) -> Path:
46		− return self.lake / "raw"
	56	+ return (self.lake_cold or self.lake) / "raw"
47	57
48	58	@property
49	59	def curated_dir(self) -> Path:

added packages/core/src/findata_core/migrations/0002_full_catalog.sql +67 −0

		`@@ -0,0 +1,67 @@`
	1	+-- Full FirstRate catalog (planned until data arrives); options dataset kind.
	2	+ALTER TABLE datasets DROP CONSTRAINT IF EXISTS datasets_kind_check;
	3	+ALTER TABLE datasets ADD CONSTRAINT datasets_kind_check CHECK (kind IN ('market_bars','fundamentals','corporate_actions','reference','derived','options'));
	4	+INSERT INTO datasets(id, name, description, source_id, kind, asset_class, interval, adjustment, license_class, status, timestamp_convention, timezone, notes) VALUES
	5	+('market_bars.stock.1m.split','US stocks — 1-minute bars (split-adjusted (vendor adj_split))','Vendor 1-minute OHLCV bars, split-adjusted (vendor adj_split). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','stock','1m','split','frd-private','planned','bar_open','America/New_York',NULL),
	6	+('market_bars.stock.1m.split_div','US stocks — 1-minute bars (split- and dividend-adjusted (vendor adj_splitdiv))','Vendor 1-minute OHLCV bars, split- and dividend-adjusted (vendor adj_splitdiv). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','stock','1m','split_div','frd-private','planned','bar_open','America/New_York',NULL),
	7	+('market_bars.stock.5m.split','US stocks — 5-minute bars (split-adjusted (vendor adj_split))','Vendor 5-minute OHLCV bars, split-adjusted (vendor adj_split). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','stock','5m','split','frd-private','planned','bar_open','America/New_York',NULL),
	8	+('market_bars.stock.5m.split_div','US stocks — 5-minute bars (split- and dividend-adjusted (vendor adj_splitdiv))','Vendor 5-minute OHLCV bars, split- and dividend-adjusted (vendor adj_splitdiv). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','stock','5m','split_div','frd-private','planned','bar_open','America/New_York',NULL),
	9	+('market_bars.stock.30m.split','US stocks — 30-minute bars (split-adjusted (vendor adj_split))','Vendor 30-minute OHLCV bars, split-adjusted (vendor adj_split). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','stock','30m','split','frd-private','planned','bar_open','America/New_York',NULL),
	10	+('market_bars.stock.30m.split_div','US stocks — 30-minute bars (split- and dividend-adjusted (vendor adj_splitdiv))','Vendor 30-minute OHLCV bars, split- and dividend-adjusted (vendor adj_splitdiv). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','stock','30m','split_div','frd-private','planned','bar_open','America/New_York',NULL),
	11	+('market_bars.stock.1h.split','US stocks — 1-hour bars (split-adjusted (vendor adj_split))','Vendor 1-hour OHLCV bars, split-adjusted (vendor adj_split). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','stock','1h','split','frd-private','planned','bar_open','America/New_York',NULL),
	12	+('market_bars.stock.1h.split_div','US stocks — 1-hour bars (split- and dividend-adjusted (vendor adj_splitdiv))','Vendor 1-hour OHLCV bars, split- and dividend-adjusted (vendor adj_splitdiv). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','stock','1h','split_div','frd-private','planned','bar_open','America/New_York',NULL),
	13	+('market_bars.stock.1d.raw','US stocks — daily bars (unadjusted)','Vendor daily OHLCV bars, unadjusted. Includes delisted tickers (suffix -DELISTED) when their archives are ingested.','firstrate','market_bars','stock','1d','raw','frd-private','planned','session','America/New_York',NULL),
	14	+('market_bars.stock.1d.split','US stocks — daily bars (split-adjusted (vendor adj_split))','Vendor daily OHLCV bars, split-adjusted (vendor adj_split). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','stock','1d','split','frd-private','planned','session','America/New_York',NULL),
	15	+('market_bars.stock.1d.split_div','US stocks — daily bars (split- and dividend-adjusted (vendor adj_splitdiv))','Vendor daily OHLCV bars, split- and dividend-adjusted (vendor adj_splitdiv). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','stock','1d','split_div','frd-private','planned','session','America/New_York',NULL),
	16	+('market_bars.etf.1m.split','US ETFs — 1-minute bars (split-adjusted (vendor adj_split))','Vendor 1-minute OHLCV bars, split-adjusted (vendor adj_split). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','etf','1m','split','frd-private','planned','bar_open','America/New_York',NULL),
	17	+('market_bars.etf.1m.split_div','US ETFs — 1-minute bars (split- and dividend-adjusted (vendor adj_splitdiv))','Vendor 1-minute OHLCV bars, split- and dividend-adjusted (vendor adj_splitdiv). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','etf','1m','split_div','frd-private','planned','bar_open','America/New_York',NULL),
	18	+('market_bars.etf.5m.split','US ETFs — 5-minute bars (split-adjusted (vendor adj_split))','Vendor 5-minute OHLCV bars, split-adjusted (vendor adj_split). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','etf','5m','split','frd-private','planned','bar_open','America/New_York',NULL),
	19	+('market_bars.etf.5m.split_div','US ETFs — 5-minute bars (split- and dividend-adjusted (vendor adj_splitdiv))','Vendor 5-minute OHLCV bars, split- and dividend-adjusted (vendor adj_splitdiv). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','etf','5m','split_div','frd-private','planned','bar_open','America/New_York',NULL),
	20	+('market_bars.etf.30m.split','US ETFs — 30-minute bars (split-adjusted (vendor adj_split))','Vendor 30-minute OHLCV bars, split-adjusted (vendor adj_split). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','etf','30m','split','frd-private','planned','bar_open','America/New_York',NULL),
	21	+('market_bars.etf.30m.split_div','US ETFs — 30-minute bars (split- and dividend-adjusted (vendor adj_splitdiv))','Vendor 30-minute OHLCV bars, split- and dividend-adjusted (vendor adj_splitdiv). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','etf','30m','split_div','frd-private','planned','bar_open','America/New_York',NULL),
	22	+('market_bars.etf.1h.split','US ETFs — 1-hour bars (split-adjusted (vendor adj_split))','Vendor 1-hour OHLCV bars, split-adjusted (vendor adj_split). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','etf','1h','split','frd-private','planned','bar_open','America/New_York',NULL),
	23	+('market_bars.etf.1h.split_div','US ETFs — 1-hour bars (split- and dividend-adjusted (vendor adj_splitdiv))','Vendor 1-hour OHLCV bars, split- and dividend-adjusted (vendor adj_splitdiv). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','etf','1h','split_div','frd-private','planned','bar_open','America/New_York',NULL),
	24	+('market_bars.etf.1d.raw','US ETFs — daily bars (unadjusted)','Vendor daily OHLCV bars, unadjusted. Includes delisted tickers (suffix -DELISTED) when their archives are ingested.','firstrate','market_bars','etf','1d','raw','frd-private','planned','session','America/New_York',NULL),
	25	+('market_bars.etf.1d.split','US ETFs — daily bars (split-adjusted (vendor adj_split))','Vendor daily OHLCV bars, split-adjusted (vendor adj_split). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','etf','1d','split','frd-private','planned','session','America/New_York',NULL),
	26	+('market_bars.etf.1d.split_div','US ETFs — daily bars (split- and dividend-adjusted (vendor adj_splitdiv))','Vendor daily OHLCV bars, split- and dividend-adjusted (vendor adj_splitdiv). Includes delisted tickers (suffix -DELISTED) when their archives are ingested. Vendor-adjusted history is rewritten after corporate actions; refreshed weekly via full_update.','firstrate','market_bars','etf','1d','split_div','frd-private','planned','session','America/New_York',NULL),
	27	+('market_bars.index.1m.raw','Indices — 1-minute bars','Vendor 1-minute bars for indices. No volume (stored as 0).','firstrate','market_bars','index','1m','raw','frd-private','planned','bar_open','America/New_York',NULL),
	28	+('market_bars.index.5m.raw','Indices — 5-minute bars','Vendor 5-minute bars for indices. No volume (stored as 0).','firstrate','market_bars','index','5m','raw','frd-private','planned','bar_open','America/New_York',NULL),
	29	+('market_bars.index.30m.raw','Indices — 30-minute bars','Vendor 30-minute bars for indices. No volume (stored as 0).','firstrate','market_bars','index','30m','raw','frd-private','planned','bar_open','America/New_York',NULL),
	30	+('market_bars.index.1h.raw','Indices — 1-hour bars','Vendor 1-hour bars for indices. No volume (stored as 0).','firstrate','market_bars','index','1h','raw','frd-private','planned','bar_open','America/New_York',NULL),
	31	+('market_bars.index.1d.raw','Indices — daily bars','Vendor daily bars for indices. No volume (stored as 0).','firstrate','market_bars','index','1d','raw','frd-private','planned','session','America/New_York',NULL),
	32	+('market_bars.fx.1m.raw','FX majors — 1-minute bars','Vendor 1-minute bars for fx majors. Volume = tick count.','firstrate','market_bars','fx','1m','raw','frd-private','planned','bar_open','America/New_York',NULL),
	33	+('market_bars.fx.5m.raw','FX majors — 5-minute bars','Vendor 5-minute bars for fx majors. Volume = tick count.','firstrate','market_bars','fx','5m','raw','frd-private','planned','bar_open','America/New_York',NULL),
	34	+('market_bars.fx.30m.raw','FX majors — 30-minute bars','Vendor 30-minute bars for fx majors. Volume = tick count.','firstrate','market_bars','fx','30m','raw','frd-private','planned','bar_open','America/New_York',NULL),
	35	+('market_bars.fx.1h.raw','FX majors — 1-hour bars','Vendor 1-hour bars for fx majors. Volume = tick count.','firstrate','market_bars','fx','1h','raw','frd-private','planned','bar_open','America/New_York',NULL),
	36	+('market_bars.fx.1d.raw','FX majors — daily bars','Vendor daily bars for fx majors. Volume = tick count.','firstrate','market_bars','fx','1d','raw','frd-private','planned','session','America/New_York',NULL),
	37	+('market_bars.crypto.1m.raw','Crypto — 1-minute bars','Vendor 1-minute bars for crypto.','firstrate','market_bars','crypto','1m','raw','frd-private','planned','bar_open','UTC',NULL),
	38	+('market_bars.crypto.5m.raw','Crypto — 5-minute bars','Vendor 5-minute bars for crypto.','firstrate','market_bars','crypto','5m','raw','frd-private','planned','bar_open','UTC',NULL),
	39	+('market_bars.crypto.30m.raw','Crypto — 30-minute bars','Vendor 30-minute bars for crypto.','firstrate','market_bars','crypto','30m','raw','frd-private','planned','bar_open','UTC',NULL),
	40	+('market_bars.crypto.1h.raw','Crypto — 1-hour bars','Vendor 1-hour bars for crypto.','firstrate','market_bars','crypto','1h','raw','frd-private','planned','bar_open','UTC',NULL),
	41	+('market_bars.crypto.1d.raw','Crypto — daily bars','Vendor daily bars for crypto.','firstrate','market_bars','crypto','1d','raw','frd-private','planned','session','UTC',NULL),
	42	+('market_bars.futures.1m.continuous_raw','Futures — 1-minute continuous, unadjusted (contin_UNadj)','Vendor continuous series per root (continuous, unadjusted (contin_UNadj)); roll dates fixed by vendor.','firstrate','market_bars','futures','1m','continuous_raw','frd-private','planned','bar_open','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	43	+('market_bars.futures.1m.continuous_ratio','Futures — 1-minute continuous, ratio back-adjusted','Vendor continuous series per root (continuous, ratio back-adjusted); roll dates fixed by vendor.','firstrate','market_bars','futures','1m','continuous_ratio','frd-private','planned','bar_open','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	44	+('market_bars.futures.1m.continuous_absolute','Futures — 1-minute continuous, absolute back-adjusted','Vendor continuous series per root (continuous, absolute back-adjusted); roll dates fixed by vendor.','firstrate','market_bars','futures','1m','continuous_absolute','frd-private','planned','bar_open','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	45	+('market_bars.futures_contracts.1m.raw','Futures individual contracts — 1-minute bars','Per-contract bars (e.g. ES_H25); symbol = ROOT_MYY. Daily files carry open interest.','firstrate','market_bars','futures','1m','raw','frd-private','planned','bar_open','America/New_York',NULL),
	46	+('market_bars.futures.5m.continuous_raw','Futures — 5-minute continuous, unadjusted (contin_UNadj)','Vendor continuous series per root (continuous, unadjusted (contin_UNadj)); roll dates fixed by vendor.','firstrate','market_bars','futures','5m','continuous_raw','frd-private','planned','bar_open','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	47	+('market_bars.futures.5m.continuous_ratio','Futures — 5-minute continuous, ratio back-adjusted','Vendor continuous series per root (continuous, ratio back-adjusted); roll dates fixed by vendor.','firstrate','market_bars','futures','5m','continuous_ratio','frd-private','planned','bar_open','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	48	+('market_bars.futures.5m.continuous_absolute','Futures — 5-minute continuous, absolute back-adjusted','Vendor continuous series per root (continuous, absolute back-adjusted); roll dates fixed by vendor.','firstrate','market_bars','futures','5m','continuous_absolute','frd-private','planned','bar_open','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	49	+('market_bars.futures_contracts.5m.raw','Futures individual contracts — 5-minute bars','Per-contract bars (e.g. ES_H25); symbol = ROOT_MYY. Daily files carry open interest.','firstrate','market_bars','futures','5m','raw','frd-private','planned','bar_open','America/New_York',NULL),
	50	+('market_bars.futures.30m.continuous_raw','Futures — 30-minute continuous, unadjusted (contin_UNadj)','Vendor continuous series per root (continuous, unadjusted (contin_UNadj)); roll dates fixed by vendor.','firstrate','market_bars','futures','30m','continuous_raw','frd-private','planned','bar_open','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	51	+('market_bars.futures.30m.continuous_ratio','Futures — 30-minute continuous, ratio back-adjusted','Vendor continuous series per root (continuous, ratio back-adjusted); roll dates fixed by vendor.','firstrate','market_bars','futures','30m','continuous_ratio','frd-private','planned','bar_open','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	52	+('market_bars.futures.30m.continuous_absolute','Futures — 30-minute continuous, absolute back-adjusted','Vendor continuous series per root (continuous, absolute back-adjusted); roll dates fixed by vendor.','firstrate','market_bars','futures','30m','continuous_absolute','frd-private','planned','bar_open','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	53	+('market_bars.futures_contracts.30m.raw','Futures individual contracts — 30-minute bars','Per-contract bars (e.g. ES_H25); symbol = ROOT_MYY. Daily files carry open interest.','firstrate','market_bars','futures','30m','raw','frd-private','planned','bar_open','America/New_York',NULL),
	54	+('market_bars.futures.1h.continuous_raw','Futures — 1-hour continuous, unadjusted (contin_UNadj)','Vendor continuous series per root (continuous, unadjusted (contin_UNadj)); roll dates fixed by vendor.','firstrate','market_bars','futures','1h','continuous_raw','frd-private','planned','bar_open','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	55	+('market_bars.futures.1h.continuous_ratio','Futures — 1-hour continuous, ratio back-adjusted','Vendor continuous series per root (continuous, ratio back-adjusted); roll dates fixed by vendor.','firstrate','market_bars','futures','1h','continuous_ratio','frd-private','planned','bar_open','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	56	+('market_bars.futures.1h.continuous_absolute','Futures — 1-hour continuous, absolute back-adjusted','Vendor continuous series per root (continuous, absolute back-adjusted); roll dates fixed by vendor.','firstrate','market_bars','futures','1h','continuous_absolute','frd-private','planned','bar_open','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	57	+('market_bars.futures_contracts.1h.raw','Futures individual contracts — 1-hour bars','Per-contract bars (e.g. ES_H25); symbol = ROOT_MYY. Daily files carry open interest.','firstrate','market_bars','futures','1h','raw','frd-private','planned','bar_open','America/New_York',NULL),
	58	+('market_bars.futures.1d.continuous_raw','Futures — daily continuous, unadjusted (contin_UNadj)','Vendor continuous series per root (continuous, unadjusted (contin_UNadj)); roll dates fixed by vendor.','firstrate','market_bars','futures','1d','continuous_raw','frd-private','planned','session','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	59	+('market_bars.futures.1d.continuous_ratio','Futures — daily continuous, ratio back-adjusted','Vendor continuous series per root (continuous, ratio back-adjusted); roll dates fixed by vendor.','firstrate','market_bars','futures','1d','continuous_ratio','frd-private','planned','session','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	60	+('market_bars.futures.1d.continuous_absolute','Futures — daily continuous, absolute back-adjusted','Vendor continuous series per root (continuous, absolute back-adjusted); roll dates fixed by vendor.','firstrate','market_bars','futures','1d','continuous_absolute','frd-private','planned','session','America/New_York','Daily bars follow the contract trading day; absolute-adjusted series may be negative.'),
	61	+('market_bars.futures_contracts.1d.raw','Futures individual contracts — daily bars','Per-contract bars (e.g. ES_H25); symbol = ROOT_MYY. Daily files carry open interest.','firstrate','market_bars','futures','1d','raw','frd-private','planned','session','America/New_York',NULL)
	62	+ON CONFLICT (id) DO NOTHING;
	63	+-- the pre-existing planned rows get consistent names
	64	+UPDATE datasets SET kind='options', name='US options — end-of-day chains (CBOE)', description='EOD chains since 2010: strike, expiry, call/put, last/bid/ask, IVs, open interest, volume, Greeks. Not split-adjusted at source. One Parquet file per underlying and quarter.' WHERE id='options.eod';
	65	+UPDATE datasets SET id='market_bars.futures.1m.continuous_raw' WHERE id='market_bars.futures.1m.continuous' AND NOT EXISTS (SELECT 1 FROM datasets WHERE id='market_bars.futures.1m.continuous_raw');
	66	+DELETE FROM datasets WHERE id='market_bars.futures.1m.continuous';
	67	+UPDATE datasets SET name='US stocks — 1-minute bars (unadjusted)', description='Raw (unadjusted) 1-minute OHLCV bars for US-listed stocks, extended hours 04:00–20:00 ET, zero-volume bars absent at source. Includes delisted tickers (suffix -DELISTED).' WHERE id='market_bars.stock.1m.raw';

modified packages/schema/src/findata_schema/__init__.py +24 −2

		`@@ -1,5 +1,5 @@`
1	1	"""findata_schema — Arrow schemas, data contracts and connector interfaces (single source of truth)."""
2		−from findata_schema.connector import ( # noqa: F401
	2	+from findata_schema.connector import (
3	3	Capabilities,
4	4	Checkpoint,
5	5	Connector,
		`@@ -8,11 +8,12 @@ from findata_schema.connector import ( # noqa: F401`
8	8	NormalizedBatch,
9	9	ValidationReport,
10	10	)
11		−from findata_schema.contracts import ( # noqa: F401
	11	+from findata_schema.contracts import (
12	12	BARS_SCHEMA,
13	13	BARS_SCHEMA_VERSION,
14	14	FUNDAMENTALS_SCHEMA,
15	15	INTERVAL_SECONDS,
	16	+ OPTIONS_SCHEMA,
16	17	RAW_BAR_COLUMNS,
17	18	Adjustment,
18	19	AssetClass,
		`@@ -20,3 +21,24 @@ from findata_schema.contracts import ( # noqa: F401`
20	21	QualityFlag,
21	22	bars_schema_json,
22	23	)
	24	+
	25	+__all__ = [
	26	+ "BARS_SCHEMA",
	27	+ "BARS_SCHEMA_VERSION",
	28	+ "FUNDAMENTALS_SCHEMA",
	29	+ "INTERVAL_SECONDS",
	30	+ "OPTIONS_SCHEMA",
	31	+ "RAW_BAR_COLUMNS",
	32	+ "Adjustment",
	33	+ "AssetClass",
	34	+ "Capabilities",
	35	+ "Checkpoint",
	36	+ "Connector",
	37	+ "FetchRequest",
	38	+ "FetchResult",
	39	+ "Interval",
	40	+ "NormalizedBatch",
	41	+ "QualityFlag",
	42	+ "ValidationReport",
	43	+ "bars_schema_json",
	44	+]

modified packages/schema/src/findata_schema/contracts.py +27 −0

		`@@ -114,3 +114,30 @@ FUNDAMENTALS_SCHEMA = pa.schema(`
114	114
115	115	def bars_schema_json() -> str:
116	116	return json.dumps([{"name": f.name, "type": str(f.type), "nullable": f.nullable} for f in BARS_SCHEMA], indent=2)
	117	+
	118	+
	119	+# End-of-day option chains (vendor CBOE feed). trade_date is the session; not split-adjusted at source.
	120	+OPTIONS_SCHEMA = pa.schema(
	121	+ [
	122	+ pa.field("underlying", pa.string(), nullable=False),
	123	+ pa.field("trade_date", pa.date32(), nullable=False),
	124	+ pa.field("strike", pa.float64(), nullable=False),
	125	+ pa.field("expiry", pa.date32(), nullable=False),
	126	+ pa.field("call_put", pa.string(), nullable=False),
	127	+ pa.field("last_price", pa.float64(), nullable=True),
	128	+ pa.field("bid", pa.float64(), nullable=True),
	129	+ pa.field("ask", pa.float64(), nullable=True),
	130	+ pa.field("bid_iv", pa.float64(), nullable=True),
	131	+ pa.field("ask_iv", pa.float64(), nullable=True),
	132	+ pa.field("open_interest", pa.float64(), nullable=True),
	133	+ pa.field("volume", pa.float64(), nullable=True),
	134	+ pa.field("delta", pa.float64(), nullable=True),
	135	+ pa.field("gamma", pa.float64(), nullable=True),
	136	+ pa.field("vega", pa.float64(), nullable=True),
	137	+ pa.field("theta", pa.float64(), nullable=True),
	138	+ pa.field("rho", pa.float64(), nullable=True),
	139	+ pa.field("source_id", pa.string(), nullable=False),
	140	+ pa.field("ingested_at", pa.timestamp("us", tz="UTC"), nullable=False),
	141	+ ],
	142	+ metadata={b"findata.schema": b"options_eod", b"findata.schema_version": b"1"},
	143	+)

modified services/ingest/src/findata_ingest/cli.py +5 −2

		`@@ -79,9 +79,12 @@ def ingest_backfill(ticker_range: str, type_: str = typer.Option("stock", "--typ`
79	79
80	80
81	81	@ingest_app.command("fundamentals")
82		−def ingest_fundamentals(tickers: str) -> None:
	82	+def ingest_fundamentals(tickers: str \| None = typer.Argument(None), all_active: bool = False) -> None:
	83	+ """Fundamentals for a ticker list, or --all-active for every active US stock (checkpointed per ticker)."""
83	84	from findata_ingest import queue
84		− typer.echo(queue.enqueue("fundamentals", "fundamentals.financials", {"tickers": tickers.split(",")}, priority=30, created_by="cli"))
	85	+ params = {"all_active": True} if all_active else {"tickers": (tickers or "").split(",")}
	86	+ typer.echo(queue.enqueue("fundamentals", "fundamentals.financials", params, priority=58, created_by="cli",
	87	+ dedupe_key="fundamentals:all_active" if all_active else None))
85	88
86	89
87	90	@ingest_app.command("derive")

modified services/ingest/src/findata_ingest/lake.py +83 −37

		`@@ -70,73 +70,116 @@ class WriterLock:`
70	70
71	71
72	72	class Lake:
73		− def __init__(self, root: Path \| None = None):
74		− self.root = (root or get_settings().lake).expanduser()
	73	+ """Two physical roots: `root` (internal SSD, hot datasets, manifests, exports) and optional `cold`
	74	+ (external SSD: raw archives + every dataset not listed in FINDATA_HOT_DATASETS). Relative paths stored in
	75	+ `lake_files` are root-agnostic (`curated/<dataset>/…`); `root_for(dataset)` picks the physical root."""
	76	+
	77	+ def __init__(self, root: Path \| None = None, cold: Path \| None = None):
	78	+ s = get_settings()
	79	+ self.root = (root or s.lake).expanduser()
	80	+ self.cold = (cold if cold is not None else s.lake_cold)
	81	+ self.cold = self.cold.expanduser() if self.cold else None
	82	+ self.hot = s.hot_dataset_ids
75	83	for sub in ("raw", "curated", "manifests", "_tmp", "_retired", "exports"):
76	84	(self.root / sub).mkdir(parents=True, exist_ok=True)
	85	+ if self.cold:
	86	+ if not (self.cold / ".findata-cold-root").exists():
	87	+ raise RuntimeError(f"cold lake root {self.cold} is not mounted (marker .findata-cold-root missing)")
	88	+ for sub in ("raw", "curated", "_tmp", "_retired"):
	89	+ (self.cold / sub).mkdir(parents=True, exist_ok=True)
	90	+
	91	+ # ------------------------------------------------------------------ roots
	92	+ def root_for(self, dataset_id: str) -> Path:
	93	+ if self.cold and dataset_id not in self.hot:
	94	+ return self.cold
	95	+ return self.root
	96	+
	97	+ def roots(self) -> list[Path]:
	98	+ return [self.root] + ([self.cold] if self.cold else [])
	99	+
	100	+ def _root_of_rel(self, rel: str) -> Path:
	101	+ parts = rel.split("/")
	102	+ if len(parts) >= 2 and parts[0] in ("curated", "_retired"):
	103	+ return self.root_for(parts[1])
	104	+ if parts and parts[0] == "raw":
	105	+ return self.cold or self.root
	106	+ return self.root
77	107
78		− # ------------------------------------------------------------------ paths
79	108	def raw_dir(self, source_id: str, dataset_id: str) -> Path:
80		− p = self.root / "raw" / source_id / dataset_id / str(datetime.now(UTC).year)
	109	+ p = (self.cold or self.root) / "raw" / source_id / dataset_id / str(datetime.now(UTC).year)
81	110	p.mkdir(parents=True, exist_ok=True)
82	111	return p
83	112
84	113	def abs(self, rel: str) -> Path:
85		− return self.root / rel
	114	+ return self._root_of_rel(rel) / rel
	115	+
	116	+ def rel(self, path: Path) -> str:
	117	+ for r in self.roots():
	118	+ try:
	119	+ return str(path.relative_to(r))
	120	+ except ValueError:
	121	+ continue
	122	+ raise ValueError(f"{path} is not inside a lake root")
86	123
87	124	def resolve(self, rel: str) -> Path:
88	125	"""Live path, or its retired location (older snapshots stay readable until GC after 7 days)."""
89		− p = self.root / rel
	126	+ p = self.abs(rel)
90	127	if p.exists():
91	128	return p
92	129	if rel.startswith("curated/"):
93		− r = self.root / "_retired" / rel[len("curated/"):]
	130	+ r = self._root_of_rel(rel) / "_retired" / rel[len("curated/"):]
94	131	if r.exists():
95	132	return r
96	133	return p
97	134
98		− def free_gb(self) -> float:
99		− st = os.statvfs(self.root)
	135	+ def free_gb(self, root: Path \| None = None) -> float:
	136	+ st = os.statvfs(root or self.root)
100	137	return st.f_bavail * st.f_frsize / 1e9
101	138
102	139	def used_bytes(self, sub: str = "curated") -> int:
103	140	total = 0
104		− for p in (self.root / sub).rglob("*"):
105		− if p.is_file():
106		− total += p.stat().st_size
	141	+ for r in self.roots():
	142	+ for p in (r / sub).rglob("*"):
	143	+ if p.is_file():
	144	+ total += p.stat().st_size
107	145	return total
108	146
109		− def check_budget(self) -> None:
	147	+ def check_budget(self, dataset_id: str \| None = None) -> None:
110	148	s = get_settings()
111		− if self.free_gb() < s.lake_hard_min_free_gb:
112		− raise RuntimeError(f"lake hard stop: only {self.free_gb():.1f} GB free (< {s.lake_hard_min_free_gb} GB)")
	149	+ root = self.root_for(dataset_id) if dataset_id else self.root
	150	+ limit = s.lake_cold_hard_min_free_gb if (self.cold and root == self.cold) else s.lake_hard_min_free_gb
	151	+ if self.free_gb(root) < limit:
	152	+ raise RuntimeError(f"lake hard stop on {root}: only {self.free_gb(root):.1f} GB free (< {limit} GB)")
113	153
114	154	# ------------------------------------------------------------------ writing
115	155	def write_table(self, dataset_id: str, table: pa.Table, layout: str, partition: dict[str, str],
116	156	sort: bool = True) -> WrittenFile:
117	157	"""Write a Parquet file into _tmp then move it atomically under curated/. Does NOT register it."""
118		− self.check_budget()
	158	+ self.check_budget(dataset_id)
119	159	if table.num_rows == 0:
120	160	raise ValueError("refusing to write an empty file")
121		− if sort:
	161	+ root = self.root_for(dataset_id)
	162	+ is_bars = "symbol_as_reported" in table.column_names and "timestamp_utc" in table.column_names
	163	+ if sort and is_bars:
122	164	table = table.sort_by([("symbol_as_reported", "ascending"), ("timestamp_utc", "ascending")])
123	165	part_dir = "/".join(f"{k}={v}" for k, v in partition.items())
124	166	rel = f"curated/{dataset_id}/{layout}/{part_dir}/part-{uuid.uuid4().hex[:12]}.parquet"
125		− final = self.root / rel
126		− tmp = self.root / "_tmp" / f"{uuid.uuid4().hex}.parquet"
	167	+ final = root / rel
	168	+ tmp = root / "_tmp" / f"{uuid.uuid4().hex}.parquet"
127	169	interval = table.column("interval")[0].as_py() if "interval" in table.column_names else "1d"
128	170	rg = ROW_GROUP_DAILY if interval == "1d" else ROW_GROUP_INTRADAY
	171	+ dict_cols = [c for c in ("symbol_as_reported", "instrument_id", "asset_class", "interval", "currency", "adjustment_type",
	172	+ "source_id", "underlying", "call_put") if c in table.column_names]
129	173	pq.write_table(table, tmp, compression="zstd", compression_level=3, row_group_size=rg,
130		− write_statistics=True, use_dictionary=["symbol_as_reported", "instrument_id", "asset_class",
131		− "interval", "currency", "adjustment_type", "source_id"])
	174	+ write_statistics=True, use_dictionary=dict_cols)
132	175	with open(tmp, "rb") as f:
133	176	os.fsync(f.fileno())
134	177	digest = _sha256(tmp)
135	178	final.parent.mkdir(parents=True, exist_ok=True)
136	179	os.replace(tmp, final)
137		− syms = table.column("symbol_as_reported")
138		− dates = table.column("session_date")
139	180	import pyarrow.compute as pc
	181	+ syms = table.column("symbol_as_reported") if is_bars else table.column("underlying")
	182	+ dates = table.column("session_date") if is_bars else table.column("trade_date")
140	183	return WrittenFile(rel_path=rel, layout=layout, symbol_min=pc.min(syms).as_py(), symbol_max=pc.max(syms).as_py(),
141	184	date_min=pc.min(dates).as_py(), date_max=pc.max(dates).as_py(), row_count=table.num_rows,
142	185	bytes=final.stat().st_size, sha256=digest)
		`@@ -176,9 +219,10 @@ class Lake:`
176	219	tmp.write_text(json.dumps(manifest, indent=1, default=str))
177	220	os.replace(tmp, mpath)
178	221	for p in retire_paths:
179		− src = self.root / p
	222	+ src = self.abs(p)
180	223	if src.exists():
181		− dst = self.root / "_retired" / p[len("curated/"):] if p.startswith("curated/") else self.root / "_retired" / p
	224	+ r = self._root_of_rel(p)
	225	+ dst = r / "_retired" / (p[len("curated/"):] if p.startswith("curated/") else p)
182	226	dst.parent.mkdir(parents=True, exist_ok=True)
183	227	shutil.move(str(src), str(dst))
184	228	return {"version": version, "snapshot_id": snapshot_id, "file_count": file_count, "row_count": row_count, "bytes": nbytes}
		`@@ -190,23 +234,25 @@ class Lake:`
190	234	known = {r["path"] for r in conn.execute("SELECT path FROM lake_files").fetchall()}
191	235	cutoff = datetime.now(UTC).timestamp() - older_than_hours * 3600
192	236	n = 0
193		− for p in (self.root / "curated").rglob("*.parquet"):
194		− rel = str(p.relative_to(self.root))
195		− if rel not in known and p.stat().st_mtime < cutoff:
196		− p.unlink()
197		− n += 1
	237	+ for r in self.roots():
	238	+ for p in (r / "curated").rglob("*.parquet"):
	239	+ rel = str(p.relative_to(r))
	240	+ if rel not in known and p.stat().st_mtime < cutoff:
	241	+ p.unlink()
	242	+ n += 1
198	243	return n
199	244
200	245	def gc_retired(self, older_than_days: int = 7) -> int:
201	246	cutoff = datetime.now(UTC).timestamp() - older_than_days * 86400
202	247	n = 0
203		− for p in (self.root / "_retired").rglob("*.parquet"):
204		− if p.stat().st_mtime < cutoff:
205		− p.unlink()
206		− n += 1
207		− for p in (self.root / "_tmp").glob("*"):
208		− if p.stat().st_mtime < datetime.now(UTC).timestamp() - 86400:
209		− p.unlink()
	248	+ for r in self.roots():
	249	+ for p in (r / "_retired").rglob("*.parquet"):
	250	+ if p.stat().st_mtime < cutoff:
	251	+ p.unlink()
	252	+ n += 1
	253	+ for p in (r / "_tmp").glob("*"):
	254	+ if p.stat().st_mtime < datetime.now(UTC).timestamp() - 86400:
	255	+ p.unlink()
210	256	return n
211	257
212	258

modified services/ingest/src/findata_ingest/reader.py +7 −4

		`@@ -126,8 +126,11 @@ def run_sql(sql: str, max_rows: int, timeout_s: int, views: dict[str, list[str]]`
126	126	raise ValueError("statement must start with SELECT / WITH")
127	127	is_explain = bool(re.match(r"^\s*explain\b", sql, re.I))
128	128	con = _con()
129		− lake_root = str(Lake().root).replace("'", "''")
130		− con.execute(f"SET allowed_directories = ['{lake_root}/curated', '{lake_root}/_retired']")
	129	+ dirs = []
	130	+ for r in Lake().roots():
	131	+ rr = str(r.resolve()).replace("'", "''")
	132	+ dirs += [f"'{rr}/curated'", f"'{rr}/_retired'"]
	133	+ con.execute(f"SET allowed_directories = [{', '.join(dirs)}]")
131	134	for name, paths in (views or {}).items():
132	135	if not re.fullmatch(r"[a-z][a-z0-9_]*", name):
133	136	raise ValueError(f"bad view name {name}")
		`@@ -155,8 +158,8 @@ def dataset_views() -> dict[str, list[str]]:`
155	158	lake = Lake()
156	159	out: dict[str, list[str]] = {}
157	160	with connection() as conn:
158		− ds = conn.execute("SELECT id FROM datasets WHERE kind IN ('market_bars','derived')").fetchall()
	161	+ ds = conn.execute("SELECT id FROM datasets WHERE kind IN ('market_bars','derived','options') AND current_version > 0").fetchall()
159	162	for d in ds:
160	163	files = live_files(conn, d["id"])
161		− out[d["id"].replace(".", "_")] = [str(lake.abs(f["path"])) for f in files]
	164	+ out[d["id"].replace(".", "_")] = [str(lake.resolve(f["path"])) for f in files]
162	165	return out

modified services/ingest/src/findata_ingest/tasks/__init__.py +2 −0

		`@@ -6,6 +6,7 @@ from findata_ingest.tasks import (`
6	6	fundamentals,
7	7	incremental,
8	8	metadata,
	9	+ options,
9	10	quality,
10	11	)
11	12
		`@@ -13,6 +14,7 @@ REGISTRY = {`
13	14	"metadata": metadata.run,
14	15	"incremental": incremental.run,
15	16	"backfill": backfill.run,
	17	+ "options": options.run,
16	18	"derive": derive.run,
17	19	"compaction": compaction.run,
18	20	"fundamentals": fundamentals.run,

modified services/ingest/src/findata_ingest/tasks/backfill.py +25 −15

		`@@ -11,14 +11,14 @@ from typing import Any`
11	11	import pyarrow as pa
12	12	import pyarrow.compute as pc
13	13	from findata_core.db import connection, transaction
14		−from findata_market_firstrate import FirstRateMarketConnector
	14	+from findata_market_firstrate import FirstRateMarketConnector, FRDNoData
15	15	from findata_schema import Interval
16	16	from findata_schema.connector import FetchRequest
17	17
18	18	from findata_ingest import queue
19	19	from findata_ingest.lake import Lake
20	20	from findata_ingest.tasks.common import (
21		− DATASET_ASSET,
	21	+ dataset_for_params,
22	22	live_delta_dates,
23	23	record_anomalies,
24	24	record_quality_report,
		`@@ -30,15 +30,20 @@ from findata_ingest.tasks.common import (`
30	30	def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:
31	31	p = job.get("params") or {}
32	32	asset = p.get("type", "stock")
33		− letter = p["ticker_range"]
	33	+ letter = p.get("ticker_range") or p.get("archive_number") or p.get("update") or p.get("contract_files") or p.get("period", "full")
34	34	timeframe = p.get("timeframe", "1min")
35		− adjustment = p.get("adjustment", "UNADJUSTED")
	35	+ adjustment = p.get("adjustment")
36	36	only = set(s.upper() for s in (p.get("symbols") or []))
37	37	max_symbols = int(p.get("max_symbols") or 0)
38		− dataset_id = job.get("dataset_id") or DATASET_ASSET[asset]
	38	+ dataset_id = job.get("dataset_id") or dataset_for_params(p)
39	39	run_id = job.get("run_id")
40	40	interval = Interval({"1min": "1m", "5min": "5m", "30min": "30m", "1hour": "1h", "1day": "1d"}[timeframe])
41	41	conn_fr = FirstRateMarketConnector()
	42	+ vendor_params = {k: v for k, v in p.items() if k in ("endpoint", "type", "period", "ticker_range", "timeframe", "adjustment",
	43	+ "archive_number", "update", "contract_files", "contracts", "year", "quarter") and v is not None}
	44	+ vendor_params.setdefault("endpoint", "data_file")
	45	+ if vendor_params["endpoint"] == "data_file":
	46	+ vendor_params.setdefault("period", "full")
42	47
43	48	ck = queue.get_checkpoint(f"job:{job['id']}", "state") or {}
44	49	done = set(ck.get("done_symbols") or [])
		`@@ -47,16 +52,19 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:`
47	52	from datetime import datetime
48	53
49	54	from findata_schema.connector import FetchResult
50		− res = FetchResult(request=FetchRequest(dataset_id, {"type": asset, "period": "full", "ticker_range": letter, "timeframe": timeframe, "adjustment": adjustment}),
51		− path=lake.root / raw_rel, sha256=ck["sha256"], bytes=ck["bytes"], fetched_at=datetime.fromisoformat(ck["fetched_at"]),
	55	+ res = FetchResult(request=FetchRequest(dataset_id, vendor_params),
	56	+ path=lake.abs(raw_rel), sha256=ck["sha256"], bytes=ck["bytes"], fetched_at=datetime.fromisoformat(ck["fetched_at"]),
52	57	vendor_last_update=None)
53	58	log.info("backfill.resume", done=len(done), archive=raw_rel)
54	59	else:
55		− req = FetchRequest(dataset_id=dataset_id, params={"endpoint": "data_file", "type": asset, "period": "full", "ticker_range": letter,
56		− "timeframe": timeframe, "adjustment": adjustment})
57		− queue.progress(job["id"], {"stage": "download", "ticker_range": letter})
58		− res = conn_fr.fetch_range(req, lake.raw_dir("firstrate", dataset_id))
59		− raw_rel = str(res.path.relative_to(lake.root))
	60	+ req = FetchRequest(dataset_id=dataset_id, params=vendor_params)
	61	+ queue.progress(job["id"], {"stage": "download", "archive": letter})
	62	+ try:
	63	+ res = conn_fr.fetch_range(req, lake.raw_dir("firstrate", dataset_id))
	64	+ except FRDNoData as e:
	65	+ log.info("backfill.no_data", reason=str(e)[:200])
	66	+ return {"skipped": True, "reason": str(e)[:300]}
	67	+ raw_rel = lake.rel(res.path)
60	68	with transaction() as conn:
61	69	record_raw_archive(conn, "firstrate", dataset_id, req.params, raw_rel, res.sha256, res.bytes, res.file_count, res.vendor_last_update, run_id)
62	70	queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": raw_rel, "sha256": res.sha256, "bytes": res.bytes,
		`@@ -92,7 +100,7 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:`
92	100	with connection() as conn:
93	101	old = conn.execute("SELECT path FROM lake_files WHERE dataset_id=%s AND layout='base' AND version_removed IS NULL AND symbol_min=%s AND symbol_max=%s",
94	102	(dataset_id, batch.symbol, batch.symbol)).fetchall()
95		− lake.publish(dataset_id, written, [o["path"] for o in old], run_id, note=f"backfill {asset} {letter} {batch.symbol}")
	103	+ lake.publish(dataset_id, written, [o["path"] for o in old], run_id, note=f"backfill {asset} {letter} {timeframe} {adjustment} {batch.symbol}")
96	104	with transaction() as conn:
97	105	upsert_coverage(conn, dataset_id, t, interval)
98	106	record_anomalies(conn, dataset_id, rep.anomalies, run_id, limit=500)
		`@@ -106,13 +114,15 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:`
106	114	queue.progress(job["id"], {"stage": "publish", **stats, "last_symbol": batch.symbol})
107	115	with transaction() as conn:
108	116	conn.execute("UPDATE raw_archives SET status='published' WHERE sha256=%s", (res.sha256,))
	117	+ if stats["rows"]:
	118	+ conn.execute("UPDATE datasets SET status='active', updated_at=now() WHERE id=%s AND status='planned'", (dataset_id,))
109	119	record_quality_report(conn, dataset_id, run_id, stats["rows"], [])
110		− if not p.get("keep_raw", False) and p.get("delete_raw_after_publish", True):
	120	+ if p.get("delete_raw_after_publish", False):
111	121	# full archives are re-downloadable; sha256 stays in raw_archives
112	122	res.path.unlink(missing_ok=True)
113	123	with transaction() as conn:
114	124	conn.execute("UPDATE raw_archives SET status='deleted_after_publish', path=NULL WHERE sha256=%s", (res.sha256,))
115		− if interval == Interval.m1 and stats["rows"]:
	125	+ if interval == Interval.m1 and stats["rows"] and dataset_id in ("market_bars.stock.1m.raw", "market_bars.etf.1m.raw"):
116	126	queue.enqueue("derive", dataset_id.replace(".1m.", ".5m."), {"source_dataset": dataset_id, "target_interval": "5m",
117	127	"symbols": sorted(done)}, priority=45, dedupe_key=f"derive:{dataset_id}:5m:backfill:{job['id']}")
118	128	log.info("backfill.done", **stats)

modified services/ingest/src/findata_ingest/tasks/common.py +8 −0

		`@@ -12,6 +12,12 @@ DATASET_ASSET = {"stock": "market_bars.stock.1m.raw", "etf": "market_bars.etf.1m`
12	12	"fx": "market_bars.fx.1m.raw", "crypto": "market_bars.crypto.1m.raw", "futures": "market_bars.futures.1m.continuous"}
13	13
14	14
	15	+def dataset_for_params(params: dict[str, Any]) -> str:
	16	+ from findata_market_firstrate.catalog import dataset_id_for
	17	+ return dataset_id_for(params.get("type", "stock"), params.get("timeframe", "1min"), params.get("adjustment"),
	18	+ contracts=bool(params.get("contracts")))
	19	+
	20	+
15	21	def record_raw_archive(conn, source_id: str, dataset_id: str, request: dict[str, Any], rel_path: str, sha256: str,
16	22	nbytes: int, file_count: int \| None, vendor_last_update: date \| None, run_id: str \| None) -> tuple[str, bool]:
17	23	"""Insert raw archive row; returns (id, is_new). Same sha256 → already known (idempotent re-fetch)."""
		`@@ -26,6 +32,8 @@ def record_raw_archive(conn, source_id: str, dataset_id: str, request: dict[str,`
26	32
27	33
28	34	def upsert_coverage(conn, dataset_id: str, table: pa.Table, interval: Interval) -> int:
	35	+ if table.num_rows and table.column("asset_class")[0].as_py() in ("fx", "crypto", "futures"):
	36	+ return 0 # no exchange-session expectation for 24h markets
29	37	rows = session_coverage(table, interval)
30	38	for r in rows:
31	39	conn.execute(

modified services/ingest/src/findata_ingest/tasks/compaction.py +5 −0

		`@@ -20,6 +20,11 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:`
20	20	max_files = int(p.get("max_delta_files", 40))
21	21	cutoff = (datetime.now(UTC) - timedelta(days=min_age)).date()
22	22	with connection() as conn:
	23	+ busy = conn.execute("SELECT count(*) AS n FROM ingest_jobs WHERE dataset_id=%s AND kind IN ('backfill','derive') AND status IN ('queued','running') AND id<>%s",
	24	+ (dataset_id, job["id"])).fetchone()["n"]
	25	+ if busy:
	26	+ log.info("compaction.deferred", reason=f"{busy} backfill/derive job(s) pending on {dataset_id}")
	27	+ return {"compacted_files": 0, "deferred": True}
23	28	deltas = conn.execute("SELECT path, date_min FROM lake_files WHERE dataset_id=%s AND layout='delta' AND version_removed IS NULL AND date_max <= %s ORDER BY date_min LIMIT %s",
24	29	(dataset_id, cutoff, max_files)).fetchall()
25	30	if not deltas:

modified services/ingest/src/findata_ingest/tasks/fundamentals.py +8 −2

		`@@ -17,9 +17,15 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:`
17	17	p = job.get("params") or {}
18	18	tickers = [t.upper() for t in (p.get("tickers") or [])]
19	19	if not tickers:
20		− n = int(p.get("limit", 50))
21	20	with connection() as conn:
22		− tickers = [r["symbol"] for r in conn.execute("SELECT symbol FROM instruments WHERE asset_class='stock' AND status='active' ORDER BY symbol LIMIT %s", (n,)).fetchall()]
	21	+ if p.get("all_active"):
	22	+ tickers = [r["symbol"] for r in conn.execute("SELECT symbol FROM instruments WHERE asset_class='stock' AND status='active' ORDER BY symbol").fetchall()]
	23	+ elif p.get("tracked"):
	24	+ row = conn.execute("SELECT value FROM settings_kv WHERE key='fundamentals_tracked_tickers'").fetchone()
	25	+ tickers = list(row["value"]) if row and row["value"] else [r["symbol"] for r in conn.execute("SELECT DISTINCT split_part(entity_id, ':', 2) symbol FROM fundamentals_facts WHERE entity_id LIKE 'frd:%'").fetchall()]
	26	+ else:
	27	+ n = int(p.get("limit", 50))
	28	+ tickers = [r["symbol"] for r in conn.execute("SELECT symbol FROM instruments WHERE asset_class='stock' AND status='active' ORDER BY symbol LIMIT %s", (n,)).fetchall()]
23	29	fr = FirstRateFundamentalsConnector()
24	30	observed = datetime.now(UTC)
25	31	run_id = job.get("run_id")

modified services/ingest/src/findata_ingest/tasks/incremental.py +15 −8

		`@@ -12,13 +12,14 @@ from typing import Any`
12	12	import pyarrow as pa
13	13	from findata_core.db import connection, transaction
14	14	from findata_market_firstrate import FirstRateMarketConnector
	15	+from findata_market_firstrate.client import FRDNoData
15	16	from findata_schema import Interval
16	17	from findata_schema.connector import FetchRequest
17	18
18	19	from findata_ingest import queue
19	20	from findata_ingest.lake import Lake
20	21	from findata_ingest.tasks.common import (
21		− DATASET_ASSET,
	22	+ dataset_for_params,
22	23	live_delta_paths_for_dates,
23	24	record_anomalies,
24	25	record_quality_report,
		`@@ -32,8 +33,8 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:`
32	33	asset = p.get("type", "stock")
33	34	period = p.get("period", "day")
34	35	timeframe = p.get("timeframe", "1min")
35		− adjustment = p.get("adjustment", "UNADJUSTED")
36		− dataset_id = job.get("dataset_id") or DATASET_ASSET[asset]
	36	+ adjustment = p.get("adjustment") or ("contin_UNadj" if asset == "futures" else ("UNADJUSTED" if asset in ("stock", "etf") else None))
	37	+ dataset_id = job.get("dataset_id") or dataset_for_params({"type": asset, "timeframe": timeframe, "adjustment": adjustment})
37	38	run_id = job.get("run_id")
38	39	conn_fr = FirstRateMarketConnector()
39	40	vendor_last = conn_fr.client.last_update(asset)
		`@@ -42,11 +43,16 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:`
42	43	log.info("incremental.skip", reason="vendor not updated since last run", vendor_last_update=str(vendor_last))
43	44	return {"skipped": True, "vendor_last_update": vendor_last.isoformat()}
44	45
45		− req = FetchRequest(dataset_id=dataset_id, params={"endpoint": "data_file", "type": asset, "period": period,
46		− "timeframe": timeframe, "adjustment": adjustment})
	46	+ req = FetchRequest(dataset_id=dataset_id, params={k: v for k, v in {"endpoint": "data_file", "type": asset, "period": period,
	47	+ "timeframe": timeframe, "adjustment": adjustment}.items() if v is not None})
47	48	queue.progress(job["id"], {"stage": "download"})
48		− res = conn_fr.fetch_range(req, lake.raw_dir("firstrate", dataset_id))
49		− rel_raw = str(res.path.relative_to(lake.root))
	49	+ try:
	50	+ res = conn_fr.fetch_range(req, lake.raw_dir("firstrate", dataset_id))
	51	+ except FRDNoData as e:
	52	+ queue.set_checkpoint(f"dataset:{dataset_id}", f"last_vendor_update:{period}", {"vendor_last_update": vendor_last.isoformat(), "no_data": True})
	53	+ log.info("incremental.no_data", reason=str(e)[:200])
	54	+ return {"skipped": True, "reason": str(e)[:300], "vendor_last_update": vendor_last.isoformat()}
	55	+ rel_raw = lake.rel(res.path)
50	56	with transaction() as conn:
51	57	raw_id, is_new = record_raw_archive(conn, "firstrate", dataset_id, req.params, rel_raw, res.sha256, res.bytes, res.file_count, res.vendor_last_update, run_id)
52	58	if not is_new:
		`@@ -92,6 +98,7 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:`
92	98	pub = lake.publish(dataset_id, written, retire, run_id, note=f"incremental {asset} {period} {timeframe} {adjustment}")
93	99	with transaction() as conn:
94	100	conn.execute("UPDATE raw_archives SET status='published' WHERE id=%s", (raw_id,))
	101	+ conn.execute("UPDATE datasets SET status='active', updated_at=now() WHERE id=%s AND status='planned'", (dataset_id,))
95	102	cov = sum(upsert_coverage(conn, dataset_id, t, interval) for t in tables.values())
96	103	record_anomalies(conn, dataset_id, anomalies, run_id)
97	104	record_quality_report(conn, dataset_id, run_id, sum(t.num_rows for t in tables.values()), anomalies)
		`@@ -100,7 +107,7 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:`
100	107	queue.set_checkpoint(cp.scope, cp.key, cp.value)
101	108	dates = [d.isoformat() for d in sorted(tables)]
102	109	# downstream: derived 5-minute bars for these sessions
103		− if interval == Interval.m1 and asset in ("stock", "etf"):
	110	+ if interval == Interval.m1 and asset in ("stock", "etf") and dataset_id.endswith(".raw"):
104	111	queue.enqueue("derive", dataset_id.replace(".1m.", ".5m."), {"source_dataset": dataset_id, "target_interval": "5m", "dates": dates},
105	112	priority=40, dedupe_key=f"derive:{dataset_id}:5m:{dates[0]}:{dates[-1]}:{pub['version']}")
106	113	result = {"vendor_last_update": vendor_last.isoformat(), "archive_sha256": res.sha256, "archive_bytes": res.bytes,

added services/ingest/src/findata_ingest/tasks/options.py +108 −0

		`@@ -0,0 +1,108 @@`
	1	+"""End-of-day option chains: one vendor archive per (year, quarter) or `period=month`; one vendor file per underlying.
	2	+Layout: options.eod/base/underlying=<U>/period=<YYYYqN\|latest>/part-*.parquet. The monthly archive replaces the
	3	+`latest` partition and drops rows already covered by a completed quarter archive (no overlap in the live set)."""
	4	+from __future__ import annotations
	5	+
	6	+from datetime import UTC, date, datetime
	7	+from typing import Any
	8	+
	9	+import pyarrow as pa
	10	+import pyarrow.compute as pc
	11	+from findata_core.db import connection, transaction
	12	+from findata_market_firstrate import (
	13	+ FirstRateMarketConnector,
	14	+ FRDNoData,
	15	+ iter_archive_files,
	16	+ parse_options_file,
	17	+ ticker_from_filename,
	18	+)
	19	+from findata_schema.connector import FetchRequest
	20	+
	21	+from findata_ingest import queue
	22	+from findata_ingest.lake import Lake
	23	+from findata_ingest.tasks.common import record_raw_archive
	24	+
	25	+DATASET = "options.eod"
	26	+
	27	+
	28	+def 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 FetchResult
	42	+ 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 = None
	57	+ 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 = 0
	69	+ for name, data in iter_archive_files(res.path):
	70	+ und = ticker_from_filename(name)
	71	+ if und in done or und.startswith("_"):
	72	+ continue
	73	+ try:
	74	+ table, raw_rows = parse_options_file(data, und, "firstrate", ingested_at)
	75	+ except ValueError as e:
	76	+ stats["skipped_files"] += 1
	77	+ log.warning("options.skip_file", file=name, error=str(e)[:120])
	78	+ continue
	79	+ 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_rows
	82	+ if table.num_rows == 0:
	83	+ done.add(und)
	84	+ continue
	85	+ 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"] += 1
	89	+ stats["rows"] += table.num_rows
	90	+ done.add(und)
	91	+ batch_rows += table.num_rows
	92	+ if len(written) >= 1000: # publish in batches so a crash loses little work
	93	+ 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}

modified services/ingest/src/findata_ingest/worker.py +49 −4

		`@@ -17,6 +17,8 @@ from findata_ingest.tasks import REGISTRY`
17	17
18	18	ET = ZoneInfo("America/New_York")
19	19	_stop = False
	20	+BULK_KINDS = ["backfill", "options", "fundamentals"]
	21	+PRIMARY_KINDS = ["metadata", "incremental", "derive", "compaction", "backup", "quality", "gc"]
20	22
21	23
22	24	def _handle(sig, frame):
		`@@ -39,6 +41,37 @@ def schedule_tick(log) -> None:`
39	41	queue.enqueue("metadata", None, {"types": ["stock", "etf"]}, priority=30, dedupe_key=f"metadata:{day}")
40	42	queue.enqueue("compaction", "market_bars.stock.1m.raw", {}, priority=60, dedupe_key=f"compaction:1m:{day}")
41	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}")
42	75	if now.hour == 5 and now.minute < 30:
43	76	queue.enqueue("fundamentals", "fundamentals.financials", {"tracked": True}, priority=50, dedupe_key=f"fundamentals:{day}")
44	77	if now.hour == 4 and now.minute < 30 and get_settings().backup_target:
		`@@ -48,18 +81,31 @@ def schedule_tick(log) -> None:`
48	81	queue.enqueue("gc", None, {}, priority=90, dedupe_key=f"gc:{day}")
49	82
50	83
51		−def main(once: bool = False, kinds: list[str] \| None = None, with_scheduler: bool = True) -> None:
	84	+def 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 N
	86	+ parallel workers for backfill/options/fundamentals (publication is serialized by Postgres row locks)."""
52	87	configure_logging()
53	88	log = get_logger("worker")
54	89	signal.signal(signal.SIGTERM, _handle)
55	90	signal.signal(signal.SIGINT, _handle)
56	91	lake = Lake()
	92	+ if role == "bulk":
	93	+ _loop(lake, log, once, kinds or BULK_KINDS, False)
	94	+ return
57	95	requeued = queue.requeue_stale()
58	96	if requeued:
59	97	log.warning("worker.requeued_stale", n=requeued)
60		− last_tick = 0.0
	98	+ if kinds is None:
	99	+ kinds = PRIMARY_KINDS if get_settings().bulk_workers > 0 else None
61	100	with WriterLock(lake.root):
62		− log.info("worker.start", lake=str(lake.root), free_gb=round(lake.free_gb(), 1))
	101	+ _loop(lake, log, once, kinds, with_scheduler)
	102	+ log.info("worker.stop")
	103	+
	104	+
	105	+def _loop(lake: Lake, log, once: bool, kinds: list[str] \| None, with_scheduler: bool) -> None:
	106	+ last_tick = 0.0
	107	+ if True:
	108	+ log.info("worker.start", lake=str(lake.root), cold=str(lake.cold), free_gb=round(lake.free_gb(), 1), kinds=kinds)
63	109	while not _stop:
64	110	if with_scheduler and time.monotonic() - last_tick > 300:
65	111	try:
		`@@ -87,4 +133,3 @@ def main(once: bool = False, kinds: list[str] \| None = None, with_scheduler: boo`
87	133	jl.error("job.failed", error=str(e)[:300], seconds=round(time.monotonic() - t0, 1))
88	134	if once:
89	135	break
90		− log.info("worker.stop")

modified services/quality/src/findata_quality/validators.py +9 −2

		`@@ -13,14 +13,21 @@ from findata_quality.calendar import SessionCalendar, xnys`
13	13	PRICE_JUMP_THRESHOLD = 0.25 # 25 % between consecutive 1-minute closes inside a session (flag only)
14	14
15	15
	16	+NO_SESSION_CALENDAR = {"fx", "crypto", "futures"} # 24h / overnight sessions: no XNYS session checks
	17	+
	18	+
16	19	def validate_bars(table: pa.Table, interval: Interval, calendar: SessionCalendar \| None = None,
17		− check_sessions: bool = True) -> tuple[pa.Table, list[dict[str, Any]]]:
	20	+ check_sessions: bool = True, asset_class: str \| None = None) -> tuple[pa.Table, list[dict[str, Any]]]:
18	21	"""Return (table with `quality_flags` filled, anomalies list). Input must follow BARS_SCHEMA columns."""
19	22	cal = calendar or xnys()
20	23	step = INTERVAL_SECONDS[interval]
21	24	df = pl.from_arrow(table)
22	25	if df.height == 0:
23	26	return table, []
	27	+ if asset_class is None and "asset_class" in df.columns:
	28	+ asset_class = df.get_column("asset_class")[0]
	29	+ if asset_class in NO_SESSION_CALENDAR:
	30	+ check_sessions = False
24	31	df = df.sort(["symbol_as_reported", "timestamp_utc"])
25	32	ts_s = pl.col("timestamp_utc").dt.epoch("s")
26	33	flags: list[pl.Expr] = [
		`@@ -28,7 +35,7 @@ def validate_bars(table: pa.Table, interval: Interval, calendar: SessionCalendar`
28	35	\| (pl.col("high") < pl.col("low"))).then(pl.lit(QualityFlag.ohlc_inconsistent.value)),
29	36	pl.when(pl.min_horizontal("open", "high", "low", "close") <= 0).then(pl.lit(QualityFlag.nonpositive_price.value)),
30	37	pl.when(pl.col("volume") < 0).then(pl.lit(QualityFlag.negative_volume.value)),
31		− pl.when(pl.col("volume") == 0).then(pl.lit(QualityFlag.zero_volume.value)),
	38	+ pl.when((pl.col("volume") == 0) & pl.lit(asset_class != "index")).then(pl.lit(QualityFlag.zero_volume.value)),
32	39	pl.when(pl.col("volume") != pl.col("volume").floor()).then(pl.lit(QualityFlag.fractional_volume.value)),
33	40	pl.when((ts_s % step) != 0).then(pl.lit(QualityFlag.off_interval.value)),
34	41	pl.when(pl.struct(["symbol_as_reported", "timestamp_utc"]).is_duplicated()).then(pl.lit(QualityFlag.duplicate.value)),

added tests/test_full_catalog.py +126 −0

		`@@ -0,0 +1,126 @@`
	1	+"""Full vendor catalog: new layouts (index without volume, futures contracts, option chains), no-data responses,
	2	+hot/cold lake roots, the complete plan, and the options task through the vendor mock."""
	3	+from __future__ import annotations
	4	+
	5	+import io
	6	+import zipfile
	7	+from datetime import UTC, date, datetime
	8	+
	9	+import httpx
	10	+import pytest
	11	+from findata_market_firstrate import (
	12	+ FRDClient,
	13	+ FRDNoData,
	14	+ build_full_plan,
	15	+ dataset_id_for,
	16	+ parse_options_file,
	17	+ ticker_from_filename,
	18	+)
	19	+from findata_market_firstrate.archive import parse_bar_file
	20	+from findata_quality import validate_bars
	21	+from findata_schema import OPTIONS_SCHEMA, Adjustment, AssetClass, Interval
	22	+
	23	+from tests.conftest import TEST_USERID
	24	+
	25	+NOW = datetime(2026, 11, 1, 12, 0, tzinfo=UTC)
	26	+
	27	+
	28	+def test_index_without_volume_and_fx_two_fields():
	29	+ t, raw = parse_bar_file(b"2026-10-29 04:00:00,1136.43,1137.03,1135.97,1135.97\n2026-10-29 04:01:00,1135.89,1135.89,1134.56,1134.83\n",
	30	+ "AEX", AssetClass.index, Interval.m1, Adjustment.raw, "firstrate", NOW)
	31	+ assert raw == 2 and t.num_rows == 2 and t.column("volume").to_pylist() == [0.0, 0.0]
	32	+ out, anomalies = validate_bars(t, Interval.m1, asset_class="index")
	33	+ assert all("zero_volume" not in f for f in out.column("quality_flags").to_pylist())
	34	+ t, _ = parse_bar_file(b"20261026,00:02:00,3.49536933,3.49536933,3.49536933,3.49536933,1\n", "AUDBRL", AssetClass.fx, Interval.m1, Adjustment.raw, "firstrate", NOW)
	35	+ assert t.column("timestamp_utc")[0].as_py() == datetime(2026, 10, 26, 4, 2, tzinfo=UTC) # 00:02 EDT
	36	+ out, anomalies = validate_bars(t, Interval.m1, asset_class="fx")
	37	+ assert not any(a["code"] in ("outside_session", "non_trading_day") for a in anomalies) # 24h market: no session checks
	38	+ t, _ = parse_bar_file(b"2026-10-30 00:00:00,119.071,119.46,119.04,119.045,208\n", "AAVE", AssetClass.crypto, Interval.m1, Adjustment.raw, "firstrate", NOW, source_tz="UTC")
	39	+ assert t.column("timestamp_utc")[0].as_py() == datetime(2026, 10, 30, 0, 0, tzinfo=UTC)
	40	+
	41	+
	42	+def test_contract_filenames_and_daily_open_interest():
	43	+ assert ticker_from_filename("A6_F25_1day.txt", contract=True) == "A6_F25"
	44	+ assert ticker_from_filename("ES_H25_1min.txt", contract=True) == "ES_H25"
	45	+ assert ticker_from_filename("ES_full_1min_contin_adj_ratio.txt") == "ES"
	46	+ t, _ = parse_bar_file(b"2024-09-11,0.6669,0.6679,0.6669,0.6679,10,5\n", "A6_F25", AssetClass.futures, Interval.d1, Adjustment.raw, "firstrate", NOW)
	47	+ assert t.num_rows == 1 and t.column("session_date")[0].as_py() == date(2024, 9, 11)
	48	+
	49	+
	50	+def test_option_chain_parser():
	51	+ data = b"2026-10-01,33,2026-10-16,c,8.14,7.3,8.8,0.0,1.102138,0,0,1.0,0.0,1e-05,0.0,0.0\n2026-10-01,34,2026-10-16,p,7.14,6.4,7.9,0.0,1.048606,12,3,-0.4,0.01,1e-05,-0.02,0.0\n"
	52	+ t, raw = parse_options_file(data, "AAL", "firstrate", NOW)
	53	+ assert raw == 2 and t.num_rows == 2 and t.schema.equals(OPTIONS_SCHEMA)
	54	+ r = t.to_pylist()[1]
	55	+ assert r["underlying"] == "AAL" and r["call_put"] == "p" and r["strike"] == 34.0 and r["expiry"] == date(2026, 10, 16) and r["open_interest"] == 12.0
	56	+
	57	+
	58	+def test_no_data_response_is_distinguished_from_errors(tmp_path):
	59	+ def handler(req: httpx.Request) -> httpx.Response:
	60	+ return httpx.Response(200, text="No update available", headers={"content-type": "text/plain", "content-length": "19"})
	61	+ c = FRDClient(TEST_USERID, min_interval_s=0, transport=httpx.MockTransport(handler))
	62	+ with pytest.raises(FRDNoData):
	63	+ c.download("full_update", {"type": "stock", "period": "week", "timeframe": "1day", "adjustment": "adj_split"}, tmp_path / "x.zip")
	64	+
	65	+
	66	+def test_dataset_ids_and_full_plan():
	67	+ assert dataset_id_for("stock", "1min", "adj_splitdiv") == "market_bars.stock.1m.split_div"
	68	+ assert dataset_id_for("etf", "1day", "UNADJUSTED") == "market_bars.etf.1d.raw"
	69	+ assert dataset_id_for("futures", "30min", "contin_adj_absolute") == "market_bars.futures.30m.continuous_absolute"
	70	+ assert dataset_id_for("futures", "1min", None, contracts=True) == "market_bars.futures_contracts.1m.raw"
	71	+ assert dataset_id_for("options", "1day", None) == "options.eod"
	72	+ plan = build_full_plan(today=date(2026, 11, 1))
	73	+ assert len(plan) == 798
	74	+ assert plan[0]["dataset_id"] == "market_bars.stock.1m.raw" and plan[0]["params"]["ticker_range"] == "A"
	75	+ assert sum(1 for j in plan if j["kind"] == "options") == 67 + 1
	76	+ assert len({j["dedupe_key"] for j in plan}) == len(plan)
	77	+ assert all(j["priority"] >= 10 for j in plan) and plan[-1]["kind"] == "options"
	78	+ small = build_full_plan(types=["index"], include_options=False, include_delisted=False, today=date(2026, 11, 1))
	79	+ assert len(small) == 5
	80	+
	81	+
	82	+def test_cold_root_placement(tmp_path, monkeypatch):
	83	+ from findata_ingest.lake import Lake
	84	+ cold = tmp_path / "ssd"
	85	+ cold.mkdir()
	86	+ with pytest.raises(RuntimeError):
	87	+ Lake(tmp_path / "hot", cold) # marker missing → refuse (volume may be unmounted)
	88	+ (cold / ".findata-cold-root").write_text("x")
	89	+ lake = Lake(tmp_path / "hot", cold)
	90	+ assert lake.root_for("market_bars.stock.1m.raw") == tmp_path / "hot"
	91	+ assert lake.root_for("options.eod") == cold and lake.root_for("market_bars.stock.1m.split_div") == cold
	92	+ assert lake.raw_dir("firstrate", "x").is_relative_to(cold)
	93	+ assert lake.abs("curated/options.eod/base/underlying=A/period=2024q1/p.parquet").is_relative_to(cold)
	94	+ assert lake.abs("curated/market_bars.stock.1m.raw/delta/session_date=2026-10-30/p.parquet").is_relative_to(tmp_path / "hot")
	95	+ assert lake.rel(cold / "raw" / "a.zip") == "raw/a.zip"
	96	+
	97	+
	98	+def test_options_task_with_vendor_mock(vendor, lake, log, monkeypatch):
	99	+ from findata_core.db import connection, fetch_one
	100	+ from findata_ingest import queue
	101	+ from findata_ingest.tasks import options as O
	102	+
	103	+ def opt_zip() -> bytes:
	104	+ buf = io.BytesIO()
	105	+ with zipfile.ZipFile(buf, "w") as zf:
	106	+ zf.writestr(zipfile.ZipInfo("AAL_month_option_chain.txt", (2026, 10, 30, 0, 0, 0)), "2026-10-01,33,2026-10-16,c,8.14,7.3,8.8,0.0,1.1,0,0,1.0,0.0,1e-05,0.0,0.0\n2026-10-02,33,2026-10-16,c,8.0,7.2,8.7,0.0,1.1,1,2,1.0,0.0,1e-05,0.0,0.0\n")
	107	+ zf.writestr(zipfile.ZipInfo("SPY_month_option_chain.txt", (2026, 10, 30, 0, 0, 0)), "2026-10-01,500,2026-10-17,p,1.5,1.4,1.6,0.2,0.21,100,50,-0.3,0.01,0.1,-0.05,0.0\n")
	108	+ return buf.getvalue()
	109	+ vendor.archive_override = opt_zip()
	110	+ jid = queue.enqueue("options", "options.eod", {"period": "month"})
	111	+ job = queue.claim_next(["options"])
	112	+ r = O.run(job, lake, log)
	113	+ queue.finish(job, True, r)
	114	+ assert r["underlyings"] == 2 and r["rows"] == 3 and r["files"] == 2
	115	+ with connection() as conn:
	116	+ rows = conn.execute("SELECT symbol_min, date_min, date_max, row_count FROM lake_files WHERE dataset_id='options.eod' AND version_removed IS NULL ORDER BY 1").fetchall()
	117	+ assert [(x["symbol_min"], x["row_count"]) for x in rows] == [("AAL", 2), ("SPY", 1)]
	118	+ assert fetch_one("SELECT status FROM datasets WHERE id='options.eod'")["status"] == "active"
	119	+ # the month archive re-run supersedes the `latest` partition (no duplicates)
	120	+ job = queue.claim_next(["options"]) or (queue.enqueue("options", "options.eod", {"period": "month"}) and queue.claim_next(["options"]))
	121	+ r2 = O.run(job, lake, log)
	122	+ queue.finish(job, True, r2)
	123	+ with connection() as conn:
	124	+ live = conn.execute("SELECT count(*) AS n, sum(row_count) AS rows FROM lake_files WHERE dataset_id='options.eod' AND version_removed IS NULL").fetchone()
	125	+ assert live["n"] == 2 and live["rows"] == 3
	126	+ vendor.archive_override = None

modified tests/test_ingest_lake.py +3 −0

		`@@ -103,6 +103,7 @@ def test_derive_5m_from_published_session(vendor, lake, log):`
103	103	job = queue.claim_next(["derive"]) # enqueued by the incremental run
104	104	assert job is not None
105	105	r = derive.run(job, lake, log)
	106	+ queue.finish(job, True, r)
106	107	assert r["rows"] > 0
107	108	t, _ = read_bars("market_bars.stock.5m.raw", ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))
108	109	assert 150 <= t.num_rows <= 192 and t.column("interval")[0].as_py() == "5m"
		`@@ -135,6 +136,8 @@ def test_backfill_resumes_from_checkpoint_and_respects_delta_dates(vendor, lake,`
135	136
136	137
137	138	def test_compaction_merges_delta_into_base_and_checks_invariant(vendor, lake, log):
	139	+ with transaction() as conn: # compaction defers while backfill/derive jobs are pending on the dataset
	140	+ conn.execute("UPDATE ingest_jobs SET status='cancelled' WHERE status IN ('queued','running') AND kind IN ('derive','backfill')")
138	141	job = _job("compaction", DS, {"min_age_days": 0})
139	142	r = compaction.run(job, lake, log)
140	143	assert r["compacted_files"] == 1 and r["base_files_written"] == 3
141	144

spb/findata

full FirstRate catalog: cold lake root (external SSD), 57 new datasets, index/contracts/options parsers, options task, bulk workers, complete download plan (798 jobs), extended scheduler

Theme